1baf98eb40
- CCLF: fixed-width parser using IP-derived field positions (cclf_layout.py), ZIP extraction, file discovery by CMS naming convention, DuckDB loading, optional pipeline execution to produce input_layer tables - BCDA: flatten ndjson → Parquet, load into bcda schema in DuckDB - Seeds: auto-discover CSV/Excel/Parquet files, load into reference_data schema - Unified staging: orchestrate all three loaders with optional Iceberg promotion - CLI: stack load cclf/bcda/seed with full options (--path, --database, etc.) - 57 tests covering parsers, loaders, CLI commands, and staging pipeline fixes #5 fixes #6 fixes #7 fixes #8
33 lines
1.0 KiB
Python
33 lines
1.0 KiB
Python
"""Tests for BCDA data loading."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from aco.load.bcda import _find_latest_export
|
|
|
|
|
|
class TestFindLatestExport:
|
|
def test_no_exports_dir_raises(self, tmp_path: Path) -> None:
|
|
with pytest.raises(FileNotFoundError, match="No exports directory"):
|
|
_find_latest_export(tmp_path)
|
|
|
|
def test_empty_exports_dir_raises(self, tmp_path: Path) -> None:
|
|
(tmp_path / "exports").mkdir()
|
|
with pytest.raises(FileNotFoundError, match="No export directories"):
|
|
_find_latest_export(tmp_path)
|
|
|
|
def test_finds_latest(self, tmp_path: Path) -> None:
|
|
import os
|
|
|
|
exports = tmp_path / "exports"
|
|
exports.mkdir()
|
|
(exports / "old_job").mkdir()
|
|
(exports / "new_job").mkdir()
|
|
# Set old_job mtime to past so new_job is clearly newer
|
|
os.utime(exports / "old_job", (1000000, 1000000))
|
|
result = _find_latest_export(tmp_path)
|
|
assert result.name == "new_job"
|