diff --git a/integrations/acquisition/covidcast/delete_batch.csv b/integrations/acquisition/covidcast/delete_batch.csv
index e0e1eb82c..5c1602218 100644
--- a/integrations/acquisition/covidcast/delete_batch.csv
+++ b/integrations/acquisition/covidcast/delete_batch.csv
@@ -1,4 +1,4 @@
 geo_id,value,stderr,sample_size,issue,time_value,geo_type,signal,source
-d_nonlatest,0,0,0,1,0,geo,sig,src
-d_latest,   0,0,0,3,0,geo,sig,src
-d_justone,  0,0,0,1,0,geo,sig,src
\ No newline at end of file
+d_nonlatest,0,0,0,1,0,county,sig,src
+d_latest,   0,0,0,3,0,county,sig,src
+d_justone,  0,0,0,1,0,county,sig,src
\ No newline at end of file
diff --git a/integrations/acquisition/covidcast/test_csv_uploading.py b/integrations/acquisition/covidcast/test_csv_uploading.py
index de3eb5f13..f975ecfa0 100644
--- a/integrations/acquisition/covidcast/test_csv_uploading.py
+++ b/integrations/acquisition/covidcast/test_csv_uploading.py
@@ -213,8 +213,8 @@ def test_uploading(self):
         "time_value": [20200419],
         "signal": [signal_name],
         "direction": [None]})], axis=1).rename(columns=uploader_column_rename)
-      expected_values_df["missing_value"].iloc[0] = Nans.OTHER
-      expected_values_df["missing_sample_size"].iloc[0] = Nans.NOT_MISSING
+      expected_values_df.loc[0, "missing_value"] = Nans.OTHER
+      expected_values_df.loc[0, "missing_sample_size"] = Nans.NOT_MISSING
       expected_values = expected_values_df.to_dict(orient="records")
       expected_response = {'result': 1, 'epidata': self.apply_lag(expected_values), 'message': 'success'}
 
diff --git a/integrations/acquisition/covidcast/test_db.py b/integrations/acquisition/covidcast/test_db.py
index 3cd7e91a7..7b9d80770 100644
--- a/integrations/acquisition/covidcast/test_db.py
+++ b/integrations/acquisition/covidcast/test_db.py
@@ -1,9 +1,8 @@
-import unittest
-
 from delphi_utils import Nans
-from delphi.epidata.acquisition.covidcast.database import Database, CovidcastRow, DBLoadStateException
-from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase
-import delphi.operations.secrets as secrets
+
+from delphi.epidata.acquisition.covidcast.database import DBLoadStateException
+from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase, CovidcastTestRow
+
 
 # all the Nans we use here are just one value, so this is a shortcut to it:
 nmv = Nans.NOT_MISSING.value
@@ -11,7 +10,7 @@
 class TestTest(CovidcastBase):
 
     def _find_matches_for_row(self, row):
-        # finds (if existing) row from both history and latest views that matches long-key of provided CovidcastRow
+        # finds (if existing) row from both history and latest views that matches long-key of provided CovidcastTestRow
         cols = "source signal time_type time_value geo_type geo_value issue".split()
         results = {}
         cur = self._db._cursor
@@ -31,8 +30,8 @@ def _find_matches_for_row(self, row):
 
     def test_insert_or_update_with_nonempty_load_table(self):
         # make rows
-        a_row = self._make_placeholder_row()[0]
-        another_row = self._make_placeholder_row(time_value=self.DEFAULT_TIME_VALUE+1, issue=self.DEFAULT_ISSUE+1)[0]
+        a_row = CovidcastTestRow.make_default_row(time_value=2020_02_02)
+        another_row = CovidcastTestRow.make_default_row(time_value=2020_02_03, issue=2020_02_03)
         # insert one
         self._db.insert_or_update_bulk([a_row])
         # put something into the load table
@@ -61,7 +60,7 @@ def test_id_sync(self):
         latest_view = 'epimetric_latest_v'
 
         # add a data point
-        base_row, _ = self._make_placeholder_row()
+        base_row = CovidcastTestRow.make_default_row()
         self._insert_rows([base_row])
         # ensure the primary keys match in the latest and history tables
         matches = self._find_matches_for_row(base_row)
@@ -71,7 +70,7 @@ def test_id_sync(self):
         old_pk_id = matches[latest_view][pk_column]
 
         # add a reissue for said data point
-        next_row, _ = self._make_placeholder_row()
+        next_row = CovidcastTestRow.make_default_row()
         next_row.issue += 1
         self._insert_rows([next_row])
         # ensure the new keys also match
diff --git a/integrations/acquisition/covidcast/test_delete_batch.py b/integrations/acquisition/covidcast/test_delete_batch.py
index 915c9341b..4624df27c 100644
--- a/integrations/acquisition/covidcast/test_delete_batch.py
+++ b/integrations/acquisition/covidcast/test_delete_batch.py
@@ -5,13 +5,10 @@
 import unittest
 from os import path
 
-# third party
-import mysql.connector
-
 # first party
-from delphi_utils import Nans
-from delphi.epidata.acquisition.covidcast.database import Database, CovidcastRow
 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
 
 # py3tester coverage target (equivalent to `import *`)
 __test_target__ = 'delphi.epidata.acquisition.covidcast.database'
@@ -56,17 +53,13 @@ def test_delete_from_tuples(self):
 
     def _test_delete_batch(self, cc_deletions):
         # load sample data
-        rows = []
-        for time_value in [0, 1]:
-            rows += [
-                # varying numeric column here (2nd to last) is `issue`
-                CovidcastRow('src', 'sig', 'day', 'geo', time_value, "d_nonlatest", 0,0,0,0,0,0, 1, 0),
-                CovidcastRow('src', 'sig', 'day', 'geo', time_value, "d_nonlatest", 0,0,0,0,0,0, 2, 0),
-                CovidcastRow('src', 'sig', 'day', 'geo', time_value, "d_latest", 0,0,0,0,0,0, 1, 0),
-                CovidcastRow('src', 'sig', 'day', 'geo', time_value, "d_latest", 0,0,0,0,0,0, 2, 0),
-                CovidcastRow('src', 'sig', 'day', 'geo', time_value, "d_latest",  0,0,0,0,0,0, 3, 0)
-            ]
-        rows.append(CovidcastRow('src', 'sig', 'day', 'geo', 0, "d_justone",  0,0,0,0,0,0, 1, 0))
+        rows = covidcast_rows_from_args(
+            time_value = [0] * 5 + [1] * 5 + [0],
+            geo_value = ["d_nonlatest"] * 2 + ["d_latest"] * 3 + ["d_nonlatest"] * 2 + ["d_latest"] * 3 + ["d_justone"],
+            issue = [1, 2] + [1, 2, 3] + [1, 2] + [1, 2, 3] + [1],
+            sanitize_fields = True
+        )
+
         self._db.insert_or_update_bulk(rows)
 
         # delete entries
diff --git a/integrations/client/test_delphi_epidata.py b/integrations/client/test_delphi_epidata.py
index 625d2859d..82c1452ec 100644
--- a/integrations/client/test_delphi_epidata.py
+++ b/integrations/client/test_delphi_epidata.py
@@ -1,26 +1,26 @@
 """Integration tests for delphi_epidata.py."""
 
 # standard library
-import unittest
 import time
-from unittest.mock import patch, MagicMock
 from json import JSONDecodeError
+from unittest.mock import MagicMock, patch
 
-# third party
-from aiohttp.client_exceptions import ClientResponseError
-import mysql.connector
+# first party
 import pytest
+from aiohttp.client_exceptions import ClientResponseError
 
-# first party
-from delphi_utils import Nans
-from delphi.epidata.client.delphi_epidata import Epidata
-from delphi.epidata.acquisition.covidcast.database import Database, CovidcastRow
-from delphi.epidata.acquisition.covidcast.covidcast_meta_cache_updater import main as update_covidcast_meta_cache
-from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase
+# third party
 import delphi.operations.secrets as secrets
+from delphi.epidata.acquisition.covidcast.covidcast_meta_cache_updater import main as update_covidcast_meta_cache
+from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase, CovidcastTestRow
+from delphi.epidata.client.delphi_epidata import Epidata
+from delphi_utils import Nans
+
 
 # py3tester coverage target
 __test_target__ = 'delphi.epidata.client.delphi_epidata'
+# all the Nans we use here are just one value, so this is a shortcut to it:
+nmv = Nans.NOT_MISSING.value
 
 def fake_epidata_endpoint(func):
   """This can be used as a decorator to enable a bogus Epidata endpoint to return 404 responses."""
@@ -30,9 +30,6 @@ def wrapper(*args):
     Epidata.BASE_URL = 'http://delphi_web_epidata/epidata/api.php'
   return wrapper
 
-# all the Nans we use here are just one value, so this is a shortcut to it:
-nmv = Nans.NOT_MISSING.value
-
 class DelphiEpidataPythonClientTests(CovidcastBase):
   """Tests the Python client."""
 
@@ -54,13 +51,11 @@ def test_covidcast(self):
 
     # insert placeholder data: three issues of one signal, one issue of another
     rows = [
-      self._make_placeholder_row(issue=self.DEFAULT_ISSUE + i, value=i, lag=i)[0]
+      CovidcastTestRow.make_default_row(issue=2020_02_02 + i, value=i, lag=i)
       for i in range(3)
     ]
     row_latest_issue = rows[-1]
