276 lines
11 KiB
Python
276 lines
11 KiB
Python
"""
|
||
Тесты POST /api/upload — загрузка файлов в сессию.
|
||
|
||
Запуск:
|
||
python3 tests/test_upload.py
|
||
"""
|
||
|
||
import io
|
||
import os
|
||
import sys
|
||
import time
|
||
import json
|
||
import gc
|
||
|
||
# Настраиваем пути
|
||
_site = os.path.join(os.path.dirname(__file__), "..", "site")
|
||
sys.path.insert(0, _site)
|
||
sys.path.insert(0, os.path.dirname(_site)) # project root for drhider
|
||
|
||
from flask import Flask
|
||
from routes.api_bp import api_bp
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Helpers
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def _new_client():
|
||
"""Создать свежий тестовый клиент."""
|
||
app = Flask(__name__)
|
||
app.register_blueprint(api_bp)
|
||
return app.test_client()
|
||
|
||
|
||
def _get_rss():
|
||
"""RSS процесса в байтах (Linux)."""
|
||
try:
|
||
with open("/proc/self/statm", "r") as f:
|
||
parts = f.read().split()
|
||
return int(parts[1]) * 4096
|
||
except Exception:
|
||
return 0
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Базовые тесты
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_no_file():
|
||
"""POST без файла → 400."""
|
||
c = _new_client()
|
||
rv = c.post("/api/upload")
|
||
assert rv.status_code == 400
|
||
data = json.loads(rv.data)
|
||
assert data["ok"] is False
|
||
assert "No file" in data["error"]
|
||
|
||
|
||
def test_no_filename():
|
||
"""POST с файлом без имени → 400."""
|
||
c = _new_client()
|
||
data = {"files": (io.BytesIO(b"hello"), "")}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 400
|
||
resp = json.loads(rv.data)
|
||
assert resp["ok"] is False
|
||
assert "filename" in resp["error"].lower()
|
||
|
||
|
||
def test_single_file():
|
||
"""POST с одним файлом → 200, сессия + count=1."""
|
||
c = _new_client()
|
||
data = {"files": (io.BytesIO(b"hello world"), "test.txt")}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 200
|
||
resp = json.loads(rv.data)
|
||
assert resp["ok"] is True
|
||
assert len(resp["session"]) == 12
|
||
assert resp["count"] == 1
|
||
|
||
|
||
def test_two_files_same_session():
|
||
"""Два файла в одну сессию → count=2."""
|
||
c = _new_client()
|
||
data1 = {"files": (io.BytesIO(b"first"), "file1.txt")}
|
||
rv1 = c.post("/api/upload", data=data1, content_type="multipart/form-data")
|
||
assert rv1.status_code == 200
|
||
sid = json.loads(rv1.data)["session"]
|
||
|
||
data2 = {
|
||
"files": (io.BytesIO(b"second"), "file2.txt"),
|
||
"session": sid,
|
||
}
|
||
rv2 = c.post("/api/upload", data=data2, content_type="multipart/form-data")
|
||
assert rv2.status_code == 200
|
||
resp2 = json.loads(rv2.data)
|
||
assert resp2["session"] == sid
|
||
assert resp2["count"] == 2
|
||
|
||
|
||
def test_fake_session_404():
|
||
"""Несуществующая сессия → 404."""
|
||
c = _new_client()
|
||
data = {
|
||
"files": (io.BytesIO(b"data"), "f.txt"),
|
||
"session": "deadbeef1234",
|
||
}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 404
|
||
resp = json.loads(rv.data)
|
||
assert resp["ok"] is False
|
||
|
||
|
||
def test_empty_file():
|
||
"""Пустой файл (0 байт) → 200."""
|
||
c = _new_client()
|
||
data = {"files": (io.BytesIO(b""), "empty.txt")}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 200
|
||
resp = json.loads(rv.data)
|
||
assert resp["ok"] is True
|
||
assert resp["count"] == 1
|
||
|
||
|
||
def test_cyrillic_filename():
|
||
"""Кириллическое имя файла → 200."""
|
||
c = _new_client()
|
||
content = "кириллица".encode("utf-8")
|
||
data = {"files": (io.BytesIO(content), "договор_№123.txt")}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 200
|
||
resp = json.loads(rv.data)
|
||
assert resp["ok"] is True
|
||
assert resp["count"] == 1
|
||
|
||
|
||
def test_content_type_json():
|
||
"""Ответ — application/json."""
|
||
c = _new_client()
|
||
data = {"files": (io.BytesIO(b"x"), "f.txt")}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 200
|
||
assert "application/json" in rv.content_type
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Большие файлы
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_large_10mb():
|
||
"""Файл 10 MB — загружается без ошибок."""
|
||
c = _new_client()
|
||
size = 10 * 1024 * 1024
|
||
big_data = b"A" * size
|
||
data = {"files": (io.BytesIO(big_data), "big_10mb.bin")}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 200
|
||
resp = json.loads(rv.data)
|
||
assert resp["ok"] is True
|
||
assert resp["count"] == 1
|
||
|
||
|
||
def test_large_50mb():
|
||
"""Файл 50 MB — загружается без ошибок."""
|
||
c = _new_client()
|
||
size = 50 * 1024 * 1024
|
||
big_data = b"B" * size
|
||
data = {"files": (io.BytesIO(big_data), "big_50mb.bin")}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 200
|
||
resp = json.loads(rv.data)
|
||
assert resp["ok"] is True
|
||
assert resp["count"] == 1
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Много мелких файлов
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_many_small_files():
|
||
"""100 файлов по 1 KB в одну сессию."""
|
||
c = _new_client()
|
||
data1 = {"files": (io.BytesIO(b"x" * 1024), "file_000.txt")}
|
||
rv1 = c.post("/api/upload", data=data1, content_type="multipart/form-data")
|
||
sid = json.loads(rv1.data)["session"]
|
||
|
||
for i in range(1, 100):
|
||
data = {
|
||
"files": (io.BytesIO(b"y" * 1024), f"file_{i:03d}.txt"),
|
||
"session": sid,
|
||
}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 200
|
||
resp = json.loads(rv.data)
|
||
assert resp["session"] == sid
|
||
assert resp["count"] == i + 1
|
||
|
||
assert resp["count"] == 100
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Утечка памяти
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_memory_after_large_files():
|
||
"""После загрузки больших файлов память не течёт."""
|
||
c = _new_client()
|
||
gc.collect()
|
||
mem_before = _get_rss()
|
||
|
||
for i in range(10):
|
||
data = {"files": (io.BytesIO(b"M" * (1024 * 1024)), f"memtest_{i}.bin")}
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
assert rv.status_code == 200
|
||
|
||
gc.collect()
|
||
mem_after = _get_rss()
|
||
diff_mb = (mem_after - mem_before) / (1024 * 1024)
|
||
assert diff_mb < 50, f"Memory grew {diff_mb:.1f} MB, expected < 50 MB"
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Бенчмарк скорости
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_upload_speed():
|
||
"""1 MB файл — загрузка < 500 ms."""
|
||
c = _new_client()
|
||
data = {"files": (io.BytesIO(b"S" * (1024 * 1024)), "speed.bin")}
|
||
|
||
start = time.time()
|
||
rv = c.post("/api/upload", data=data, content_type="multipart/form-data")
|
||
elapsed = time.time() - start
|
||
|
||
assert rv.status_code == 200
|
||
assert elapsed < 0.5, f"Upload took {elapsed:.3f}s, expected < 0.5s"
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Основной блок
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
if __name__ == "__main__":
|
||
passed = 0
|
||
failed = 0
|
||
tests = [
|
||
test_no_file,
|
||
test_no_filename,
|
||
test_single_file,
|
||
test_two_files_same_session,
|
||
test_fake_session_404,
|
||
test_empty_file,
|
||
test_cyrillic_filename,
|
||
test_content_type_json,
|
||
test_large_10mb,
|
||
test_large_50mb,
|
||
test_many_small_files,
|
||
test_memory_after_large_files,
|
||
test_upload_speed,
|
||
]
|
||
for test in tests:
|
||
try:
|
||
test()
|
||
print(f" ✅ {test.__name__}")
|
||
passed += 1
|
||
except Exception as e:
|
||
print(f" ❌ {test.__name__}: {e}")
|
||
failed += 1
|
||
print(f"\n{'='*50}")
|
||
print(f"Результат: {passed} OK, {failed} FAIL из {len(tests)}")
|
||
if failed == 0:
|
||
print("✅ ВСЕ ТЕСТЫ ПРОЙДЕНЫ")
|
||
else:
|
||
print("❌ ЕСТЬ ПРОВАЛЫ")
|
||
sys.exit(1)
|