💚 Improve coverage

Author: akrherz

src/pywwa/common.py

@@ -205,11 +205,14 @@ def email_error(exp, message="", trimstr=100):
     msg["subject"] = (
         f"[pyWWA] {sys.argv[0].rsplit('/', maxsplit=1)[-1]} Traceback -- {hn}"
     )
-    msg["From"] = SETTINGS.get("pywwa_errors_from", "ldm@localhost")
-    msg["To"] = SETTINGS.get("pywwa_errors_to", "ldm@localhost")
+    msg["From"] = SETTINGS.get("pywwa_errors_from", "root@localhost")
+    msg["To"] = SETTINGS.get("pywwa_errors_to", "root@localhost")
     if not CTX["disable_email"]:
         df = smtp.sendmail(
-            SETTINGS.get("pywwa_smtp", "smtp"), msg["From"], msg["To"], msg
+            SETTINGS.get("pywwa_smtp", "localhost"),
+            msg["From"],
+            msg["To"],
+            msg,
         )
         df.addErrback(LOG.error)
     else:

src/pywwa/workflows/nexrad3_attr.py

@@ -33,15 +33,15 @@ def load_station_table(txn):
     LOG.info("Station Table size %s", len(ST.keys()))
 
 
-def process_data(data):
+def process_data(data: bytes) -> dict:
     """I am called when data is ahoy"""
     bio = BytesIO()
     bio.write(data)
     bio.seek(0)
-    process(bio)
+    return process(bio)
 
 
-def process(bio):
+def process(bio: BytesIO) -> dict:
     """Process our data, please"""
     l3 = Level3File(bio)
     ctx = {
@@ -57,7 +57,7 @@ def process(bio):
             if "text" in line:
                 ctx["lines"].append(line["text"])
     df = PGCONN.runInteraction(really_process, ctx)
-    df.addErrback(common.email_error, ctx)
+    df.addErrback(common.email_error, str(ctx))
     return ctx
 
 

src/pywwa/workflows/sps.py

@@ -2,13 +2,11 @@
 
 from functools import partial
 
-# 3rd Party
 import click
 from pyiem.database import get_sqlalchemy_conn
 from pyiem.nws.products.sps import parser
 from pyiem.nws.ugc import UGCProvider
 
-# Local
 from pywwa import LOG, common
 from pywwa.database import get_database, load_nwsli
 from pywwa.ldm import bridge

src/pywwa/workflows/vtec.py

@@ -13,14 +13,12 @@
 
 from functools import partial
 
-# 3rd Party
 import click
 from pyiem.database import get_sqlalchemy_conn
 from pyiem.nws.products.vtec import parser as vtecparser
 from pyiem.nws.ugc import UGCProvider
 from twisted.mail.smtp import SMTPSenderFactory
 
-# Local
 from pywwa import LOG, common
 from pywwa.database import get_database, load_nwsli
 from pywwa.ldm import bridge

src/pywwa/workflows/xteus.py

@@ -1,16 +1,14 @@
 """XTEUS Product Parser!"""
 
-# 3rd Party
 import click
 from pyiem.nws.products.xteus import parser
 
-# Local
 from pywwa import common
 from pywwa.database import get_database
 from pywwa.ldm import bridge
 
 
-def process_data(txn, data):
+def process_data(txn, data: str):
     """Process the product"""
     prod = parser(data, utcnow=common.utcnow())
     if prod.warnings:

tests/test_common.py

@@ -1,5 +1,6 @@
 """Test pywwa.common"""
 
+import pytest_twisted
 from pyiem.util import utc
 
 from pywwa import CTX, common
@@ -31,9 +32,10 @@ def test_should_email():
     assert not common.email_error(None, None)
 
 
+@pytest_twisted.inlineCallbacks
 def test_email_error():
     """Test that we can email an error."""
     common.EMAIL_TIMESTAMPS = []
-    common.email_error(None, None)
+    yield common.email_error(None, None)
     CTX["disable_email"] = True
-    common.email_error(None, None)
+    yield common.email_error(None, None)

tests/workflows/test_dsm_parser.py

@@ -15,8 +15,8 @@ def test_load_stations(cursor):
     """Test station loading."""
     # Need to set one station to a bad tzname to exercise an exception
     cursor.execute(
-        "UPDATE stations SET tzname = 'BoG0S' where id = 'MSP' and "
-        "network = 'MN_ASOS'"
+        "UPDATE stations SET tzname = 'BoG0S' where id = 'DSM' and "
+        "network = 'IA_ASOS'"
     )
     assert cursor.rowcount == 1
     dsm.load_stations(cursor)

tests/workflows/test_nexrad3_attr.py

@@ -1,11 +1,8 @@
 """Test nexrad3_attr."""
 
-# 3rd Party
 import pytest
 
 from pywwa.testing import get_example_filepath
-
-# Local
 from pywwa.workflows import nexrad3_attr
 
 
@@ -29,7 +26,7 @@ def test_process(cursor):
 def test_210910_badvil(cursor):
     """Test that a missing VIL does not cause issues."""
     with open(get_example_filepath("NCR_20210911_0023"), "rb") as fh:
-        ctx = nexrad3_attr.process(fh)
+        ctx = nexrad3_attr.process_data(fh.read())
     assert ctx["nexrad"] == "LAS"
     processed = nexrad3_attr.really_process(cursor, ctx)
     assert processed == 2

tests/workflows/test_sps.py

@@ -0,0 +1,13 @@
+"""Exercise the sps workflow."""
+
+import pytest
+
+from pywwa.testing import get_example_file
+from pywwa.workflows import sps
+
+
+@pytest.mark.parametrize("database", ["postgis"])
+def test_real_process(cursor):
+    """Can we process a real SPS product?"""
+    data = get_example_file("SPS.txt")
+    sps.real_process({}, cursor, data)