From c2f85f10d185cbe3f24c1c4ebf35382220be66f6 Mon Sep 17 00:00:00 2001 From: dev-belly <288855003+dev-belly@users.noreply.github.com> Date: Tue, 13 Oct 2026 17:15:25 +0800 Subject: [PATCH] Reject conflicting master records and preserve cleaning evidence on failure --- README.md | 16 ++- docs/MASTER_DATA_CASE.zh-CN.md | 82 +++++++++++ src/data_cleaning.py | 127 +++++++++++++++-- tests/test_data_cleaning.py | 244 +++++++++++++++++++++++++++++++++ 4 files changed, 451 insertions(+), 18 deletions(-) create mode 100644 docs/MASTER_DATA_CASE.zh-CN.md diff --git a/README.md b/README.md index 04edbda..d0167bd 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ [![CI](https://github.com/dev-belly/AuditLens/actions/workflows/ci.yml/badge.svg)](https://github.com/dev-belly/AuditLens/actions/workflows/ci.yml) [![Python 3.11+](https://img.shields.io/badge/python-3.11%2B-blue.svg)](https://www.python.org/downloads/) [![License: MIT](https://img.shields.io/badge/license-MIT-green.svg)](LICENSE) -[![Tests: 465](https://img.shields.io/badge/tests-465%20passing-brightgreen.svg)](#testing) +[![Tests: 501](https://img.shields.io/badge/tests-501%20passing-brightgreen.svg)](#testing) **Financial anomaly detection and audit analytics over a 30,000-voucher general ledger.** @@ -58,7 +58,7 @@ Three design commitments drive everything else: | Anomalies caught by neither | **15** | | Isolation Forest | ROC-AUC **0.816**, precision 0.291, recall 0.286 | | Benford first-digit MAD | **0.00218** — close conformity | -| Test suite | **465 tests**, including dashboard render, warehouse and review-plan checks | +| Test suite | **501 tests**, including dashboard render, warehouse and review-plan checks | ### The most important number here is 98.2%, and it is a warning @@ -565,7 +565,7 @@ AuditLens/ │ ├── components.py # KPI cards, risk badges, charts, tables │ └── views/ # the six pages ├── sql/audit_queries.sql # 15 named business queries -├── tests/ # 465 tests, incl. cross-process reproducibility +├── tests/ # 501 tests, incl. cross-process reproducibility ├── docs/ │ ├── architecture.md # design decisions, data contracts, what is deliberately excluded │ ├── methodology.md # every threshold, every weight, every mistake @@ -579,7 +579,7 @@ AuditLens/ │ ├── notebook_content.py # what the three notebooks contain, kept apart from the builder │ ├── capture_screenshots.py # headless captures of all six dashboard pages, via DevTools Protocol │ └── ci_summary.py # headline figures for the CI job summary -├── .github/workflows/ci.yml # pipeline + 465 tests; Ubuntu 3.11/3.12 and Windows 3.12 +├── .github/workflows/ci.yml # pipeline + 501 tests; Ubuntu 3.11/3.12 and Windows 3.12 └── Makefile ``` @@ -655,6 +655,8 @@ Stated plainly. A portfolio project that overclaims is worse than one that does ## Interview talking points +Chinese walkthrough: [Master-data integrity: business decisions, failure recovery and interview practice](docs/MASTER_DATA_CASE.zh-CN.md). The cleaner rejects conflicting supplier/employee IDs before lossy date parsing, and stages all four cleaning outputs before publication with rollback on handled failures. This is a single-writer CLI guarantee; concurrent snapshots and crash recovery require a separate versioning contract. + If you are reading this as a portfolio project, these are the things worth being able to defend. Each is a real decision in the code. @@ -750,7 +752,7 @@ builder, so two consecutive builds are byte-identical. ```bash pip install -r requirements.txt python src/run_pipeline.py # ~30 seconds, deterministic -python -m pytest tests/ -q # 465 tests +python -m pytest tests/ -q # 501 tests ``` Two consecutive runs produce byte-identical reports under `outputs/reports/`. This is @@ -758,11 +760,11 @@ enforced by `tests/test_reproducibility.py`, not assumed. ## Testing -465 tests. `make test` runs the lot; `make test-fast` skips the dashboard render suite. +501 tests. `make test` runs the lot; `make test-fast` skips the dashboard render suite. | File | Tests | What it pins down | |---|---|---| -| `test_data_cleaning.py` | 57 | Each repair path: missing values, duplicates, invalid dates and amounts, debit/credit balance, orphan vendor IDs, inconsistent account names, currency normalisation | +| `test_data_cleaning.py` | 93 | Repairs, full-record master integrity, source-preserving conflicts, and staged cleaning publication with failure rollback | | `test_audit_rules.py` | 38 | One class per procedure, plus mixed-size split-cluster regressions and the text-encoded-label fallback in `evaluate()` | | `test_benford.py` | 56 | Expected frequencies, MAD bands, χ², the per-bucket z-score, and the schema stability of the insufficient-data branch | | `test_risk_scoring.py` | 55 | Component scores, the noisy-OR combination, band boundaries, `risk_reasons` wording | diff --git a/docs/MASTER_DATA_CASE.zh-CN.md b/docs/MASTER_DATA_CASE.zh-CN.md new file mode 100644 index 0000000..640c5bc --- /dev/null +++ b/docs/MASTER_DATA_CASE.zh-CN.md @@ -0,0 +1,82 @@ +# 面试案例：先证明主数据可信，再谈异常检测 + +本案例来自 AuditLens 的真实代码缺口与修复。数据是合成数据，项目是个人作品；没有真实客户上线、审计签字或收益证明。下面的第一人称讲法供理解代码后练习使用，个人分工应按事实改写。 + +## 60 秒讲法 + +> 我用 AuditLens 做了一条审计分析流程：清洗总账、执行规则、排序风险，再形成复核清单。我关注的一个问题是，规则本身正确，输入主数据却可能不可信。 +> +> 原来同一个供应商编号出现两个银行账号时，清洗直接保留第一条。只要交换文件行顺序，选中的账号就变了。供应商关联和后续复核建立在这个结果上，所以这不是普通重复行问题。 +> +> 修复是先按已有规则整理文本，再比较同一编号的所有源字段。完全一致的才合并；有冲突就停止，提示来源行号和字段，避免在日志里暴露账号。四个清洗结果也先暂存，发布失败会恢复旧结果。 +> +> 我用交换输入顺序、空值、日期解析和中途写入失败来验证这个约束。它仍然是单进程分析工具；如果要支持历史主数据或并发读取，需要再设计版本和快照合同。 + +## 3 分钟讲法：把追问引向可以验证的业务决策 + +1. **业务链路。** 总账通过供应商、员工编号关联主数据。银行账号进入共享账号特征，员工角色进入职责分离例外的 SQL 复核结果。主数据必须先有一致的解释，否则后面漂亮的分数没有可靠基础。 +2. **复现。** 构造两条 `vendor_id=V0001` 的记录，其他字段相同，账号分别为 `BANK_A` 和 `BANK_B`。旧实现按编号 `keep="first"`；逆序输入就得到另一个账号。问题不是选错了一个固定答案，而是系统没有选择答案的业务依据。 +3. **选择。** 当前输入合同是“一次抽取中，每个编号对应一条一致记录”，没有生效区间、权威修订号或变更审批记录。因此不能擅自取最新、第一条、最后一条，也不能把空值自动补齐后假装记录一致。冲突需要回到来源核对。 +4. **实现。** 先去除缺编号的记录，比较其余重复编号的所有字段。已知文本沿用空白归一化，空值彼此等价。检查发生在日期的有损解析之前，避免两个不同坏日期都变成 `NaT` 后被当成同一条。通过验证的供应商编号才用于孤立引用检查。 +5. **失败处理。** 只有全部校验和四份序列化完成，才替换现有文件。发布到一半失败，恢复已替换的文件；第一次运行没有旧结果，则删除刚发布的部分结果。回滚也失败时保留旧内容备份，明确报告恢复路径。 +6. **验证与边界。** 测试检查源数据没有被修改、冲突在两种顺序下都拒绝、第三份表或报告写失败不覆盖旧证据，并注入不同位置的替换失败。正常合成演示继续经过完整管线与 CI 检查。没有声称多文件替换对并发读者是一个原子快照，也没有声称可承受断电或进程被强杀。 + +## 当前合同：什么可以合并 + +| 输入差异 | 处理 | 理由 | +|---|---|---| +| 同一编号、所有字段一致 | 合并一份 | 不存在要裁决的属性差异 | +| 已知文本字段仅前后空白或重复空白不同 | 归一化后可合并 | 沿用项目已有文本清洗规则 | +| `None`、`NaN`、`pd.NA` | 视为同一缺失值 | 缺失标记的存储形式不构成业务差异 | +| 已知文本字段为空字符串或全空白 | 视为缺失 | 沿用现有空白语义 | +| 某字段缺失与该字段有值 | 拒绝 | 没有证据证明可以补齐成同一条 | +| 银行账号、角色或任何其他源字段不同 | 拒绝 | 不按白名单忽略新增字段 | +| 两个不同无效日期字符串 | 拒绝 | 不能用解析失败抹掉原始差异 | +| `2024-01-01` 与 `01/01/2024` | 拒绝 | 保守保留源格式差异，需上游统一日期合同 | +| 缺失或空白编号 | 去除 | 保持既有行为；本次未新增缺编号的治理报表 | + +这是一项保守的输入校验。唯一编号记录的无效注册日期仍沿用原有解析为 `NaT` 的行为；本次没有把所有日期质量问题一并改成硬错误。 + +## 面试官追问与可落到代码的回答 + +**为什么不取注册日期最新的一条？** 注册日期是供应商属性，不是该记录的更新时间。把它用作版本顺序会编造业务规则。历史版本要有明确生效时间、失效时间或权威修订号，并约束同一时点只匹配一条。 + +**为什么比较所有列，连新增字段也检查？** 只比较名称和账号，未来新增风险状态或权限列时容易再次静默丢失差异。当前合同是完整记录一致；新增字段默认参与验证，若某列确实只是采集元信息，应先单独定义其合同。 + +**为什么空值不从另一条补齐？** 两条记录可能来自不同版本或不同系统。没有来源优先级时，自动补齐会组合出源文件里不存在的记录。缺失和有值先作为冲突交给上游确认。 + +**停止整条流程是不是影响可用性？** 是取舍：主数据在多个规则和关联中复用，继续运行可能输出错误证据。当前选择停止更新，保留上一次结果。上一次结果也不能当成最新结果；调用方要看到失败并核对输入批次。本项目尚未实现批次状态服务。 + +**你说原子写入，究竟原子在哪里？** 单个文件在同一文件系统内用替换操作发布；四份文件不是一个原子事务。实现保证正常捕获的失败触发回滚，不能保证并发读者在发布途中看不到混合版本。需要更强保证时，可以发布到独立版本目录，再切换一个清单入口，所有读者都依照该入口读取。 + +**如果回滚也因为权限或磁盘问题失败呢？** 报告“不完整回滚”，保留恢复旧内容所需的备份路径；不能继续报告成功。测试专门覆盖这个情况。文件备份也有清理失败风险，所以临时文件清理失败会记录日志。 + +**员工权限有没有真正的审批系统？** 没有。这是审计分析中的员工属性一致性校验。测试中的额外权限列说明所有源字段都参与比较，不代表项目实现了账号授权或审批权限执行。 + +**这是你自己做的还是 AI 帮你写的？** 按真实分工回答：“AI 协助了我实际使用到的部分；我能复现问题、解释为什么不能任意选一条，并现场修改和验证这条约束。”如果某部分还解释不清，应明确说明，先完成下面的练习，不能把生成代码的过程包装成客户经验。 + +## 现场复现与阅读路径 + +使用 README 中安装好的 Python 环境，在仓库根目录执行： + +```shell +python -m pytest tests/test_data_cleaning.py -q -k "MasterDataIntegrity or CleaningPublication" +python -m pytest tests/test_data_cleaning.py -q +python src/run_pipeline.py +``` + +| 阅读入口 | 应解释的内容 | +|---|---| +| [`normalise_text`](../src/data_cleaning.py) | 哪些字段整理空白，为什么空白是缺失 | +| [`_deduplicate_master` / `clean_master_data`](../src/data_cleaning.py) | 全字段一致性、来源行号、日期解析顺序 | +| [`_publish_cleaning_outputs` / `run_cleaning`](../src/data_cleaning.py) | 校验、暂存、备份、发布、回滚的顺序 | +| [`TestMasterDataIntegrity`](../tests/test_data_cleaning.py) | 输入顺序、未列出的列、空值和源格式差异 | +| [`TestCleaningPublication`](../tests/test_data_cleaning.py) | 完整旧结果、首次运行、回滚失败的证据 | +| [`build_vendor_features`](../src/feature_engineering.py) / [SQL 审计查询](../sql/audit_queries.sql) | 银行账号与员工角色实际进入哪些分析结果 | +| [CI](../.github/workflows/ci.yml) | 跨平台测试、完整管线和报告复现检查 | + +## 两个练习与解题方向 + +**练习一：添加 `supplier_status`，同一编号分别为 `active` / `blocked`。** 先预测结果，再修改测试数据并运行。解题方向：新增列也参与完整记录比较；异常应包含列名，不包含属性值。交换行序仍应拒绝。 + +**练习二：第三份文件发布失败，第一次运行和已有结果两种情况下磁盘应该是什么状态？** 不看实现先列出预期，再读替换失败测试。解题方向：已有结果的四份内容恢复原字节；首次运行四份都不存在；两种情况均清理暂存文件。追加“回滚自身失败”时，预期改为明确异常并保留恢复备份，不能继续声称结果一致。 diff --git a/src/data_cleaning.py b/src/data_cleaning.py index 9e52957..0567a60 100644 --- a/src/data_cleaning.py +++ b/src/data_cleaning.py @@ -28,7 +28,10 @@ from __future__ import annotations +import os +import shutil import sys +import tempfile from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any @@ -508,25 +511,130 @@ def clean_transactions( return df, report +def _deduplicate_master(df: pd.DataFrame, key: str, source: str) -> pd.DataFrame: + """Require one consistent, complete record per master-data ID. + + Missing cells compare equal to other missing cells, but not to populated + cells. Compare before lossy conversions such as invalid dates becoming NaT. + Source row numbers include the CSV header; no attribute values are exposed. + """ + df = df.reset_index(drop=True).dropna(subset=[key]) + duplicates = df.loc[df.duplicated(subset=[key], keep=False)] + for identifier, group in duplicates.groupby(key, sort=False): + first = group.iloc[0] + equal = group.eq(first, axis="columns") | (group.isna() & first.isna()) + conflicts = equal.fillna(False).all().loc[lambda fields: ~fields].index.tolist() + if conflicts: + rows = ", ".join(str(index + 2) for index in group.index[:6]) + if len(group) > 6: + rows += ", ..." + raise ValueError( + f"Conflicting master data in {source}: {key}={identifier!r}; " + f"source rows {rows}; fields: {', '.join(conflicts)}. " + "Reconcile the source records; no record was selected." + ) + return df.drop_duplicates(subset=[key], keep="first").reset_index(drop=True) + + def clean_master_data( vendors: pd.DataFrame, employees: pd.DataFrame ) -> tuple[pd.DataFrame, pd.DataFrame]: """Clean the vendor and employee master files. - Master data is small and authoritative, so cleaning is limited to trimming - text, parsing the registration date and de-duplicating on the primary key. + Trim known text fields and drop missing IDs. Repeated IDs are merged only + when all source fields agree after text normalisation. Otherwise reject the + extract instead of silently selecting a bank account or employee permission. + Date parsing follows validation so distinct invalid dates remain conflicts. """ vendors = normalise_text(vendors, ("vendor_id", "vendor_name", "vendor_category", "bank_account", "country", "risk_level")) - vendors["registration_date"] = pd.to_datetime(vendors["registration_date"], errors="coerce", format="mixed") - vendors = vendors.dropna(subset=["vendor_id"]).drop_duplicates(subset=["vendor_id"], keep="first") - employees = normalise_text(employees, ("employee_id", "employee_name", "department", "role")) - employees = employees.dropna(subset=["employee_id"]).drop_duplicates(subset=["employee_id"], keep="first") + + vendors = _deduplicate_master(vendors, "vendor_id", VENDORS_RAW.name) + employees = _deduplicate_master(employees, "employee_id", EMPLOYEES_RAW.name) + vendors["registration_date"] = pd.to_datetime(vendors["registration_date"], errors="coerce", format="mixed") LOGGER.info("Cleaned master data: %s vendors, %s employees", len(vendors), len(employees)) return vendors.reset_index(drop=True), employees.reset_index(drop=True) +def _publish_cleaning_outputs( + transactions: pd.DataFrame, + vendors: pd.DataFrame, + employees: pd.DataFrame, + report: DataQualityReport, +) -> None: + """Stage the whole result, then publish with rollback on handled failures. + + Each temporary file lives beside its destination, so replacement stays on + one filesystem. This is for a single CLI writer: it is not a multi-file + atomic snapshot for concurrent readers or a transaction surviving a crash. + """ + temporary_paths: set[Path] = set() + staged: dict[Path, Path] = {} + backups: dict[Path, Path | None] = {} + published: list[Path] = [] + + def temporary_for(target: Path) -> Path: + target.parent.mkdir(parents=True, exist_ok=True) + descriptor, name = tempfile.mkstemp( + prefix=".auditlens-clean-", suffix=target.suffix, dir=target.parent, + ) + os.close(descriptor) + path = Path(name) + temporary_paths.add(path) + return path + + try: + for target, frame in ( + (TRANSACTIONS_CLEAN, transactions), + (VENDORS_CLEAN, vendors), + (EMPLOYEES_CLEAN, employees), + ): + staged[target] = temporary_for(target) + frame.to_parquet(staged[target], index=False) + staged[DATA_QUALITY_REPORT] = temporary_for(DATA_QUALITY_REPORT) + save_json(report.to_dict(), staged[DATA_QUALITY_REPORT]) + + # Preserve every previous result before replacing any live output. + for target in staged: + backups[target] = None + if target.exists(): + backup = temporary_for(target) + shutil.copyfile(target, backup) + backups[target] = backup + for target, stage in staged.items(): + os.replace(stage, target) + published.append(target) + except BaseException as error: + recovery_paths: list[str] = [] + for target in reversed(published): + backup = backups[target] + try: + if backup is None: + target.unlink(missing_ok=True) + else: + os.replace(backup, target) + except OSError: + if backup is not None: + # Do not delete the only remaining copy of the old result. + temporary_paths.discard(backup) + recovery_paths.append(f"{target} from {backup}") + else: + recovery_paths.append(f"remove newly published {target}") + if recovery_paths: + raise RuntimeError( + "Cleaning publication failed and rollback was incomplete. " + "Recover previous outputs: " + "; ".join(recovery_paths) + ) from error + raise + finally: + for path in temporary_paths: + try: + path.unlink(missing_ok=True) + except OSError: + LOGGER.warning("Could not remove cleaning temporary file %s", path) + + def run_cleaning() -> dict[str, Any]: """Execute the cleaning pipeline end to end and persist its outputs. @@ -536,13 +644,10 @@ def run_cleaning() -> dict[str, Any]: ensure_directories() transactions_raw, vendors_raw, employees_raw = load_raw_data() - transactions, report = clean_transactions(transactions_raw, vendors_raw) vendors, employees = clean_master_data(vendors_raw, employees_raw) + transactions, report = clean_transactions(transactions_raw, vendors) - transactions.to_parquet(TRANSACTIONS_CLEAN, index=False) - vendors.to_parquet(VENDORS_CLEAN, index=False) - employees.to_parquet(EMPLOYEES_CLEAN, index=False) - save_json(report.to_dict(), DATA_QUALITY_REPORT) + _publish_cleaning_outputs(transactions, vendors, employees, report) LOGGER.info("Wrote %s", TRANSACTIONS_CLEAN) LOGGER.info("Wrote %s", DATA_QUALITY_REPORT) diff --git a/tests/test_data_cleaning.py b/tests/test_data_cleaning.py index cff18ec..7b96d60 100644 --- a/tests/test_data_cleaning.py +++ b/tests/test_data_cleaning.py @@ -15,12 +15,17 @@ from __future__ import annotations +import json +from pathlib import Path + import pandas as pd import pytest +from src import data_cleaning as cleaning from src.data_cleaning import ( CURRENCY_ALIASES, DataQualityReport, + clean_master_data, clean_transactions, coerce_numeric, flag_orphan_vendors, @@ -515,3 +520,242 @@ def test_cleaning_is_idempotent(self) -> None: assert len(twice) == len(once) assert report.duplicates_removed == 0 + + +def _masters(vendor_rows: list[dict] | None = None, employee_rows: list[dict] | None = None): + vendor = { + "vendor_id": "V0001", "vendor_name": "Test Vendor", "bank_account": "BANK_A", + "registration_date": "2024-01-01", "risk_level": "low", "is_dormant": False, + } + employee = {"employee_id": "E0001", "employee_name": "Test Employee", "role": "creator"} + vendors = pd.DataFrame([{**vendor, **row} for row in (vendor_rows if vendor_rows is not None else [{}])]) + employees = pd.DataFrame([{**employee, **row} for row in (employee_rows if employee_rows is not None else [{}])]) + return vendors, employees + + +class TestMasterDataIntegrity: + def test_identical_records_merge_without_mutating_sources(self) -> None: + vendors, employees = _masters([{}, {}], [{}, {}]) + originals = vendors.copy(deep=True), employees.copy(deep=True) + clean_vendors, clean_employees = clean_master_data(vendors, employees) + assert len(clean_vendors) == len(clean_employees) == 1 + pd.testing.assert_frame_equal(vendors, originals[0]) + pd.testing.assert_frame_equal(employees, originals[1]) + + def test_known_text_fields_use_existing_whitespace_normalisation(self) -> None: + vendors, employees = _masters( + [{}, {"vendor_id": " V0001 ", "vendor_name": " Test Vendor ", "bank_account": " BANK_A "}], + [{}, {"employee_id": " E0001 ", "role": " creator "}], + ) + cleaned = clean_master_data(vendors, employees) + assert [len(frame) for frame in cleaned] == [1, 1] + + def test_missing_ids_still_get_removed(self) -> None: + vendors, employees = _masters( + [{}, {"vendor_id": None}, {"vendor_id": " "}], + [{}, {"employee_id": None}, {"employee_id": " "}], + ) + cleaned = clean_master_data(vendors, employees) + assert [len(frame) for frame in cleaned] == [1, 1] + + def test_missing_cells_are_equivalent_including_unlisted_columns(self) -> None: + vendors, employees = _masters([ + {"bank_account": None, "registration_date": None, "source_note": None}, + {"bank_account": " ", "registration_date": pd.NA, "source_note": float("nan")}, + ]) + clean_vendors, _ = clean_master_data(vendors, employees) + assert len(clean_vendors) == 1 + assert pd.isna(clean_vendors.loc[0, "bank_account"]) + assert pd.isna(clean_vendors.loc[0, "registration_date"]) + + @pytest.mark.parametrize("reverse", [False, True]) + @pytest.mark.parametrize("field,value", [ + ("bank_account", "BANK_B"), ("risk_level", "high"), + ("registration_date", "2024-02-01"), ("is_dormant", True), + ("source_revision", "revision_b"), + ]) + def test_conflicting_vendor_records_reject_in_either_order(self, field, value, reverse) -> None: + vendors, employees = _masters([ + {"source_revision": "revision_a"}, {"source_revision": "revision_a", field: value}, + ]) + if reverse: + vendors = vendors.iloc[::-1] + with pytest.raises(ValueError, match="Conflicting master data") as error: + clean_master_data(vendors, employees) + message = str(error.value) + assert "vendors_raw.csv" in message and "V0001" in message and field in message + assert "BANK_A" not in message and "BANK_B" not in message + assert "source rows 2, 3" in message + + @pytest.mark.parametrize("reverse", [False, True]) + @pytest.mark.parametrize("field,value", [("role", "approver"), ("can_approve", True)]) + def test_conflicting_employee_attributes_reject_in_either_order(self, field, value, reverse) -> None: + vendors, employees = _masters(employee_rows=[ + {"can_approve": False}, {"can_approve": False, field: value}, + ]) + if reverse: + employees = employees.iloc[::-1] + with pytest.raises(ValueError, match="employees_raw.csv") as error: + clean_master_data(vendors, employees) + assert "E0001" in str(error.value) and field in str(error.value) + + def test_missing_and_populated_attributes_conflict_instead_of_filling(self) -> None: + vendors, employees = _masters([{"bank_account": None}, {}]) + with pytest.raises(ValueError, match="bank_account"): + clean_master_data(vendors, employees) + + @pytest.mark.parametrize("dates", [ + ("invalid_a", "invalid_b"), ("2024-01-01", "01/01/2024"), (None, "invalid_a"), + ]) + def test_date_conversion_cannot_erase_source_differences(self, dates) -> None: + vendors, employees = _masters([{"registration_date": date} for date in dates]) + with pytest.raises(ValueError, match="registration_date"): + clean_master_data(vendors, employees) + + def test_source_row_positions_survive_missing_ids_and_duplicate_indexes(self) -> None: + vendors, employees = _masters([{}, {"vendor_id": None}, {"bank_account": "BANK_B"}]) + vendors.index = [9, 9, 9] + with pytest.raises(ValueError, match="source rows 2, 4"): + clean_master_data(vendors, employees) + + +@pytest.fixture +def cleaning_outputs(monkeypatch, tmp_path): + targets = [ + tmp_path / "processed" / "transactions.parquet", + tmp_path / "processed" / "vendors.parquet", + tmp_path / "processed" / "employees.parquet", + tmp_path / "reports" / "quality.json", + ] + for name, path in zip( + ("TRANSACTIONS_CLEAN", "VENDORS_CLEAN", "EMPLOYEES_CLEAN", "DATA_QUALITY_REPORT"), targets, + ): + monkeypatch.setattr(cleaning, name, path) + vendors, employees = _masters() + monkeypatch.setattr(cleaning, "load_raw_data", lambda: (_frame([{}]), vendors, employees)) + monkeypatch.setattr(cleaning, "ensure_directories", lambda: None) + return targets + + +def _seed_outputs(targets: list[Path]) -> dict[Path, bytes]: + previous = {path: f"previous complete result {index}".encode() for index, path in enumerate(targets)} + for path, payload in previous.items(): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(payload) + return previous + + +def _assert_no_temporary_files(targets: list[Path]) -> None: + for parent in {path.parent for path in targets}: + assert not list(parent.glob(".auditlens-clean-*")) + + +class TestCleaningPublication: + def test_normalised_master_ids_are_used_for_orphan_detection(self, cleaning_outputs, monkeypatch) -> None: + vendors, employees = _masters([{"vendor_id": " V0001 "}]) + monkeypatch.setattr(cleaning, "load_raw_data", lambda: (_frame([{}]), vendors, employees)) + result = cleaning.run_cleaning() + assert result["report"].orphan_vendor_rows == 0 + assert not result["transactions"]["orphan_vendor_flag"].any() + + def test_success_publishes_matching_tables_and_report(self, cleaning_outputs) -> None: + _seed_outputs(cleaning_outputs) + result = cleaning.run_cleaning() + for key, path in zip(("transactions", "vendors", "employees"), cleaning_outputs): + pd.testing.assert_frame_equal(pd.read_parquet(path), result[key]) + assert json.loads(cleaning_outputs[3].read_text("utf-8")) == result["report"].to_dict() + _assert_no_temporary_files(cleaning_outputs) + + def test_master_conflict_preserves_all_previous_evidence(self, cleaning_outputs, monkeypatch) -> None: + previous = _seed_outputs(cleaning_outputs) + vendors, employees = _masters([{}, {"bank_account": "BANK_B"}]) + monkeypatch.setattr(cleaning, "load_raw_data", lambda: (_frame([{}]), vendors, employees)) + with pytest.raises(ValueError, match="bank_account"): + cleaning.run_cleaning() + assert {path: path.read_bytes() for path in cleaning_outputs} == previous + _assert_no_temporary_files(cleaning_outputs) + + @pytest.mark.parametrize("failed_stage", ["employees", "report"]) + def test_serialisation_failure_preserves_all_previous_outputs( + self, cleaning_outputs, monkeypatch, failed_stage, + ) -> None: + previous = _seed_outputs(cleaning_outputs) + if failed_stage == "employees": + original = pd.DataFrame.to_parquet + calls = 0 + + def fail_third(frame, path, **kwargs): + nonlocal calls + calls += 1 + if calls == 3: + Path(path).write_bytes(b"partial serialisation") + raise OSError("serialisation failed") + return original(frame, path, **kwargs) + + monkeypatch.setattr(pd.DataFrame, "to_parquet", fail_third) + else: + def fail_report(payload, path): + path.write_bytes(b"partial report") + raise OSError("serialisation failed") + + monkeypatch.setattr(cleaning, "save_json", fail_report) + with pytest.raises(OSError, match="serialisation failed"): + cleaning.run_cleaning() + assert {path: path.read_bytes() for path in cleaning_outputs} == previous + _assert_no_temporary_files(cleaning_outputs) + + @pytest.mark.parametrize("has_previous", [False, True]) + @pytest.mark.parametrize("failed_replace", [1, 2, 4]) + def test_publication_failure_rolls_back_complete_result( + self, cleaning_outputs, monkeypatch, has_previous, failed_replace, + ) -> None: + previous = _seed_outputs(cleaning_outputs) if has_previous else {} + original = cleaning.os.replace + calls = 0 + + def fail_once(source, target): + nonlocal calls + calls += 1 + if calls == failed_replace: + raise OSError("publication failed") + return original(source, target) + + monkeypatch.setattr(cleaning.os, "replace", fail_once) + with pytest.raises(OSError, match="publication failed"): + cleaning.run_cleaning() + assert {path: path.read_bytes() for path in cleaning_outputs if path.exists()} == previous + _assert_no_temporary_files(cleaning_outputs) + + def test_backup_failure_never_publishes_any_output(self, cleaning_outputs, monkeypatch) -> None: + previous = _seed_outputs(cleaning_outputs) + + def fail_copy(source, target): + raise OSError("backup failed") + + monkeypatch.setattr(cleaning.shutil, "copyfile", fail_copy) + with pytest.raises(OSError, match="backup failed"): + cleaning.run_cleaning() + assert {path: path.read_bytes() for path in cleaning_outputs} == previous + _assert_no_temporary_files(cleaning_outputs) + + def test_failed_rollback_retains_old_bytes_and_reports_recovery_path( + self, cleaning_outputs, monkeypatch, + ) -> None: + previous = _seed_outputs(cleaning_outputs) + original = cleaning.os.replace + calls = 0 + + def fail_publication_and_rollback(source, target): + nonlocal calls + calls += 1 + if calls in (2, 3): + raise OSError("blocked replacement") + return original(source, target) + + monkeypatch.setattr(cleaning.os, "replace", fail_publication_and_rollback) + with pytest.raises(RuntimeError, match="rollback was incomplete") as error: + cleaning.run_cleaning() + backups = list(cleaning_outputs[0].parent.glob(".auditlens-clean-*")) + assert len(backups) == 1 and backups[0].read_bytes() == previous[cleaning_outputs[0]] + assert str(backups[0]) in str(error.value) + assert all(path.read_bytes() == previous[path] for path in cleaning_outputs[1:])