diff --git a/lambda/config_handler/__init__.py b/lambda/config_handler/__init__.py index 5ed71a3..df76891 100644 --- a/lambda/config_handler/__init__.py +++ b/lambda/config_handler/__init__.py @@ -2,6 +2,7 @@ import os import boto3 import json import functools +import logging @functools.lru_cache() def get(topic: str, parameter: str, default=None) -> str: @@ -26,6 +27,8 @@ def get(topic: str, parameter: str, default=None) -> str: try: secret_data = json.loads(sm.get_secret_value(SecretId=topic)['SecretString']) return secret_data[parameter] + except sm.exceptions.ClientError as error: + logging.warning(f"Error trying to use secret manager for {topic} {parameter} : {error}") except (KeyError, sm.exceptions.ResourceNotFoundException): pass except: diff --git a/lambda/station_api_to_iot_core/__init__.py b/lambda/station_api_to_iot_core/__init__.py index afc720a..31b780f 100644 --- a/lambda/station_api_to_iot_core/__init__.py +++ b/lambda/station_api_to_iot_core/__init__.py @@ -10,7 +10,7 @@ from io import BytesIO import boto3 import config_handler -CALLSIGN_BLOCK_LIST = ["CHANGEME_RDZTTGO"] +CALLSIGN_BLOCK_LIST = ["CHANGEME_RDZTTGO", "14FRS119_LE JUDA"] import es @@ -56,7 +56,7 @@ def lambda_handler(event, context): payload = json.loads(event["body"]) except: return {"statusCode": 400, "body": "JSON decode issue"} - print(payload) + print(json.dumps(event)) if "user-agent" in event["headers"]: event["time_server"] = datetime.datetime.now().isoformat() payload["user-agent"] = event["headers"]["user-agent"] diff --git a/lambda/station_api_to_iot_core/__main__.py b/lambda/station_api_to_iot_core/__main__.py index d5097f5..655f0f0 100644 --- a/lambda/station_api_to_iot_core/__main__.py +++ b/lambda/station_api_to_iot_core/__main__.py @@ -1,4 +1,10 @@ -from . import * +import unittest +import station_api_to_iot_core +from unittest.mock import MagicMock, call, patch +import datetime +import copy +import json + payload = { "version": "2.0", "routeKey": "PUT /sondes/telemetry", @@ -36,8 +42,35 @@ payload = { "timeEpoch": 1612051825409, }, "body": """ - {"software_name": "radiosonde_auto_rx", "software_version": "1.5.8-beta2", "uploader_callsign": "CHANGEME_RDZTTGO", "uploader_position": [null,null,null], "uploader_antenna": "Dipole", "uploader_contact_email": "none@none.com", "mobile": false} + {"software_name": "radiosonde_auto_rx", "software_version": "1.5.8-beta2", "uploader_callsign": "A", "uploader_position": [null,null,null], "uploader_antenna": "Dipole", "uploader_contact_email": "none@none.com", "mobile": false} """, "isBase64Encoded": False, } -print(lambda_handler(payload, {})) \ No newline at end of file + +class TestStation(unittest.TestCase): + def setUp(self): + station_api_to_iot_core.es.request = MagicMock() + station_api_to_iot_core.sns.publish = MagicMock() + + @patch('builtins.print') + def test_blocked_call(self, mocked_print): + blocked_call_payload = copy.deepcopy(payload) + body = json.loads(blocked_call_payload["body"]) + body["uploader_callsign"] = "CHANGEME_RDZTTGO" + blocked_call_payload["body"] = json.dumps(body) + station_api_to_iot_core.lambda_handler(blocked_call_payload,{}) + station_api_to_iot_core.es.request.assert_not_called() + station_api_to_iot_core.sns.publish.assert_not_called() + mocked_print.assert_called() + json.loads(mocked_print.call_args.args[0]) + + @patch('builtins.print') + def test_call(self, mocked_print): + blocked_call_payload = copy.deepcopy(payload) + body = json.loads(blocked_call_payload["body"]) + blocked_call_payload["body"] = json.dumps(body) + station_api_to_iot_core.lambda_handler(blocked_call_payload,{}) + station_api_to_iot_core.es.request.assert_called() + station_api_to_iot_core.sns.publish.assert_called() +if __name__ == '__main__': + unittest.main()