179 lines
7.7 KiB
Python
179 lines
7.7 KiB
Python
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
|