Replace unittest.TestCase with custom base class. Remove unnecessary code. Adjust test case's code.

Author: dmytrotsko

integrations/acquisition/covidcast/test_covidcast_meta_caching.py

@@ -2,17 +2,13 @@
 
 # standard library
 import json
-import unittest
 
 # third party
-import mysql.connector
 import requests
 
 # first party
 from delphi_utils import Nans
-from delphi.epidata.client.delphi_epidata import Epidata
-import delphi.operations.secrets as secrets
-import delphi.epidata.acquisition.covidcast.database as live
+from delphi.epidata.common.covidcast_test_base import CovidcastTestBase
 from delphi.epidata.maintenance.covidcast_meta_cache_updater import main
 
 # py3tester coverage target (equivalent to `import *`)
@@ -23,86 +19,35 @@
 
 # use the local instance of the Epidata API
 BASE_URL = 'http://delphi_web_epidata/epidata'
+AUTH = ("epidata", "key")
 
 
-class CovidcastMetaCacheTests(unittest.TestCase):
+class CovidcastMetaCacheTests(CovidcastTestBase):
   """Tests covidcast metadata caching."""
 
-  def setUp(self):
-    """Perform per-test setup."""
-
-    # connect to the `epidata` database
-    cnx = mysql.connector.connect(
-        user='user',
-        password='pass',
-        host='delphi_database_epidata',
-        database='covid')
-    cur = cnx.cursor()
-
-    # clear all tables
-    cur.execute("truncate table epimetric_load")
-    cur.execute("truncate table epimetric_full")
-    cur.execute("truncate table epimetric_latest")
-    cur.execute("truncate table geo_dim")
-    cur.execute("truncate table signal_dim")
-    # reset the `covidcast_meta_cache` table (it should always have one row)
-    cur.execute('update covidcast_meta_cache set timestamp = 0, epidata = "[]"')
-    cnx.commit()
-    cur.close()
-
-    # make connection and cursor available to test cases
-    self.cnx = cnx
-    self.cur = cnx.cursor()
-
-    # use the local instance of the epidata database
-    secrets.db.host = 'delphi_database_epidata'
-    secrets.db.epi = ('user', 'pass')
-
-    epidata_cnx = mysql.connector.connect(
-        user='user',
-        password='pass',
-        host='delphi_database_epidata',
-        database='epidata')
-    epidata_cur = epidata_cnx.cursor()
-
-    epidata_cur.execute("DELETE FROM `api_user`")
-    epidata_cur.execute('INSERT INTO `api_user`(`api_key`, `email`) VALUES("key", "email")')
-    epidata_cnx.commit()
-    epidata_cur.close()
-    epidata_cnx.close()
-
-    # use the local instance of the Epidata API
-    Epidata.BASE_URL = BASE_URL
-    Epidata.auth = ('epidata', 'key')
-
-  def tearDown(self):
-    """Perform per-test teardown."""
-    self.cur.close()
-    self.cnx.close()
-
   @staticmethod
   def _make_request():
     params = {'cached': 'true'}
-    response = requests.get(f"{Epidata.BASE_URL}/covidcast_meta", params=params, auth=Epidata.auth)
+    response = requests.get(f"{BASE_URL}/covidcast_meta", params=params, auth=AUTH)
     response.raise_for_status()
     return response.json()
 
   def test_caching(self):
     """Populate, query, cache, query, and verify the cache."""
 
     # insert dummy data
