fix date error in the schedule

.github/workflows/test.yml

@@ -33,6 +33,10 @@ jobs:
 
       - run: make init
 
+      - name: Unit Tests
+        run: |
+          make test-unit
+
       - name: Integration Tests
         run: |
           make test-integration

Makefile

@@ -27,7 +27,10 @@ compose-down::
 	docker compose down  --rmi 'all'
 
 
-test:: test-integration test-acceptance
+test:: test-unit test-integration test-acceptance
+
+test-unit::
+	python -m pytest tests/unit
 
 test-integration::
 	python -m pytest tests/integration

dags/collection_config.py

@@ -7,15 +7,17 @@
 defined here rather than being hardcoded into the DAG generation logic.
 """
 
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from typing import Optional
 
 from dateutil.rrule import rrulestr
 from pydantic import BaseModel
 
 # fixed anchor for expanding schedule_rrule recurrences from; only matters for cadences whose
-# phase depends on a start date (e.g. fortnightly) - harmless for DAILY/MONTHLY;BYDAY rules
-RRULE_SERIES_START = datetime(2025, 1, 1)
+# phase depends on a start date (e.g. fortnightly) - harmless for DAILY/MONTHLY;BYDAY rules.
+# Must be timezone-aware since Airflow's logical_date is UTC-aware, and dateutil can't compare
+# naive and aware datetimes.
+RRULE_SERIES_START = datetime(2025, 1, 1, tzinfo=timezone.utc)
 
 
 class CollectionDagConfig(BaseModel):

tests/acceptance/test_dags_render.py

@@ -9,6 +9,12 @@ def test_dag_rendering():
     dag_bag = DagBag(dag_folder=DAG_FOLDER, include_examples=False)
     assert len(dag_bag.dags) > 0, "No DAGs found in DAG bag!"
 
+    # DagBag swallows per-file import errors rather than raising, so a broken DAG module
+    # wouldn't otherwise fail this test - it would just silently be missing from dag_bag.dags
+    if dag_bag.import_errors:
+        errors = "\n".join(f"{path}: {error}" for path, error in dag_bag.import_errors.items())
+        pytest.fail(f"DAG files failed to import:\n{errors}")
+
     # Collect any DAGs that failed to load
     failed_dags = []
 

tests/conftest.py

@@ -6,44 +6,45 @@
 import pytest
 from moto import mock_aws
 
+CONFIG_PATH = Path("dags/config.json")
+CONFIG_BACKUP_PATH = CONFIG_PATH.with_suffix(".json.backup")
 
-@pytest.fixture(scope="session", autouse=True)
-def create_test_config():
-    """
-    Create a test config.json file for DAG tests.
+# Matches the output of: bin/generate_dag_config.py --env development
+TEST_CONFIG = {"env": "development", "schedule": "0 0 * * *", "max_active_tasks": 50, "collection_selection": "explicit", "collections": ["ancient-woodland", "organisation"]}
+
+# Several DAG modules call get_secrets("emr_execution_role", env) at import time, which hits real
+# AWS Secrets Manager unless overridden. aws_secrets_manager.get_secret_emr_compatible supports a
+# SECRET_<name> environment variable escape hatch specifically for this - see its docstring.
+SECRET_ENV_VAR = "SECRET__DEVELOPMENT_PD_BATCH_DEPLOYMENT_VARIABLES_SECRET"
+TEST_SECRET_VALUE = json.dumps({"emr_execution_role": "arn:aws:iam::000000000000:role/test-emr-execution-role"})
 
-    This fixture runs automatically before any tests and creates the config file
-    that DAGs expect to find in dags/config.json. If a config already exists,
-    it will be backed up and restored after tests complete.
 
-    The test configuration matches the output of:
-        bin/generate_dag_config.py --env development
+def pytest_configure(config):
+    """
+    Create a test config.json file, and fake the AWS secret DAGs read at import time.
 
-    To customize the test config, modify the test_config dictionary below.
+    Some tests import DAG modules directly at module scope (e.g. `from dags.digital_land_builder
+    import dag`), which happens during collection - before any pytest fixture would run. This
+    hook runs earlier still, so both are guaranteed to be in place before anything is collected.
+    If a config already exists, it's backed up and restored in pytest_unconfigure.
     """
