91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
"""Session fixture for the ani-web test suite.
|
|
|
|
Spins up a throwaway server (127.0.0.1:8399, temp data dir, seeded
|
|
users admin/alice/bob/dave/erin) once per pytest run, tears it down
|
|
afterwards. The real server on :8321 and its data are never touched.
|
|
|
|
Usage: .venv/bin/pytest tests/ -q
|
|
"""
|
|
import json
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).parent.parent
|
|
PORT = 8399
|
|
BASE = f"http://127.0.0.1:{PORT}"
|
|
USERS = (("admin", "adminpw12345"), ("alice", "alicepw12345"), ("bob", "bobpw12345"),
|
|
("dave", "davepw12345"), ("erin", "erinpw12345"))
|
|
|
|
|
|
def _wait_ready():
|
|
for _ in range(120):
|
|
try:
|
|
urllib.request.urlopen(BASE + "/login", timeout=1)
|
|
return
|
|
except Exception:
|
|
time.sleep(0.5)
|
|
raise RuntimeError("server did not start")
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def server():
|
|
# fail fast instead of silently testing a leftover server.
|
|
# SO_REUSEADDR: the suite's own TIME_WAIT sockets (no listener, but
|
|
# thousands of connections) would otherwise wedge this bind for ~60s.
|
|
s = socket.socket()
|
|
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
try:
|
|
s.bind(("127.0.0.1", PORT))
|
|
except OSError:
|
|
pytest.exit(f"port {PORT} is busy - is a leftover test server running?")
|
|
finally:
|
|
s.close()
|
|
|
|
data = Path(tempfile.mkdtemp(prefix="aniweb-test-"))
|
|
os.environ["ANIWEB_DATA"] = str(data)
|
|
# test-tuned throttle: small IP budget + short window so the per-IP test
|
|
# can both trip the gate and watch it expire inside one test run
|
|
os.environ["ANIWEB_IP_FAILS"] = "10"
|
|
os.environ["ANIWEB_LOCK_WINDOW"] = "8"
|
|
sys.path.insert(0, str(ROOT))
|
|
from server import db, users # noqa: E402
|
|
db.init() # schema before users.add / seeding
|
|
for name, pw in USERS:
|
|
users.add(name, pw)
|
|
c = db.connect()
|
|
seed_t = time.time()
|
|
c.execute("INSERT INTO history VALUES (?,?,?,?,?)",
|
|
("admin", "seed-1", 1, seed_t, json.dumps(
|
|
{"title": "Seed Anime", "slug_id": "seed-1", "season_id": 1,
|
|
"ep_id": 1, "lang": "eng", "ep": "1", "pos": 10.0, "t": seed_t})))
|
|
c.commit()
|
|
c.close()
|
|
|
|
proc = subprocess.Popen(
|
|
[sys.executable, "-m", "uvicorn", "server.app:app",
|
|
"--host", "127.0.0.1", "--port", str(PORT)],
|
|
cwd=ROOT, env=os.environ,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
start_new_session=True)
|
|
try:
|
|
_wait_ready()
|
|
yield
|
|
finally:
|
|
os.killpg(proc.pid, signal.SIGTERM)
|
|
try:
|
|
proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
proc.wait()
|
|
shutil.rmtree(data, ignore_errors=True)
|