-    self.cur.execute(f'''
+    self._db._cursor.execute(f'''
       INSERT INTO `signal_dim` (`signal_key_id`, `source`, `signal`)
       VALUES
         (42, 'src', 'sig');
     ''')
-    self.cur.execute(f'''
+    self._db._cursor.execute(f'''
       INSERT INTO `geo_dim` (`geo_key_id`, `geo_type`, `geo_value`)
       VALUES
         (96, 'state', 'pa'), 
         (97, 'state', 'wa');
     ''')
-    self.cur.execute(f'''
+    self._db._cursor.execute(f'''
       INSERT INTO
         `epimetric_latest` (`epimetric_id`, `signal_key_id`, `geo_key_id`, `time_type`,
 	      `time_value`, `value_updated_timestamp`,
@@ -115,13 +60,10 @@ def test_caching(self):
         (16, 42, 97, 'day', 20200422,
           789, 1, 2, 3, 20200423, 1, {Nans.NOT_MISSING}, {Nans.NOT_MISSING}, {Nans.NOT_MISSING})
     ''')
-    self.cnx.commit()
+    self._db._connection.commit()
 
     # make sure the live utility is serving something sensible
-    cvc_database = live.Database()
-    cvc_database.connect()
-    epidata1 = cvc_database.compute_covidcast_meta()
-    cvc_database.disconnect(False)
+    epidata1 = self._db.compute_covidcast_meta()
     self.assertEqual(len(epidata1),1)
     self.assertEqual(epidata1, [
       {
@@ -146,26 +88,26 @@ def test_caching(self):
 
     # make sure the API covidcast_meta is still blank, since it only serves
     # the cached version and we haven't cached anything yet
-    epidata2 = Epidata.covidcast_meta()
+    epidata2 = self.epidata_client.covidcast_meta()
     self.assertEqual(epidata2['result'], -2, json.dumps(epidata2))
 
     # update the cache
     args = None
     main(args)
 
     # fetch the cached version
-    epidata3 = Epidata.covidcast_meta()
+    epidata3 = self.epidata_client.covidcast_meta()
 
     # cached version should now equal live version
     self.assertEqual(epidata1, epidata3)
 
     # insert dummy data timestamped as of now
-    self.cur.execute('''
+    self._db._cursor.execute('''
       update covidcast_meta_cache set
         timestamp = UNIX_TIMESTAMP(NOW()),
         epidata = '[{"hello": "world"}]'
     ''')
-    self.cnx.commit()
+    self._db._connection.commit()
 
     # fetch the cached version (manually)
     epidata4 = self._make_request()
@@ -180,12 +122,12 @@ def test_caching(self):
     })
 
     # insert dummy data timestamped as 2 hours old
-    self.cur.execute('''
+    self._db._cursor.execute('''
       update covidcast_meta_cache set
         timestamp = UNIX_TIMESTAMP(NOW()) - 3600 * 2,
         epidata = '[{"hello": "world"}]'
     ''')
-    self.cnx.commit()
+    self._db._connection.commit()
 
     # fetch the cached version (manually)
     epidata5 = self._make_request()

integrations/acquisition/covidcast/test_csv_uploading.py

@@ -3,80 +3,23 @@
 # standard library
 from datetime import date
 import os
-import unittest
-import argparse
 
 # third party
-import mysql.connector
 import pandas as pd
 import numpy as np
 
 # first party
 from delphi_utils import Nans
-from delphi.epidata.client.delphi_epidata import Epidata
 from delphi.epidata.acquisition.covidcast.csv_to_database import main, get_argument_parser
-import delphi.operations.secrets as secrets
+from delphi.epidata.common.covidcast_test_base import CovidcastTestBase
 
 # py3tester coverage target (equivalent to `import *`)
 __test_target__ = 'delphi.epidata.acquisition.covidcast.csv_to_database'
 
 
-class CsvUploadingTests(unittest.TestCase):
+class CsvUploadingTests(CovidcastTestBase):
   """Tests covidcast CSV uploading."""
 
-  def setUp(self):
-    """Perform per-test setup."""
-
-    # connect to the `epidata` database and clear the `covidcast` table
-    cnx = mysql.connector.connect(
-        user='user',
-        password='pass',
-        host='delphi_database_epidata',
-        database='covid')
-    cur = cnx.cursor()
-
-    # clear all tables
-    cur.execute("truncate table epimetric_load")
-    cur.execute("truncate table epimetric_full")
-    cur.execute("truncate table epimetric_latest")
-    cur.execute("truncate table geo_dim")
-    cur.execute("truncate table signal_dim")
-    # reset the `covidcast_meta_cache` table (it should always have one row)
-    cur.execute('update covidcast_meta_cache set timestamp = 0, epidata = "[]"')
-
-    cnx.commit()
-    cur.close()
-
-    # make connection and cursor available to test cases
-    self.cnx = cnx
-    self.cur = cnx.cursor()
-
-    # use the local instance of the epidata database
-    secrets.db.host = 'delphi_database_epidata'
-    secrets.db.epi = ('user', 'pass')
-
-    epidata_cnx = mysql.connector.connect(
-        user='user',
-        password='pass',
-        host='delphi_database_epidata',
-        database='epidata')
-    epidata_cur = epidata_cnx.cursor()
-
-    epidata_cur.execute("DELETE FROM `api_user`")
-    epidata_cur.execute('INSERT INTO `api_user`(`api_key`, `email`) VALUES("key", "email")')
-    epidata_cnx.commit()
-    epidata_cur.close()
-    epidata_cnx.close()
-
-    # use the local instance of the Epidata API
-    Epidata.BASE_URL = 'http://delphi_web_epidata/epidata'
-    Epidata.auth = ('epidata', 'key')
-
-  def tearDown(self):
-    """Perform per-test teardown."""
-    self.cur.close()
-    self.cnx.close()
-
   @staticmethod
   def apply_lag(expected_epidata):
     expected_issue_day=date.today()
@@ -91,11 +34,11 @@ def apply_lag(expected_epidata):
     return expected_epidata
 
   def verify_timestamps_and_defaults(self):
-    self.cur.execute('''
+    self._db._cursor.execute('''
 select value_updated_timestamp from epimetric_full
 UNION ALL
 select value_updated_timestamp from epimetric_latest''')
-    for (value_updated_timestamp,) in self.cur:
+    for (value_updated_timestamp,) in self._db._cursor:
       self.assertGreater(value_updated_timestamp, 0)
 
   def test_uploading(self):
@@ -130,7 +73,7 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
+      response = self.epidata_client.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_values = pd.concat([values, pd.DataFrame({ "geo_type": "state", "source": "src-name", "time_type": "day", "time_value": [20200419] * 3, "signal": [signal_name] * 3, "direction": [None] * 3})], axis=1).rename(columns=uploader_column_rename).to_dict(orient="records")
       expected_response = {'result': 1, 'epidata': self.apply_lag(expected_values), 'message': 'success'}
@@ -158,7 +101,7 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
+      response = self.epidata_client.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_values = pd.concat([values, pd.DataFrame({
         "geo_type": "state",
@@ -195,7 +138,7 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
+      response = self.epidata_client.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_response = {'epidata': [], 'result': -2, 'message': 'no results'}
 
@@ -220,7 +163,7 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
+      response = self.epidata_client.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_values_df = pd.concat([values, pd.DataFrame({
         "geo_type": "state",
@@ -256,7 +199,7 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
+      response = self.epidata_client.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_values = pd.concat([values, pd.DataFrame({
         "geo_type": "state",
@@ -290,7 +233,7 @@ def test_uploading(self):
 
       # upload CSVs
       main(args)
-      response = Epidata.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
+      response = self.epidata_client.covidcast('src-name', signal_name, 'day', 'state', 20200419, '*')
 
       expected_response = {'epidata': [], 'result': -2, 'message': 'no results'}
 

integrations/acquisition/covidcast/test_delete_batch.py

@@ -6,39 +6,16 @@
 from os import path
 
 # first party
-import delphi.operations.secrets as secrets
-from delphi.epidata.acquisition.covidcast.database import Database
-from delphi.epidata.acquisition.covidcast.test_utils import covidcast_rows_from_args
+from delphi.epidata.common.covidcast_test_base import covidcast_rows_from_args, CovidcastTestBase
 
 # py3tester coverage target (equivalent to `import *`)
 __test_target__ = 'delphi.epidata.acquisition.covidcast.database'
 
 Example = namedtuple("example", "given expected")
 
-class DeleteBatch(unittest.TestCase):
+class DeleteBatch(CovidcastTestBase):
     """Tests batch deletions"""
 
-
-    def setUp(self):
-        """Perform per-test setup."""
-
-        # use the local instance of the epidata database
-        secrets.db.host = 'delphi_database_epidata'
-        secrets.db.epi = ('user', 'pass')
-
-        # will use secrets as set above
-        self._db = Database()
-        self._db.connect()
-
-        for table in "epimetric_load  epimetric_latest  epimetric_full  geo_dim  signal_dim".split():
-            self._db._cursor.execute(f"TRUNCATE TABLE {table}")
-
-
-    def tearDown(self):
-        """Perform per-test teardown."""
-        self._db.disconnect(False)
-        del self._db
-
     @unittest.skip("Database user would require FILE privileges")
     def test_delete_from_file(self):
         self._test_delete_batch(path.join(path.dirname(__file__), "delete_batch.csv"))