Coverage for tests/test_mock_mobius.py: 100%

1import json 

2from datetime import datetime 

3 

4import pytest 

5from flask import Flask 

6from flask.testing import FlaskClient 

7from werkzeug.test import TestResponse 

8 

9from mobius_adapter.utils import Reading, to_mobius_payload 

10from mock_mobius import app as mobius_app 

11 

12 

13class TestFlaskApp: 

14 @pytest.fixture() 

15 def app(self) -> Flask: # type: ignore 

16 app = mobius_app.create_app("config_testing.json") 

17 # set up 

18 yield app 

19 # clean up 

20 

21 @pytest.fixture() 

22 def app_client(self, app: Flask) -> FlaskClient: 

23 return app.test_client() 

24 

25 def test_publish_data(self, app_client, reading: Reading): 

26 sensorId = "sensorId1_foo" 

27 sensorPath = "sensorPath1_foo" 

28 

29 payload: dict = to_mobius_payload(reading, sensorId) 

30 print(payload) 

31 response: TestResponse = app_client.post(f"/{sensorPath}", json=payload) 

32 

33 assert ( 

34 response.status_code == 200 

35 ), "Invalid response code from server when submitting valid payload" 

36 

37 response = app_client.post("/", json=payload) 

38 

39 assert ( 

40 response.status_code == 404 

41 ), "Invalid response code from server when submitting on invalid route" 

42 

43 def test_db_consistency(self, app_client, reading: Reading): 

44 """ 

45 This test is meant to check if data stored 

46 as Reading in database is consistent 

47 """ 

48 

49 sensorId = "sensorId1_foo" 

50 sensorPath = "sensorPath1_foo" 

51 payload: dict = to_mobius_payload(reading, sensorId) 

52 

53 response: TestResponse = app_client.post(f"/{sensorPath}", json=payload) 

54 

55 assert ( 

56 response.status_code == 200 

57 ), "Invalid response code from server when submitting valid payload" 

58 

59 data = app_client.get(f"/{sensorPath}").data 

60 data_dict: dict = json.loads(data.decode()) 

61 

62 print(data_dict) 

63 assert ( 

64 "m2m:rsp" in data_dict and "m2m:cin" in data_dict["m2m:rsp"] 

65 ), "Invalid structure of response from server when querying /<SENSOR_PATH>" 

66 

67 assert ( 

68 len(data_dict["m2m:rsp"]["m2m:cin"]) > 0 

69 ), "Reading doesn't get saved on the database" 

70 

71 def test_db_query_limits(self, app_client, reading: Reading): 

72 sensorId = "sensorId1_foo" 

73 sensorPath = "sensorPath1_foo" 

74 

75 reading_timestamps = [ 

76 datetime(2022, 7, 10, 12, 45, x) for x in [10, 20, 30, 40, 50] 

77 ] 

78 reading_timestamps_iso = [int(x.timestamp()) for x in reading_timestamps] 

79 

80 for time in reading_timestamps_iso: 

81 reading.readingID = time 

82 print(reading.publishedAt) 

83 payload: dict = to_mobius_payload(reading, sensorId) 

84 

85 response: TestResponse = app_client.post(f"/{sensorPath}", json=payload) 

86 

87 assert ( 

88 response.status_code == 200 

89 ), "Invalid response code from server when submitting valid payload" 

90 

91 inf_limit: str = reading_timestamps[0].strftime("%Y%m%dT%H%M%S") 

92 sup_limit: str = reading_timestamps[-1].strftime("%Y%m%dT%H%M%S") 

93 response = app_client.get(f"/{sensorPath}?crb={sup_limit}&cra={inf_limit}") 

94 

95 assert ( 

96 response.status_code == 200 

97 ), "Invalid response code from server when querying with time limits" 

98 

99 decoded_response: dict = json.loads(response.data) 

100 assert ( 

101 "m2m:rsp" in decoded_response and "m2m:cin" in decoded_response["m2m:rsp"] 

102 ), "Invalid structure of response from server when querying /<SENSOR_PATH>" 

103 

104 assert ( 

105 len(decoded_response["m2m:rsp"]["m2m:cin"]) == 3 

106 ), f"Invalid response from server, expected 3 results \ 

107 but got {len(decoded_response['m2m:rsp']['m2m:cin'])}" 

108 

109 response = app_client.get( 

110 f"/{sensorPath}?crb={sup_limit}&cra={inf_limit}&lim=2" 

111 ) 

112 

113 assert ( 

114 response.status_code == 200 

115 ), "Invalid response code from server when querying \ 

116 with time limits and quantity limits" 

117 

118 decoded_response: dict = json.loads(response.data) 

119 assert ( 

120 "m2m:rsp" in decoded_response and "m2m:cin" in decoded_response["m2m:rsp"] 

121 ), "Invalid structure of response from server when querying /<SENSOR_PATH>" 

122 

123 assert ( 

124 len(decoded_response["m2m:rsp"]["m2m:cin"]) == 2 

125 ), f"Invalid response from server, expected 2 results \ 

126 but got {len(decoded_response['m2m:rsp']['m2m:cin'])}" 

127 

128 def test_db_query_last(self, app_client): 

129 sensorPath = "sensorPath1_foo" 

130 

131 response: TestResponse = app_client.get(f"/{sensorPath}") 

132 decoded_response: dict = json.loads(response.data) 

133 

134 readings = [x for x in decoded_response["m2m:rsp"]["m2m:cin"]] 

135 

136 readings.sort( 

137 key=lambda x: x["con"]["metadata"]["readingTimestamp"], reverse=True 

138 ) 

139 last_reading: dict = readings[0] 

140 

141 response = app_client.get(f"/{sensorPath}/la") 

142 

143 assert ( 

144 response.status_code == 200 

145 ), "Invalid response code from server when querying last sensor reading" 

146 

147 decoded_response = json.loads(response.data) 

148 

149 assert ( 

150 "m2m:cin" in decoded_response 

151 ), "Invalid response structure when querying last sensor reading: \ 

152 expected 'm2m:cin' in response json" 

153 

154 assert ( 

155 decoded_response["m2m:cin"] == last_reading 

156 ), "Invalid response when querying last sensor reading: it is not the latest"