153 lines
4.5 KiB
Python
153 lines
4.5 KiB
Python
import asyncio
|
|
import io
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import uuid
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
|
|
|
|
TEST_ID = uuid.uuid4().hex
|
|
TEST_DB = Path(tempfile.gettempdir()) / f"opencloud-{TEST_ID}.sqlite3"
|
|
TEST_UPLOADS = Path(tempfile.gettempdir()) / f"opencloud-{TEST_ID}-uploads"
|
|
|
|
ADMIN_EMAIL = "admin@example.com"
|
|
ADMIN_PASSWORD = "admin-password-123"
|
|
|
|
os.environ["ENVIRONMENT"] = "test"
|
|
os.environ["DATABASE_URL"] = f"sqlite+aiosqlite:///{TEST_DB}"
|
|
os.environ["SECRET_KEY"] = "test-secret-key-that-is-long-enough-for-opencloud"
|
|
os.environ["PUBLIC_BASE_URL"] = "http://testserver"
|
|
os.environ["FRONTEND_URL"] = "http://testserver"
|
|
os.environ["CORS_ORIGINS"] = "http://testserver"
|
|
os.environ["UPLOAD_DIR"] = str(TEST_UPLOADS)
|
|
os.environ["EMAIL_DELIVERY_MODE"] = "console"
|
|
os.environ["ADMIN_EMAIL"] = ADMIN_EMAIL
|
|
os.environ["ADMIN_PASSWORD"] = ADMIN_PASSWORD
|
|
os.environ["ADMIN_USERNAME"] = "测试管理员"
|
|
|
|
from app.database import Base, engine # noqa: E402
|
|
from app.main import app # noqa: E402
|
|
from app.routers import auth as auth_router # noqa: E402
|
|
from app.seed import seed # noqa: E402
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# shared helpers — usable from any test file
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def bearer(token: str) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def png_bytes() -> bytes:
|
|
output = io.BytesIO()
|
|
Image.new("RGB", (200, 200), color=(125, 196, 240)).save(output, format="PNG")
|
|
return output.getvalue()
|
|
|
|
|
|
def capture_email(monkeypatch) -> dict[str, str]:
|
|
"""Patch send_confirmation_email and return a dict that will hold the token."""
|
|
captured: dict[str, str] = {}
|
|
|
|
async def capture(recipient: str, token: str) -> None:
|
|
captured["token"] = token
|
|
|
|
monkeypatch.setattr(auth_router, "send_confirmation_email", capture)
|
|
return captured
|
|
|
|
|
|
def register_user(client: TestClient, email: str, password: str, username: str):
|
|
return client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": password, "username": username},
|
|
)
|
|
|
|
|
|
def confirm_email(client: TestClient, token: str):
|
|
return client.post("/api/v1/auth/confirm-email", json={"token": token})
|
|
|
|
|
|
def login_user(client: TestClient, email: str, password: str):
|
|
return client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": email, "password": password},
|
|
)
|
|
|
|
|
|
def register_and_confirm(
|
|
client: TestClient,
|
|
monkeypatch,
|
|
email: str,
|
|
password: str,
|
|
username: str,
|
|
) -> dict[str, str]:
|
|
"""Create a fully registered + confirmed user. Returns token & user_id."""
|
|
captured = capture_email(monkeypatch)
|
|
r = register_user(client, email, password, username)
|
|
assert r.status_code == 201
|
|
token = captured.get("token")
|
|
assert token
|
|
cr = confirm_email(client, token)
|
|
assert cr.status_code == 200
|
|
lr = login_user(client, email, password)
|
|
assert lr.status_code == 200
|
|
body = lr.json()
|
|
return {"token": body["access_token"], "user_id": body["user"]["id"]}
|
|
|
|
|
|
def admin_token(client: TestClient) -> str:
|
|
r = login_user(client, ADMIN_EMAIL, ADMIN_PASSWORD)
|
|
return r.json()["access_token"]
|
|
|
|
|
|
def upload_cloud(client: TestClient, token: str, **overrides):
|
|
data = {"cloud_type_id": "1", "is_hidden": "false"}
|
|
data.update(overrides)
|
|
return client.post(
|
|
"/api/v1/clouds",
|
|
headers=bearer(token),
|
|
files={"image": ("cloud.png", png_bytes(), "image/png")},
|
|
data=data,
|
|
)
|
|
|
|
|
|
def approve_cloud(client: TestClient, admin_tok: str, cloud_id: str) -> None:
|
|
r = client.patch(
|
|
"/api/v1/admin/clouds/status",
|
|
headers=bearer(admin_tok),
|
|
json={"ids": [cloud_id], "status": "approved"},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _prepare_database() -> None:
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(Base.metadata.create_all)
|
|
await seed()
|
|
|
|
|
|
async def _close_database() -> None:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def client() -> Iterator[TestClient]:
|
|
asyncio.run(_prepare_database())
|
|
with TestClient(app) as test_client:
|
|
yield test_client
|
|
asyncio.run(_close_database())
|
|
TEST_DB.unlink(missing_ok=True)
|
|
shutil.rmtree(TEST_UPLOADS, ignore_errors=True)
|