102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
import os
|
|
import sys
|
|
|
|
# SQLite in-memory for tests — no PostgreSQL needed
|
|
os.environ.setdefault("DATABASE_URL", "sqlite:///./test.db")
|
|
os.environ.setdefault("SIGNING_KEY", "test-signing-key-32chars-padding!!")
|
|
os.environ.setdefault("ADMIN_USERNAME", "admin")
|
|
os.environ.setdefault("ADMIN_PASSWORD", "testpass123")
|
|
os.environ.setdefault("PUBLIC_HOST", "http://localhost")
|
|
os.environ.setdefault("ENABLE_STARTUP_MAINTENANCE", "0")
|
|
os.environ.setdefault("LOG_LEVEL", "ERROR")
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
# Patch docker before importing app modules
|
|
_docker_mock = MagicMock()
|
|
_docker_mock.containers.get.side_effect = Exception("no docker in tests")
|
|
_docker_mock.containers.list.return_value = []
|
|
_docker_mock.containers.run.return_value = MagicMock(id="test-container-id", status="running", name="test")
|
|
|
|
sys.modules.setdefault("docker", MagicMock())
|
|
|
|
with patch("docker.from_env", return_value=_docker_mock):
|
|
with patch("runtime.ensure_schema_compatibility", lambda: None):
|
|
from database import Base, get_db
|
|
import main as app_module
|
|
|
|
engine = create_engine(
|
|
"sqlite:///./test.db",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
TestingSessionLocal = sessionmaker(bind=engine)
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
# Create admin user
|
|
from auth import hash_password
|
|
from models import User
|
|
with TestingSessionLocal() as db:
|
|
if not db.query(User).filter(User.username == "admin").first():
|
|
import datetime as _dt
|
|
db.add(User(
|
|
username="admin",
|
|
password_hash=hash_password("testpass123"),
|
|
is_admin=True,
|
|
active=True,
|
|
expires_at=_dt.datetime(2099, 1, 1, tzinfo=_dt.timezone.utc),
|
|
))
|
|
db.commit()
|
|
|
|
|
|
def override_get_db():
|
|
db = TestingSessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
app_module.app.dependency_overrides[get_db] = override_get_db
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def client():
|
|
with patch("runtime.docker_client", return_value=_docker_mock), \
|
|
patch("runtime.ensure_schema_compatibility", lambda: None):
|
|
with TestClient(app_module.app, raise_server_exceptions=False, base_url="https://testserver") as c:
|
|
yield c
|
|
|
|
|
|
def _extract_csrf(client) -> str:
|
|
"""GET / → берём CSRF из HTML и ставим куку вручную."""
|
|
import re
|
|
r = client.get("/", follow_redirects=True)
|
|
assert r.status_code == 200
|
|
m = re.search(r'name=["\']csrf_token["\'][^>]*value=["\']([^"\']+)["\']'
|
|
r'|value=["\']([^"\']+)["\'][^>]*name=["\']csrf_token["\']', r.text)
|
|
if not m:
|
|
m = re.search(r'csrf_token["\']?\s*[=:]\s*["\']([^"\']{10,})["\']', r.text)
|
|
assert m, f"csrf_token not found in HTML: {r.text[:500]}"
|
|
csrf = m.group(1) or m.group(2)
|
|
client.cookies.set("portal_csrf", csrf, domain="testserver")
|
|
return csrf
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def auth_client(client):
|
|
"""Client with admin session cookie."""
|
|
csrf = _extract_csrf(client)
|
|
r = client.post("/login", data={
|
|
"username": "admin",
|
|
"password": "testpass123",
|
|
"csrf_token": csrf,
|
|
}, follow_redirects=True)
|
|
assert r.status_code == 200, f"login failed: {r.status_code} {r.text[:300]}"
|
|
return client
|