refactor: удалён мёртвый legacy-код загрузки напрямую (/upload, /unzip-upload, _check_ext, _unzip, ALLOWED, константы фронта); загрузка только через ВМ (/api/upload_refs)
This commit is contained in:
+2
-4
@@ -27,14 +27,12 @@ pytest установлен в venv: `/home/naeel/nubes/contracts/.venv/bin/pyth
|
||||
| `test_spec_events.py` | `db/spec_events.py`: `apply_ops` (ADD/UPDATE/DELETE/UNRESOLVED/unknown), `clear_current`, `reset`, `_hash`, `get_next_seq` |
|
||||
| `test_spec_current.py` | `db/spec_current.py`: `list_by_contract` (порядок), `get_elements_json` |
|
||||
| `test_process_pipeline.py` | `services/process.py`: `run_pipeline` с Fake LLM — **full_replace не дублирует строки**, **трансляция target_id→target_hash**, `_elements_to_text` |
|
||||
| `test_routes.py` | HTTP-эндпоинты (Flask test client): `/health`, `/api/spec-current`, `/upload`, `/unzip-upload`, `/api/groups` |
|
||||
| `test_upload_security.py` | `routes/upload_bp.py`: `_check_ext` (допустимые/недопустимые), `_unzip` (path traversal нейтрализуется, битый zip) |
|
||||
| `test_routes.py` | HTTP-эндпоинты (Flask test client): `/health`, `/api/spec-current`, `/api/groups` |
|
||||
|
||||
## Ключевые проверки
|
||||
|
||||
- `test_full_replace_no_duplicates` — два `full_replace` подряд → в `spec_current` 3 строки, а не 6; история `spec_events` (6 событий) сохранена.
|
||||
- `test_update_applies` — `target_id: "r1"` транслируется в `target_hash` → UPDATE применяется (status `applied`), а не уходит в UNRESOLVED.
|
||||
- `test_path_traversal_neutralized` — `../etc/passwd` внутри ZIP → имя файла `passwd` без пути.
|
||||
|
||||
## Изоляция
|
||||
|
||||
@@ -56,7 +54,7 @@ pytest -m "not load" -q # быстрые юнит-тесты без на
|
||||
| `load/test_load_pipeline.py` | массовый `run_pipeline`: N контрактов × M допников (Fake LLM) — проверка, что `full_replace` не дублирует строки в масштабе |
|
||||
| `load/test_load_apply_ops.py` | массовые `apply_ops`: N ADD → N UPDATE → N DELETE, целостность `spec_events` |
|
||||
| `load/test_load_concurrency.py` | конкурентная запись в SQLite (WAL + `busy_timeout`), проверка отсутствия потери данных |
|
||||
| `load/stress_http.py` | standalone HTTP-стресс (`/upload`, `/health`) — параллельные запросы, RPS/ошибки/таймауты |
|
||||
| `load/stress_http.py` | standalone HTTP-стресс (`/health`; загрузка `/upload` удалена — реальный путь `/api/upload_refs`) |
|
||||
|
||||
Масштаб через переменные окружения (дефолты — большие):
|
||||
|
||||
|
||||
@@ -1,7 +1,4 @@
|
||||
"""Интеграционные тесты HTTP-эндпоинтов (Flask test client)."""
|
||||
import io
|
||||
import zipfile
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@@ -45,40 +42,6 @@ class TestSpecCurrent:
|
||||
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")
|
||||
|
||||
@@ -1,50 +0,0 @@
|
||||
"""Тесты безопасности 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"
|
||||
Reference in New Issue
Block a user