-    rows.append(
-      self._make_placeholder_row(signal="sig2")[0]
-    )
+    rows.append(CovidcastTestRow.make_default_row(signal="sig2"))
     self._insert_rows(rows)
 
     with self.subTest(name='request two signals'):
@@ -70,10 +65,11 @@ def test_covidcast(self):
       )
 
       expected = [
-        self.expected_from_row(row_latest_issue),
-        self.expected_from_row(rows[-1])
+        row_latest_issue.as_api_compatibility_row_dict(),
+        rows[-1].as_api_compatibility_row_dict()
       ]
 
+      self.assertEqual(response['epidata'], expected)
       # check result
       self.assertEqual(response, {
         'result': 1,
@@ -89,10 +85,10 @@ def test_covidcast(self):
 
       expected = [{
         rows[0].signal: [
-          self.expected_from_row(row_latest_issue, self.DEFAULT_MINUS + ['signal']),
+          row_latest_issue.as_api_compatibility_row_dict(ignore_fields=['signal']),
         ],
         rows[-1].signal: [
-          self.expected_from_row(rows[-1], self.DEFAULT_MINUS + ['signal']),
+          rows[-1].as_api_compatibility_row_dict(ignore_fields=['signal']),
         ],
       }]
 
@@ -109,12 +105,12 @@ def test_covidcast(self):
         **self.params_from_row(rows[0])
       )
 
-      expected = self.expected_from_row(row_latest_issue)
+      expected = [row_latest_issue.as_api_compatibility_row_dict()]
 
       # check result
       self.assertEqual(response_1, {
         'result': 1,
-        'epidata': [expected],
+        'epidata': expected,
         'message': 'success',
       })
 
@@ -124,13 +120,13 @@ def test_covidcast(self):
         **self.params_from_row(rows[0], as_of=rows[1].issue)
       )
 
-      expected = self.expected_from_row(rows[1])
+      expected = [rows[1].as_api_compatibility_row_dict()]
 
       # check result
       self.maxDiff=None
       self.assertEqual(response_1a, {
         'result': 1,
-        'epidata': [expected],
+        'epidata': expected,
         'message': 'success',
       })
 
@@ -141,8 +137,8 @@ def test_covidcast(self):
       )
 
       expected = [
-        self.expected_from_row(rows[0]),
-        self.expected_from_row(rows[1])
+        rows[0].as_api_compatibility_row_dict(),
+        rows[1].as_api_compatibility_row_dict()
       ]
 
       # check result
@@ -158,12 +154,12 @@ def test_covidcast(self):
         **self.params_from_row(rows[0], lag=2)
       )
 
-      expected = self.expected_from_row(row_latest_issue)
+      expected = [row_latest_issue.as_api_compatibility_row_dict()]
 
       # check result
       self.assertDictEqual(response_3, {
           'result': 1,
-          'epidata': [expected],
+          'epidata': expected,
           'message': 'success',
       })
     with self.subTest(name='long request'):
@@ -223,16 +219,16 @@ def test_geo_value(self):
     # insert placeholder data: three counties, three MSAs
     N = 3
     rows = [
-      self._make_placeholder_row(geo_type="county", geo_value=str(i)*5, value=i)[0]
+      CovidcastTestRow.make_default_row(geo_type="county", geo_value=str(i)*5, value=i)
       for i in range(N)
     ] + [
-      self._make_placeholder_row(geo_type="msa", geo_value=str(i)*5, value=i*10)[0]
+      CovidcastTestRow.make_default_row(geo_type="msa", geo_value=str(i)*5, value=i*10)
       for i in range(N)
     ]
     self._insert_rows(rows)
 
     counties = [
-      self.expected_from_row(rows[i]) for i in range(N)
+      rows[i].as_api_compatibility_row_dict() for i in range(N)
     ]
 
     def fetch(geo):
@@ -241,41 +237,48 @@ def fetch(geo):
       )
 
     # test fetch all
-    r = fetch('*')
-    self.assertEqual(r['message'], 'success')
-    self.assertEqual(r['epidata'], counties)
+    request = fetch('*')
+    self.assertEqual(request['message'], 'success')
+    self.assertEqual(request['epidata'], counties)
     # test fetch a specific region
-    r = fetch('11111')
-    self.assertEqual(r['message'], 'success')
-    self.assertEqual(r['epidata'], [counties[1]])
+    request = fetch('11111')
+    self.assertEqual(request['message'], 'success')
+    self.assertEqual(request['epidata'], [counties[1]])
     # test fetch a specific yet not existing region
-    r = fetch('55555')
-    self.assertEqual(r['message'], 'no results')
+    request = fetch('55555')
+    self.assertEqual(request['message'], 'no results')
     # test fetch a multiple regions
-    r = fetch(['11111', '22222'])
-    self.assertEqual(r['message'], 'success')
-    self.assertEqual(r['epidata'], [counties[1], counties[2]])
+    request = fetch(['11111', '22222'])
+    self.assertEqual(request['message'], 'success')
+    self.assertEqual(request['epidata'], [counties[1], counties[2]])
     # test fetch a multiple regions in another variant
-    r = fetch(['00000', '22222'])
-    self.assertEqual(r['message'], 'success')
-    self.assertEqual(r['epidata'], [counties[0], counties[2]])
+    request = fetch(['00000', '22222'])
+    self.assertEqual(request['message'], 'success')
+    self.assertEqual(request['epidata'], [counties[0], counties[2]])
     # test fetch a multiple regions but one is not existing
-    r = fetch(['11111', '55555'])
-    self.assertEqual(r['message'], 'success')
-    self.assertEqual(r['epidata'], [counties[1]])
+    request = fetch(['11111', '55555'])
+    self.assertEqual(request['message'], 'success')
+    self.assertEqual(request['epidata'], [counties[1]])
     # test fetch a multiple regions but specify no region
-    r = fetch([])
-    self.assertEqual(r['message'], 'no results')
+    request = fetch([])
+    self.assertEqual(request['message'], 'no results')
 
   def test_covidcast_meta(self):
     """Test that the covidcast_meta endpoint returns expected data."""
 
+    DEFAULT_TIME_VALUE = 2020_02_02
+    DEFAULT_ISSUE = 2020_02_02
+
     # insert placeholder data: three dates, three issues. values are:
     # 1st issue: 0 10 20
     # 2nd issue: 1 11 21
     # 3rd issue: 2 12 22
     rows = [
-      self._make_placeholder_row(time_value=self.DEFAULT_TIME_VALUE + t, issue=self.DEFAULT_ISSUE + i, value=t*10 + i)[0]
+      CovidcastTestRow.make_default_row(
+        time_value=DEFAULT_TIME_VALUE + t,
+        issue=DEFAULT_ISSUE + i,
+        value=t*10 + i
+      )
       for i in range(3) for t in range(3)
     ]
     self._insert_rows(rows)
@@ -299,14 +302,14 @@ def test_covidcast_meta(self):
       signal=rows[0].signal,
       time_type=rows[0].time_type,
       geo_type=rows[0].geo_type,
-      min_time=self.DEFAULT_TIME_VALUE,
-      max_time=self.DEFAULT_TIME_VALUE + 2,
+      min_time=DEFAULT_TIME_VALUE,
+      max_time=DEFAULT_TIME_VALUE + 2,
       num_locations=1,
       min_value=2.,
       mean_value=12.,
       max_value=22.,
       stdev_value=8.1649658, # population stdev, not sample, which is 10.
-      max_issue=self.DEFAULT_ISSUE + 2,
+      max_issue=DEFAULT_ISSUE + 2,
       min_lag=0,
       max_lag=0, # we didn't set lag when inputting data
     )
@@ -322,10 +325,10 @@ def test_async_epidata(self):
     # insert placeholder data: three counties, three MSAs
     N = 3
     rows = [
-      self._make_placeholder_row(geo_type="county", geo_value=str(i)*5, value=i)[0]
+      CovidcastTestRow.make_default_row(geo_type="county", geo_value=str(i)*5, value=i)
       for i in range(N)
     ] + [
-      self._make_placeholder_row(geo_type="msa", geo_value=str(i)*5, value=i*10)[0]
+      CovidcastTestRow.make_default_row(geo_type="msa", geo_value=str(i)*5, value=i*10)
       for i in range(N)
     ]
     self._insert_rows(rows)
diff --git a/integrations/server/test_covidcast.py b/integrations/server/test_covidcast.py
index bcca3b199..c3b50206d 100644
--- a/integrations/server/test_covidcast.py
+++ b/integrations/server/test_covidcast.py
@@ -1,7 +1,7 @@
 """Integration tests for the `covidcast` endpoint."""
 
 # standard library
-import json
+from typing import Callable
 import unittest
 
 # third party
@@ -10,13 +10,11 @@
 
 # first party
 from delphi_utils import Nans
-from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase
+from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase, CovidcastTestRow
 
 # use the local instance of the Epidata API
 BASE_URL = 'http://delphi_web_epidata/epidata/api.php'
 
-
-
 class CovidcastTests(CovidcastBase):
   """Tests the `covidcast` endpoint."""
 
@@ -24,28 +22,26 @@ def localSetUp(self):
     """Perform per-test setup."""
     self._db._cursor.execute('update covidcast_meta_cache set timestamp = 0, epidata = "[]"')
 
