fix(tests): added some tests

frbvoe/backend/voe.py

@@ -1,30 +1,52 @@
 """VOEvent Server Blueprint."""
 
+from pymongo.errors import PyMongoError
 from sanic import Blueprint
 from sanic.log import logger
 from sanic.request import Request
-from sanic.response import HTTPResponse
 from sanic.response import json as json_response
 from sanic_ext import openapi
-from pymongo.errors import PyMongoError
 
+from frbvoe.models.comet import Comet
+from frbvoe.models.email import Email
 from frbvoe.models.voe import VOEvent
-from frbvoe.utilities.senders import send_to_comet
 
 voe = Blueprint("voe", url_prefix="/")
 
 
-# Post at /voe
-@voe.post("voe")
-@openapi.response(201, description="Validates data from the telescope.")
+# Post at /create_voe
+@voe.post("create_voe")
+@openapi.response(201, description="Validates VOEvent data from a host observatory.")
 # Add the validated payload to the MongoDB Database
 async def create_voe(request: Request, voe_event: VOEvent):
+    """Process a VOEvent.
 
-    mongo = request.app.ctx.mongo
-    request.app.add_task(send_to_comet(voe_event.model_dump()))
+    Args:
+        request (Request): The request object.
+        voe_event (VOEvent): The VOEvent to be processed.
+
+    Returns:
+        JSON response: A JSON response containing the inserted ID if successful,
+        or an error message if there was an issue.
+
+    Raises:
+        Exception: If there is an error inserting the VOEvent into MongoDB.
+        PyMongoError: If there is an error with the PyMongo library.
+
+    """
+    # Send VOEvent to Comet
+    comet_report = Comet(voe_event.model_dump())
+    comet_report.send
+
+    # Send VOEvent to Email
+    email_report = Email(voe_event.model_dump())
+    email_report.send
+
+    # Add VOEvent to MongoDB
     # Database Name: frbvoe
     # Collection Name: voe
     # Document -> VOEvent Payload Dict.
+    mongo = request.app.ctx.mongo
     try:
         insert_result = await mongo["frbvoe"]["voe"].insert_one(voe_event.model_dump())
     except (Exception, PyMongoError) as mongo_error:
@@ -35,12 +57,20 @@ async def create_voe(request: Request, voe_event: VOEvent):
 
 
 async def delete_voe(request: Request):
+    """Delete a VOE document from the database.
+
+    Args:
+        request (Request): The HTTP request object.
+
+    Returns:
+        Response: The HTTP response object.
 
-    # /voe?id=<id>
+    Raises:
+        HTTPException: If the request is missing the 'id' parameter.
+        HTTPException: If there is an error deleting the document from the database.
+    """
     if "id" not in request.args.keys():
-        return json_response(
-            {"message": "Your request needs an ID."}, status=400
-        )
+        return json_response({"message": "Your request needs an ID."}, status=400)
     id = request.args["id"]
     mongo = request.app.ctx.mongo
     try:

frbvoe/models/comet.py

@@ -3,10 +3,10 @@
 from typing import Any, Dict
 
 import picologging as logging
+import requests
 from pydantic import Field
 
 from frbvoe.models.voe import VOEvent
-from frbvoe.utilities.comet import send
 
 logging.basicConfig()
 log = logging.getLogger()
@@ -24,12 +24,16 @@ class Comet(VOEvent):
     )
 
     @property
-    def send_xml(comet_report: Dict[str, Any]):
+    def send(comet_report: Dict[str, Any]):
         """Sends a report using the given VOEvent and comet URL.
 
         Args:
             voevent (Dict[str, Any]): The VOEvent to send.
             comet_url (str): The URL of the comet to send the report to.
         """
         log.info("Sending VOE payload to Comet as a report.")
-        send(comet_report)
+        # vp.dump(voevent=comet_report, xml_declaration=False, file="temp_voe.txt")
+        response = requests.post(
+            "http://comet:8098/", json=comet_report
+        )  # TODO: check comet endpoint
+        return response.status_code == 200

frbvoe/models/email.py

@@ -6,7 +6,7 @@
 from pydantic import Field, SecretStr
 
 from frbvoe.models.voe import VOEvent
-from frbvoe.utilities.email import send
+from frbvoe.utilities.email import send_email
 
 logging.basicConfig()
 log = logging.getLogger()
@@ -24,7 +24,7 @@ class Email(VOEvent):
     )
 
     @property
-    def send_email(email_report: Dict[str, Any]):
+    def send(email_report: Dict[str, Any]):
         """Sends the VOEvent email.
 
         Args:
@@ -34,4 +34,4 @@ def send_email(email_report: Dict[str, Any]):
             status (str): The status of the email.
         """
         log.info("Emailing VOE payload to subscribers.")
-        send(email_report)
+        send_email(email_report)

frbvoe/server.py

@@ -9,8 +9,8 @@
 from sanic.log import logger
 from sanic.worker.loader import AppLoader
 
-from frbvoe.models.voe import VOEvent
 from frbvoe.backend.voe import voe as voe_blueprint
+from frbvoe.models.voe import VOEvent
 
 
 async def mongo(app: Sanic, loop: AbstractEventLoop) -> None:
@@ -72,7 +72,7 @@ def create(name: str = "frbvoe", debug: bool = False) -> Sanic:
     app.config.HEALTH_ENDPOINT = True
     app.ctx.debug = debug
     app.config.FALL
-    BACK_ERROR_FORMAT = "json"
+    # BACK_ERROR_FORMAT = "json"
     # ? Blueprints
     app.blueprint(voe_blueprint)
     # ? Listeners

