Integration tests refactoring

integrations/acquisition/covidcast/test_covidcast_meta_caching.py

@@ -2,17 +2,10 @@
 
 # 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 *`)
@@ -21,88 +14,26 @@
   'covidcast_meta_cache_updater'
 )
 
-# use the local instance of the Epidata API
-BASE_URL = 'http://delphi_web_epidata/epidata'
-
 
-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.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('''
       INSERT INTO `signal_dim` (`signal_key_id`, `source`, `signal`)
       VALUES
         (42, 'src', 'sig');
     ''')
-    self.cur.execute(f'''
+    self._db._cursor.execute('''
       INSERT INTO `geo_dim` (`geo_key_id`, `geo_type`, `geo_value`)
       VALUES
-        (96, 'state', 'pa'), 
+        (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,14 +46,11 @@ 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)
-    self.assertEqual(len(epidata1),1)
+    epidata1 = self._db.compute_covidcast_meta()
+    self.assertEqual(len(epidata1), 1)
     self.assertEqual(epidata1, [
       {
         'data_source': 'src',
@@ -142,33 +70,33 @@ def test_caching(self):
         'max_lag': 1,
       }
     ])
-    epidata1={'result':1, 'message':'success', 'epidata':epidata1}
+    epidata1 = {'result': 1, 'message': 'success', 'epidata': epidata1}
 
     # 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()
+    epidata4 = self._make_request(endpoint="covidcast_meta", json=True, params={'cached': 'true'}, auth=self.epidata_client.auth, raise_for_status=True)
 
     # make sure the cache was actually served
     self.assertEqual(epidata4, {
@@ -180,15 +108,15 @@ 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()
+    epidata5 = self._make_request(endpoint="covidcast_meta", json=True, params={'cached': 'true'}, auth=self.epidata_client.auth, raise_for_status=True)
 
     # make sure the cache was returned anyhow
     self.assertEqual(epidata4, epidata5)

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,16 +73,16 @@ 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_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'}
 
       self.assertEqual(response, expected_response)
       self.verify_timestamps_and_defaults()
 
       # Verify that files were archived
-      path = data_dir + f'/archive/successful/src-name/20200419_state_test.csv.gz'
+      path = data_dir + '/archive/successful/src-name/20200419_state_test.csv.gz'
       self.assertIsNotNone(os.stat(path))
 
       self.tearDown()
@@ -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_db.py

@@ -1,13 +1,13 @@
 from delphi_utils import Nans
 
 from delphi.epidata.acquisition.covidcast.database import DBLoadStateException
-from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase, CovidcastTestRow
+from delphi.epidata.common.covidcast_test_base import CovidcastTestBase, CovidcastTestRow
 
 
 # all the Nans we use here are just one value, so this is a shortcut to it:
 nmv = Nans.NOT_MISSING.value
 
-class TestTest(CovidcastBase):
+class TestDatabase(CovidcastTestBase):
 
     def _find_matches_for_row(self, row):
         # finds (if existing) row from both history and latest views that matches long-key of provided CovidcastTestRow