81 lines
3.7 KiB
Python
81 lines
3.7 KiB
Python
import codecs
|
|
import csv
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from src.extract import FIELDS, read_records
|
|
from src.quality import build_report, save_outputs
|
|
from src.transform import clean_orders
|
|
|
|
|
|
def read_csv(path):
|
|
assert path.read_bytes().startswith(codecs.BOM_UTF8)
|
|
with path.open(encoding="utf-8-sig", newline="") as source:
|
|
return list(csv.reader(source))
|
|
|
|
|
|
def test_export_and_replace_snapshot(tmp_path):
|
|
path = tmp_path / "raw.csv"
|
|
header = ("category", "order_id", "quantity", "order_date", "unit_price")
|
|
bad = ['дом, "сад"\nспорт', "2", "", "?", "0"]
|
|
good = [" КНИГИ ", "7", "3", "29.02.2024", "0,07"]
|
|
extra = good + ["лишнее"]
|
|
rows = [good, bad, bad, extra, [], ["", "4", "1", "2024-01-01", "10"]]
|
|
with path.open("w", encoding="utf-8", newline="") as output:
|
|
writer = csv.writer(output)
|
|
writer.writerow(header)
|
|
writer.writerows(rows)
|
|
records = list(read_records(path))
|
|
accepted, rejected = clean_orders(records)
|
|
directory = tmp_path / "outputs"
|
|
save_outputs(accepted[::-1], rejected[::-1], len(records), directory)
|
|
clean = directory / "clean_orders.csv"
|
|
rejects = directory / "rejected_rows.csv"
|
|
report_file = directory / "quality_report.json"
|
|
clean_header = ["order_id", "order_date", "category", "quantity", "unit_price_kopecks", "revenue_kopecks"]
|
|
reject_header = ["record_number", "source_header", "source_values", "reasons"]
|
|
assert read_csv(clean) == [
|
|
clean_header, ["7", "2024-02-29", "Книги", "3", "7", "21"],
|
|
["4", "2024-01-01", "Не указана", "1", "1000", "1000"],
|
|
]
|
|
columns, *exported = read_csv(rejects)
|
|
assert columns == reject_header
|
|
assert [int(row[0]) for row in exported] == [2, 3, 4, 5]
|
|
for row, rejection in zip(exported, rejected, strict=True):
|
|
assert json.loads(row[1]) == list(header)
|
|
assert json.loads(row[2]) == rows[int(row[0]) - 1]
|
|
assert json.loads(row[3]) == rejection.reasons
|
|
|
|
report = json.loads(report_file.read_text(encoding="utf-8"))
|
|
assert (report["source_count"], report["accepted_count"], report["rejected_count"]) == (6, 2, 4)
|
|
assert report["accepted_share"] == 2 / 6
|
|
assert report["final_reason_counts"] == {
|
|
"order_date: ожидается YYYY-MM-DD или DD.MM.YYYY": 1, "Точный дубль": 1,
|
|
"Число полей: ожидалось 5, получено 6": 1, "Пустая запись CSV": 1,
|
|
}
|
|
assert report["problem_counts"] == report["final_reason_counts"] | {
|
|
"quantity: пропущено обязательное значение": 1, "unit_price: цена должна быть больше нуля": 1,
|
|
}
|
|
before = {file: file.read_bytes() for file in directory.iterdir()}
|
|
save_outputs(accepted, rejected, len(records), directory)
|
|
assert before == {file: file.read_bytes() for file in directory.iterdir()}
|
|
|
|
save_outputs([], [], 0, directory)
|
|
assert read_csv(clean) == [clean_header]
|
|
assert read_csv(rejects) == [reject_header]
|
|
report = json.loads(report_file.read_text(encoding="utf-8"))
|
|
assert report["accepted_share"] is None
|
|
assert report["source_count"] == report["accepted_count"] == report["rejected_count"] == 0
|
|
assert report["final_reason_counts"] == report["problem_counts"] == {}
|
|
|
|
|
|
def test_invalid_accounting(tmp_path):
|
|
source = tmp_path / "raw.csv"
|
|
source.write_text(",".join(FIELDS) + "\n1,2024-01-01,x,1,1\n", encoding="utf-8")
|
|
accepted, rejected = clean_orders(read_records(source))
|
|
with pytest.raises(ValueError, match="сумме"):
|
|
build_report(accepted, rejected, 2)
|
|
with pytest.raises(ValueError, match="без повторов"):
|
|
build_report(accepted * 2, rejected, 2)
|