116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import pytest
|
|
|
|
app_path = Path(__file__).resolve().parent.parent / "site" / "app.py"
|
|
spec = importlib.util.spec_from_file_location("site_app", app_path)
|
|
site_app = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(site_app)
|
|
|
|
app = site_app.app
|
|
_mock_storage = site_app._mock_storage
|
|
_mock_storage_lock = site_app._mock_storage_lock
|
|
|
|
from upload.backend.session import cleanup, get_files
|
|
import httpx
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
app.config["TESTING"] = True
|
|
app.config["UPLOAD_HTTPX_TRANSPORT"] = httpx.WSGITransport(app=app)
|
|
with app.test_client() as client:
|
|
yield client
|
|
|
|
|
|
def test_health(client):
|
|
res = client.get("/health")
|
|
assert res.status_code == 200
|
|
assert res.data == b"ok"
|
|
|
|
|
|
def test_index_page(client):
|
|
res = client.get("/")
|
|
assert res.status_code == 200
|
|
assert b"upload-btn" in res.data
|
|
assert b"file-picker" in res.data
|
|
|
|
|
|
def test_mock_buffer_crud(client):
|
|
key = "test_item_1"
|
|
# PUT
|
|
res = client.put(f"/mock-buffer/{key}", data=b"binary_payload_123")
|
|
assert res.status_code == 201
|
|
|
|
# GET
|
|
res = client.get(f"/mock-buffer/{key}")
|
|
assert res.status_code == 200
|
|
assert res.data == b"binary_payload_123"
|
|
|
|
# Status
|
|
res = client.get("/mock-buffer/status")
|
|
assert res.status_code == 200
|
|
data = res.get_json()
|
|
assert data["count"] >= 1
|
|
assert any(item["key"] == key for item in data["items"])
|
|
|
|
# DELETE
|
|
res = client.delete(f"/mock-buffer/{key}")
|
|
assert res.status_code == 204
|
|
|
|
# GET after delete -> 404
|
|
res = client.get(f"/mock-buffer/{key}")
|
|
assert res.status_code == 404
|
|
|
|
|
|
def test_full_transit_flow_mock(client):
|
|
"""Тестирует пофайловый транзит через mock-буфер и приём в сессию RAM."""
|
|
with _mock_storage_lock:
|
|
_mock_storage.clear()
|
|
|
|
# Шаг 1: Браузер кладёт файл в mock-буфер
|
|
key = "token_uuid_0"
|
|
file_bytes = b"%PDF-1.4 test document content for transit"
|
|
res = client.put(f"/mock-buffer/{key}", data=file_bytes)
|
|
assert res.status_code == 201
|
|
|
|
# Проверяем, что файл в буфере
|
|
with _mock_storage_lock:
|
|
assert key in _mock_storage
|
|
|
|
# Шаг 2: Браузер вызывает /api/upload_refs для этого файла
|
|
# Используем относительный URL пути /mock-buffer/key
|
|
res = client.post("/api/upload_refs", json={
|
|
"files": [
|
|
{"name": "contract.pdf", "size": len(file_bytes), "url": f"/mock-buffer/{key}"}
|
|
]
|
|
})
|
|
assert res.status_code == 200
|
|
data = res.get_json()
|
|
assert data["ok"] is True
|
|
sid = data["session"]
|
|
assert data["count"] == 1
|
|
assert data["added"] == 1
|
|
|
|
# Шаг 3: Проверяем, что файл переместился в RAM сессии
|
|
files = get_files(sid)
|
|
assert len(files) == 1
|
|
assert files[0] == ("contract.pdf", file_bytes)
|
|
|
|
# Шаг 4: Проверяем, что файл удалился из mock-буфера (RAM буфера очищен)
|
|
with _mock_storage_lock:
|
|
assert key not in _mock_storage
|
|
|
|
# Шаг 5: Проверяем эндпоинт проверки сессии
|
|
res = client.get(f"/api/session/{sid}/files")
|
|
assert res.status_code == 200
|
|
s_data = res.get_json()
|
|
assert s_data["ok"] is True
|
|
assert s_data["session"] == sid
|
|
assert len(s_data["files"]) == 1
|
|
assert s_data["files"][0]["name"] == "contract.pdf"
|
|
assert s_data["files"][0]["size"] == len(file_bytes)
|
|
|
|
cleanup(sid)
|