ExternalVendorCode/sondehub-infra
1
0
Fork 0
mirror of https://github.com/projecthorus/sondehub-infra.git synced 2024-12-19 21:27:55 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Bump required sondemonitor version to 6.2.8.8, add tests

Browse Source
This commit is contained in:
Mark Jessop 2023-10-27 19:55:40 +10:30
parent be1facf8a6
commit 1811f8b9fa
2 changed files with 22 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
lambda/sonde_api_to_iot_core/__init__.py
View File

@ -312,8 +312,8 @@ def telemetry_filter(telemetry):
return (False, f"dxlAPRS-SHUE versions below 1.1.2 due not send correct serial numbers for M10 radiosondes. Please update to 1.1.2 or later")
if "DFM" in telemetry["type"]:
if telemetry["software_name"] == "SondeMonitor":
if parse_sondemonitor_version(telemetry["software_version"]) < (6,2,8,7):
return (False,f"SondeMonitor version is out of date and doesn't handle DFM radiosondes correctly. Please update to 6.2.8.7 or later")
if parse_sondemonitor_version(telemetry["software_version"]) < (6,2,8,8):
return (False,f"SondeMonitor version is out of date and doesn't handle DFM radiosondes correctly. Please update to 6.2.8.8 or later")
if telemetry["software_name"] == "rdzTTGOsonde":
ttgo_branch, ttgo_version = parse_rdz_ttgo_version(telemetry["software_version"])
if ttgo_branch == "devel":

21
lambda/sonde_api_to_iot_core/__main__.py
View File

@ -181,7 +181,26 @@ class TestIngestion(unittest.TestCase):
sns.publish.assert_not_called()
body_decode = json.loads(output["body"])
self.assertEqual(body_decode["message"], "some or all payloads could not be processed")
def test_good_sondemonitor_payload(self):
payload = copy.deepcopy(example_body)
payload[0]["datetime"] = datetime.datetime.now().isoformat()
payload[0]["software_name"] = "SondeMonitor"
payload[0]["software_version"] = "6.2.8.8"
payload[0]["type"] = "DFM"
output = lambda_handler(compress_payload(payload), fakeContext())
sns.publish.assert_called()
self.assertEqual(output["body"], "^v^ telm logged")
self.assertEqual(output["statusCode"], 200)
def test_bad_sondemonitor_payload(self):
payload = copy.deepcopy(example_body)
payload[0]["datetime"] = datetime.datetime.now().isoformat()
payload[0]["software_name"] = "SondeMonitor"
payload[0]["software_version"] = "6.2.8.7"
payload[0]["type"] = "DFM"
output = lambda_handler(compress_payload(payload), fakeContext())
sns.publish.assert_not_called()
body_decode = json.loads(output["body"])
self.assertEqual(body_decode["message"], "some or all payloads could not be processed")
if __name__ == '__main__':
unittest.main()