audit-labs/audit-tools

 1from types import SimpleNamespace
 2
 3import pandas as pd
 4import pytest
 5
 6from sampling.sampling_tool.cli import run
 7from sampling.sampling_tool.io import AuditSamplingError
 8
 9
10def _options(input_path, out_path, **kwargs):
11    values = {
12        "input": str(input_path),
13        "sheet": None,
14        "id_column": "ID",
15        "method": "validate-only",
16        "sample_size": None,
17        "stratify_column": None,
18        "strata_counts": None,
19        "strata_proportions": None,
20        "seed": None,
21        "out": str(out_path),
22        "exclude_blank_id": False,
23        "dedupe_id": "fail",
24        "filters": {},
25        "allow_shortfall": False,
26    }
27    values.update(kwargs)
28    return SimpleNamespace(**values)
29
30
31def test_duplicate_ids_fail_by_default_and_write_duplicate_file(tmp_path):
32    source = tmp_path / "population.csv"
33    pd.DataFrame({"ID": ["A", "A", "B"], "Status": ["Closed"] * 3}).to_csv(
34        source, index=False
35    )
36
37    options = _options(source, tmp_path / "out")
38    with pytest.raises(AuditSamplingError):
39        run(options)
40
41    run_dir = next((tmp_path / "out").glob("sample_*"))
42    duplicates = pd.read_csv(run_dir / "duplicate_ids.csv")
43    assert duplicates["ID"].tolist() == ["A", "A"]
44
45
46def test_blank_ids_are_excluded_when_requested(tmp_path):
47    source = tmp_path / "population.csv"
48    pd.DataFrame({"ID": ["A", "", "B"], "Status": ["Closed"] * 3}).to_csv(
49        source, index=False
50    )
51
52    run_dir = run(_options(source, tmp_path / "out", exclude_blank_id=True))
53
54    validated = pd.read_csv(run_dir / "population_validated.csv")
55    excluded = pd.read_csv(run_dir / "excluded_rows.csv")
56    assert len(validated) == 2
57    assert excluded["_exclusion_reason"].tolist() == ["Blank ID"]
58
59
60def test_filters_reduce_population_and_write_excluded_rows(tmp_path):
61    source = tmp_path / "population.csv"
62    pd.DataFrame(
63        {"ID": ["A", "B", "C"], "Status": ["Closed", "Open", "Closed"]}
64    ).to_csv(source, index=False)
65
66    run_dir = run(_options(source, tmp_path / "out", filters={"Status": "Closed"}))
67
68    validated = pd.read_csv(run_dir / "population_validated.csv")
69    excluded = pd.read_csv(run_dir / "excluded_rows.csv")
70    assert validated["ID"].tolist() == ["A", "C"]
71    assert excluded["ID"].tolist() == ["B"]
72
73
74def test_validate_only_writes_no_sample_csv(tmp_path):
75    source = tmp_path / "population.csv"
76    pd.DataFrame({"ID": ["A", "B", "C"], "Status": ["Closed"] * 3}).to_csv(
77        source, index=False
78    )
79
80    run_dir = run(_options(source, tmp_path / "out"))
81
82    assert not (run_dir / "sample.csv").exists()