-    config_path = Path("dags/config.json")
+    if CONFIG_PATH.exists():
+        CONFIG_PATH.rename(CONFIG_BACKUP_PATH)
 
-    # Create test configuration (matches development environment)
-    test_config = {"env": "development", "schedule": "0 0 * * *", "max_active_tasks": 50, "collection_selection": "explicit", "collections": ["ancient-woodland", "organisation"]}
+    with open(CONFIG_PATH, "w") as f:
+        json.dump(TEST_CONFIG, f, indent=4)
 
-    # Check if config already exists (backup if it does)
-    backup_path = None
-    if config_path.exists():
-        backup_path = config_path.with_suffix(".json.backup")
-        config_path.rename(backup_path)
+    os.environ[SECRET_ENV_VAR] = TEST_SECRET_VALUE
 
-    # Write test config
-    with open(config_path, "w") as f:
-        json.dump(test_config, f, indent=4)
 
-    yield config_path
+def pytest_unconfigure(config):
+    if CONFIG_PATH.exists():
+        CONFIG_PATH.unlink()
 
-    # Cleanup: remove test config and restore backup if it existed
-    if config_path.exists():
-        config_path.unlink()
+    if CONFIG_BACKUP_PATH.exists():
+        CONFIG_BACKUP_PATH.rename(CONFIG_PATH)
 
-    if backup_path and backup_path.exists():
-        backup_path.rename(config_path)
+    os.environ.pop(SECRET_ENV_VAR, None)
 
 
 @pytest.fixture(scope="function")

tests/unit/test_collection_config.py

@@ -0,0 +1,51 @@
+import pendulum
+import pytest
+
+from dags.collection_config import DEFAULT_COLLECTION_CONFIG, collection_schedule_matches, get_collection_dag_config
+
+
+def test_get_collection_dag_config_returns_default_for_unknown_collection():
+    assert get_collection_dag_config("central-activities-zone") is DEFAULT_COLLECTION_CONFIG
+
+
+def test_get_collection_dag_config_returns_override_for_title_boundary():
+    config = get_collection_dag_config("title-boundary")
+    assert config.schedule_rrule == "FREQ=MONTHLY;BYDAY=1MO"
+    assert config.transform_batch_size == 100
+    assert config.max_executors == 50
+
+
+def test_default_collection_schedule_is_daily():
+    assert DEFAULT_COLLECTION_CONFIG.schedule_rrule == "FREQ=DAILY"
+
+
+def test_collection_schedule_matches_accepts_timezone_aware_logical_date():
+    """Regression test: Airflow's logical_date is UTC-aware. dateutil raises "can't compare
+    offset-naive and offset-aware datetimes" if the rrule's own anchor isn't aware too."""
+    logical_date = pendulum.datetime(2026, 8, 3, tz="UTC")
+    collection_schedule_matches("title-boundary", logical_date)  # should not raise
+
+
+@pytest.mark.parametrize(
+    "logical_date,expected",
+    [
+        (pendulum.datetime(2026, 8, 3, tz="UTC"), True),  # first Monday of August 2026
+        (pendulum.datetime(2026, 8, 2, tz="UTC"), False),  # first Sunday - not a match
+        (pendulum.datetime(2026, 8, 10, tz="UTC"), False),  # second Monday - not a match
+        (pendulum.datetime(2026, 9, 7, tz="UTC"), True),  # first Monday of September 2026
+    ],
+)
+def test_collection_schedule_matches_title_boundary_first_monday(logical_date, expected):
+    assert collection_schedule_matches("title-boundary", logical_date) is expected
+
+
+@pytest.mark.parametrize(
+    "logical_date",
+    [
+        pendulum.datetime(2026, 8, 1, tz="UTC"),
+        pendulum.datetime(2026, 8, 15, tz="UTC"),
+        pendulum.datetime(2026, 12, 25, tz="UTC"),
+    ],
+)
+def test_collection_schedule_matches_default_daily_collection_always_matches(logical_date):
+    assert collection_schedule_matches("central-activities-zone", logical_date) is True