Separate Union City Model Runs as Testing Fixture (#67)

* Feat: Make Union City model run a pytest Fixture to enable re-use
* Added OMX compare helper functions to reuse in testing
* Updated file search for OMX files to use glob; refactored OMX opening/shutting
* pep8/docstrings

Closes #66 

Co-authored-by: Kevin Bragg <[email protected]>

tests/test_docs.py

@@ -14,10 +14,12 @@ def test_docs_build():
 
     # Build the docs
     try:
-        subprocess.run(["mkdocs", "build"], check=True, cwd=base_dir,capture_output=True)
+        subprocess.run(
+            ["mkdocs", "build"], check=True, cwd=base_dir, capture_output=True
+        )
     except subprocess.CalledProcessError as e:
-        msg = e.stderr.decode('utf-8')
+        msg = e.stderr.decode("utf-8")
         pytest.fail(f"Documentation Failed to Build.\n {msg}")
 
     # Check that the docs were built successfully
-    assert os.path.exists(os.path.join(base_dir, "site", "index.html"))
+    assert os.path.exists(os.path.join(base_dir, "site", "index.html"))

tests/test_union_city.py

@@ -1,6 +1,12 @@
+import glob
 import os
-from unittest.mock import MagicMock
 import sys
+
+from unittest.mock import MagicMock
+from typing import Collection, Union
+
+import openmatrix as omx
+import pandas as pd
 import pytest
 
 
@@ -53,55 +59,110 @@ def test_example_download():
     assert not (os.path.exists(os.path.join(example_dir, name, "test_data.zip")))
 
 
+def diff_omx(ref_omx: str, run_omx: str) -> Collection[Collection[str]]:
+    """
+    Compare two OMX files, return missing and different matrices from reference.
+
+    Args:
+        ref_omx: reference OMX file
+        run_omx: run OMX file
+    """
+    _ref_f = omx.open_file(ref_omx, "r")
+    _run_f = omx.open_file(run_omx, "r")
+    _ref_matrix_names = _ref_f.list_matrices()
+    _run_matrix_names = _run_f.list_matrices()
+
+    missing_matrices = [f for f in _ref_matrix_names if f not in _run_matrix_names]
+    different_matrices = []
+    for m_key in _ref_matrix_names:
+        _ref_matrix = _ref_f[m_key].read()
+        _run_matrix = _run_f[m_key].read()
+        if not (_ref_matrix == _run_matrix).all():
+            different_matrices.append(m_key)
+
+    _ref_f.close()
+    _run_f.close()
+    return missing_matrices, different_matrices
+
+
+@pytest.fixture(scope="module")
 @pytest.mark.skipci
-def test_highway():
+def union_city():
+    """Union City model run testing fixture."""
     from tm2py.controller import RunController
     from tm2py.examples import get_example
-    import openmatrix as _omx
 
     union_city_root = os.path.join(os.getcwd(), _EXAMPLES_DIR, "UnionCity")
     get_example(
         example_name="UnionCity", example_subdir=_EXAMPLES_DIR, root_dir=os.getcwd()
     )
     controller = RunController(
         [
-            os.path.join(_EXAMPLES_DIR, r"scenario_config.toml"),
-            os.path.join(_EXAMPLES_DIR, r"model_config.toml"),
+            os.path.join(_EXAMPLES_DIR, "scenario_config.toml"),
+            os.path.join(_EXAMPLES_DIR, "model_config.toml"),
         ],
-        run_dir=union_city_root
+        run_dir=union_city_root,
     )
     controller.run()
+    return controller
+
+
+@pytest.mark.skipci
+def test_highway_skims(union_city):
+    """Test that the OMX highway skims match the reference."""
+    run_dir = union_city.run_dir
+
+    ref_dir_hwy_skims = os.path.join(run_dir, "ref_skim_matrices", "highway")
+    ref_skim_files = glob.glob(os.path.join(ref_dir_hwy_skims, "*.omx"))
+
+    run_dir_hwy_skims = os.path.join(run_dir, "skim_matrices", "highway")
+    run_skim_files = glob.glob(os.path.join(run_dir_hwy_skims, "*.omx"))
+
+    # check that the expected files are all there
+    ref_skim_names = [os.path.basename(f) for f in ref_skim_files]
+    run_skim_names = [os.path.basename(f) for f in run_skim_files]
 
-    root = os.path.join(controller.run_dir, r"skim_matrices\highway")
-    ref_root = os.path.join(controller.run_dir, r"ref_skim_matrices\highway")
-    open_files = []
-    file_names = [name for name in os.listdir(root) if name.endswith(".omx")]
+    assert set(ref_skim_names) == set(
+        run_skim_names
+    ), f"Skim matrix names do not match expected\
+        reference. \n Expected: {ref_skim_names}\n Actual: {run_skim_names}"
+
+    missing_skims = []
     different_skims = []
-    try:
-        for name in file_names:
-            skims = _omx.open_file(os.path.join(root, name))
-            open_files.append(skims)
-            ref_skims = _omx.open_file(os.path.join(ref_root, name))
-            open_files.append(ref_skims)
-            for key in skims.list_matrices():
-                data = skims[key].read()
-                ref_data = ref_skims[key].read()
-                if not (data == ref_data).all():
-                    different_skims.append(key)
-    finally:
-        for f in open_files:
-            f.close()
-    assert (
-        len(different_skims) == 0
-    ), f"there are {len(different_skims)} different skims: {','.join(different_skims)}"
-
-    count_different_lines = 0
-    with open(os.path.join(root, "HWYSKIM_MAZMAZ_DA.csv")) as data:
-        with open(os.path.join(ref_root, "HWYSKIM_MAZMAZ_DA.csv")) as ref_data:
-            for line in data:
-                ref_line = next(ref_data)
-                if ref_line != line:
-                    count_different_lines += 1
-    assert (
-        count_different_lines == 0
-    ), f"HWYSKIM_MAZMAZ_DA.csv differs on {count_different_lines} lines"
+
+    for ref_skim_f, run_skim_f in zip(ref_skim_files, run_skim_files):
+        _missing_ms, _diff_ms = diff_omx(ref_skim_f, run_skim_f)
+        missing_skims.extend([ref_skim_f + _m for _m in _missing_ms])
+        different_skims.extend(ref_skim_f + _m for _m in _diff_ms)
+
+    assert len(missing_skims) == 0, f"Missing skims: {missing_skims}"
+    assert len(different_skims) == 0, f"Different skims: {different_skims}"
+
+
+def assert_csv_equal(ref_csv: str, run_csv: str):
+    """Compare two csv files, return results of pd.testing.assert_frame_equal().
+
+    Args:
+        ref_csv (str): Reference CSV location
+        run_csv (str): Model run CSV location
+
+    Returns:
+        Results of pd.testing.assert_frame_equal()
+    """
+    ref_df = pd.read_csv(ref_csv)
+    run_df = pd.read_csv(run_csv)
+    return pd.testing.assert_frame_equal(ref_df, run_df)
+
+
+@pytest.mark.skipci
+def test_maz_da_skims(union_city):
+    """Test that the DA MAZ skims match the reference."""
+    run_dir = union_city.run_dir
+
+    ref_dir_hwy_skims = os.path.join(run_dir, "ref_skim_matrices", "highway")
+    run_dir_hwy_skims = os.path.join(run_dir, "skim_matrices", "highway")
+
+    ref_csv = os.path.join(ref_dir_hwy_skims, "HWYSKIM_MAZMAZ_DA.csv")
+    run_csv = os.path.join(run_dir_hwy_skims, "HWYSKIM_MAZMAZ_DA.csv")
+
+    return assert_csv_equal(ref_csv, run_csv)