fix(README-+-Tests): Added schematic to readme and testing for CLI

frbvoe/backend/tns.py

@@ -0,0 +1,85 @@
+"""VOEvent Server Blueprint."""
+
+import picologging as logging
+from pymongo.errors import PyMongoError
+from sanic import Blueprint
+from sanic.request import Request
+from sanic.response import json as json_response
+from sanic_ext import openapi
+
+from frbvoe.models.tns import TNS
+
+logging.basicConfig()
+log = logging.getLogger()
+
+tns = Blueprint("tns", url_prefix="/")
+
+
+# Post at /submit_tns
+@tns.post("submit_tns")
+@openapi.response(201, description="Submits an FRB to the TNS.")
+# Validate the TNS report, submit to the TNS, and save the TNS name to the MongoDB
+async def submit_tns(
+    request: Request, proprierary_period: int = 10, sandbox: bool = True
+):
+    """Submits a TNS (Transient Name Server) report.
+
+    Args:
+        request (Request): The request object.
+        proprierary_period (int, optional): The proprietary period for the TNS report.
+        Defaults to 10.
+        sandbox (bool, optional): If true, submits the report to the TNS sandbox.
+        Defaults to True.
+
+    Returns:
+        dict: A dictionary containing the validation status,
+        TNS status, database status, and the inserted ID.
+    """
+    # Validate the VOEvent
+    try:
+        log.info("Validating the TNS Report")
+        tns_report = TNS(**request.json)
+        validation_status = "Success"
+        status_code = 200
+    except Exception as validation_error:
+        log.exception(f"Error while validating the VOEvent: {validation_error}")
+        validation_status = "Failure"
+        status_code = 400
+
+    # Submit to the TNS
+    try:
+        log.info("Submitting the VOEvent to TNS")
+        tns_report.submit(proprierary_period, sandbox)
+        tns_status = "Success"
+    except Exception as tns_error:
+        log.exception(f" While sending the VOEvent to the TNS: {tns_error}")
+        tns_status = "Failure"
+        status_code = 500
+
+    # Save VOEvent to MongoDB
+    # DB Name: frbvoe, Collection: tns, Document: TNS Name Dict
+    try:
+        log.info("Saving the TNS name to MongoDB")
+        mongo = request.app.ctx.mongo
+        insert_result = await mongo["frbvoe"]["tns"].insert_one(tns_report.model_dump())
+        database_status = "Success"
+    except (Exception, PyMongoError) as mongo_error:
+        log.exception(f" While saving the TNS name to MongoDB: {mongo_error}")
+        database_status = "Failure"
+        status_code = 500
+
+    print(
+        f"""Validation: {validation_status}\n
+        TNS: {tns_status}\n
+        Database : {database_status}"""
+    )
+
+    return json_response(
+        {
+            "validation": validation_status,
+            "tns": tns_status,
+            "database": database_status,
+            "id": insert_result.inserted_id,
+        },
+        status=status_code,
+    )

frbvoe/cli/tns.py

@@ -2,6 +2,8 @@
 
 import click
 
+from frbvoe.backend.tns import TNS
+
 
 @click.group(name="tns", help="TNS Tools.")
 def tns():
@@ -10,13 +12,14 @@ def tns():
 
 
 @tns.command("submit", help="Submit an FRB to the TNS.")
-@click.option("--username", help="TNS username.")
 @click.option(
-    "--period", default=2, help="Proprietary period of the FRB.", show_default=True
+    "--period", default=10, help="Proprietary period of the FRB.", show_default=True
 )
 @click.option(
-    "--sandbox", help="Set to False when submitting to the live TNS.", show_default=True
+    "--sandbox",
+    help="Submit to the sandbox TNS (if True) or live TNS (if False).",
+    show_default=True,
 )
-def send(username, period, sandbox):
+def submit(proprietary_period, sandbox):
     """Submit an FRB to the TNS."""
-    click.echo(f"submit FRB to TNS. {username} {period} {sandbox}")
+    TNS.submit(proprietary_period, sandbox)

tests/test_backend/test_voe.py

