adding a test event model to track data loaded from ivy, and hopefully cross-referenced with data loaded from Firebase.

communicator/errors.py

@@ -0,0 +1,32 @@
+class ApiError(Exception):
+    """
+    Follows the RFC 7807 standard https://tools.ietf.org/html/rfc7807
+    Example Usage:
+        {
+        "type": "https://example.com/probs/out-of-credit",
+        "title": "You do not have enough credit.",
+        "detail": "Your current balance is 30, but that costs 50.",
+        "instance": "/account/12345/msgs/abc",
+        "balance": 30,
+        "accounts": ["/account/12345",
+                     "/account/67890"]
+        }
+    """
+
+    def __init__(self, type, title, status, detail, instance):
+        self.type = type
+        self.title = title
+        self.status = status
+        self.detail = detail
+        self.instance = instance
+
+
+class CommError(Exception):
+    """A standard error from which to extend, that can be easily converted to
+     an API error if needed. The code should be a unique numeric value, and
+     the detail is optional."""
+
+    def __init__(self, code: int, title, detail=""):
+        self.code = code
+        self.title = title
+        self.detail = detail

communicator/models/test_event.py

@@ -0,0 +1,26 @@
+from communicator import db
+
+
+class TestEvent(db.Model):
+    student_id = db.Column(db.Integer, primary_key=True)
+    date = db.Column(db.DateTime, primary_key=True)
+    location = db.Column(db.Integer)
+    phone = db.Column(db.String)
+    email = db.Column(db.String)
+    result_code = db.Column(db.String)
+    notified = db.Column(db.Boolean, default=False)
+    firebase_record = db.Column(db.Boolean, default=False)  # Does this record exist in Firebase?
+    ivy_record = db.Column(db.Boolean, default=False)  # Has this record come in from the IVY?
+
+    @classmethod
+    def from_ivy_dict(cls, dictionary):
+        """Creates a Test Result from a record read in from the IVY CSV File"""
+        instance = cls()
+        instance.student_id = dictionary["Student ID"]
+        instance.phone = dictionary["Student Cellphone"]
+        instance.email = dictionary["Student Email"]
+        instance.date = dictionary["Test Date Time"]
+        instance.location = dictionary["Test Kiosk Loc"]
+        instance.result_code = dictionary["Test Result Code"]
+        instance.from_ivy_dict = True
+        return instance

communicator/services/ivy_service.py

@@ -0,0 +1,23 @@
+import csv
+
+from communicator.errors import CommError
+from communicator.models.test_event import TestEvent
+
+
+class IvyService(object):
+
+    @staticmethod
+    def import_ivy_file(file_name):
+        rows = []
+        with open(file_name, 'r') as csv_file:
+            reader = csv.DictReader(csv_file, delimiter='|')
+            for row in reader:
+                rows.append(row)
+        return rows
+
+    @staticmethod
+    def to_test_event_record(data):
+        try:
+            return TestEvent.from_ivy_dict(data)
+        except KeyError as e:
+            raise CommError("100", f"Invalid CSV Record, missing column {e}");

tests/data/incorrect.csv

@@ -0,0 +1,3 @@
+987654321|555/555-5555|rkc7h@virginia.edu|202009030809|4321|8726520277
+987654322|555/555-5556|testpositive@virginia.edu|202009060919|4321|8269722523
+987655321|555/555-5558|testnegetive@virginia.edu|202009070719|4321|1142270225

tests/data/results.csv

@@ -0,0 +1,4 @@
+Student ID|Student Cellphone|Student Email|Test Date Time|Test Kiosk Loc|Test Result Code
+987654321|555/555-5555|rkc7h@virginia.edu|202009030809|4321|8726520277
+987654322|555/555-5556|testpositive@virginia.edu|202009060919|4321|8269722523
+987655321|555/555-5558|testnegetive@virginia.edu|202009070719|4321|1142270225

tests/services/ivy_service_test.py

@@ -0,0 +1,29 @@
+import unittest
+
+from communicator.errors import CommError
+from communicator.services.ivy_service import IvyService
+
+
+class IvyServiceTest(unittest.TestCase):
+
+    def test_read_file_and_build_records(self):
+        data = IvyService.import_ivy_file('../data/results.csv')
+        self.assertEquals(3, len(data))
+        # Quick spot check on values
+        self.assertEquals("987654321", data[0]["Student ID"])
+        self.assertEquals("testpositive@virginia.edu", data[1]["Student Email"])
+        self.assertEquals("1142270225", data[2]["Test Result Code"])
+        records = []
+        for d in data:
+            records.append(IvyService.to_test_event_record(d))
+        self.assertEquals("987654321", records[0].student_id)
+        self.assertEquals("testpositive@virginia.edu", records[1].email)
+        self.assertEquals("1142270225", records[2].result_code)
+
+    def test_invalid_file(self):
+        with self.assertRaises(CommError):
+            data = IvyService.import_ivy_file('../data/incorrect.csv')
+            records = []
+            for d in data:
+                records.append(IvyService.to_test_event_record(d))
+