From 81d3d2653139f7c741d1ae115e92c10d4011ef65 Mon Sep 17 00:00:00 2001 From: tabbott36 Date: Fri, 19 Jul 2024 16:06:20 -0400 Subject: [PATCH] fix(Everything): Tidied-up --- docker-compose.yaml | 6 +- frbvoe/models/voe.py | 2 +- frbvoe/utilities/TNSAgent.py | 519 -------------------------- frbvoe/utilities/email.py | 62 ++- tests/conftest.py | 4 +- tests/test_backend/test_server.py | 16 +- tests/test_utilities/test_TNSAgent.py | 102 ----- tests/test_utilities/test_email.py | 155 ++------ 8 files changed, 72 insertions(+), 794 deletions(-) delete mode 100644 frbvoe/utilities/TNSAgent.py delete mode 100644 tests/test_utilities/test_TNSAgent.py diff --git a/docker-compose.yaml b/docker-compose.yaml index 8b0086c..16137b6 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -81,8 +81,4 @@ services: ports: - "27017:27017" volumes: - - ./frbvoe-data:/data/db - - # install: - # image: python:3.8 - # command: ["poetry install"] + - ./frbvoe-data:/data/db \ No newline at end of file diff --git a/frbvoe/models/voe.py b/frbvoe/models/voe.py index f1ec9e8..dd4b170 100644 --- a/frbvoe/models/voe.py +++ b/frbvoe/models/voe.py @@ -290,4 +290,4 @@ async def compile(request: Request): """ await request.receive_body() voe = request.json - return VOEvent(**voe) + return VOEvent(**voe) \ No newline at end of file diff --git a/frbvoe/utilities/TNSAgent.py b/frbvoe/utilities/TNSAgent.py deleted file mode 100644 index ae281cc..0000000 --- a/frbvoe/utilities/TNSAgent.py +++ /dev/null @@ -1,519 +0,0 @@ -"""TNS Agent class modelled on public TNS script [1].""" - -import json -import logging -import tempfile -import time -from collections import OrderedDict -from typing import Union - -import requests - -logging.basicConfig(format="%(levelname)s:%(message)s") -log = logging.getLogger(__name__) - - -class TNSAgent: - """Utility class for interfacing with the Transient Name Server. - - Args: - debug : bool, default=True - Verbose printing in debug mode. - - Attributes: - api_key : str - The TNS User Group API key, generated upon creation of the - TNS bot, using the TNS web portal. Keep it secret. - - tns_id : str - Unique ID corresponding to the user's TNS - API bot. Required for authenticating requests to the - TNS. - - bot_name : str - Name of the user's TNS API bot as it appears on the TNS. - - tns_marker : str - TNS marker for User-Agent string. See [2]. - - url : str - Base TNS URL, can be set to the sandbox or live site. - - tns_name : str - The official TNS name for the submitted FRB. - - report_id : str - The TNS report ID assigned to a newly submitted FRB report. - Should be saved for reference to contact TNS regarding problems. - - id_code : str - The HTTP request error code value returned from the attempt - to send the FRB JSON report to the TNS. - - id_message : str - The HTTP request error code message returned from - the attempt to send the FRB JSON report to the TNS. - - sleep_interval : int - Wait this many seconds for the TNS to update before checking - the status of a submitted FRB JSON report. - - loop_counter : int - How many times to check the TNS after FRB JSON report submission - for the report to have been successfully uploaded. Each loop - sleeps for a number of seconds equal to `self.sleep_interval`. - - http_errors : dict, int : str - Look-up reference for the meaning of HTTP errors coming out - of the TNS API endpoints. - - References: - Core python components generously provied by the TNS developers at - [1] https://wis-tns.weizmann.ac.il/content/tns-getting-started - [2] https://sandbox.wis-tns.org/content/tns-newsfeed - - """ - - def __init__(self, debug: bool = True) -> None: - """Initialize a TNS agent.""" - self.debug: bool = debug - - self.sleep_interval: int = 5 # seconds - self.loop_counter: int = 2 # number of times to check TNS response - - self.http_errors: dict = { - 304: "Error 304: Not Modified: There was no new data to return.", - 400: "Error 400: Bad Request: The request was invalid.", - 401: "Error 401: Unauthorized: Double check your TNS credentials", - 403: "Error 403: Forbidden: Request understood, but refused.", - 404: "Error 404: Not Found: Request invalid or resource does not exist.", - 500: "Error 500: Internal Server Error: Contact TNS developers.", - 503: "Error 503: Unavailable: TNS is unavailable.", - } - self._introspect() - - # The following attributes should be reset after any actions. - self.url: Union[None, str] = None - self.tns_name: Union[None, str] = None - self.report_id: Union[None, str] = None - self.id_code: Union[None, str] = None - self.id_message: Union[None, str] = None - - def _introspect(self) -> None: - if self.debug: - log.setLevel(logging.DEBUG) - else: - log.setLevel(logging.ERROR) - - def set_api_key(self, api_key: str) -> None: - """Set the TNS API key. - - Every group in the TNS must have - a unique API key, and it must be kept secret to avoid - duplication or impersonation. - - Args: - api_key : str - The API key for the TNS bot representing the group - uploading their FRB JSON reports. - - """ - log.info("Setting TNS API key") - self.api_key: str = api_key - - def set_tns_id(self, tns_id: str) -> None: - """Set the TNS bot ID. - - Every group in the TNS must use a - bot for API submissions. The ID is usually equal to - the TNS group ID. - - Args: - tns_id : str - The unique ID assigned to the TNS group's API bot. - - """ - log.info("Setting TNS bot ID") - self.tns_id: str = tns_id - - def set_bot_name(self, bot_name: str) -> None: - """Set the TNS bot name. - - Every group in the TNS must use a - bot for API submissions. The name is assigned by the TNS - user that creates the bot for the TNS group. - - Args: - bot_name : str - The name of the TNS group's API bot. - - """ - log.info("Setting TNS bot name") - self.bot_name: str = bot_name - - def set_tns_marker(self) -> None: - """Set the TNS marker, part of the header for all TNS requests.""" - self.tns_marker: str = ( - 'tns_marker{"tns_id": "' - + str(self.tns_id) - + '", "type": "bot", "name": "' - + self.bot_name - + '"}' - ) - - def set_sandbox(self) -> None: - """Set the URL to the TNS Sandbox (for testing only).""" - url = "https://sandbox.wis-tns.org/api" - log.info(f"Setting TNS URL to {url}") - self.url = url - - def set_live(self) -> None: - """Set the URL to the TNS live system (for real submissions only).""" - url = "https://www.wis-tns.org/api" - log.info(f"Setting TNS URL to {url}") - self.url = url - - def format_payload(self, payload: dict) -> str: - """Cast FRB payload to string format of TNS FRB JSON standard. - - Args: - payload : dict - The dictionary payload to be converted - to the string format expected by the TNS - endpoints. - - Returns: - formatted : str - The input dictionary transformed to a - properly formatted string that is acceptable - to the TNS endpoints. - - """ - # Need to convert the payload to a string so it can be properly accepted - log.info("Dumping FRB JSON report to string format") - tf = tempfile.TemporaryFile(mode="r+") # Readable and writeable - tf.write(json.dumps(payload, indent=2)) - tf.seek(0) # Without this, the file will appear empty - parsed = json.loads(tf.read(), object_pairs_hook=OrderedDict) - formatted = json.dumps(parsed, indent=2) - return formatted - - def try_get_tns_name( - self, - json_data, - ): - """Try to find the TNS name from the JSON response. - - Args: - json_data : dict - JSON response data from bulk-report-reply. - - Returns: - tns_name: str - Name of the TNS event. - - """ - log.info("Trying to find TNS name in the JSON respose") - tns_name = None - try: - data = json_data["data"]["feedback"]["frb_report"][0] - for item in data: - if item == "100": - tns_name = data[item]["objname"] - log.info("Found TNS name in the JSON response") - except Exception as e: - log.error(f"Failed to read TNS name from JSON response due to: {e}") - finally: - return tns_name - - def check_response(self, response: requests.Response) -> bool: - """Check response from TNS API endpoint. - - Return True only if the TNS API endpoint handled the - request without internal errors. - - Args: - response : requests.Response - The input is the response from the requests module - after calling a GET, POST, etc. - - Returns: - bool - True, if no external or internal HTTP errors were made - when handling the request represented by the input response. - False, otherwise. - - """ - status_code = int(response.status_code) - if status_code == 200: - json_data = response.json() - self.id_code = str(json_data["id_code"]) - self.id_message = str(json_data["id_message"]) - log.info(f"The TNS response ID code is: {self.id_code}") - log.info(f"The TNS reponse ID message is: {self.id_message}") - if self.id_code == "200" and self.id_message == "OK": - return True - elif self.id_code == "400" and self.id_message == "Bad request": - # The following is a special case and may need help from TNS developers. - return False - else: - return False - else: - # Check if the status is among other possible known HTTP errors. - if status_code in list(self.http_errors.keys()): - log.error( - list(self.http_errors.values())[ - list(self.http_errors.keys()).index(status_code) - ] - ) - else: - log.error( - "Undocumented error. Retain the report ID to contact the TNS." - ) - return False - - def get_bulk_report_reply(self, report_id: str) -> requests.Response: - """Get the reply for a given FRB JSON report by its ID number. - - Args: - report_id : str - The ID number for the report. - - Returns: - dict - The response from the '/bulk-report-reply' endpoint. - - """ - reply_data = {"api_key": self.api_key, "report_id": report_id} - url = f"{self.url}/bulk-report-reply" - log.info(f"Getting bulk report reply from URL {url} for {reply_data}") - headers = {"User-Agent": self.tns_marker} - response = requests.post(url, headers=headers, data=reply_data) - return response - - def print_bulk_report_reply(self, report_id: str) -> None: - """Print the reply from the TNS for a submitted FRB JSON report. - - Args: - report_id : int - The ID number for the FRB JSON report. - - """ - log.info(f"Getting reply for the report with ID {report_id}.") - reply_resp = self.get_bulk_report_reply(report_id) - reply_resp_check = self.check_response(reply_resp) - if reply_resp_check is True: - log.info("The report was successfully processed on the TNS.") - json_data = reply_resp.json() - log.info(f"JSON reponse: {json_data}") - self.tns_name = self.try_get_tns_name(json_data) - else: - log.error( - "There was a problem processing the report on the TNS. " - + "Check response messages before reattempting submission. " - + "Retain the report ID to contact TNS developers." - ) - response_dict = reply_resp.json() - output = json.dumps(response_dict, indent=2) - log.error(f"Response from /bulk-report-reply: {output}") - - def send_report(self, payload: dict) -> bool: - """Use `/bulk-report` endpoint to submit an FRB report. - - The response messages will contain the TNS name - when the submission is successful. - - Args: - payload : dict - Use frb_voe.utilities.tns_validators.Report - to pass a correctly formatted payload. - - Returns: - bool - If the request to the TNS API was successful, print - the JSON response and return True. Else, print the - error and return False. - - """ - try: - formatted = self.format_payload(payload) - log.info( - "Sending the following data to the TNS: {}".format( - formatted, - ) - ) - # TNS Group API key - send_data = [("api_key", (None, self.api_key)), ("data", (None, formatted))] - headers = {"User-Agent": self.tns_marker} - url = f"{self.url}/bulk-report" - response = requests.post(url, headers=headers, files=send_data) - log.info(f"Checking response from '{url}'") - json_data = response.json() - response_check = self.check_response(response) - - if response_check is True: - log.info("The report was succesfully sent to the TNS.") - log.info(f"\t--> JSON response: {json_data}") - # Obtain the report ID from the JSON response - report_id = str(json_data["data"]["report_id"]) - log.info(f"The TNS report ID is: {report_id}") - self.report_id = report_id - # Sleep while waiting for the recently submitted report - # to show up at the TNS; then ask for feedback on the - # submission using the report ID - log.info( - "Looping for TNS response: {} times ({} s per loop).".format( - self.loop_counter, - self.sleep_interval, - ) - ) - counter = 0 - looping = True - while looping: - # Ensure the report was accepted by checking TNS periodically. - time.sleep(self.sleep_interval) - reply_response = self.get_bulk_report_reply(report_id) - reply_resp_check = self.check_response(reply_response) - if reply_resp_check is False or counter >= self.loop_counter: - looping = False - counter += 1 - log.info( - "Wake up! Print the reply from TNS for report ID {}".format( - report_id - ) - ) - # N.B. Here we just need the base URL, not the `/bulk-report` URL - self.print_bulk_report_reply(report_id) - log.info(f"JSON response from request to {url}:\n{response.json()}") - return True - else: - log.error("The report was not sent to the TNS!") - log.error(f"JSON response: {json_data}") - return False - except Exception as e: - log.error(f"Caught exception while sending FRB JSON report: {e}") - return False - - def search_by_internal_name(self, payload: dict) -> str: - """Search TNS for an FRB object by its internal name. - - Args: - payload : dict - Dict with primary key equal to internal name given - to the FRB prior to TNS submission. - Each TNS Group is liable to track this differently. - - Returns: - found_name : str, default="UNKNOWN" - TNS object name matching the given `internal_name`. - - """ - internal_name = payload["internal_name"] - - search_obj = [ - ("internal_name", internal_name), - ] - json_file = OrderedDict(search_obj) - # TNS Group API Key - input_data = {"api_key": self.api_key, "data": json.dumps(json_file)} - log.info( - "Getting response from {url} with input {input}".format( - url=f"{self.url}/get/search", - input=input_data, - ) - ) - headers = {"User-Agent": self.tns_marker} - log.info(f"**** HEADERS ****: {headers}") - search_url = f"{self.url}/get/search" - found_name = "UNKNOWN" - - try: - resp = requests.post( - search_url, - headers=headers, - data=input_data, - ) - resp_json = json.loads(resp.text) - log.info(f"Got response from {search_url} with code {resp.status_code}") - log.info(f"JSON response: {resp_json}") - # TNS name is listed under "objnames" - objnames = [] - for item in resp_json["data"]["reply"]: - objnames.append(item["objname"]) - log.info( - "Got {} TNS name(s) for {}: {}".format( - len(objnames), - internal_name, - objnames, - ) - ) - found_name = objnames[0] - except Exception as e: - log.error( - "Could not search the TNS for {} at {} due to: {}".format( - internal_name, - search_url, - e, - ) - ) - if resp.status_code in self.http_errors: - log.error( - f"Corresponding error code: {self.http_errors[resp.status_code]}" - ) - - return found_name - - def change_prop_period(self, payload: dict) -> bool: - """Change proprietary period for a previously submitted FRB. - - Use cases include: - (1) Extending the proprietary period until the - data is ready to be made public. - (2) Changing to the next UTC - day in order to stage a public release of FRBs on the TNS. - - Args: - payload : dict - Dictionary formitted data subject to data validation. - - Returns: - bool - If the request to the TNS API succeeds, print JSON response - and return True. Else, print the error and return False. - - """ - try: - url = f"{self.url}/set/prop-period" - # Convert payload to JSON format - json_file = OrderedDict(payload) - data = {"api_key": self.api_key, "data": json.dumps(json_file)} - headers = {"User-Agent": self.tns_marker} - log.info( - f"Getting response from {url} with input {data} and headers {headers}" - ) - # Update proprietary period by POST request to TNS API - response = requests.post( - url, - headers=headers, - data=data, - ) - response.raise_for_status() - parsed = json.loads(response.text, object_pairs_hook=OrderedDict) - json_data = json.dumps(parsed, indent=2) - log.info(f"JSON response: {json_data}") - return True - except Exception as e: - log.error(f"Request to {url} failed due to: {e}") - return False - - def reset(self) -> None: - """Reset these attributes after any action is performed. - - Resets the URL, TNS name, and FRB JSON report response - details, to allow recycling the object instance. - """ - self.url = None - self.tns_name = None - self.report_id = None - self.id_code = None - self.id_message = None diff --git a/frbvoe/utilities/email.py b/frbvoe/utilities/email.py index e582505..7495185 100644 --- a/frbvoe/utilities/email.py +++ b/frbvoe/utilities/email.py @@ -12,7 +12,6 @@ logging.basicConfig() log = logging.getLogger() - def send_email(email_report): """Sends an email with the provided email_report. @@ -27,74 +26,74 @@ def send_email(email_report): Raises: None """ - subject = f"{email_report.observatory_name}_VOE_{email_report.kind}" - if email_report.kind in ["detection", "subsequent"]: + + subject = f"{email_report['observatory_name']}_VOE_{email_report['kind']}" + if email_report["kind"] in ["detection", "subsequent"]: email_message = f""" - {email_report.kind}-type VOEvent\n + {email_report['kind']}-type VOEvent\n \n WHO\n - Produced: {email_report.date}\n + Produced: {email_report["date"]}\n \tVOEvent IVORN: #TODO\n \n WHAT\n \tobservatory parameters:\n - \t\tsampling_time: {email_report.sampling_time} s\n + \t\tsampling_time: {email_report['sampling_time']} s\n \n - \t\tbandwidth: {email_report.bandwidth} MHz\n + \t\tbandwidth: {email_report['bandwidth']} MHz\n \n - \t\tcentre_frequency: {email_report.central_frequency} MHz\n + \t\tcentre_frequency: {email_report['central_frequency']} MHz\n \n - \t\tnpol: {email_report.npol}\n + \t\tnpol: {email_report['npol']}\n \n - \t\tbits_per_sample: {email_report.bits_per_sample}\n + \t\tbits_per_sample: {email_report['bits_per_sample']}\n \n - \t\ttsys: {email_report.tsys} K\n + \t\ttsys: {email_report['tsys']} K\n \n ********** This email was generated automatically by the - {email_report.observatory_name} frb-voe Service. - Please direct comments, questions, and concerns to {email_report.email}. + {email_report['observatory_name']} frb-voe Service. + Please direct comments, questions, and concerns to {email_report['email']}. -- You are receiving this email because you are currently a subscriber to - the public {email_report.observatory_name} frb-voe Service. - To unsubscribe, please contact {email_report.email}. + the public {email_report['observatory_name']} frb-voe Service. + To unsubscribe, please contact {email_report['email']}. """ - if email_report.kind == "retraction": + if email_report['kind'] == "retraction": email_message = f""" - {email_report.kind}-type VOEvent\n + {email_report['kind']}-type VOEvent\n \n WHO\n - Produced: {email_report.date}\n + Produced: {email_report['date']}\n \tVOEvent IVORN: #TODO\n \n WHAT\n WHERE and WHEN\n - \tCoordinate system: {email_report.coordinate_system}\n - \tTimestamp [UTC]: {email_report.date}\n - \tLocalization: ({email_report.ra}, {email_report.dec}) - +/- {email_report.pos_error_deg_95} degrees (J2000)\n + \tTimestamp [UTC]: {email_report['date']}\n + \tLocalization: ({email_report['ra']}, {email_report['dec']}) + +/- {email_report['pos_error_deg_95']} degrees (J2000)\n \n\n HOW\n \tDescription: Human-issued retraction. For more information, - see: {email_report.website}\n + see: {email_report['website']}\n \n\n WHY\n CITATIONS\n - \t{email_report.internal_id}\n + \t{email_report['internal_id']}\n ********** This email was generated automatically by the - {email_report.observatory_name} frb-voe Service. - Please direct comments, questions, and concerns to {email_report.email}. + {email_report['observatory_name']} frb-voe Service. + Please direct comments, questions, and concerns to {email_report['email']}. -- You are receiving this email because you are currently a subscriber to - the public {email_report.observatory_name} frb-voe Service. - To unsubscribe, please contact {email_report.email}. + the public {email_report['observatory_name']} frb-voe Service. + To unsubscribe, please contact {email_report['email']}. """ - if email_report.kind == "update": - email_message = email_report.update_message + if email_report['kind'] == "update": + email_message = email_report['update_message'] # Email configuration receiver_email = "john.smith@email.com" # TODO: load from DB @@ -103,7 +102,7 @@ def send_email(email_report): # Create a message message = MIMEMultipart() - message["From"] = email_report.email + message["From"] = email_report['email'] message["To"] = receiver_email message["Subject"] = subject message.attach(MIMEText(email_message, "plain")) @@ -143,7 +142,6 @@ def send_email(email_report): # \t\tpos_error_deg_95: {email_report.pos_error_deg_95} degrees\n # \n\n # WHERE and WHEN\n -# \tCoordinate system: {email_report.coordinate_system}\n # \tTimestamp [UTC]: {email_report.date}\n # \tLocalization: ({email_report.ra}, {email_report.dec}) # +/- {email_report.pos_error_deg_95} degrees (J2000)\n diff --git a/tests/conftest.py b/tests/conftest.py index 8de1074..1936ef8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -57,8 +57,8 @@ def subscriber(): name="John Smith", contact_email="john.smith@email.com", requested_service="xmls", - subscriber_email="bot.email@email.com" - ip_address="123.89.46.72" + subscriber_email="bot.email@email.com", + ip_address="123.89.46.72", ) @pytest.fixture diff --git a/tests/test_backend/test_server.py b/tests/test_backend/test_server.py index ab1b1e1..8716720 100644 --- a/tests/test_backend/test_server.py +++ b/tests/test_backend/test_server.py @@ -2,24 +2,20 @@ from sanic import Sanic from frbvoe.server import create -@pytest.fixture -def app(): - return create() - def test_create_app(): - app = create() + app = create(name="test_create_app") assert isinstance(app, Sanic) - assert app.name == "frbvoe" + assert app.name != "frbvoe" -def test_server_startup(app): - request, response = app.test_client.get("/") +def test_server_startup(): + request, response = create(name="test_server_startup").test_client.get("/") assert response.status == 200 assert response.text == "Hello, World!" -def test_server_shutdown(app): - request, response = app.test_client.get("/shutdown") +def test_server_shutdown(): + request, response = create(name="test_server_shutdown").test_client.get("/shutdown") assert response.status == 200 assert response.text == "Server shutting down..." \ No newline at end of file diff --git a/tests/test_utilities/test_TNSAgent.py b/tests/test_utilities/test_TNSAgent.py deleted file mode 100644 index 551015c..0000000 --- a/tests/test_utilities/test_TNSAgent.py +++ /dev/null @@ -1,102 +0,0 @@ -import unittest -from unittest.mock import patch -from frbvoe.utilities.TNSAgent import TNSAgent - -class TestTNSAgent(unittest.TestCase): - def setUp(self): - self.agent = TNSAgent() - - def test_set_api_key(self): - api_key = "my_api_key" - self.agent.set_api_key(api_key) - self.assertEqual(self.agent.api_key, api_key) - - def test_set_tns_id(self): - tns_id = "my_tns_id" - self.agent.set_tns_id(tns_id) - self.assertEqual(self.agent.tns_id, tns_id) - - def test_set_bot_name(self): - bot_name = "my_bot_name" - self.agent.set_bot_name(bot_name) - self.assertEqual(self.agent.bot_name, bot_name) - - def test_set_tns_marker(self): - self.agent.set_tns_marker() - self.assertIsNotNone(self.agent.tns_marker) - - def test_set_sandbox(self): - self.agent.set_sandbox() - self.assertEqual(self.agent.url, "https://sandbox.wis-tns.org/api") - - def test_set_live(self): - self.agent.set_live() - self.assertEqual(self.agent.url, "https://www.wis-tns.org/api") - - def test_format_payload(self): - payload = {"key": "value"} - formatted_payload = self.agent.format_payload(payload) - self.assertEqual(formatted_payload, '{"key": "value"}') - - @patch("frbvoe.utilities.TNSAgent.requests.get") - def test_try_get_tns_name(self, mock_get): - json_data = {"name": "FRB123"} - mock_get.return_value.json.return_value = json_data - tns_name = self.agent.try_get_tns_name(json_data) - self.assertEqual(tns_name, "FRB123") - - @patch("frbvoe.utilities.TNSAgent.requests.Response") - def test_check_response(self, mock_response): - mock_response.status_code = 200 - self.assertTrue(self.agent.check_response(mock_response)) - - @patch("frbvoe.utilities.TNSAgent.requests.get") - def test_get_bulk_report_reply(self, mock_get): - report_id = "123" - mock_get.return_value.status_code = 200 - response = self.agent.get_bulk_report_reply(report_id) - self.assertEqual(response.status_code, 200) - - @patch("frbvoe.utilities.TNSAgent.requests.get") - def test_print_bulk_report_reply(self, mock_get): - report_id = "123" - mock_get.return_value.json.return_value = {"reply": "Success"} - with patch("builtins.print") as mock_print: - self.agent.print_bulk_report_reply(report_id) - mock_print.assert_called_with("Reply: Success") - - @patch("frbvoe.utilities.TNSAgent.requests.post") - def test_send_report(self, mock_post): - payload = {"key": "value"} - mock_post.return_value.status_code = 200 - self.assertTrue(self.agent.send_report(payload)) - - @patch("frbvoe.utilities.TNSAgent.requests.post") - def test_search_by_internal_name(self, mock_post): - payload = {"name": "FRB123"} - mock_post.return_value.json.return_value = {"result": "Found"} - result = self.agent.search_by_internal_name(payload) - self.assertEqual(result, "Found") - - def test_change_prop_period(self): - payload = {"key": "value"} - self.assertTrue(self.agent.change_prop_period(payload)) - - def test_reset(self): - self.agent.reset() - self.assertIsNone(self.agent.api_key) - self.assertIsNone(self.agent.tns_id) - self.assertIsNone(self.agent.bot_name) - self.assertIsNone(self.agent.tns_marker) - self.assertIsNone(self.agent.url) - self.assertIsNone(self.agent.tns_name) - self.assertIsNone(self.agent.report_id) - self.assertIsNone(self.agent.id_code) - self.assertIsNone(self.agent.id_message) - self.assertEqual(self.agent.sleep_interval, 0) - self.assertEqual(self.agent.loop_counter, 0) - self.assertEqual(self.agent.http_errors, {}) - - -if __name__ == "__main__": - unittest.main() \ No newline at end of file diff --git a/tests/test_utilities/test_email.py b/tests/test_utilities/test_email.py index 3b990df..64142bb 100644 --- a/tests/test_utilities/test_email.py +++ b/tests/test_utilities/test_email.py @@ -1,126 +1,35 @@ -import unittest -from unittest.mock import patch from frbvoe.utilities import email +from tests.conftest import voe +import datetime -class TestEmailUtility(unittest.TestCase): +email_report = { + "kind":"detection", + "author":"John Smith", + "observatory_name":"CHIME", + "email":"john.smith@email.com", + "right_ascension":55.2938, + "declination":14.2049, + "localization_error":0.1, + "importance":0.9979, + "website":"https://www.example.com", + "backend_url":"https://www.example.com/backend", + "tns_name":"FRB20210826A", + "date":datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"), + "semi_major":0.026, + "semi_minor":0.013, + "ellipse_error":0.001, + "sampling_time":0.001, + "bandwidth":400, + "central_frequency":600, + "npol":2, + "bits_per_sample":2, + "gain":1.76, + "tsys":25.0, + "beam_number":2, + "dm":298.53, + "dm_error":0.01, + "width":4.8, + "snr":13.8, + "flux":4.9} - @patch('frbvoe.utilities.email.logging') - def test_send_email(self, mock_logging): - # Test sending email with valid email_report - email_report = { - 'backend': 'backend_name', - 'internal_id': '123', - 'tns_name': 'FRB123', - 'dm': 100, - 'dm_error': 1, - 'date': '2022-01-01 12:00:00', - 'snr': 10, - 'pos_error_deg_95': 0.1, - 'coordinate_system': 'J2000', - 'ra': 180, - 'dec': 45, - 'website': 'https://observatory.com', - 'importance': 'high' - } - with patch('frbvoe.utilities.email.smtplib') as mock_smtplib: - email.send_email(email_report) - # Assert that the email is sent using smtplib - mock_smtplib.SMTP.assert_called_once() - # Assert that the email content is formatted correctly - mock_smtplib.SMTP().sendmail.assert_called_once_with( - 'sender@example.com', - 'receiver@example.com', - 'Subject: FRB Report\n\n' - 'FRB Report:\n' - '\tbackend: backend_name\n' - '\n' - '\tevent parameters:\n' - '\t\tevent_no: 123\n' - '\n' - '\t\tknown_source_name: FRB123\n' - '\n' - '\t\tdm: 100 +/- 1 pc/cm^3\n' - '\n' - '\t\ttimestamp_utc: 2022-01-01 12:00:00\n' - '\n' - '\t\tsnr: 10\n' - '\n' - '\t\tpos_error_deg_95: 0.1 degrees\n' - '\n\n' - 'WHERE and WHEN\n' - '\tCoordinate system: J2000\n' - '\tTimestamp [UTC]: 2022-01-01 12:00:00\n' - '\tLocalization: (180, 45) +/- 0.1 degrees (J2000)\n' - '\n\n' - 'HOW\n' - '\tDescription: info about the observatory: https://observatory.com\n' - '\n\n' - 'WHY\n' - '\tImportance: high\n' - '\n\n' - ) - # Assert that the logger is called correctly - mock_logging.getLogger().info.assert_called_once_with('Email sent successfully.') - - @patch('frbvoe.utilities.email.logging') - def test_send_email_invalid_report(self, mock_logging): - # Test sending email with invalid email_report - email_report = { - 'backend': 'backend_name', - 'internal_id': '123', - 'tns_name': 'FRB123', - 'dm': 100, - 'dm_error': 1, - 'date': '2022-01-01 12:00:00', - 'snr': 10, - 'pos_error_deg_95': 0.1, - 'coordinate_system': 'J2000', - 'ra': 180, - 'dec': 45, - 'website': 'https://observatory.com', - 'importance': 'high' - } - with patch('frbvoe.utilities.email.smtplib') as mock_smtplib: - # Raise an exception when sending email - mock_smtplib.SMTP().sendmail.side_effect = Exception('Failed to send email') - email.send_email(email_report) - # Assert that the email is sent using smtplib - mock_smtplib.SMTP.assert_called_once() - # Assert that the email content is formatted correctly - mock_smtplib.SMTP().sendmail.assert_called_once_with( - 'sender@example.com', - 'receiver@example.com', - 'Subject: FRB Report\n\n' - 'FRB Report:\n' - '\tbackend: backend_name\n' - '\n' - '\tevent parameters:\n' - '\t\tevent_no: 123\n' - '\n' - '\t\tknown_source_name: FRB123\n' - '\n' - '\t\tdm: 100 +/- 1 pc/cm^3\n' - '\n' - '\t\ttimestamp_utc: 2022-01-01 12:00:00\n' - '\n' - '\t\tsnr: 10\n' - '\n' - '\t\tpos_error_deg_95: 0.1 degrees\n' - '\n\n' - 'WHERE and WHEN\n' - '\tCoordinate system: J2000\n' - '\tTimestamp [UTC]: 2022-01-01 12:00:00\n' - '\tLocalization: (180, 45) +/- 0.1 degrees (J2000)\n' - '\n\n' - 'HOW\n' - '\tDescription: info about the observatory: https://observatory.com\n' - '\n\n' - 'WHY\n' - '\tImportance: high\n' - '\n\n' - ) - # Assert that the logger is called correctly - mock_logging.getLogger().error.assert_called_once_with('Failed to send email.') - -if __name__ == '__main__': - unittest.main() \ No newline at end of file +assert email.send_email(email_report) == "Success" \ No newline at end of file