frbvoe/utilities/email.py

@@ -11,7 +11,7 @@
 log = logging.getLogger()
 
 
-def send(email_report: Dict[str, Any]):
+def send_email(email_report: Dict[str, Any]):
     """Sends an email with the provided email_report.
 
     Args:

tests/test_models/test_comet.py

@@ -1,42 +1,39 @@
+"""Test for the TNS model."""
+
 import pytest
 from pydantic import ValidationError
 
 from frbvoe.models.comet import Comet
 
-
-def test_report(voe):
-    voe.payload["bandwidth"]
+sample_request = {
+    "kind": "detection",
+    "date": "2025-01-13 16:55:08.844845",
+    "email": "john.smith@email.com",
+    "semi_major": 0.026,
+    "semi_minor": 0.013,
+    "sampling_time": 0.001,
+    "bandwidth": 400,
+    "central_frequency": 600,
+    "npol": 2,
+    "bits_per_sample": 2,
+    "gain": 1.76,
+    "tsys": 25.0,
+    "internal_id": "20210826A",
+    "dm": 298.53,
+    "dm_error": 0.01,
+    "width": 4.8,
+    "snr": 13.8,
+    "flux": 4.9,
+    "right_ascension": 55.2938,
+    "declination": 14.2049,
+    "pos_error_deg_95": 0.1,
+    "importance": 0.9979,
+    "website": "https://www.example.com",
+    "tns_name": "FRB20210826A",
+    "update_message": "",
+}
 
 
-def test_bad_voe():
+def test_comet():
     with pytest.raises(ValidationError):
-        Comet(
-            kind="detection",
-            author="John Smith",
-            email="john.smith@email.com",
-            coordinate_system="celestial",
-            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="2020-01-13 16:55:08.844845",
-            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,
-        )
+        Comet(**sample_request)

tests/test_models/test_email.py

@@ -1 +1,39 @@
 """Tests for the Email model."""
+
+import pytest
+from pydantic import ValidationError
+
+from frbvoe.models.email import Email
+
+sample_request = {
+    "kind": "detection",
+    "date": "2025-01-13 16:55:08.844845",
+    "email": "john.smith@email.com",
+    "semi_major": 0.026,
+    "semi_minor": 0.013,
+    "sampling_time": 0.001,
+    "bandwidth": 400,
+    "central_frequency": 600,
+    "npol": 2,
+    "bits_per_sample": 2,
+    "gain": 1.76,
+    "tsys": 25.0,
+    "internal_id": "20210826A",
+    "dm": 298.53,
+    "dm_error": 0.01,
+    "width": 4.8,
+    "snr": 13.8,
+    "flux": 4.9,
+    "right_ascension": 55.2938,
+    "declination": 14.2049,
+    "pos_error_deg_95": 0.1,
+    "importance": 0.9979,
+    "website": "https://www.example.com",
+    "tns_name": "FRB20210826A",
+    "update_message": "",
+}
+
+
+def test_email():
+    with pytest.raises(ValidationError):
+        Email(**sample_request)

tests/test_models/test_tns.py

@@ -1 +1,39 @@
 """Test for the TNS model."""
+
+import pytest
+from pydantic import ValidationError
+
+from frbvoe.models.tns import TNS
+
+sample_request = {
+    "kind": "detection",
+    "date": "2025-01-13 16:55:08.844845",
+    "email": "john.smith@email.com",
+    "semi_major": 0.026,
+    "semi_minor": 0.013,
+    "sampling_time": 0.001,
+    "bandwidth": 400,
+    "central_frequency": 600,
+    "npol": 2,
+    "bits_per_sample": 2,
+    "gain": 1.76,
+    "tsys": 25.0,
+    "internal_id": "20210826A",
+    "dm": 298.53,
+    "dm_error": 0.01,
+    "width": 4.8,
+    "snr": 13.8,
+    "flux": 4.9,
+    "right_ascension": 55.2938,
+    "declination": 14.2049,
+    "pos_error_deg_95": 0.1,
+    "importance": 0.9979,
+    "website": "https://www.example.com",
+    "tns_name": "FRB20210826A",
+    "update_message": "",
+}
+
+
+def test_tns():
+    with pytest.raises(ValidationError):
+        TNS(**sample_request)

tests/test_models/test_voe.py

@@ -1 +1,43 @@
 """Test for VOE model."""
+
+# from typing import Dict
+
+import pytest
+from pydantic import ValidationError
+
+from frbvoe.models.voe import VOEvent
+
+sample_request = {
+    "kind": "detection",
+    "date": "2025-01-13 16:55:08.844845",
+    "email": "john.smith@email.com",
+    "semi_major": 0.026,
+    "semi_minor": 0.013,
+    "sampling_time": 0.001,
+    "bandwidth": 400,
+    "central_frequency": 600,
+    "npol": 2,
+    "bits_per_sample": 2,
+    "gain": 1.76,
+    "tsys": 25.0,
+    "internal_id": "20210826A",
+    "dm": 298.53,
+    "dm_error": 0.01,
+    "width": 4.8,
+    "snr": 13.8,
+    "flux": 4.9,
+    "right_ascension": 55.2938,
+    "declination": 14.2049,
+    "pos_error_deg_95": 0.1,
+    "importance": 0.9979,
+    "website": "https://www.example.com",
+    "tns_name": "FRB20210826A",
+    "update_message": "",
+}
+
+
+def test_voe():
+    with pytest.raises(ValidationError):
+        VOEvent(**sample_request)
+    # with pytest.raises(TypeError):
+    #     VOEvent(**sample_request).payload is Dict