Coverage for tests/tests/helpers.py: 0%

1#!/usr/bin/env python3 

2"""RAMSES RF - a RAMSES-II protocol decoder & analyser.""" 

3 

4import json 

5import logging 

6import warnings 

7from collections.abc import AsyncGenerator, Callable 

8from pathlib import Path 

9from random import shuffle 

10from typing import Any 

11 

12import pytest 

13import voluptuous as vol 

14 

15from ramses_rf import Gateway 

16from ramses_rf.database import MessageIndex 

17from ramses_rf.helpers import shrink 

18from ramses_rf.schemas import SCH_GLOBAL_CONFIG, SCH_GLOBAL_SCHEMAS 

19from ramses_tx.schemas import SCH_GLOBAL_TRAITS_DICT 

20 

21SCH_GLOBAL_TRAITS = vol.Schema(SCH_GLOBAL_TRAITS_DICT, extra=vol.PREVENT_EXTRA) 

22 

23# import tracemalloc 

24# tracemalloc.start() 

25 

26warnings.filterwarnings("ignore", category=DeprecationWarning) 

27 

28logging.disable(logging.WARNING) # usu. WARNING 

29 

30 

31TEST_DIR = Path(__file__).resolve().parent # TEST_DIR = f"{os.path.dirname(__file__)}" 

32 

33 

34def shuffle_dict(old_dict: dict) -> dict: 

35 keys = list(old_dict.keys()) 

36 shuffle(keys) 

37 new_dict = dict() 

38 for key in keys: 

39 new_dict.update({key: old_dict[key]}) 

40 return new_dict 

41 

42 

43@pytest.fixture 

44async def gwy() -> AsyncGenerator[Gateway, None]: # NOTE: async to get running loop 

45 """Return a vanilla system (with a known, minimal state).""" 

46 gwy = Gateway("/dev/null", config={}) 

47 gwy._disable_sending = True 

48 gwy.msg_db = MessageIndex() # required to add heat dummy 3220 msg 

49 try: 

50 yield gwy 

51 finally: 

52 await gwy.stop() # close sqlite3 connection 

53 

54 

55def assert_expected( 

56 actual: dict[str, Any], expected: dict[str, Any] | None = None 

57) -> None: 

58 """Compare an actual system state dict against the corresponding expected state.""" 

59 

60 def assert_expected(actual_: dict[str, Any], expect_: dict[str, Any]) -> None: 

61 assert actual_ == expect_ 

62 

63 if expected: 

64 assert_expected(shrink(actual), shrink(expected)) 

65 

66 

67def assert_expected_set(gwy: Gateway, expected: dict) -> None: 

68 """Compare the actual system state against the expected system state.""" 

69 

70 assert_expected(gwy.schema, expected.get("schema")) 

71 assert_expected(gwy.params, expected.get("params")) 

72 assert_expected(gwy.status, expected.get("status")) 

73 assert_expected(gwy.known_list, expected.get("known_list")) 

74 

75 if gwy.msg_db: 

76 gwy.msg_db.stop() # close sqlite3 connection 

77 

78 

79def assert_raises(exception: type[Exception], fnc: Callable, *args: Any) -> None: 

80 try: 

81 fnc(*args) 

82 except exception: # as err: 

83 pass # or: assert True 

84 else: 

85 assert False 

86 

87 

88async def load_test_gwy(dir_name: Path, **kwargs: Any) -> Gateway: 

89 """Create a system state from a packet log (using an optional configuration).""" 

90 

91 kwargs = SCH_GLOBAL_CONFIG({k: v for k, v in kwargs.items() if k[:1] != "_"}) 

92 

93 try: 

94 with open(f"{dir_name}/config.json") as f: 

95 config = json.load(f) 

96 except FileNotFoundError: 

97 config = {} 

98 

99 if config: 

100 kwargs.update(config) 

101 

102 path = f"{dir_name}/packet.log" 

103 gwy = Gateway(None, input_file=path, **kwargs) 

104 await gwy.start() 

105 

106 await gwy._protocol.wait_for_connection_lost() # until packet log is EOF 

107 

108 # if hasattr( 

109 # gwy.pkt_transport.serial, "mock_devices" 

110 # ): # needs ser instance, so after gwy.start() 

111 # gwy.pkt_transport.serial.mock_devices = [MockDeviceCtl(gwy, CTL_ID)] 

112 

113 return gwy 

114 

115 

116def load_expected_results(dir_name: Path) -> dict[str, Any]: 

117 """Return the expected (global) schema/params/status & traits (aka known_list).""" 

118 

119 try: 

120 with open(f"{dir_name}/schema.json") as f: 

121 schema = json.load(f) 

122 except FileNotFoundError: 

123 schema = {} 

124 schema = SCH_GLOBAL_SCHEMAS(schema) 

125 

126 try: 

127 with open(f"{dir_name}/known_list.json") as f: 

128 known_list = json.load(f)["known_list"] 

129 except FileNotFoundError: 

130 known_list = {} 

131 known_list = SCH_GLOBAL_TRAITS({"known_list": shrink(known_list)})["known_list"] 

132 

133 try: 

134 with open(f"{dir_name}/params.json") as f: 

135 params = json.load(f)["params"] 

136 except FileNotFoundError: 

137 params = {} 

138 

139 try: 

140 with open(f"{dir_name}/status.json") as f: 

141 status = json.load(f)["status"] 

142 except FileNotFoundError: 

143 status = {} 

144 

145 # TODO: do known_list, status 

146 return { 

147 "schema": schema, 

148 "known_list": known_list, 

149 "params": params, 

150 "status": status, 

151 }