import csv from pathlib import Path import pytest from src.extract import CSVFileError, FIELDS, read_records from src.transform import MAX_INTEGER, clean_orders def order(**changes): return dict(order_id="1", order_date="2024-02-29", category="Книги", quantity="3", unit_price="0,07") | changes def process(tmp_path, *rows, header=FIELDS): path = tmp_path / "orders.csv" with path.open("w", encoding="utf-8-sig", newline="") as output: writer = csv.writer(output) writer.writerow(header) for row in rows: writer.writerow([row[name] for name in header] if isinstance(row, dict) else row) before = path.read_bytes() accepted, rejected = clean_orders(read_records(path)) assert path.read_bytes() == before numbers = [item.record_number for item in accepted] numbers.extend(item.source.record_number for item in rejected) assert sorted(numbers) == list(range(1, len(rows) + 1)) return accepted, rejected @pytest.mark.parametrize("value", ["2024-02-29", " 29.02.2024 "]) def test_normalization_and_exact_money(tmp_path, value): accepted, rejected = process(tmp_path, order(order_date=value, category=" "), header=FIELDS[::-1]) assert not rejected item, = accepted assert (item.order_date, item.category) == ("2024-02-29", "Не указана") assert (item.unit_price_kopecks, item.revenue_kopecks) == (7, 21) @pytest.mark.parametrize(("name", "value"), [ ("order_date", "2023-02-29"), ("order_date", "1.2.2024"), ("order_date", "20240229"), ("quantity", "1.2"), ("quantity", "0"), ("quantity", "-2"), ("order_id", "0"), ("unit_price", "NaN"), ("unit_price", "Infinity"), ("unit_price", "-Infinity"), ("unit_price", "0"), ("unit_price", "-1"), ("unit_price", "1,230"), ("unit_price", "1 000.20"), ("unit_price", "1e2"), ("unit_price", "92233720368547758.08"), ("quantity", str(MAX_INTEGER + 1)), ("order_id", "9" * 5000), ]) def test_invalid_field(tmp_path, name, value): accepted, rejected = process(tmp_path, order(**{name: value})) assert not accepted reason, = rejected[0].reasons assert reason.startswith(name + ":") def test_sqlite_boundary_and_product_overflow(tmp_path): price = "92233720368547758.07" accepted, rejected = process(tmp_path, order(quantity="1", unit_price=price), order(quantity="2", unit_price=price)) assert accepted[0].revenue_kopecks == MAX_INTEGER assert rejected[0].reasons == ["revenue_kopecks: превышен предел INTEGER SQLite"] def test_multiple_missing_fields_and_exact_repeat(tmp_path): bad = order(order_id=" ", order_date="", quantity="", unit_price="") _, rejected = process(tmp_path, bad, bad) assert rejected[0].source.values == tuple(bad.values()) assert [s.split(":")[0] for s in rejected[0].reasons] == [ "order_id", "order_date", "quantity", "unit_price" ] assert rejected[1].reasons == ["Точный дубль"] def test_duplicate_stages_and_conflict(tmp_path): first = order(category=" ТоВаРы\n ДЛЯ ДоМа ") equivalent = first | {"order_date": "29.02.2024", "order_id": "001", "category": "товары для дома", "unit_price": "0.07"} conflict = order(order_id="2") accepted, rejected = process( tmp_path, first, first, equivalent, conflict, conflict | {"quantity": "03"}, conflict | {"quantity": "4"}, order(quantity="много") ) assert [(item.record_number, item.category) for item in accepted] == [(1, "Товары для дома")] assert [(r.source.record_number, r.reasons) for r in rejected] == [ (2, ["Точный дубль"]), (3, ["Нормализованный дубль"]), (4, ["Конфликт order_id"]), (5, ["Конфликт order_id"]), (6, ["Конфликт order_id"]), (7, ["quantity: ожидается положительное целое число"]), ] def test_empty_short_and_long_records(tmp_path): accepted, rejected = process(tmp_path, [], ["1"], list(order().values()) + ["лишнее"], order()) assert accepted[0].record_number == 4 assert [r.reasons for r in rejected] == [ ["Пустая запись CSV"], ["Число полей: ожидалось 5, получено 1"], ["Число полей: ожидалось 5, получено 6"], ] def test_header_only(tmp_path): assert process(tmp_path) == ([], []) @pytest.mark.parametrize("header", ["", "order_id,quantity", ",".join((*FIELDS, "extra")), ",".join((*FIELDS[:-1], "quantity"))]) def test_invalid_header(tmp_path, header): path = tmp_path / "orders.csv" path.write_text(header, encoding="utf-8") with pytest.raises(CSVFileError, match="неверный заголовок"): clean_orders(read_records(path)) def test_missing_or_damaged_file(tmp_path): path = tmp_path / "orders.csv" with pytest.raises(CSVFileError, match="файл не найден"): clean_orders(read_records(path)) path.write_text(",".join(FIELDS) + '\n1,2024-02-29,"незакрытое поле', encoding="utf-8") with pytest.raises(CSVFileError, match="ошибка структуры CSV"): clean_orders(read_records(path)) def test_invalid_utf8(tmp_path): path = tmp_path / "orders.csv" path.write_bytes(",".join(FIELDS).encode() + b"\n1,2024-01-01,\xff,1,1\n") with pytest.raises(CSVFileError, match="UTF-8") as error: list(read_records(path)) assert str(path) in str(error.value) def test_access_denied(tmp_path, monkeypatch): path = tmp_path / "private.csv" def deny_access(self, *args, **kwargs): raise PermissionError("access denied") monkeypatch.setattr(Path, "open", deny_access) with pytest.raises(CSVFileError, match="ошибка чтения") as error: list(read_records(path)) assert str(path) in str(error.value) assert isinstance(error.value.__cause__, PermissionError) def test_logical_record_numbers_across_multiline_field(tmp_path): path = tmp_path / "orders.csv" path.write_bytes((",".join(FIELDS) + '\r\n1,2024-02-29,"дом,\r\nсад",1,10\r\n' '\r\n2,2024-03-01,книги,2,3\r\n').encode()) records = list(read_records(path)) assert [r.record_number for r in records] == [1, 2, 3] assert records[0].values[2] == "дом,\r\nсад" assert records[1].structure_error == "Пустая запись CSV" assert records[2].values[0] == "2" def test_order_and_reasons_across_processing_stages(tmp_path): conflict = order(order_id="9") bad = order(quantity="-1", unit_price="0.001") accepted, rejected = process( tmp_path, conflict, order(order_id="8"), bad, conflict, conflict | {"order_date": "29.02.2024"}, order(), bad, conflict | {"quantity": "2"}, order(order_id="08"), ) assert [(o.record_number, o.order_id) for o in accepted] == [(2, 8), (6, 1)] assert [(r.source.record_number, r.reasons) for r in rejected] == [ (1, ["Конфликт order_id"]), (3, ["quantity: ожидается положительное целое число", "unit_price: больше двух дробных знаков"]), (4, ["Точный дубль"]), (5, ["Конфликт order_id"]), (7, ["Точный дубль"]), (8, ["Конфликт order_id"]), (9, ["Нормализованный дубль"]), ] def test_large_price_preserves_last_kopeck(tmp_path): accepted, rejected = process(tmp_path, order(unit_price="90071992547409,93", quantity="3")) assert not rejected assert accepted[0].unit_price_kopecks == 9_007_199_254_740_993 assert accepted[0].revenue_kopecks == 27_021_597_764_222_979