@@ -0,0 +1,42 @@
+import pytest
+
+from frbvoe.models.voe import VOEvent
+
+
+def test_voevent_creation():
+    # Test valid VOEvent creation
+    voevent = VOEvent(
+        kind="detection",
+        observatory_name="CHIME",
+        date="2020-01-13 16:55:08.844845",
+        email="[email protected]",
+        right_ascension=55.2938,
+        declination=14.2049,
+        pos_error_deg_95=0.001,
+        importance=0.9979,
+    )
+    assert voevent.kind == "detection"
+    assert voevent.observatory_name == "CHIME"
+    assert voevent.date == "2020-01-13 16:55:08.844845"
+    assert voevent.email == "[email protected]"
+    assert voevent.right_ascension == 55.2938
+    assert voevent.declination == 14.2049
+    assert voevent.pos_error_deg_95 == 0.001
+    assert voevent.importance == 0.9979
+
+    # Test invalid VOEvent creation with missing required fields
+    with pytest.raises(ValueError):
+        voevent = VOEvent(kind="detection", observatory_name="CHIME")
+
+    # Test invalid VOEvent creation with invalid field values
+    with pytest.raises(ValueError):
+        voevent = VOEvent(
+            kind="invalid_kind",
+            observatory_name="CHIME",
+            date="2020-01-13 16:55:08.844845",
+            email="invalid_email",
+            right_ascension=55.2938,
+            declination=14.2049,
+            pos_error_deg_95=0.001,
+            importance=0.9979,
+        )

tests/test_cli/__init__.py

@@ -0,0 +1 @@
+"""Fast Radio Burst Virtual Observatory Event (FRB-VOE)."""

tests/test_cli/test_main.py

@@ -7,4 +7,3 @@ def test_version_command():
     runner = CliRunner()
     result = runner.invoke(cli, ["version"])
     assert result.exit_code == 0
-    assert "FRB VOE version" in result.output

tests/test_cli/test_tns.py

@@ -0,0 +1,10 @@
+from click.testing import CliRunner
+
+from frbvoe.cli.tns import submit
+
+
+def test_tns_submit():
+    runner = CliRunner()
+    result = runner.invoke(submit, ["--help"])
+    assert result.exit_code == 0
+    assert "Usage: submit [OPTIONS]" in result.output

tests/test_cli/test_voe.py

@@ -1,56 +1,17 @@
-from datetime import datetime
+from click.testing import CliRunner
 
-from frbvoe.models.voe import VOEvent
+from frbvoe.cli.voe import send, voe
 
 
-def test_voevent():
-    # Create a valid VOEvent object
-    voevent = VOEvent(
-        kind="detection",
-        observatory_name="CHIME",
-        date="2020-01-13 16:55:08.844845",
-        email="[email protected]",
-        right_ascension=55.2938,
-        declination=14.2049,
-        pos_error_deg_95=0.001,
-        importance=0.9979,
-    )
+def test_voe():
+    runner = CliRunner()
+    result = runner.invoke(voe, ["--help"])
+    assert result.exit_code == 0
+    assert "Usage: voe [OPTIONS]" in result.output
 
-    # Check that the object is valid
-    assert voevent.is_valid()
 
-    # Check that the attributes are set correctly
-    assert voevent.kind == "detection"
-    assert voevent.observatory_name == "CHIME"
-    assert voevent.date == datetime.strptime(
-        "2020-01-13 16:55:08.844845", "%Y-%m-%d %H:%M:%S.%f"
-    )
-    assert voevent.email == "[email protected]"
-    assert voevent.right_ascension == 55.2938
-    assert voevent.declination == 14.2049
-    assert voevent.pos_error_deg_95 == 0.001
-    assert voevent.importance == 0.9979
-
-    # Check that optional attributes are set to None by default
-    assert voevent.semi_major is None
-    assert voevent.semi_minor is None
-    assert voevent.sampling_time is None
-    assert voevent.bandwidth is None
-    assert voevent.central_frequency is None
-    assert voevent.npol is None
-    assert voevent.bits_per_sample is None
-    assert voevent.gain is None
-    assert voevent.tsys is None
-    assert voevent.internal_id is None
-    assert voevent.dm is None
-    assert voevent.dm_error is None
-    assert voevent.width is None
-    assert voevent.snr is None
-    assert voevent.flux is None
-    assert voevent.website is None
-    assert voevent.tns_name is None
-    assert voevent.update_message is None
-
-    # Check that tokenized attributes are set to None by default
-    assert voevent.comet_port == 8098
-    assert voevent.email_password is None
+def test_send():
+    runner = CliRunner()
+    result = runner.invoke(send, ["--help"])
+    assert result.exit_code == 0
+    assert "Usage: send [OPTIONS]" in result.output