stramline, add tests for path_format check

tests/integrity_test.py

@@ -7,6 +7,7 @@ from pathlib import Path
 import os
 import pandas as pd
 import datetime as dt
+import pytest
 
 
 def test_list_ensembles(tmp_path: Path) -> None:
@@ -72,8 +73,6 @@ def test_has_valid_time() -> None:
                 '2026-03-26 12:55:18.229966', '2024-03-26 12:55:18.229966'] # this is invalid...
     record_G = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf2", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                 '2026-03-26 12:55:18.229966', str(dt.datetime.now() + dt.timedelta(days=2, hours=3, minutes=5, seconds=30))] # created and updated later
-
-    data = [record_A, record_B, record_C, record_D, record_E]
     cols = ["name",
             "ensemble",
             "code",
@@ -83,19 +82,12 @@ def test_has_valid_time() -> None:
             "parameter_file",
             "created_at",
             "updated_at"]
+    data = [record_A, record_B, record_C, record_D, record_E]
+
     df = pd.DataFrame(data,columns=cols)
     for _, result in df.iterrows():
         assert integ.has_valid_times(result)
     data = [record_F, record_G]
-    cols = ["name",
-            "ensemble",
-            "code",
-            "path",
-            "project",
-            "parameters",
-            "parameter_file",
-            "created_at",
-            "updated_at"]
     df = pd.DataFrame(data,columns=cols)
     for _, result in df.iterrows():
         assert not integ.has_valid_times(result)
@@ -119,7 +111,6 @@ def test_are_keys_unique(tmp_path: Path) -> None:
     record_G = ["f_A", "ensA", "sfcf", "archive/SF_A/f_A/Project_A.json.gz::asdfasdfasdf2", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
                 '2026-03-26 12:55:18.229966', str(dt.datetime.now() + dt.timedelta(days=2, hours=3, minutes=5, seconds=30))] # created and updated later
 
-    data = [record_A, record_B, record_C, record_D, record_E, record_F]
     cols = ["name",
             "ensemble",
             "code",
@@ -129,6 +120,8 @@ def test_are_keys_unique(tmp_path: Path) -> None:
             "parameter_file",
             "created_at",
             "updated_at"]
+
+    data = [record_A, record_B, record_C, record_D, record_E, record_F]
     df = pd.DataFrame(data,columns=cols)
     conn = sqlite3.connect(db)
     df.to_sql('backlogs', conn)
@@ -137,6 +130,32 @@ def test_are_keys_unique(tmp_path: Path) -> None:
 
     db = tmp_path / 'test_fail.db'
     data = [record_A, record_B, record_C, record_D, record_E, record_F, record_G]
+
+    df = pd.DataFrame(data,columns=cols)
+    conn = sqlite3.connect(db)
+    df.to_sql('backlogs', conn)
+    conn.close()
+    assert not integ.are_keys_unique(db, 'backlogs', 'path')
+
+
+def test_check_path_format() -> None:
+    record_A = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_A.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
+            '2025-03-26 12:55:18.229966', '2025-03-26 12:55:18.229966'] # only created
+    record_B = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_B.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
+                '2025-03-26 12:55:18.229966', '2025-04-26 12:55:18.229966'] # created and updated
+    record_C = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_A.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
+                '2026-03-26 12:55:18.229966', '2026-04-14 12:55:18.229966'] # created and updated later
+    record_D = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_B.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
+                '2026-03-26 12:55:18.229966', '2026-03-27 12:55:18.229966']
+    record_E = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_A.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
+                '2024-03-26 12:55:18.229966', '2024-03-26 12:55:18.229966'] # only created, earlier
+    record_F = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_B.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdf", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
+                '2026-03-26 12:55:18.229966', '2024-03-26 12:55:18.229966'] # this is invalid...
+    record_G = ["f_A", "ensA", "sfcf", "archive/ensA/f_A/Project_A.json.gz::asdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfasdfas", "SF_A", '{"par_A": 5.0, "par_B": 5.0}', "projects/SF_A/input.in",
+                '2026-03-26 12:55:18.229966', str(dt.datetime.now() + dt.timedelta(days=2, hours=3, minutes=5, seconds=30))] # created and updated later
+
+    projects = ['Project_A', 'Project_B']
+    ensembles = ['ensA']
     cols = ["name",
             "ensemble",
             "code",
@@ -146,8 +165,25 @@ def test_are_keys_unique(tmp_path: Path) -> None:
             "parameter_file",
             "created_at",
             "updated_at"]
+
+    data = [record_A, record_B, record_C, record_D, record_E, record_F]
     df = pd.DataFrame(data,columns=cols)
-    conn = sqlite3.connect(db)
+    for _, result in df.iterrows():
-    df.to_sql('backlogs', conn)
+        integ.check_path_format(result, ensembles, projects)
-    conn.close()
+
-    assert not integ.are_keys_unique(db, 'backlogs', 'path')
+    projects = ['Project_A', 'Project_B']
+    ensembles = ['ensB']
+    for _, result in df.iterrows():
+        with pytest.raises(ValueError):
+            integ.check_path_format(result, ensembles, projects)
+
+    projects = ['Project_A', 'Project_B']
+    ensembles = ['ensA', 'ensB']
+    for _, result in df.iterrows():
+        integ.check_path_format(result, ensembles, projects)
+
+    data = [record_G]
+    df = pd.DataFrame(data,columns=cols)
+    for _, result in df.iterrows():
+        with pytest.raises(ValueError):
+            integ.check_path_format(result, ensembles, projects)