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

add additional callsign to block

Browse Source
This commit is contained in:
xssfox 2024-02-01 14:58:04 +11:00
parent 822ca559da
commit 749204dd22
3 changed files with 41 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
lambda/config_handler/__init__.py
View File

@ -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:

4
lambda/station_api_to_iot_core/__init__.py
View File

@ -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"]

39
lambda/station_api_to_iot_core/__main__.py
View File

@ -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, {}))
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()