From 8f8fabf959a80cd1cab06529f90c3a699e805937 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Wed, 26 Aug 2026 16:47:30 +0300 Subject: [PATCH] =?UTF-8?q?test:=20=D0=BD=D0=B0=D0=B1=D0=BE=D1=80=20pytest?= =?UTF-8?q?-=D1=82=D0=B5=D1=81=D1=82=D0=BE=D0=B2=20=D0=BF=D0=BE=D0=B4=20si?= =?UTF-8?q?te/=20(unit=20+=20=D0=B8=D0=BD=D1=82=D0=B5=D0=B3=D1=80=D0=B0?= =?UTF-8?q?=D1=86=D0=B8=D0=BE=D0=BD=D0=BD=D1=8B=D0=B5=20+=20=D0=B1=D0=B5?= =?UTF-8?q?=D0=B7=D0=BE=D0=BF=D0=B0=D1=81=D0=BD=D0=BE=D1=81=D1=82=D1=8C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/conftest.py | 37 ++++++++++++ tests/test_classify.py | 92 ++++++++++++++++++++++++++++ tests/test_connection.py | 35 +++++++++++ tests/test_grouping.py | 83 +++++++++++++++++++++++++ tests/test_llm.py | 25 ++++++++ tests/test_llm_client.py | 40 ++++++++++++ tests/test_metrics.py | 87 +++++++++++++++++++++++++++ tests/test_parse.py | 28 +++++++++ tests/test_process_pipeline.py | 96 +++++++++++++++++++++++++++++ tests/test_routes.py | 85 ++++++++++++++++++++++++++ tests/test_spec_current.py | 31 ++++++++++ tests/test_spec_events.py | 107 +++++++++++++++++++++++++++++++++ tests/test_upload_security.py | 50 +++++++++++++++ 13 files changed, 796 insertions(+) create mode 100644 tests/conftest.py create mode 100644 tests/test_classify.py create mode 100644 tests/test_connection.py create mode 100644 tests/test_grouping.py create mode 100644 tests/test_llm.py create mode 100644 tests/test_llm_client.py create mode 100644 tests/test_metrics.py create mode 100644 tests/test_parse.py create mode 100644 tests/test_process_pipeline.py create mode 100644 tests/test_routes.py create mode 100644 tests/test_spec_current.py create mode 100644 tests/test_spec_events.py create mode 100644 tests/test_upload_security.py diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..4a2ca46 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,37 @@ +"""Общие фикстуры для тестов contracts-flask (модули site/). + +Запуск: pytest tests/ -q +""" +import os +import sys + +import pytest + +# site/ в sys.path — импортируем config/db/services/routes напрямую +_SITE = os.path.join(os.path.dirname(__file__), "..", "site") +if _SITE not in sys.path: + sys.path.insert(0, _SITE) + + +@pytest.fixture(autouse=True, scope="session") +def _isolate_global_db(tmp_path_factory): + """Не дать импорту app.py пересоздать реальную /tmp/contracts.db.""" + from db import connection as conn + conn.DB_PATH = str(tmp_path_factory.mktemp("dbs") / "session.db") + + +@pytest.fixture +def db(tmp_path, monkeypatch): + """Изолированная БД: отдельный DB_PATH + свежая схема (init_db).""" + from db import connection as conn + monkeypatch.setattr(conn, "DB_PATH", str(tmp_path / "contracts_test.db")) + conn.init_db() + yield conn + # cleanup: закрыть thread-local соединение + c = getattr(conn._local, "conn", None) + if c is not None: + try: + c.close() + except Exception: + pass + conn._local.conn = None diff --git a/tests/test_classify.py b/tests/test_classify.py new file mode 100644 index 0000000..83de0d9 --- /dev/null +++ b/tests/test_classify.py @@ -0,0 +1,92 @@ +"""Unit-тесты services/classify.py — фильтры + JSON-парсинг + выжимка.""" +import json + +import pytest + +from services.classify import ( + _is_garbage_by_filename, + _is_garbage_by_header, + _safe_json_parse, + _smart_extract, +) + + +class TestGarbageFilename: + def test_invoice(self): + assert _is_garbage_by_filename("счет-фактура №123.docx") is True + + def test_act(self): + assert _is_garbage_by_filename("Акт сверки.pdf") is True + + def test_upd(self): + assert _is_garbage_by_filename("УПД-2025.docx") is True + + def test_not_garbage(self): + assert _is_garbage_by_filename("договор.docx") is False + assert _is_garbage_by_filename("спецификация.pdf") is False + + +class TestGarbageHeader: + def test_invoice_header(self): + assert _is_garbage_by_header("СЧЕТ-ФАКТУРА № 123 от 01.01.2026") is True + + def test_act_header(self): + assert _is_garbage_by_header("АКТ ОКАЗАННЫХ УСЛУГ за май") is True + + def test_not_garbage(self): + assert _is_garbage_by_header("ДОГОВОР № 03700_1 об оказании услуг") is False + + +class TestSafeJsonParse: + def test_valid(self): + d, fix = _safe_json_parse('{"a":1}') + assert d == {"a": 1} + assert fix is False + + def test_markdown(self): + d, fix = _safe_json_parse('```json\n{"a":1}\n```') + assert d == {"a": 1} + assert fix is False + + def test_chatter(self): + d, _ = _safe_json_parse('Вот ответ: {"a":1}. Спасибо!') + assert d == {"a": 1} + + def test_trailing_comma(self): + d, fix = _safe_json_parse('{"a":1,}') + assert d == {"a": 1} + assert fix is True + + def test_trailing_comma_in_list(self): + d, fix = _safe_json_parse('{"a":[1,2,]}') + assert d == {"a": [1, 2]} + assert fix is True + + def test_empty_raises(self): + with pytest.raises(ValueError): + _safe_json_parse("") + + def test_none_raises(self): + with pytest.raises(ValueError): + _safe_json_parse(None) + + +class TestSmartExtract: + def test_paragraphs(self): + ej = [{"type": "paragraph", "text": "Договор № 03700"}] + out = _smart_extract(ej) + assert "Договор" in out + + def test_table(self): + ej = [{"type": "table", "rows": [["Аренда", "50000"]]}] + out = _smart_extract(ej) + assert "Аренда" in out + + def test_string_json(self): + ej = json.dumps([{"type": "paragraph", "text": "Спецификация"}]) + out = _smart_extract(ej) + assert "Спецификация" in out + + def test_empty(self): + assert _smart_extract(None) == "" + assert _smart_extract("") == "" diff --git a/tests/test_connection.py b/tests/test_connection.py new file mode 100644 index 0000000..c75dcdd --- /dev/null +++ b/tests/test_connection.py @@ -0,0 +1,35 @@ +"""Unit-тесты db/connection.py — SQL-конвертация (без БД).""" +from db.connection import _pg_to_sqlite, _extract_table + + +class TestPgToSqlite: + def test_placeholder(self): + assert _pg_to_sqlite("SELECT * FROM x WHERE a = %s") == "SELECT * FROM x WHERE a = ?" + + def test_jsonb(self): + out = _pg_to_sqlite("UPDATE t SET j = %s::jsonb WHERE id = %s") + assert "::jsonb" not in out + + def test_boolean(self): + out = _pg_to_sqlite("CREATE TABLE t (flag BOOLEAN)") + assert "BOOLEAN" not in out + assert "INTEGER" in out + + def test_ilike(self): + assert _pg_to_sqlite("WHERE name ILIKE 'x'") == "WHERE name LIKE 'x'" + + def test_true_false(self): + out = _pg_to_sqlite("SET a = TRUE, b = FALSE") + assert "TRUE" not in out + assert "FALSE" not in out + + +class TestExtractTable: + def test_insert(self): + assert _extract_table("INSERT INTO documents (id) VALUES (1)") == "documents" + + def test_insert_no_space(self): + assert _extract_table("INSERT INTO prompts(id) VALUES (1)") == "prompts" + + def test_no_insert(self): + assert _extract_table("SELECT * FROM documents") is None diff --git a/tests/test_grouping.py b/tests/test_grouping.py new file mode 100644 index 0000000..048e314 --- /dev/null +++ b/tests/test_grouping.py @@ -0,0 +1,83 @@ +"""Unit-тесты services/grouping.py — нормализация + группировка (mock db).""" +from services import grouping + + +class TestNormalizeNumber: + def test_basic(self): + assert grouping.normalize_number("МЭС-123-2024") == "МЭС1232024" + + def test_spaces_slashes(self): + assert grouping.normalize_number("МЭС 123/2024") == "МЭС1232024" + + def test_case(self): + assert grouping.normalize_number("мэс-123") == "МЭС123" + + def test_empty(self): + assert grouping.normalize_number("") == "" + assert grouping.normalize_number(None) == "" + + +class _FakeDocs: + def __init__(self, docs): + self.docs = docs + + def list_by_batch(self, batch_id): + return self.docs + + +class TestGroupDocuments: + def _doc(self, id, doc_type, own, parent, date, cp): + return { + "id": id, "filename": f"{id}.docx", "doc_type": doc_type, + "own_number": own, "parent_number": parent, "doc_date": date, + "counterparty": cp, "classify_status": "classified", + } + + def test_contract_plus_supplement(self, monkeypatch): + docs = [ + self._doc("d1", "contract", "03700_1", None, "2025-01-01", "ЗАО X"), + self._doc("d2", "supplement", "1", "03700_1", "2025-02-01", "ЗАО X"), + ] + monkeypatch.setattr(grouping, "db_docs", _FakeDocs(docs)) + r = grouping.group_documents("b1") + assert r["ok"] is True + assert r["total_docs"] == 2 + groups = [g for g in r["groups"] if g["contract_number"] != "__unresolved__"] + assert len(groups) == 1 + assert groups[0]["contract_number"] == "03700_1" + assert len(groups[0]["documents"]) == 2 + + def test_unmatched_goes_unresolved(self, monkeypatch): + docs = [self._doc("d1", "other", None, None, None, "")] + monkeypatch.setattr(grouping, "db_docs", _FakeDocs(docs)) + r = grouping.group_documents("b1") + unresolved = [g for g in r["groups"] if g["contract_number"] == "__unresolved__"] + assert len(unresolved) == 1 + + +class TestApplyGroups: + def test_apply(self, monkeypatch): + created = [] + + class FakeContracts: + def insert(self, number, client=""): + return {"id": f"c_{number}"} + + class FakeSupps: + def insert(self, cid, did, stype): + created.append((cid, did, stype)) + return {"id": "s"} + + monkeypatch.setattr(grouping, "db_contracts", FakeContracts()) + monkeypatch.setattr(grouping, "db_supplements", FakeSupps()) + + groups = [ + {"contract_number": "03700_1", "counterparty": "X", + "documents": [{"id": "d1"}, {"id": "d2"}]}, + {"contract_number": "__unresolved__", "counterparty": "", + "documents": [{"id": "d3"}]}, + ] + r = grouping.apply_groups("b1", groups) + assert r["ok"] is True + assert r["created"] == 2 + assert created == [("c_03700_1", "d1", "initial"), ("c_03700_1", "d2", "additional")] diff --git a/tests/test_llm.py b/tests/test_llm.py new file mode 100644 index 0000000..2659dc0 --- /dev/null +++ b/tests/test_llm.py @@ -0,0 +1,25 @@ +"""Unit-тесты services/llm.py — call_llm с FakeLLMClient.""" +from services.llm import call_llm +from services.llm_client import FakeLLMClient + + +def _build(cur, txt): + return (f"prompt:{txt}", "pid-1") + + +class TestCallLLM: + def test_plain_json(self): + llm = FakeLLMClient({"default": '{"mode":"partial","ops":[]}'}) + res, pid = call_llm([], "текст", _build, llm_client=llm) + assert pid == "pid-1" + assert res == {"mode": "partial", "ops": []} + + def test_markdown_json(self): + llm = FakeLLMClient({"default": '```json\n{"mode":"full_replace","ops":[]}\n```'}) + res, _ = call_llm([], "текст", _build, llm_client=llm) + assert res["mode"] == "full_replace" + + def test_markdown_generic(self): + llm = FakeLLMClient({"default": '```\n{"a":1}\n```'}) + res, _ = call_llm([], "текст", _build, llm_client=llm) + assert res == {"a": 1} diff --git a/tests/test_llm_client.py b/tests/test_llm_client.py new file mode 100644 index 0000000..5d404e5 --- /dev/null +++ b/tests/test_llm_client.py @@ -0,0 +1,40 @@ +"""Unit-тесты services/llm_client.py.""" +from services.llm_client import FakeLLMClient, HttpxLLMClient + + +class TestFakeLLMClient: + def test_exact_match(self): + c = FakeLLMClient({"hello": "world"}) + assert c.complete("hello") == "world" + + def test_partial_match(self): + c = FakeLLMClient({"needle": "found"}) + assert c.complete("a needle in haystack") == "found" + + def test_default_fallback(self): + c = FakeLLMClient({"default": "fallback"}) + assert c.complete("whatever") == "fallback" + + def test_calls_recorded(self): + c = FakeLLMClient() + c.complete("x") + c.complete("y") + assert c.calls == ["x", "y"] + + def test_add(self): + c = FakeLLMClient() + c.add("k", "v") + assert c.complete("k") == "v" + + +class TestHttpxLLMClient: + def test_init_defaults(self): + c = HttpxLLMClient(url="http://x", key="k") + assert c.model == "gpt-oss-120b" + assert c.timeout == 120 + assert c.max_tokens == 8000 + + def test_init_custom(self): + c = HttpxLLMClient(url="http://x", key="k", model="m", timeout=10) + assert c.model == "m" + assert c.timeout == 10 diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..fc89512 --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,87 @@ +"""Unit-тесты services/metrics.py — чистая логика, без БД.""" +from decimal import Decimal + +import pytest + +from services.metrics import ( + check_arithmetic, + normalize_date, + _to_decimal, + ClassifyMetrics, +) + + +class TestNormalizeDate: + def test_dd_mm_yyyy(self): + assert normalize_date("01.01.2025") == "2025-01-01" + + def test_already_iso(self): + assert normalize_date("2025-01-01") == "2025-01-01" + + def test_empty(self): + assert normalize_date("") is None + assert normalize_date(None) is None + + def test_unrecognized_passthrough(self): + assert normalize_date("01 января 2025") == "01 января 2025" + + +class TestToDecimal: + def test_int(self): + assert _to_decimal("50000") == Decimal("50000") + + def test_russian_number(self): + assert _to_decimal("50 000,00") == Decimal("50000.00") + + def test_nbsp(self): + assert _to_decimal("1\u00a0000,50") == Decimal("1000.50") + + def test_none(self): + assert _to_decimal(None) is None + assert _to_decimal("") is None + + def test_garbage(self): + assert _to_decimal("не число") is None + + +class TestCheckArithmetic: + def test_ok(self): + ops = [{"action": "ADD", "new_row": {"name": "x", "price": 100, "qty": 2, "sum": 200}}] + assert check_arithmetic(ops) == [] + + def test_mismatch(self): + ops = [{"action": "ADD", "new_row": {"name": "x", "price": 100, "qty": 2, "sum": 250}}] + m = check_arithmetic(ops) + assert len(m) == 1 + assert m[0]["expected_sum"] == 200.0 + assert m[0]["actual_sum"] == 250.0 + assert m[0]["diff"] == 50.0 + + def test_update_values(self): + ops = [{"action": "UPDATE", "new_values": {"price": 10, "qty": 3, "sum": 30}}] + assert check_arithmetic(ops) == [] + + def test_skip_incomplete(self): + # нет qty → пропуск + ops = [{"action": "ADD", "new_row": {"name": "x", "price": 100, "sum": 200}}] + assert check_arithmetic(ops) == [] + + def test_skip_non_add_update(self): + ops = [{"action": "DELETE", "target_hash": "h"}] + assert check_arithmetic(ops) == [] + + +class TestClassifyMetrics: + def test_fix_rate(self): + m = ClassifyMetrics() + m.record(False) + m.record(True) + m.record(False) + assert m.total == 3 + assert m.fixes == 1 + assert m.fix_rate == pytest.approx(1 / 3) + + def test_empty(self): + m = ClassifyMetrics() + assert m.fix_rate == 0.0 + assert m.summary()["total_classifications"] == 0 diff --git a/tests/test_parse.py b/tests/test_parse.py new file mode 100644 index 0000000..2a1cb47 --- /dev/null +++ b/tests/test_parse.py @@ -0,0 +1,28 @@ +"""Unit-тесты services/parse.py.""" +from services.parse import parse_file, _parse_text + + +class TestParseText: + def test_plain(self): + r = parse_file("test.txt", "строка1\nстрока2\nстрока3".encode()) + assert r["status"] == "parsed" + assert r["element_count"] == 3 + + def test_text_fn(self): + r = _parse_text(b"a\nb\nc") + assert r["status"] == "parsed" + assert r["element_count"] == 3 + assert r["elements"][0]["text"] == "a" + + def test_no_extension_falls_back_to_text(self): + r = parse_file("noext", b"line1\nline2") + assert r["status"] == "parsed" + assert r["element_count"] == 2 + + def test_docx_invalid_returns_error(self): + r = parse_file("test.docx", b"not a real docx") + assert r["status"] == "error" + + def test_pdf_invalid_returns_error(self): + r = parse_file("test.pdf", b"not a real pdf") + assert r["status"] == "error" diff --git a/tests/test_process_pipeline.py b/tests/test_process_pipeline.py new file mode 100644 index 0000000..436cbb2 --- /dev/null +++ b/tests/test_process_pipeline.py @@ -0,0 +1,96 @@ +"""Интеграционные тесты services/process.py — run_pipeline с Fake LLM.""" +import json + +from db import contracts, documents, supplements, spec_current +from db.connection import query +from services import process + + +def _seed_contract(n_supps=2): + """Создать contract + n документов (parsed) + n supplements.""" + c = contracts.insert("03700_1") + for i in range(n_supps): + d = documents.insert(f"d{i}.docx", "application/octet-stream", "raw", batch_id="b1") + documents.set_parsed(d["id"], [{"type": "paragraph", "text": f"строка {i}"}]) + supplements.insert(c["id"], d["id"], "initial" if i == 0 else "additional") + return c + + +class TestFullReplace: + def test_no_duplicates(self, db, monkeypatch): + c = _seed_contract(2) + + def fake_call(current_spec, doc_text, build_prompt_fn, llm_client=None): + ops = [ + {"action": "ADD", "new_row": {"name": "Аренда стойко-места", "price": 50000, "qty": 1, "sum": 50000}}, + {"action": "ADD", "new_row": {"name": "IP-адрес IPv4", "price": 300, "qty": 8, "sum": 2400}}, + {"action": "ADD", "new_row": {"name": "Канал 1 Гбит/с", "price": 20000, "qty": 1, "sum": 20000}}, + ] + return ({"mode": "full_replace", "ops": ops}, "pid") + + monkeypatch.setattr("services.llm.call_llm", fake_call) + + events = list(process.run_pipeline(c["id"], "", lambda cur, txt: ("p", "pid"))) + + rows = spec_current.list_by_contract(c["id"]) + # 2 full_replace, но без дублей — всегда 3 строки, не 6 + assert len(rows) == 3 + assert {r["name"] for r in rows} == {"Аренда стойко-места", "IP-адрес IPv4", "Канал 1 Гбит/с"} + + # история сохранена: 2 full_replace × 3 ADD = 6 событий + ev = query("SELECT count(*) AS c FROM spec_events WHERE contract_id = %s", (c["id"],)) + assert ev[0]["c"] == 6 + + # не было ошибок извлечения + assert not any(e.get("type") == "extract_error" for e in events) + + +class TestTargetIdTranslation: + def test_update_applies(self, db, monkeypatch): + c = _seed_contract(2) + + def fake_call(current_spec, doc_text, build_prompt_fn, llm_client=None): + if not current_spec: + return ({"mode": "partial", "ops": [ + {"action": "ADD", "new_row": {"name": "Аренда", "price": 50000, "qty": 1, "sum": 50000}}, + ]}, "pid") + return ({"mode": "partial", "ops": [ + {"action": "UPDATE", "target_id": "r1", "new_values": {"price": 55000, "sum": 55000}}, + ]}, "pid") + + monkeypatch.setattr("services.llm.call_llm", fake_call) + + list(process.run_pipeline(c["id"], "", lambda cur, txt: ("p", "pid"))) + + rows = spec_current.list_by_contract(c["id"]) + assert len(rows) == 1 + assert rows[0]["price"] == 55000 # UPDATE применился, а не ушёл в UNRESOLVED + + upd = query("SELECT status FROM spec_events WHERE contract_id = %s AND action = 'UPDATE'", (c["id"],)) + assert upd[0]["status"] == "applied" + + +class TestNoSupplements: + def test_error_event(self, db): + evs = list(process.run_pipeline("nonexistent", "", lambda cur, txt: ("p", "pid"))) + assert any(e.get("type") == "error" for e in evs) + + +class TestElementsToText: + def test_list(self): + ej = [ + {"type": "paragraph", "text": "Привет"}, + {"type": "table", "rows": [["a", "b"], ["c", "d"]]}, + ] + assert process._elements_to_text(ej) == "Привет\na | b\nc | d" + + def test_dict_value(self): + ej = {"Value": [{"type": "paragraph", "text": "X"}]} + assert process._elements_to_text(ej) == "X" + + def test_string_json(self): + ej = json.dumps([{"type": "paragraph", "text": "Y"}]) + assert process._elements_to_text(ej) == "Y" + + def test_plain_string(self): + assert process._elements_to_text("просто текст") == "просто текст" diff --git a/tests/test_routes.py b/tests/test_routes.py new file mode 100644 index 0000000..31cd79a --- /dev/null +++ b/tests/test_routes.py @@ -0,0 +1,85 @@ +"""Интеграционные тесты HTTP-эндпоинтов (Flask test client).""" +import io +import zipfile + +import pytest + + +@pytest.fixture +def client(tmp_path, monkeypatch): + from db import connection as conn + monkeypatch.setattr(conn, "DB_PATH", str(tmp_path / "app.db")) + from app import create_app + app = create_app() + app.config["TESTING"] = True + yield app.test_client() + c = getattr(conn._local, "conn", None) + if c is not None: + try: + c.close() + except Exception: + pass + conn._local.conn = None + + +class TestHealth: + def test_health(self, client): + from config import VERSION + r = client.get("/health") + assert r.status_code == 200 + data = r.get_json() + assert data["ok"] is True + assert data["version"] == VERSION + + +class TestSpecCurrent: + def test_missing_contract_id(self, client): + r = client.get("/api/spec-current") + assert r.status_code == 400 + + def test_empty(self, client): + r = client.get("/api/spec-current?contract_id=missing") + assert r.status_code == 200 + data = r.get_json() + assert data["ok"] is True + assert data["rows"] == [] + + +class TestUpload: + def test_no_file(self, client): + r = client.post("/upload", data={}) + assert r.status_code == 400 + + def test_unsupported_ext(self, client): + data = {"files": (io.BytesIO(b"x"), "test.exe")} + r = client.post("/upload", data=data, content_type="multipart/form-data") + assert r.status_code == 400 + + def test_docx_creates_document(self, client): + data = {"files": (io.BytesIO(b"not a real docx"), "test.docx")} + r = client.post("/upload", data=data, content_type="multipart/form-data") + assert r.status_code == 200 + j = r.get_json() + assert j["ok"] is True + assert j["doc_id"] + + +class TestUnzipUpload: + def test_unzip(self, client): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("a.docx", b"hello") + zf.writestr("b.pdf", b"world") + buf.seek(0) + data = {"files": (buf, "test.zip")} + r = client.post("/unzip-upload", data=data, content_type="multipart/form-data") + assert r.status_code == 200 + j = r.get_json() + assert j["ok"] is True + assert len(j["files"]) == 2 + + +class TestGroups: + def test_missing_batch(self, client): + r = client.get("/api/groups") + assert r.status_code == 400 diff --git a/tests/test_spec_current.py b/tests/test_spec_current.py new file mode 100644 index 0000000..22358e2 --- /dev/null +++ b/tests/test_spec_current.py @@ -0,0 +1,31 @@ +"""Интеграционные тесты db/spec_current.py.""" +import json + +from db import spec_current, documents, spec_events + + +class TestListByContract: + def test_empty(self, db): + assert spec_current.list_by_contract("nope") == [] + + def test_ordered_by_name(self, db): + ops = [ + {"action": "ADD", "new_row": {"name": "Б", "price": 2}}, + {"action": "ADD", "new_row": {"name": "А", "price": 1}}, + ] + spec_events.apply_ops("c1", "s1", "d1", ops, "pid", {}) + rows = spec_current.list_by_contract("c1") + assert [r["name"] for r in rows] == ["А", "Б"] + + +class TestGetElementsJson: + def test_none(self, db): + assert spec_current.get_elements_json("missing") is None + + def test_parsed(self, db): + d = documents.insert("f.docx", "m", "raw", batch_id="b1") + documents.set_parsed(d["id"], [{"type": "paragraph", "text": "x"}]) + ej = spec_current.get_elements_json(d["id"]) + assert ej is not None + parsed = json.loads(ej) + assert parsed[0]["text"] == "x" diff --git a/tests/test_spec_events.py b/tests/test_spec_events.py new file mode 100644 index 0000000..1de1ff9 --- /dev/null +++ b/tests/test_spec_events.py @@ -0,0 +1,107 @@ +"""Интеграционные тесты db/spec_events.py (SQLite, изолированная БД).""" +from db.connection import query +from db import spec_events, spec_current + + +CID = "11111111-1111-1111-1111-111111111111" +SID = "22222222-2222-2222-2222-222222222222" +DID = "33333333-3333-3333-3333-333333333333" + + +def _count(table, contract_id): + rows = query(f"SELECT count(*) AS c FROM {table} WHERE contract_id = %s", (contract_id,)) + return rows[0]["c"] + + +class TestApplyOpsAdd: + def test_add(self, db): + ops = [{"action": "ADD", "new_row": {"name": "Аренда стойко-места", "price": 50000, "qty": 1, "sum": 50000, "date_start": "2025-01-01"}}] + s = spec_events.apply_ops(CID, SID, DID, ops, "pid", {}) + assert s == {"added": 1, "updated": 0, "deleted": 0} + rows = spec_current.list_by_contract(CID) + assert len(rows) == 1 + assert rows[0]["name"] == "Аренда стойко-места" + assert rows[0]["price"] == 50000 + assert rows[0]["date_start"] == "2025-01-01" + + def test_add_empty_name_unresolved(self, db): + ops = [{"action": "ADD", "new_row": {"name": ""}}] + s = spec_events.apply_ops(CID, SID, DID, ops, "pid", {}) + assert s == {"added": 0, "updated": 0, "deleted": 0} + assert spec_current.list_by_contract(CID) == [] + + def test_add_multiple(self, db): + ops = [ + {"action": "ADD", "new_row": {"name": "A", "price": 1}}, + {"action": "ADD", "new_row": {"name": "B", "price": 2}}, + ] + s = spec_events.apply_ops(CID, SID, DID, ops, "pid", {}) + assert s["added"] == 2 + assert len(spec_current.list_by_contract(CID)) == 2 + + +class TestApplyOpsUpdateDelete: + def test_update_and_delete(self, db): + spec_events.apply_ops(CID, SID, DID, [{"action": "ADD", "new_row": {"name": "Аренда", "price": 50000, "qty": 1, "sum": 50000}}], "pid", {}) + h = spec_events._hash("Аренда") + + s = spec_events.apply_ops(CID, SID, DID, [{"action": "UPDATE", "target_hash": h, "new_values": {"price": 55000}}], "pid", {}) + assert s["updated"] == 1 + assert spec_current.list_by_contract(CID)[0]["price"] == 55000 + + s = spec_events.apply_ops(CID, SID, DID, [{"action": "DELETE", "target_hash": h}], "pid", {}) + assert s["deleted"] == 1 + assert spec_current.list_by_contract(CID) == [] + + def test_update_empty_hash_unresolved(self, db): + s = spec_events.apply_ops(CID, SID, DID, [{"action": "UPDATE", "new_values": {"price": 1}}], "pid", {}) + assert s["updated"] == 0 + rows = query("SELECT status FROM spec_events WHERE contract_id = %s", (CID,)) + assert rows[0]["status"] == "unresolved" + + +class TestUnresolvedAndUnknown: + def test_unresolved_action(self, db): + ops = [{"action": "UNRESOLVED", "new_values": {"name": "X"}, "reason": "нет соответствия"}] + spec_events.apply_ops(CID, SID, DID, ops, "pid", {}) + assert spec_current.list_by_contract(CID) == [] + rows = query("SELECT action, status FROM spec_events WHERE contract_id = %s", (CID,)) + assert rows[0]["action"] == "UNRESOLVED" + assert rows[0]["status"] == "unresolved" + + def test_unknown_action(self, db): + spec_events.apply_ops(CID, SID, DID, [{"action": "WHATEVER", "new_row": {"name": "X"}}], "pid", {}) + assert spec_current.list_by_contract(CID) == [] + rows = query("SELECT action FROM spec_events WHERE contract_id = %s", (CID,)) + assert rows[0]["action"] == "UNRESOLVED" + + +class TestClearAndReset: + def test_clear_current_keeps_events(self, db): + spec_events.apply_ops(CID, SID, DID, [{"action": "ADD", "new_row": {"name": "A", "price": 1}}], "pid", {}) + assert _count("spec_current", CID) == 1 + spec_events.clear_current(CID) + assert _count("spec_current", CID) == 0 + assert _count("spec_events", CID) == 1 # история сохранена + + def test_reset_clears_both(self, db): + spec_events.apply_ops(CID, SID, DID, [{"action": "ADD", "new_row": {"name": "A", "price": 1}}], "pid", {}) + spec_events.reset(CID) + assert _count("spec_current", CID) == 0 + assert _count("spec_events", CID) == 0 + + +class TestHashAndSeq: + def test_hash_normalizes(self): + assert spec_events._hash(" Арена стойко-места ") == spec_events._hash("арена стойко-места") + + def test_hash_includes_date(self): + assert spec_events._hash("Аренда", "01.01.2025") != spec_events._hash("Аренда", "01.02.2025") + + def test_hash_len(self): + assert len(spec_events._hash("x")) == 16 + + def test_seq(self, db): + assert spec_events.get_next_seq(CID) == 1 + spec_events.apply_ops(CID, SID, DID, [{"action": "ADD", "new_row": {"name": "A"}}], "pid", {}) + assert spec_events.get_next_seq(CID) == 2 diff --git a/tests/test_upload_security.py b/tests/test_upload_security.py new file mode 100644 index 0000000..6b57ddc --- /dev/null +++ b/tests/test_upload_security.py @@ -0,0 +1,50 @@ +"""Тесты безопасности routes/upload_bp.py — чистые функции, без HTTP.""" +import io +import zipfile + +from routes.upload_bp import _check_ext, _unzip + + +class TestCheckExt: + def test_allowed(self): + assert _check_ext("a.pdf") is None + assert _check_ext("a.docx") is None + assert _check_ext("a.doc") is None + assert _check_ext("a.zip") is None + + def test_rejected(self): + assert _check_ext("a.exe") is not None + assert _check_ext("a.txt") is not None + assert _check_ext("noext") is not None + + def test_uppercase(self): + assert _check_ext("A.PDF") is None + + +class TestUnzip: + def _zip(self, entries): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + for name, content in entries.items(): + zf.writestr(name, content) + return buf.getvalue() + + def test_valid(self): + ok, files, err = _unzip(self._zip({"a.docx": b"hello", "b.pdf": b"world"})) + assert ok is True + assert len(files) == 2 + assert files[0]["filename"] == "a.docx" + assert files[0]["data_b64"] + + def test_path_traversal_neutralized(self): + ok, files, err = _unzip(self._zip({"../etc/passwd": b"evil"})) + assert ok is True + # basename убирает каталог — имя файла не содержит путь + assert files[0]["filename"] == "passwd" + assert ".." not in files[0]["filename"] + assert "/" not in files[0]["filename"] + + def test_invalid_zip(self): + ok, files, err = _unzip(b"not a zip at all") + assert ok is False + assert err == "invalid ZIP archive"