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

1#!/usr/bin/env python3 

2"""RAMSES RF - Test the payload parsers and corresponding output. 

3 

4Includes gwy dicts (schema, traits, params, status). 

5""" 

6 

7from pathlib import Path, PurePath 

8 

9import pytest 

10 

11from ramses_rf import Gateway 

12from ramses_rf.helpers import shrink 

13from ramses_tx.message import Message 

14from ramses_tx.packet import Packet 

15 

16from .helpers import ( 

17 TEST_DIR, 

18 assert_expected, 

19 assert_expected_set, 

20 load_expected_results, 

21 load_test_gwy, 

22 shuffle_dict, 

23) 

24 

25WORK_DIR = f"{TEST_DIR}/systems" 

26 

27 

28def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: 

29 def id_fnc(param: Path) -> str: 

30 return PurePath(param).name 

31 

32 folders = [f for f in Path(WORK_DIR).iterdir() if f.is_dir() and f.name[:1] != "_"] 

33 metafunc.parametrize("dir_name", folders, ids=id_fnc) 

34 

35 

36def test_payload_from_log_file(dir_name: Path) -> None: 

37 """Assert that each message payload is as expected (different to other tests).""" 

38 # RP --- 02:044328 18:200214 --:------ 2309 003 0007D0 # {'ufh_idx': '00', 'setpoint': 20.0} 

39 

40 def proc_log_line(log_line: str) -> None: 

41 if "#" not in log_line: 

42 return 

43 pkt_line, pkt_eval = log_line.split("#", maxsplit=1) 

44 

45 if not pkt_line[27:].strip(): 

46 return 

47 

48 pkt = Packet.from_file(pkt_line[:26], pkt_line[27:]) 

49 msg = Message(pkt) 

50 

51 assert msg.payload == eval(pkt_eval) 

52 

53 with open(f"{dir_name}/packet.log") as f: 

54 while line := (f.readline()): 

55 if line.strip(): 

56 proc_log_line(line) 

57 

58 

59async def test_schemax_with_log_file(dir_name: Path) -> None: 

60 """Compare the schema built from a log file with the expected results.""" 

61 

62 expected: dict = load_expected_results(dir_name) or {} 

63 gwy: Gateway = await load_test_gwy( 

64 dir_name, **expected["schema"], known_list=expected["known_list"] 

65 ) 

66 

67 schema, packets = gwy.get_state() 

68 

69 # sert_expected_set(gwy, expected) 

70 assert_expected(shrink(schema), shrink(expected["schema"])) 

71 

72 await gwy.stop() 

73 

74 

75async def test_systemx_from_log_file(dir_name: Path) -> None: 

76 """Compare the system built from a log file with the expected results.""" 

77 

78 expected: dict = load_expected_results(dir_name) or {} 

79 gwy: Gateway = await load_test_gwy(dir_name) 

80 

81 assert_expected_set(gwy, expected) 

82 # sert shrink(gwy.schema) == shrink(schema) 

83 

84 for dev in gwy.devices: 

85 _ = dev.schema 

86 _ = dev.traits 

87 _ = dev.params 

88 _ = dev.status 

89 

90 for tcs in gwy.systems: 

91 _ = tcs.schema 

92 _ = tcs.traits 

93 _ = tcs.params 

94 _ = tcs.status 

95 

96 await gwy.stop() 

97 

98 

99# async def test_restor1_from_log_file(dir_name: Path) -> None: 

100# """Compare the system built from a get_state log file with the expected results.""" 

101 

102# expected: dict = load_expected_results(dir_name) or {} 

103# gwy: Gateway = Gateway(None, input_file=io.StringIO()) # empty file 

104 

105# # schema, packets = gwy.get_state(include_expired=True) 

106# await gwy._restore_cached_packets(packets) 

107 

108# assert_expected_set(gwy, expected) 

109# # sert shrink(gwy.schema) == shrink(schema) 

110 

111# await gwy.stop() 

112 

113 

114async def test_restore_from_log_file(dir_name: Path) -> None: 

115 """Compare the system built from a get_state log file with the expected results.""" 

116 

117 expected: dict = load_expected_results(dir_name) or {} 

118 gwy: Gateway = await load_test_gwy(dir_name) 

119 

120 schema, packets = gwy.get_state(include_expired=True) 

121 await gwy._restore_cached_packets(packets) 

122 

123 assert_expected_set(gwy, expected) 

124 # sert shrink(gwy.schema) == shrink(schema) 

125 

126 for dev in gwy.devices: # SQLite refactor should pass this test 

127 if dev._gwy.msg_db: 

128 # depends on ramses_rf/entity_base.py def _msgs() from msg_db 

129 assert len(dev._msgs) == len(dev._msgs_), "_msgs not equal to _msgs_" 

130 # and make sure that every code from _msgs_ is in _msgb 

131 assert set(dev._msgs_).issubset(set(dev._msgs)), ( 

132 "_msgs_ not a subset of _msgs" 

133 ) 

134 # original test can't be met for UFC CTL 3150, see database.py qry() 

135 # assert sorted(dev._msgs) == sorted(dev._msgs_), ( 

136 # f"Assert 1: {dev} _msgs != _msgs_" 

137 # ) 

138 # don't expect them to match 100% because _msgs() creation requires filter: 

139 # sql = """ 

140 # SELECT dtm from messages WHERE verb in (' I', 'RP') 

141 # AND (src = ? OR dst = ?) """ 

142 # assert len(dev._gwy.msg_db.qry_field(sql, (dev.id[:12], dev.id[:12]))) == len( 

143 # dev._msgs_ 

144 # ), f"Assert 2: {dev} qry != _msgs_" 

145 

146 await gwy.stop() 

147 

148 

149async def test_shuffle_from_log_file(dir_name: Path) -> None: 

150 """Compare the system built from a shuffled log file with the expected results.""" 

151 

152 expected: dict = load_expected_results(dir_name) or {} 

153 gwy: Gateway = await load_test_gwy(dir_name) 

154 

155 schema, packets = gwy.get_state(include_expired=True) 

156 packets = shuffle_dict(packets) 

157 await gwy._restore_cached_packets(packets) 

158 

159 assert_expected_set(gwy, expected) 

160 # sert shrink(gwy.schema) == shrink(schema) 

161 

162 packets = shuffle_dict(packets) 

163 await gwy._restore_cached_packets(packets) 

164 

165 assert_expected_set(gwy, expected) 

166 # sert shrink(gwy.schema) == shrink(schema) 

167 

168 await gwy.stop()