Server: update test_utils to use CovidcastRow

Author: dshemetov

src/acquisition/covidcast/test_utils.py

@@ -1,7 +1,9 @@
+from typing import Sequence
 import unittest
 
 from delphi_utils import Nans
-from delphi.epidata.acquisition.covidcast.database import Database, CovidcastRow
+from delphi.epidata.acquisition.covidcast.covidcast_row import CovidcastRow
+from delphi.epidata.acquisition.covidcast.database import Database
 import delphi.operations.secrets as secrets
 
 # all the Nans we use here are just one value, so this is a shortcut to it:
@@ -31,36 +33,20 @@ def tearDown(self):
         # close and destroy conenction to the database
         self._db.disconnect(False)
         del self._db
+        self.localTearDown()
 
-    DEFAULT_TIME_VALUE=2000_01_01
-    DEFAULT_ISSUE=2000_01_01
-    def _make_placeholder_row(self, **kwargs):
-        settings = {
-            'source': 'src',
-            'signal': 'sig',
-            'geo_type': 'state',
-            'geo_value': 'pa',
-            'time_type': 'day',
-            'time_value': self.DEFAULT_TIME_VALUE,
-            'value': 0.0,
-            'stderr': 1.0,
-            'sample_size': 2.0,
-            'missing_value': nmv,
-            'missing_stderr': nmv,
-            'missing_sample_size': nmv,
-            'issue': self.DEFAULT_ISSUE,
-            'lag': 0
-        }
-        settings.update(kwargs)
-        return (CovidcastRow(**settings), settings)
+    def localTearDown(self):
+        # stub; override in subclasses to perform custom teardown. 
+        # runs after database changes have been committed
+        pass
 
-    def _insert_rows(self, rows):
+    def _insert_rows(self, rows: Sequence[CovidcastRow]):
         # inserts rows into the database using the full acquisition process, including 'dbjobs' load into history & latest tables
         n = self._db.insert_or_update_bulk(rows)
         print(f"{n} rows added to load table & dispatched to v4 schema")
         self._db._connection.commit() # NOTE: this isnt expressly needed for our test cases, but would be if using external access (like through client lib) to ensure changes are visible outside of this db session
 
-    def params_from_row(self, row, **kwargs):
+    def params_from_row(self, row: CovidcastRow, **kwargs):
         ret = {
             'data_source': row.source,
             'signals': row.signal,
@@ -71,13 +57,3 @@ def params_from_row(self, row, **kwargs):
         }
         ret.update(kwargs)
         return ret
-
-    DEFAULT_MINUS=['time_type', 'geo_type', 'source']
-    def expected_from_row(self, row, minus=DEFAULT_MINUS):
-        expected = dict(vars(row))
-        # remove columns commonly excluded from output
-        # nb may need to add source or *_type back in for multiplexed queries
-        for key in ['id', 'direction_updated_timestamp'] + minus:
-            del expected[key]
-        return expected
-