-  def request_based_on_row(self, row, extract_response=lambda x: x.json(), **kwargs):
+  def request_based_on_row(self, row: CovidcastTestRow, extract_response: Callable = lambda x: x.json(), **kwargs):
     params = self.params_from_row(row, endpoint='covidcast', **kwargs)
     response = requests.get(BASE_URL, params=params)
     response.raise_for_status()
     response = extract_response(response)
 
-    expected = self.expected_from_row(row)
-
-    return response, expected
+    return response
 
   def _insert_placeholder_set_one(self):
-    row, settings = self._make_placeholder_row()
+    row = CovidcastTestRow.make_default_row()
     self._insert_rows([row])
     return row
 
   def _insert_placeholder_set_two(self):
     rows = [
-      self._make_placeholder_row(geo_type='county', geo_value=str(i)*5, value=i*1., stderr=i*10., sample_size=i*100.)[0]
+      CovidcastTestRow.make_default_row(geo_type='county', geo_value=str(i)*5, value=i*1., stderr=i*10., sample_size=i*100.)
       for i in [1, 2, 3]
     ] + [
       # geo value intended to overlap with counties above
-      self._make_placeholder_row(geo_type='msa', geo_value=str(i-3)*5, value=i*1., stderr=i*10., sample_size=i*100.)[0]
+      CovidcastTestRow.make_default_row(geo_type='msa', geo_value=str(i-3)*5, value=i*1., stderr=i*10., sample_size=i*100.)
       for i in [4, 5, 6]
     ]
     self._insert_rows(rows)
@@ -53,11 +49,11 @@ def _insert_placeholder_set_two(self):
 
   def _insert_placeholder_set_three(self):
     rows = [
-      self._make_placeholder_row(geo_type='county', geo_value='11111', time_value=2000_01_01+i, value=i*1., stderr=i*10., sample_size=i*100., issue=2000_01_03, lag=2-i)[0]
+      CovidcastTestRow.make_default_row(geo_type='county', geo_value='11111', time_value=2000_01_01+i, value=i*1., stderr=i*10., sample_size=i*100., issue=2000_01_03, lag=2-i)
       for i in [1, 2, 3]
     ] + [
       # time value intended to overlap with 11111 above, with disjoint geo values
-      self._make_placeholder_row(geo_type='county', geo_value=str(i)*5, time_value=2000_01_01+i-3, value=i*1., stderr=i*10., sample_size=i*100., issue=2000_01_03, lag=5-i)[0]
+      CovidcastTestRow.make_default_row(geo_type='county', geo_value=str(i)*5, time_value=2000_01_01+i-3, value=i*1., stderr=i*10., sample_size=i*100., issue=2000_01_03, lag=5-i)
       for i in [4, 5, 6]
     ]
     self._insert_rows(rows)
@@ -65,11 +61,11 @@ def _insert_placeholder_set_three(self):
 
   def _insert_placeholder_set_four(self):
     rows = [
-      self._make_placeholder_row(source='src1', signal=str(i)*5, value=i*1., stderr=i*10., sample_size=i*100.)[0]
+      CovidcastTestRow.make_default_row(source='src1', signal=str(i)*5, value=i*1., stderr=i*10., sample_size=i*100.)
       for i in [1, 2, 3]
     ] + [
       # signal intended to overlap with the signal above
-      self._make_placeholder_row(source='src2', signal=str(i-3)*5, value=i*1., stderr=i*10., sample_size=i*100.)[0]
+      CovidcastTestRow.make_default_row(source='src2', signal=str(i-3)*5, value=i*1., stderr=i*10., sample_size=i*100.)
       for i in [4, 5, 6]
     ]
     self._insert_rows(rows)
@@ -77,11 +73,11 @@ def _insert_placeholder_set_four(self):
 
   def _insert_placeholder_set_five(self):
     rows = [
-      self._make_placeholder_row(time_value=2000_01_01, value=i*1., stderr=i*10., sample_size=i*100., issue=2000_01_03+i)[0]
+      CovidcastTestRow.make_default_row(time_value=2000_01_01, value=i*1., stderr=i*10., sample_size=i*100., issue=2000_01_03+i)
       for i in [1, 2, 3]
     ] + [
       # different time_values, same issues
-      self._make_placeholder_row(time_value=2000_01_01+i-3, value=i*1., stderr=i*10., sample_size=i*100., issue=2000_01_03+i-3)[0]
+      CovidcastTestRow.make_default_row(time_value=2000_01_01+i-3, value=i*1., stderr=i*10., sample_size=i*100., issue=2000_01_03+i-3)
       for i in [4, 5, 6]
     ]
     self._insert_rows(rows)
@@ -94,10 +90,13 @@ def test_round_trip(self):
     row = self._insert_placeholder_set_one()
 
     # make the request
-    response, expected = self.request_based_on_row(row)
+    response = self.request_based_on_row(row)
+
+    expected = [row.as_api_compatibility_row_dict()]
+
     self.assertEqual(response, {
       'result': 1,
-      'epidata': [expected],
+      'epidata': expected,
       'message': 'success',
     })
 
@@ -154,32 +153,25 @@ def test_csv_format(self):
 
     # make the request
     # NB 'format' is a Python reserved word
-    response, _ = self.request_based_on_row(
+    response = self.request_based_on_row(
       row,
       extract_response=lambda resp: resp.text,
       **{'format':'csv'}
     )
-    expected_response = (
-      "geo_value,signal,time_value,direction,issue,lag,missing_value," +
-      "missing_stderr,missing_sample_size,value,stderr,sample_size\n" +
-      ",".join("" if x is None else str(x) for x in [
-        row.geo_value,
-        row.signal,
-        row.time_value,
-        row.direction,
-        row.issue,
-        row.lag,
-        row.missing_value,
-        row.missing_stderr,
-        row.missing_sample_size,
-        row.value,
-        row.stderr,
-        row.sample_size
-      ]) + "\n"
+
+    # This is a hardcoded mess because of api.php.
+    column_order = [
+      "geo_value", "signal", "time_value", "direction", "issue", "lag", "missing_value",
+      "missing_stderr", "missing_sample_size", "value", "stderr", "sample_size"
+    ]
+    expected = (
+      row.as_api_compatibility_row_df()
+         .assign(direction = None)
+         .to_csv(columns=column_order, index=False)
     )
 
     # assert that the right data came back
-    self.assertEqual(response, expected_response)
+    self.assertEqual(response, expected)
 
   def test_raw_json_format(self):
     """Test generate raw json data."""
@@ -188,10 +180,12 @@ def test_raw_json_format(self):
     row = self._insert_placeholder_set_one()
 
     # make the request
-    response, expected = self.request_based_on_row(row, **{'format':'json'})
+    response = self.request_based_on_row(row, **{'format':'json'})
+
+    expected = [row.as_api_compatibility_row_dict()]
 
     # assert that the right data came back
-    self.assertEqual(response, [expected])
+    self.assertEqual(response, expected)
 
   def test_fields(self):
     """Test fields parameter"""
@@ -200,7 +194,9 @@ def test_fields(self):
     row = self._insert_placeholder_set_one()
 
     # limit fields
-    response, expected = self.request_based_on_row(row, fields='time_value,geo_value')
+    response = self.request_based_on_row(row, fields='time_value,geo_value')
+
+    expected = row.as_api_compatibility_row_dict()
     expected_all = {
       'result': 1,
       'epidata': [{
@@ -213,15 +209,14 @@ def test_fields(self):
     self.assertEqual(response, expected_all)
 
     # limit using invalid fields
-    response, _ = self.request_based_on_row(row, fields='time_value,geo_value,doesnt_exist')
+    response = self.request_based_on_row(row, fields='time_value,geo_value,doesnt_exist')
 
     # assert that the right data came back (only valid fields)
     self.assertEqual(response, expected_all)
 
 
     # limit exclude fields: exclude all except time_value and geo_value
-
-    response, _ = self.request_based_on_row(row, fields=(
+    response = self.request_based_on_row(row, fields=(
         '-value,-stderr,-sample_size,-direction,-issue,-lag,-signal,' +
         '-missing_value,-missing_stderr,-missing_sample_size'
     ))
@@ -234,18 +229,15 @@ def test_location_wildcard(self):
 
     # insert placeholder data
     rows = self._insert_placeholder_set_two()
-    expected_counties = [
-      self.expected_from_row(r) for r in rows[:3]
-    ]
-
+    expected = [row.as_api_compatibility_row_dict() for row in rows[:3]]
     # make the request
-    response, _ = self.request_based_on_row(rows[0], geo_value="*")
+    response = self.request_based_on_row(rows[0], geo_value="*")
 
     self.maxDiff = None
     # assert that the right data came back
     self.assertEqual(response, {
       'result': 1,
-      'epidata': expected_counties,
+      'epidata': expected,
       'message': 'success',
     })
 
@@ -254,18 +246,16 @@ def test_time_values_wildcard(self):
 
     # insert placeholder data
     rows = self._insert_placeholder_set_three()
-    expected_time_values = [
-      self.expected_from_row(r) for r in rows[:3]
-    ]
+    expected = [row.as_api_compatibility_row_dict() for row in rows[:3]]
 
     # make the request
-    response, _ = self.request_based_on_row(rows[0], time_values="*")
+    response = self.request_based_on_row(rows[0], time_values="*")
 
     self.maxDiff = None
     # assert that the right data came back
     self.assertEqual(response, {
       'result': 1,
-      'epidata': expected_time_values,
+      'epidata': expected,
       'message': 'success',
     })
 
@@ -274,18 +264,16 @@ def test_issues_wildcard(self):
 
     # insert placeholder data
     rows = self._insert_placeholder_set_five()
-    expected_issues = [
-      self.expected_from_row(r) for r in rows[:3]
-    ]
+    expected = [row.as_api_compatibility_row_dict() for row in rows[:3]]
 
     # make the request
-    response, _ = self.request_based_on_row(rows[0], issues="*")
+    response = self.request_based_on_row(rows[0], issues="*")
 
     self.maxDiff = None
     # assert that the right data came back
     self.assertEqual(response, {
       'result': 1,
-      'epidata': expected_issues,
+      'epidata': expected,
       'message': 'success',
     })
 
@@ -294,12 +282,10 @@ def test_signal_wildcard(self):
 
     # insert placeholder data
     rows = self._insert_placeholder_set_four()
-    expected_signals = [
-      self.expected_from_row(r) for r in rows[:3]
-    ]
+    expected_signals = [row.as_api_compatibility_row_dict() for row in rows[:3]]
 
     # make the request
-    response, _ = self.request_based_on_row(rows[0], signals="*")
+    response = self.request_based_on_row(rows[0], signals="*")
 
     self.maxDiff = None
     # assert that the right data came back
@@ -314,35 +300,33 @@ def test_geo_value(self):
 
     # insert placeholder data
     rows = self._insert_placeholder_set_two()
-    expected_counties = [
-      self.expected_from_row(r) for r in rows[:3]
-    ]
+    expected = [row.as_api_compatibility_row_dict() for row in rows[:3]]
 
     def fetch(geo_value):
       # make the request
-      response, _ = self.request_based_on_row(rows[0], geo_value=geo_value)
+      response = self.request_based_on_row(rows[0], geo_value=geo_value)
 
       return response
 
     # test fetch a specific region
     r = fetch('11111')
     self.assertEqual(r['message'], 'success')
-    self.assertEqual(r['epidata'], [expected_counties[0]])
+    self.assertEqual(r['epidata'], expected[0:1])
     # test fetch a specific yet not existing region
     r = fetch('55555')
     self.assertEqual(r['message'], 'no results')
     # test fetch multiple regions
     r = fetch('11111,22222')
     self.assertEqual(r['message'], 'success')
-    self.assertEqual(r['epidata'], [expected_counties[0], expected_counties[1]])
+    self.assertEqual(r['epidata'], expected[0:2])
     # test fetch multiple noncontiguous regions
     r = fetch('11111,33333')
     self.assertEqual(r['message'], 'success')
-    self.assertEqual(r['epidata'], [expected_counties[0], expected_counties[2]])
+    self.assertEqual(r['epidata'], [expected[0], expected[2]])
     # test fetch multiple regions but one is not existing
     r = fetch('11111,55555')
     self.assertEqual(r['message'], 'success')
-    self.assertEqual(r['epidata'], [expected_counties[0]])
+    self.assertEqual(r['epidata'], expected[0:1])
     # test fetch empty region
     r = fetch('')
     self.assertEqual(r['message'], 'no results')
@@ -352,12 +336,10 @@ def test_location_timeline(self):
 
     # insert placeholder data
     rows = self._insert_placeholder_set_three()
-    expected_timeseries = [
-      self.expected_from_row(r) for r in rows[:3]
-    ]
+    expected_timeseries = [row.as_api_compatibility_row_dict() for row in rows[:3]]
 
     # make the request
-    response, _ = self.request_based_on_row(rows[0], time_values='20000101-20000105')
+    response = self.request_based_on_row(rows[0], time_values='20000101-20000105')
 
     # assert that the right data came back
     self.assertEqual(response, {
@@ -383,15 +365,15 @@ def test_unique_key_constraint(self):
   def test_nullable_columns(self):
     """Missing values should be surfaced as null."""
 
-    row, _ = self._make_placeholder_row(
+    row = CovidcastTestRow.make_default_row(
       stderr=None, sample_size=None,
       missing_stderr=Nans.OTHER.value, missing_sample_size=Nans.OTHER.value
     )
     self._insert_rows([row])
 
     # make the request
-    response, expected = self.request_based_on_row(row)
-    expected.update(stderr=None, sample_size=None)
+    response = self.request_based_on_row(row)
+    expected = row.as_api_compatibility_row_dict()
 
     # assert that the right data came back
     self.assertEqual(response, {
@@ -405,18 +387,19 @@ def test_temporal_partitioning(self):
 
     # insert placeholder data
     rows = [
-      self._make_placeholder_row(time_type=tt)[0]
+      CovidcastTestRow.make_default_row(time_type=tt)
       for tt in "hour day week month year".split()
     ]
     self._insert_rows(rows)
 
     # make the request
-    response, expected = self.request_based_on_row(rows[1], time_values="0-99999999")
+    response = self.request_based_on_row(rows[1], time_values="*")
+    expected = [rows[1].as_api_compatibility_row_dict()]
 
     # assert that the right data came back
     self.assertEqual(response, {
       'result': 1,
-      'epidata': [expected],
+      'epidata': expected,
       'message': 'success',
     })
 
@@ -427,37 +410,37 @@ def test_date_formats(self):
     rows = self._insert_placeholder_set_three()
 
     # make the request
-    response, expected = self.request_based_on_row(rows[0], time_values="20000102", geo_value="*")
+    response = self.request_based_on_row(rows[0], time_values="20000102", geo_value="*")
 
     # assert that the right data came back
     self.assertEqual(len(response['epidata']), 2)
 
     # make the request
-    response, expected = self.request_based_on_row(rows[0], time_values="2000-01-02", geo_value="*")
+    response = self.request_based_on_row(rows[0], time_values="2000-01-02", geo_value="*")
 
     # assert that the right data came back
     self.assertEqual(len(response['epidata']), 2)
 
     # make the request
-    response, expected = self.request_based_on_row(rows[0], time_values="20000102,20000103", geo_value="*")
+    response = self.request_based_on_row(rows[0], time_values="20000102,20000103", geo_value="*")
 
     # assert that the right data came back
-    self.assertEqual(len(response['epidata']), 4)
+    self.assertEqual(len(response['epidata']), 2 * 2)
 
     # make the request
-    response, expected = self.request_based_on_row(rows[0], time_values="2000-01-02,2000-01-03", geo_value="*")
+    response = self.request_based_on_row(rows[0], time_values="2000-01-02,2000-01-03", geo_value="*")
 
     # assert that the right data came back
-    self.assertEqual(len(response['epidata']), 4)
+    self.assertEqual(len(response['epidata']), 2 * 2)
 
     # make the request
-    response, expected = self.request_based_on_row(rows[0], time_values="20000102-20000104", geo_value="*")
+    response = self.request_based_on_row(rows[0], time_values="20000102-20000104", geo_value="*")
 
     # assert that the right data came back
-    self.assertEqual(len(response['epidata']), 6)
+    self.assertEqual(len(response['epidata']), 2 * 3)
 
     # make the request
-    response, expected = self.request_based_on_row(rows[0], time_values="2000-01-02:2000-01-04", geo_value="*")
+    response = self.request_based_on_row(rows[0], time_values="2000-01-02:2000-01-04", geo_value="*")
 
     # assert that the right data came back
-    self.assertEqual(len(response['epidata']), 6)
+    self.assertEqual(len(response['epidata']), 2 * 3)
diff --git a/integrations/server/test_covidcast_endpoints.py b/integrations/server/test_covidcast_endpoints.py
index 54974a874..41d942456 100644
--- a/integrations/server/test_covidcast_endpoints.py
+++ b/integrations/server/test_covidcast_endpoints.py
@@ -1,30 +1,23 @@
 """Integration tests for the custom `covidcast/*` endpoints."""
 
 # standard library
-from typing import Iterable, Dict, Any
-import unittest
 from io import StringIO
-
-# from typing import Optional
-from dataclasses import dataclass
+from typing import Sequence
 
 # third party
-import mysql.connector
+from more_itertools import windowed
 import requests
 import pandas as pd
-from delphi_utils import Nans
 
 from delphi.epidata.acquisition.covidcast.covidcast_meta_cache_updater import main as update_cache
-
-from delphi.epidata.acquisition.covidcast.database import Database
-from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase
+from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase, CovidcastTestRow
 
 # use the local instance of the Epidata API
 BASE_URL = "http://delphi_web_epidata/epidata/covidcast"
+BASE_URL_OLD = "http://delphi_web_epidata/epidata/api.php"
 
 
 class CovidcastEndpointTests(CovidcastBase):
-
     """Tests the `covidcast/*` endpoint."""
 
     def localSetUp(self):
@@ -32,19 +25,36 @@ def localSetUp(self):
         # reset the `covidcast_meta_cache` table (it should always have one row)
         self._db._cursor.execute('update covidcast_meta_cache set timestamp = 0, epidata = "[]"')
 
-    def _fetch(self, endpoint="/", **params):
+    def _fetch(self, endpoint="/", is_compatibility=False, **params):
         # make the request
-        response = requests.get(
-            f"{BASE_URL}{endpoint}",
-            params=params,
-        )
+        if is_compatibility:
+            url = BASE_URL_OLD
+            # only set endpoint if it's not already set
+            # only set endpoint if it's not already set
+            params.setdefault("endpoint", "covidcast")
+            if params.get("source"):
+                params.setdefault("data_source", params.get("source"))
+        else:
+            url = f"{BASE_URL}{endpoint}"
+        response = requests.get(url, params=params)
         response.raise_for_status()
         return response.json()
 
+    def _diff_rows(self, rows: Sequence[float]):
+        return [
+            float(x - y) if x is not None and y is not None else None
+            for x, y in zip(rows[1:], rows[:-1])
+        ]
+
+    def _smooth_rows(self, rows: Sequence[float]):
+        return [
+            sum(e)/len(e) if None not in e else None 
+            for e in windowed(rows, 7)
+        ]
+
     def test_basic(self):
         """Request a signal from the / endpoint."""
-
-        rows = [self._make_placeholder_row(time_value=20200401 + i, value=i)[0] for i in range(10)]
+        rows = [CovidcastTestRow.make_default_row(time_value=2020_04_01 + i, value=i) for i in range(10)]
         first = rows[0]
         self._insert_rows(rows)
 
@@ -56,11 +66,25 @@ def test_basic(self):
             out = self._fetch("/", signal=first.signal_pair(), geo=first.geo_pair(), time="day:*")
             self.assertEqual(len(out["epidata"]), len(rows))
 
+    def test_compatibility(self):
+        """Request at the /api.php endpoint."""
+        rows = [CovidcastTestRow.make_default_row(source="src", signal="sig", time_value=2020_04_01 + i, value=i) for i in range(10)]
+        first = rows[0]
+        self._insert_rows(rows)
+
+        with self.subTest("validation"):
+            out = self._fetch("/", is_compatibility=True)
+            self.assertEqual(out["result"], -1)
+
+        with self.subTest("simple"):
+            out = self._fetch("/", signal=first.signal_pair(), geo=first.geo_pair(), time="day:*", is_compatibility=True)
+            self.assertEqual(len(out["epidata"]), len(rows))
+
     def test_trend(self):
         """Request a signal from the /trend endpoint."""
 
         num_rows = 30
-        rows = [self._make_placeholder_row(time_value=20200401 + i, value=i)[0] for i in range(num_rows)]
+        rows = [CovidcastTestRow.make_default_row(time_value=2020_04_01 + i, value=i) for i in range(num_rows)]
         first = rows[0]
         last = rows[-1]
         ref = rows[num_rows // 2]
@@ -68,6 +92,7 @@ def test_trend(self):
 
         out = self._fetch("/trend", signal=first.signal_pair(), geo=first.geo_pair(), date=last.time_value, window="20200401-20201212", basis=ref.time_value)
 
+
         self.assertEqual(out["result"], 1)
         self.assertEqual(len(out["epidata"]), 1)
         trend = out["epidata"][0]
@@ -90,11 +115,12 @@ def test_trend(self):
         self.assertEqual(trend["max_value"], last.value)
         self.assertEqual(trend["max_trend"], "steady")
 
+
     def test_trendseries(self):
         """Request a signal from the /trendseries endpoint."""
 
         num_rows = 3
-        rows = [self._make_placeholder_row(time_value=20200401 + i, value=num_rows - i)[0] for i in range(num_rows)]
+        rows = [CovidcastTestRow.make_default_row(time_value=2020_04_01 + i, value=num_rows - i) for i in range(num_rows)]
         first = rows[0]
         last = rows[-1]
         self._insert_rows(rows)
@@ -127,6 +153,7 @@ def match_row(trend, row):
             self.assertEqual(trend["max_date"], first.time_value)
             self.assertEqual(trend["max_value"], first.value)
             self.assertEqual(trend["max_trend"], "steady")
+
         with self.subTest("trend1"):
             trend = trends[1]
             match_row(trend, rows[1])
@@ -159,10 +186,10 @@ def test_correlation(self):
         """Request a signal from the /correlation endpoint."""
 
         num_rows = 30
-        reference_rows = [self._make_placeholder_row(signal="ref", time_value=20200401 + i, value=i)[0] for i in range(num_rows)]
+        reference_rows = [CovidcastTestRow.make_default_row(signal="ref", time_value=20200401 + i, value=i) for i in range(num_rows)]
         first = reference_rows[0]
         self._insert_rows(reference_rows)
-        other_rows = [self._make_placeholder_row(signal="other", time_value=20200401 + i, value=i)[0] for i in range(num_rows)]
+        other_rows = [CovidcastTestRow.make_default_row(signal="other", time_value=20200401 + i, value=i) for i in range(num_rows)]
         other = other_rows[0]
         self._insert_rows(other_rows)
         max_lag = 3
@@ -185,7 +212,7 @@ def test_correlation(self):
     def test_csv(self):
         """Request a signal from the /csv endpoint."""
 
-        rows = [self._make_placeholder_row(time_value=20200401 + i, value=i)[0] for i in range(10)]
+        rows = [CovidcastTestRow.make_default_row(time_value=2020_04_01 + i, value=i) for i in range(10)]
         first = rows[0]
         self._insert_rows(rows)
 
@@ -199,13 +226,15 @@ def test_csv(self):
         self.assertEqual(df.shape, (len(rows), 10))
         self.assertEqual(list(df.columns), ["geo_value", "signal", "time_value", "issue", "lag", "value", "stderr", "sample_size", "geo_type", "data_source"])
 
+
     def test_backfill(self):
         """Request a signal from the /backfill endpoint."""
 
+        TEST_DATE_VALUE = 2020_04_01
         num_rows = 10
-        issue_0 = [self._make_placeholder_row(time_value=20200401 + i, value=i, sample_size=1, lag=0, issue=20200401 + i)[0] for i in range(num_rows)]
-        issue_1 = [self._make_placeholder_row(time_value=20200401 + i, value=i + 1, sample_size=2, lag=1, issue=20200401 + i + 1)[0] for i in range(num_rows)]
-        last_issue = [self._make_placeholder_row(time_value=20200401 + i, value=i + 2, sample_size=3, lag=2, issue=20200401 + i + 2)[0] for i in range(num_rows)] # <-- the latest issues
+        issue_0 = [CovidcastTestRow.make_default_row(time_value=TEST_DATE_VALUE + i, value=i, sample_size=1, lag=0, issue=TEST_DATE_VALUE + i) for i in range(num_rows)]
+        issue_1 = [CovidcastTestRow.make_default_row(time_value=TEST_DATE_VALUE + i, value=i + 1, sample_size=2, lag=1, issue=TEST_DATE_VALUE + i + 1) for i in range(num_rows)]
+        last_issue = [CovidcastTestRow.make_default_row(time_value=TEST_DATE_VALUE + i, value=i + 2, sample_size=3, lag=2, issue=TEST_DATE_VALUE + i + 2) for i in range(num_rows)] # <-- the latest issues
         self._insert_rows([*issue_0, *issue_1, *last_issue])
         first = issue_0[0]
 
@@ -231,7 +260,7 @@ def test_meta(self):
         """Request a signal from the /meta endpoint."""
 
         num_rows = 10
-        rows = [self._make_placeholder_row(time_value=20200401 + i, value=i, source="fb-survey", signal="smoothed_cli")[0] for i in range(num_rows)]
+        rows = [CovidcastTestRow.make_default_row(time_value=2020_04_01 + i, value=i, source="fb-survey", signal="smoothed_cli") for i in range(num_rows)]
         self._insert_rows(rows)
         first = rows[0]
         last = rows[-1]
@@ -271,8 +300,8 @@ def test_coverage(self):
         """Request a signal from the /coverage endpoint."""
 
         num_geos_per_date = [10, 20, 30, 40, 44]
-        dates = [20200401 + i for i in range(len(num_geos_per_date))]
-        rows = [self._make_placeholder_row(time_value=dates[i], value=i, geo_value=str(geo_value))[0] for i, num_geo in enumerate(num_geos_per_date) for geo_value in range(num_geo)]
+        dates = [2020_04_01 + i for i in range(len(num_geos_per_date))]
+        rows = [CovidcastTestRow.make_default_row(time_value=dates[i], value=i, geo_value=str(geo_value)) for i, num_geo in enumerate(num_geos_per_date) for geo_value in range(num_geo)]
         self._insert_rows(rows)
         first = rows[0]
 
diff --git a/requirements.api.txt b/requirements.api.txt
index d5cc0e63b..6ccafc1e1 100644
--- a/requirements.api.txt
+++ b/requirements.api.txt
@@ -2,6 +2,7 @@ epiweeks==2.1.2
 Flask==2.2.2
 itsdangerous<2.1
 jinja2==3.0.3
+more_itertools==8.4.0
 mysqlclient==2.1.1
 newrelic
 orjson==3.4.7
diff --git a/src/acquisition/covidcast/covidcast_row.py b/src/acquisition/covidcast/covidcast_row.py
new file mode 100644
index 000000000..23e19eb57
--- /dev/null
+++ b/src/acquisition/covidcast/covidcast_row.py
@@ -0,0 +1,132 @@
+from dataclasses import asdict, dataclass
+from typing import Any, ClassVar, Dict, List, Optional
+
+import pandas as pd
+
+
+PANDAS_DTYPES = {
+    "source": str,
+    "signal": str,
+    "time_type": str,
+    "time_value": "Int64",
+    "geo_type": str,
+    "geo_value": str,
+    "value": float,
+    "stderr": float,
+    "sample_size": float,
+    "missing_value": "Int8",
+    "missing_stderr": "Int8",
+    "missing_sample_size": "Int8",
+    "issue": "Int64",
+    "lag": "Int64",
+    "id": "Int64",
+    "direction": "Int8",
+    "direction_updated_timestamp": "Int64",
+    "value_updated_timestamp": "Int64",
+}
+
+@dataclass
+class CovidcastRow:
+    """A container for the values of a single covidcast database row.
+
+    Used for:
+    - inserting rows into the database
+    - creating test rows with default fields for testing
+    - converting from and to formats (dict, csv, df, kwargs)
+    - creating consistent views, with consistent data types (dict, csv, df)
+
+    The rows are specified in 'v4_schema.sql'. The datatypes are made to match database. When writing to Pandas, the dtypes match the JIT model.py schema.
+    """
+
+    # Arguments.
+    source: str
+    signal: str
+    time_type: str
+    geo_type: str
+    time_value: int
+    geo_value: str
+    value: float
+    stderr: float
+    sample_size: float
+    # The following three fields are Nans enums from delphi_utils.nans.
+    missing_value: int
+    missing_stderr: int
+    missing_sample_size: int
+    issue: int
+    lag: int
+    # The following three fields are only the database, but are not ingested at acquisition and not returned by the API.
+    epimetric_id: Optional[int] = None
+    direction: Optional[int] = None
+    value_updated_timestamp: Optional[int] = 0
+
+    # Classvars.
+    _db_row_ignore_fields: ClassVar = []
+    _api_row_ignore_fields: ClassVar = ["epimetric_id", "value_updated_timestamp"]
+    _api_row_compatibility_ignore_fields: ClassVar = _api_row_ignore_fields + ["source", "time_type", "geo_type"]
+
+    _pandas_dtypes: ClassVar = PANDAS_DTYPES
+
+    def as_dict(self, ignore_fields: Optional[List[str]] = None) -> dict:
+        d = asdict(self)
+        if ignore_fields:
+            for key in ignore_fields:
+                del d[key]
+        return d
+    
+    def as_api_row_dict(self, ignore_fields: Optional[List[str]] = None) -> dict:
+        """Returns a dict view into the row with the fields returned by the API server."""
+        return self.as_dict(ignore_fields=self._api_row_ignore_fields + (ignore_fields or []))
+
+    def as_api_compatibility_row_dict(self, ignore_fields: Optional[List[str]] = None) -> dict:
+        """Returns a dict view into the row with the fields returned by the old API server (the PHP server)."""
+        return self.as_dict(ignore_fields=self._api_row_compatibility_ignore_fields + (ignore_fields or []))
+
+    def as_db_row_dict(self, ignore_fields: Optional[List[str]] = None) -> dict:
+        """Returns a dict view into the row with the fields returned by the database."""
+        return self.as_dict(ignore_fields=self._db_row_ignore_fields + (ignore_fields or []))
+
+    def as_dataframe(self, ignore_fields: Optional[List[str]] = None) -> pd.DataFrame:
+        df = pd.DataFrame.from_records([self.as_dict(ignore_fields=ignore_fields)])
+        # This is to mirror the types in model.py.
+        df = set_df_dtypes(df, self._pandas_dtypes)
+        return df
+
+    def as_api_row_df(self, ignore_fields: Optional[List[str]] = None) -> pd.DataFrame:
+        """Returns a dataframe view into the row with the fields returned by the API server."""
+        return self.as_dataframe(ignore_fields=self._api_row_ignore_fields + (ignore_fields or []))
+
+    def as_api_compatibility_row_df(self, ignore_fields: Optional[List[str]] = None) -> pd.DataFrame:
+        """Returns a dataframe view into the row with the fields returned by the old API server (the PHP server)."""
+        return self.as_dataframe(ignore_fields=self._api_row_compatibility_ignore_fields + (ignore_fields or []))
+
+    def as_db_row_df(self, ignore_fields: Optional[List[str]] = None) -> pd.DataFrame:
+        """Returns a dataframe view into the row with the fields returned by an all-field database query."""
+        return self.as_dataframe(ignore_fields=self._db_row_ignore_fields + (ignore_fields or []))
+
+    def signal_pair(self):
+        return f"{self.source}:{self.signal}"
+
+    def geo_pair(self):
+        return f"{self.geo_type}:{self.geo_value}"
+
+    def time_pair(self):
+        return f"{self.time_type}:{self.time_value}"
+
+
+
+def check_valid_dtype(dtype):
+    try:
+        pd.api.types.pandas_dtype(dtype)
+    except TypeError:
+        raise ValueError(f"Invalid dtype {dtype}")
+
+
+def set_df_dtypes(df: pd.DataFrame, dtypes: Dict[str, Any]) -> pd.DataFrame:
+    """Set the dataframe column datatypes."""
+    [check_valid_dtype(d) for d in dtypes.values()]
+
+    df = df.copy()
+    for k, v in dtypes.items():
+        if k in df.columns:
+            df[k] = df[k].astype(v)
+    return df
diff --git a/src/acquisition/covidcast/csv_importer.py b/src/acquisition/covidcast/csv_importer.py
index 3c559be84..0fa936802 100644
--- a/src/acquisition/covidcast/csv_importer.py
+++ b/src/acquisition/covidcast/csv_importer.py
@@ -15,11 +15,27 @@
 # first party
 from delphi_utils import Nans
 from delphi.utils.epiweek import delta_epiweeks
-from delphi.epidata.acquisition.covidcast.database import CovidcastRow
+from delphi.epidata.acquisition.covidcast.covidcast_row import CovidcastRow
 from delphi.epidata.acquisition.covidcast.logger import get_structured_logger
 
-DFRow = NamedTuple('DFRow', [('geo_id', str), ('value', float), ('stderr', float), ('sample_size', float), ('missing_value', int), ('missing_stderr', int), ('missing_sample_size', int)])
-PathDetails = NamedTuple('PathDetails', [('issue', int), ('lag', int), ('source', str), ('signal', str), ('time_type', str), ('time_value', int), ('geo_type', str)])
+DataFrameRow = NamedTuple('DFRow', [
+  ('geo_id', str),
+  ('value', float),
+  ('stderr', float),
+  ('sample_size', float),
+  ('missing_value', int),
+  ('missing_stderr', int),
+  ('missing_sample_size', int)
+])
+PathDetails = NamedTuple('PathDetails', [
+  ('issue', int),
+  ('lag', int),
+  ('source', str),
+  ("signal", str),
+  ('time_type', str),
+  ('time_value', int),
+  ('geo_type', str),
+])
 
 
 @dataclass
@@ -268,7 +284,7 @@ def validate_missing_code(row, attr_quantity, attr_name, filepath=None, logger=N
 
 
   @staticmethod
-  def extract_and_check_row(row: DFRow, geo_type: str, filepath: Optional[str] = None) -> Tuple[Optional[CsvRowValue], Optional[str]]:
+  def extract_and_check_row(row: DataFrameRow, geo_type: str, filepath: Optional[str] = None) -> Tuple[Optional[CsvRowValue], Optional[str]]:
     """Extract and return `CsvRowValue` from a CSV row, with sanity checks.
 
     Also returns the name of the field which failed sanity check, or None.
@@ -397,10 +413,4 @@ def load_csv(filepath: str, details: PathDetails) -> Iterator[Optional[Covidcast
         csv_row_values.missing_sample_size,
         details.issue,
         details.lag,
-        # These four fields are unused by database acquisition
-        # TODO: These will be used when CovidcastRow is updated.
-        # id=None,
-        # direction=None,
-        # direction_updated_timestamp=0,
-        # value_updated_timestamp=0,
       )
diff --git a/src/acquisition/covidcast/database.py b/src/acquisition/covidcast/database.py
index d21a27c35..3beedac82 100644
--- a/src/acquisition/covidcast/database.py
+++ b/src/acquisition/covidcast/database.py
@@ -2,69 +2,20 @@
 
 See src/ddl/covidcast.sql for an explanation of each field.
 """
+import threading
+from math import ceil
+from multiprocessing import cpu_count
+from queue import Queue, Empty
+from typing import List
 
 # third party
 import json
 import mysql.connector
-import numpy as np
-from math import ceil
-
-from queue import Queue, Empty
-import threading
-from multiprocessing import cpu_count
 
 # first party
 import delphi.operations.secrets as secrets
-
 from delphi.epidata.acquisition.covidcast.logger import get_structured_logger
-
-class CovidcastRow():
-  """A container for all the values of a single covidcast row."""
-
-  @staticmethod
-  def fromCsvRowValue(row_value, source, signal, time_type, geo_type, time_value, issue, lag):
-    if row_value is None: return None
-    return CovidcastRow(source, signal, time_type, geo_type, time_value,
-                        row_value.geo_value,
-                        row_value.value,
-                        row_value.stderr,
-                        row_value.sample_size,
-                        row_value.missing_value,
-                        row_value.missing_stderr,
-                        row_value.missing_sample_size,
-                        issue, lag)
-
-  @staticmethod
-  def fromCsvRows(row_values, source, signal, time_type, geo_type, time_value, issue, lag):
-    # NOTE: returns a generator, as row_values is expected to be a generator
-    return (CovidcastRow.fromCsvRowValue(row_value, source, signal, time_type, geo_type, time_value, issue, lag)
-            for row_value in row_values)
-
-  def __init__(self, source, signal, time_type, geo_type, time_value, geo_value, value, stderr, 
-               sample_size, missing_value, missing_stderr, missing_sample_size, issue, lag):
-    self.id = None
-    self.source = source
-    self.signal = signal
-    self.time_type = time_type
-    self.geo_type = geo_type
-    self.time_value = time_value
-    self.geo_value = geo_value      # from CSV row
-    self.value = value              # ...
-    self.stderr = stderr            # ...
-    self.sample_size = sample_size  # ...
-    self.missing_value = missing_value # ...
-    self.missing_stderr = missing_stderr # ...
-    self.missing_sample_size = missing_sample_size # from CSV row
-    self.direction_updated_timestamp = 0
-    self.direction = None
-    self.issue = issue
-    self.lag = lag
-
-  def signal_pair(self):
-    return f"{self.source}:{self.signal}"
-
-  def geo_pair(self):
-    return f"{self.geo_type}:{self.geo_value}"
+from delphi.epidata.acquisition.covidcast.covidcast_row import CovidcastRow
 
 
 class DBLoadStateException(Exception):
@@ -156,7 +107,7 @@ def do_analyze(self):
   def insert_or_update_bulk(self, cc_rows):
     return self.insert_or_update_batch(cc_rows)
 
-  def insert_or_update_batch(self, cc_rows, batch_size=2**20, commit_partial=False, suppress_jobs=False):
+  def insert_or_update_batch(self, cc_rows: List[CovidcastRow], batch_size=2**20, commit_partial=False, suppress_jobs=False):
     """
     Insert new rows into the load table and dispatch into dimension and fact tables.
     """
diff --git a/src/acquisition/covidcast/test_utils.py b/src/acquisition/covidcast/test_utils.py
index 181dfac68..96db2c164 100644
--- a/src/acquisition/covidcast/test_utils.py
+++ b/src/acquisition/covidcast/test_utils.py
@@ -1,12 +1,151 @@
+from dataclasses import fields
+from datetime import date
+from typing import Any, Dict, Iterable, List, Optional, Sequence
 import unittest
 
+import pandas as pd
+
 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
+from delphi.epidata.server.utils.dates import day_to_time_value, time_value_to_day
 import delphi.operations.secrets as secrets
 
 # all the Nans we use here are just one value, so this is a shortcut to it:
 nmv = Nans.NOT_MISSING.value
 
+
+class CovidcastTestRow(CovidcastRow):
+    @staticmethod
+    def make_default_row(**kwargs) -> "CovidcastTestRow":
+        default_args = {
+            "source": "src",
+            "signal": "sig",
+            "time_type": "day",
+            "geo_type": "county",
+            "time_value": 2020_02_02,
+            "geo_value": "01234",
+            "value": 10.0,
+            "stderr": 10.0,
+            "sample_size": 10.0,
+            "missing_value": Nans.NOT_MISSING.value,
+            "missing_stderr": Nans.NOT_MISSING.value,
+            "missing_sample_size": Nans.NOT_MISSING.value,
+            "issue": 2020_02_02,
+            "lag": 0,
+        }
+        default_args.update(kwargs)
+        return CovidcastTestRow(**default_args)
+
+    def __post_init__(self):
+        # Convert time values to ints by default.
+        if isinstance(self.time_value, date):
+            self.time_value = day_to_time_value(self.time_value)
+        if isinstance(self.issue, date):
+            self.issue = day_to_time_value(self.issue)
+        if isinstance(self.value_updated_timestamp, date):
+            self.value_updated_timestamp = day_to_time_value(self.value_updated_timestamp)
+
+    def _sanitize_fields(self, extra_checks: bool = True):
+        if self.issue and self.issue < self.time_value:
+            self.issue = self.time_value
+
+        if self.issue:
+            self.lag = (time_value_to_day(self.issue) - time_value_to_day(self.time_value)).days
+        else:
+            self.lag = None
+
+        # This sanity checking is already done in CsvImporter, but it's here so the testing class gets it too.
+        if pd.isna(self.value) and self.missing_value == Nans.NOT_MISSING:
+            self.missing_value = Nans.NOT_APPLICABLE.value if extra_checks else Nans.OTHER.value
+
+        if pd.isna(self.stderr) and self.missing_stderr == Nans.NOT_MISSING:
+            self.missing_stderr = Nans.NOT_APPLICABLE.value if extra_checks else Nans.OTHER.value
+
+        if pd.isna(self.sample_size) and self.missing_sample_size == Nans.NOT_MISSING:
+            self.missing_sample_size = Nans.NOT_APPLICABLE.value if extra_checks else Nans.OTHER.value
+
+        return self
+
+
+def covidcast_rows_from_args(sanitize_fields: bool = False, test_mode: bool = True, **kwargs: Dict[str, Iterable]) -> List[CovidcastTestRow]:
+    """A convenience constructor for test rows.
+
+    Example:
+    covidcast_rows_from_args(value=[1, 2, 3], time_value=[1, 2, 3]) will yield
+    [CovidcastTestRow.make_default_row(value=1, time_value=1), CovidcastTestRow.make_default_row(value=2, time_value=2), CovidcastTestRow.make_default_row(value=3, time_value=3)]
+    with all the defaults from CovidcastTestRow.
+    """
+    # If any iterables were passed instead of lists, convert them to lists.
+    kwargs = {key: list(value) for key, value in kwargs.items()}
+    # All the arg values must be lists of the same length.
+    assert len(set(len(lst) for lst in kwargs.values())) == 1
+
+    if sanitize_fields:
+        return [CovidcastTestRow.make_default_row(**_kwargs)._sanitize_fields(extra_checks=test_mode) for _kwargs in transpose_dict(kwargs)]
+    else:
+        return [CovidcastTestRow.make_default_row(**_kwargs) for _kwargs in transpose_dict(kwargs)]
+
+
+def covidcast_rows_from_records(records: Iterable[dict], sanity_check: bool = False) -> List[CovidcastTestRow]:
+    """A convenience constructor.
+
+    Default is different from from_args, because from_records is usually called on faux-API returns in tests,
+    where we don't want any values getting default filled in.
+
+    You can use csv.DictReader before this to read a CSV file.
+    """
+    records = list(records)
+    return [CovidcastTestRow.make_default_row(**record) if not sanity_check else CovidcastTestRow.make_default_row(**record)._sanitize_fields() for record in records]
+
+
+def covidcast_rows_as_dicts(rows: Iterable[CovidcastTestRow], ignore_fields: Optional[List[str]] = None) -> List[dict]:
+    return [row.as_dict(ignore_fields=ignore_fields) for row in rows]
+
+
+def covidcast_rows_as_dataframe(rows: Iterable[CovidcastTestRow], ignore_fields: Optional[List[str]] = None) -> pd.DataFrame:
+    if ignore_fields is None:
+        ignore_fields = []
+
+    columns = [field.name for field in fields(CovidcastTestRow) if field.name not in ignore_fields]
+
+    if rows:
+        df = pd.concat([row.as_dataframe(ignore_fields=ignore_fields) for row in rows], ignore_index=True)
+        return df[columns]
+    else:
+        return pd.DataFrame(columns=columns)
+
+
+def covidcast_rows_as_api_row_df(rows: Iterable[CovidcastTestRow]) -> pd.DataFrame:
+    return covidcast_rows_as_dataframe(rows, ignore_fields=CovidcastTestRow._api_row_ignore_fields)
+
+
+def covidcast_rows_as_api_compatibility_row_df(rows: Iterable[CovidcastTestRow]) -> pd.DataFrame:
+    return covidcast_rows_as_dataframe(rows, ignore_fields=CovidcastTestRow._api_row_compatibility_ignore_fields)
+
+
+def covidcast_rows_as_db_row_df(rows: Iterable[CovidcastTestRow]) -> pd.DataFrame:
+    return covidcast_rows_as_dataframe(rows, ignore_fields=CovidcastTestRow._db_row_ignore_fields)
+
+
+def transpose_dict(d: Dict[Any, List[Any]]) -> List[Dict[Any, Any]]:
+    """Given a dictionary whose values are lists of the same length, turn it into a list of dictionaries whose values are the individual list entries.
+
+    Example:
+    >>> transpose_dict(dict([["a", [2, 4, 6]], ["b", [3, 5, 7]], ["c", [10, 20, 30]]]))
+    [{"a": 2, "b": 3, "c": 10}, {"a": 4, "b": 5, "c": 20}, {"a": 6, "b": 7, "c": 30}]
+    """
+    return [dict(zip(d.keys(), values)) for values in zip(*d.values())]
+
+
+def assert_frame_equal_no_order(df1: pd.DataFrame, df2: pd.DataFrame, index: List[str], **kwargs: Any) -> None:
+    """Assert that two DataFrames are equal, ignoring the order of rows."""
+    # Remove any existing index. If it wasn't named, drop it. Set a new index and sort it.
+    df1 = df1.reset_index().drop(columns="index").set_index(index).sort_index()
+    df2 = df2.reset_index().drop(columns="index").set_index(index).sort_index()
+    pd.testing.assert_frame_equal(df1, df2, **kwargs)
+
+
 class CovidcastBase(unittest.TestCase):
     def setUp(self):
         # use the local test instance of the database
@@ -22,45 +161,29 @@ def setUp(self):
         self.localSetUp()
         self._db._connection.commit()
 
-    def localSetUp(self):
-        # stub; override in subclasses to perform custom setup. 
-        # runs after tables have been truncated but before database changes have been committed
-        pass
-
     def tearDown(self):
         # close and destroy conenction to the database
+        self.localTearDown()
         self._db.disconnect(False)
         del self._db
 
-    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 localSetUp(self):
+        # stub; override in subclasses to perform custom setup. 
+        # runs after tables have been truncated but before database changes have been committed
+        pass
+
+    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[CovidcastTestRow]):
         # 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: CovidcastTestRow, **kwargs):
         ret = {
             'data_source': row.source,
             'signals': row.signal,
@@ -70,14 +193,4 @@ def params_from_row(self, row, **kwargs):
             'geo_value': row.geo_value,
         }
         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
-
+        return ret
\ No newline at end of file
diff --git a/src/server/_query.py b/src/server/_query.py
index 2b716e954..267a78eb1 100644
--- a/src/server/_query.py
+++ b/src/server/_query.py
@@ -9,8 +9,9 @@
     Sequence,
     Tuple,
     Union,
-    cast
+    cast,
 )
+from flask import Response
 
 from flask import request
 from sqlalchemy import text
@@ -54,7 +55,7 @@ def filter_values(
     param_key: str,
     params: Dict[str, Any],
     formatter=lambda x: x,
-):
+) -> str:
     if not values:
         return "FALSE"
     # builds a SQL expression to filter strings (ex: locations)
@@ -69,7 +70,7 @@ def filter_strings(
     values: Optional[Sequence[str]],
     param_key: str,
     params: Dict[str, Any],
-):
+) -> str:
     return filter_values(field, values, param_key, params)
 
 
@@ -78,7 +79,7 @@ def filter_integers(
     values: Optional[Sequence[IntRange]],
     param_key: str,
     params: Dict[str, Any],
-):
+) -> str:
     return filter_values(field, values, param_key, params)
 
 
@@ -87,7 +88,7 @@ def filter_dates(
     values: Optional[TimeValues],
     param_key: str,
     params: Dict[str, Any],
-):
+) -> str:
     ranges = time_values_to_ranges(values)
     return filter_values(field, ranges, param_key, params, date_string)
 
@@ -199,7 +200,7 @@ def parse_row(
     fields_string: Optional[Sequence[str]] = None,
     fields_int: Optional[Sequence[str]] = None,
     fields_float: Optional[Sequence[str]] = None,
-):
+) -> Dict[str, Any]:
     keys = set(row.keys())
     parsed = dict()
     if fields_string:
@@ -235,7 +236,7 @@ def limit_query(query: str, limit: int) -> str:
     return full_query
 
 
-def run_query(p: APrinter, query_tuple: Tuple[str, Dict[str, Any]]):
+def run_query(p: APrinter, query_tuple: Tuple[str, Dict[str, Any]]) -> Iterable[Row]:
     query, params = query_tuple
     # limit rows + 1 for detecting whether we would have more
     full_query = text(limit_query(query, p.remaining_rows + 1))
@@ -255,7 +256,7 @@ def execute_queries(
     fields_int: Sequence[str],
     fields_float: Sequence[str],
     transform: Callable[[Dict[str, Any], Row], Dict[str, Any]] = _identity_transform,
-):
+) -> Response:
     """
     execute the given queries and return the response to send them
     """
@@ -314,14 +315,14 @@ def execute_query(
     fields_int: Sequence[str],
     fields_float: Sequence[str],
     transform: Callable[[Dict[str, Any], Row], Dict[str, Any]] = _identity_transform,
-):
+) -> Response:
     """
     execute the given query and return the response to send it
     """
     return execute_queries([(query, params)], fields_string, fields_int, fields_float, transform)
 
 
-def _join_l(value: Union[str, List[str]]):
+def _join_l(value: Union[str, List[str]]) -> str:
     return ", ".join(value) if isinstance(value, (list, tuple)) else value
 
 
diff --git a/src/server/endpoints/covidcast_utils/model.py b/src/server/endpoints/covidcast_utils/model.py
index abab0033b..4cc78888e 100644
--- a/src/server/endpoints/covidcast_utils/model.py
+++ b/src/server/endpoints/covidcast_utils/model.py
@@ -202,7 +202,7 @@ def _load_data_sources():
 
 
 data_sources, data_sources_df = _load_data_sources()
-data_source_by_id = {d.source: d for d in data_sources}
+data_sources_by_id = {d.source: d for d in data_sources}
 
 
 def _load_data_signals(sources: List[DataSource]):
@@ -231,7 +231,7 @@ def _load_data_signals(sources: List[DataSource]):
 data_signals_by_key = {d.key: d for d in data_signals}
 # also add the resolved signal version to the signal lookup
 for d in data_signals:
-    source = data_source_by_id.get(d.source)
+    source = data_sources_by_id.get(d.source)
     if source and source.uses_db_alias:
         data_signals_by_key[(source.db_source, d.signal)] = d
 
@@ -261,7 +261,7 @@ def create_source_signal_alias_mapper(source_signals: List[SourceSignalSet]) ->
     alias_to_data_sources: Dict[str, List[DataSource]] = {}
     transformed_sets: List[SourceSignalSet] = []
     for ssset in source_signals:
-        source = data_source_by_id.get(ssset.source)
+        source = data_sources_by_id.get(ssset.source)
         if not source or not source.uses_db_alias:
             transformed_sets.append(ssset)
             continue
diff --git a/tests/acquisition/covidcast/test_covidcast_row.py b/tests/acquisition/covidcast/test_covidcast_row.py
new file mode 100644
index 000000000..9462fd4ed
--- /dev/null
+++ b/tests/acquisition/covidcast/test_covidcast_row.py
@@ -0,0 +1,92 @@
+import unittest
+
+from pandas import DataFrame
+from pandas.testing import assert_frame_equal
+
+from delphi_utils.nancodes import Nans
+from delphi.epidata.acquisition.covidcast.covidcast_row import CovidcastRow, set_df_dtypes
+from delphi.epidata.acquisition.covidcast.test_utils import (
+    CovidcastTestRow,
+    covidcast_rows_as_api_compatibility_row_df, 
+    covidcast_rows_as_api_row_df, 
+    covidcast_rows_from_args, 
+    transpose_dict, 
+)
+
+# py3tester coverage target (equivalent to `import *`)
+__test_target__ = 'delphi.epidata.acquisition.covidcast.covidcast_row'
+
+
+class TestCovidcastRows(unittest.TestCase):
+    expected_df = set_df_dtypes(DataFrame({
+        "source": ["src"] * 10,
+        "signal": ["sig_base"] * 5 + ["sig_other"] * 5,
+        "time_type": ["day"] * 10,
+        "geo_type": ["county"] * 10,
+        "time_value": [2021_05_01 + i for i in range(5)] * 2,
+        "geo_value": ["01234"] * 10,
+        "value": range(10),
+        "stderr": [10.0] * 10,
+        "sample_size": [10.0] * 10,
+        "missing_value": [Nans.NOT_MISSING] * 10,
+        "missing_stderr": [Nans.NOT_MISSING] * 10,
+        "missing_sample_size": [Nans.NOT_MISSING] * 10,
+        "issue": [2021_05_01 + i for i in range(5)] * 2,
+        "lag": [0] * 10,
+        "direction": [None] * 10
+    }), CovidcastRow._pandas_dtypes)
+
+    def test_transpose_dict(self):
+        assert transpose_dict(
+            {
+                "a": [2, 4, 6],
+                "b": [3, 5, 7],
+                "c": [10, 20, 30]
+            }
+        ) == [
+            {"a": 2, "b": 3, "c": 10}, 
+            {"a": 4, "b": 5, "c": 20}, 
+            {"a": 6, "b": 7, "c": 30}
+        ]
+
+
+    def test_CovidcastRow(self):
+        df = CovidcastTestRow.make_default_row(
+            signal="sig_base",
+            value=0.0,
+            time_value=2021_05_01,
+            issue=2021_05_01,
+        ).as_api_row_df()
+        expected_df = self.expected_df.iloc[0:1]
+        assert_frame_equal(df, expected_df)
+
+        df = CovidcastTestRow.make_default_row(
+            signal="sig_base",
+            value=0.0,
+            time_value=2021_05_01,
+            issue=2021_05_01,
+        ).as_api_compatibility_row_df()
+        expected_df = self.expected_df.iloc[0:1][df.columns]
+        assert_frame_equal(df, expected_df)
+
+
+    def test_covidcast_rows(self):
+        covidcast_rows = covidcast_rows_from_args(
+            signal=["sig_base"] * 5 + ["sig_other"] * 5,
+            time_value=[2021_05_01 + i for i in range(5)] * 2,
+            value=list(range(10)),
+            sanitize_fields = True
+        )
+        df = covidcast_rows_as_api_row_df(covidcast_rows)
+        expected_df = self.expected_df
+        assert_frame_equal(df, expected_df)
+
+        covidcast_rows = covidcast_rows_from_args(
+            signal=["sig_base"] * 5 + ["sig_other"] * 5,
+            time_value=[2021_05_01 + i for i in range(5)] * 2,
+            value=list(range(10)),
+            sanitize_fields = True
+        )
+        df = covidcast_rows_as_api_compatibility_row_df(covidcast_rows)
+        expected_df = self.expected_df[df.columns]
+        assert_frame_equal(df, expected_df)
diff --git a/tests/server/test_pandas.py b/tests/server/test_pandas.py
index 083162a47..12a9c18cd 100644
--- a/tests/server/test_pandas.py
+++ b/tests/server/test_pandas.py
@@ -9,7 +9,6 @@
 from delphi.epidata.server._pandas import as_pandas
 from delphi.epidata.server._config import MAX_RESULTS
 
-
 # py3tester coverage target
 __test_target__ = "delphi.epidata.server._pandas"