276 lines
9.7 KiB
Python
276 lines
9.7 KiB
Python
"""Admin panel (/admin) integration tests: auth gate, create, edit, ban.
|
|
|
|
Run against the Docker Compose infrastructure (`docker compose up -d`).
|
|
Covers the server-rendered FastAPI + Jinja2 + HTMX panel mounted in the api.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
from contract_check.core.security.passwords import hash_password
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
async def _seed_user(
|
|
db_session,
|
|
*,
|
|
email: str,
|
|
password: str,
|
|
role: str = "user",
|
|
is_active: bool = True,
|
|
credits_left: int = 0,
|
|
) -> str:
|
|
"""Insert a user and return its id (cleaned up by the per-test session rollback
|
|
semantics; we also delete explicitly to keep tables tidy across tests)."""
|
|
result = await db_session.execute(
|
|
text(
|
|
"INSERT INTO users (email, password_hash, role, is_active, credits_left) "
|
|
"VALUES (:e, :p, :r, :a, :c) RETURNING id"
|
|
),
|
|
{
|
|
"e": email,
|
|
"p": hash_password(password),
|
|
"r": role,
|
|
"a": is_active,
|
|
"c": credits_left,
|
|
},
|
|
)
|
|
await db_session.commit()
|
|
user_id = result.first()[0]
|
|
return str(user_id)
|
|
|
|
|
|
async def _login_as(client: httpx.AsyncClient, email: str, password: str) -> httpx.Response:
|
|
return await client.post(
|
|
"/admin/login", data={"email": email, "password": password}, follow_redirects=False
|
|
)
|
|
|
|
|
|
async def test_admin_login_sets_cookie_for_admin(client: httpx.AsyncClient, db_session) -> None:
|
|
email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=email, password="adminpass-123", role="admin")
|
|
|
|
r = await _login_as(client, email, "adminpass-123")
|
|
assert r.status_code == 303
|
|
assert r.headers["location"] == "/admin/users"
|
|
assert "cc_admin_token" in r.cookies
|
|
|
|
|
|
async def test_admin_login_rejects_non_admin_role(client: httpx.AsyncClient, db_session) -> None:
|
|
email = f"user-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=email, password="userpass-123", role="user")
|
|
|
|
r = await _login_as(client, email, "userpass-123")
|
|
assert r.status_code == 303
|
|
assert "forbidden" in r.headers["location"]
|
|
assert "cc_admin_token" not in r.cookies
|
|
|
|
|
|
async def test_admin_routes_redirect_without_session(client: httpx.AsyncClient) -> None:
|
|
r = await client.get("/admin/users", follow_redirects=False)
|
|
assert r.status_code == 303
|
|
assert r.headers["location"].startswith("/admin/login")
|
|
|
|
|
|
async def test_admin_creates_user_and_redirects_to_detail(
|
|
client: httpx.AsyncClient, db_session
|
|
) -> None:
|
|
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=admin_email, password="adminpass-123", role="admin")
|
|
await _login_as(client, admin_email, "adminpass-123")
|
|
|
|
new_email = f"new-{uuid.uuid4().hex[:8]}@test.local"
|
|
r = await client.post(
|
|
"/admin/users",
|
|
data={
|
|
"email": new_email,
|
|
"password": "newpass-1234",
|
|
"role": "admin",
|
|
"credits_left": "7",
|
|
"is_active": "on",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303
|
|
location = r.headers["location"]
|
|
assert location.startswith("/admin/users/")
|
|
|
|
# Persisted with the chosen role + credits.
|
|
row = (
|
|
await db_session.execute(
|
|
text(
|
|
"SELECT credits_left, role, is_active, password_hash IS NOT NULL "
|
|
"FROM users WHERE email = :e"
|
|
),
|
|
{"e": new_email},
|
|
)
|
|
).first()
|
|
assert row is not None
|
|
assert row[0] == 7
|
|
assert row[1] == "admin"
|
|
assert row[2] is True
|
|
assert row[3] is True
|
|
|
|
# Detail page renders the created user.
|
|
detail = await client.get(location, follow_redirects=False)
|
|
assert detail.status_code == 200
|
|
assert new_email in detail.text
|
|
|
|
|
|
async def test_admin_create_rejects_duplicate_email(client: httpx.AsyncClient, db_session) -> None:
|
|
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
|
existing = f"dup-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=admin_email, password="adminpass-123", role="admin")
|
|
await _seed_user(db_session, email=existing, password="somepass-123")
|
|
await _login_as(client, admin_email, "adminpass-123")
|
|
|
|
r = await client.post(
|
|
"/admin/users",
|
|
data={"email": existing, "password": "anotherpass-123"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 200 # form re-rendered, not a redirect/5xx
|
|
assert "уже существует" in r.text
|
|
|
|
|
|
async def test_admin_create_rejects_short_password(client: httpx.AsyncClient, db_session) -> None:
|
|
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=admin_email, password="adminpass-123", role="admin")
|
|
await _login_as(client, admin_email, "adminpass-123")
|
|
|
|
r = await client.post(
|
|
"/admin/users",
|
|
data={"email": f"short-{uuid.uuid4().hex[:8]}@test.local", "password": "x"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 200
|
|
assert "Пароль короче" in r.text
|
|
|
|
|
|
async def test_admin_new_form_resolves_before_dynamic_route(
|
|
client: httpx.AsyncClient, db_session
|
|
) -> None:
|
|
"""`/admin/users/new` must hit the form route, not be parsed as {user_id}."""
|
|
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=admin_email, password="adminpass-123", role="admin")
|
|
await _login_as(client, admin_email, "adminpass-123")
|
|
|
|
r = await client.get("/admin/users/new", follow_redirects=False)
|
|
assert r.status_code == 200
|
|
assert "Новый пользователь" in r.text
|
|
|
|
|
|
async def test_admin_toggle_active_bans_user(client: httpx.AsyncClient, db_session) -> None:
|
|
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=admin_email, password="adminpass-123", role="admin")
|
|
await _login_as(client, admin_email, "adminpass-123")
|
|
|
|
target_email = f"ban-{uuid.uuid4().hex[:8]}@test.local"
|
|
target_id = await _seed_user(db_session, email=target_email, password="targetpass-12")
|
|
|
|
r = await client.post(
|
|
f"/admin/users/{target_id}/toggle-active",
|
|
headers={"hx-request": "true"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
is_active = (
|
|
await db_session.execute(
|
|
text("SELECT is_active FROM users WHERE id = :u"), {"u": target_id}
|
|
)
|
|
).scalar_one()
|
|
assert is_active is False
|
|
|
|
|
|
async def test_admin_user_detail_shows_profile_defaults(
|
|
client: httpx.AsyncClient, db_session
|
|
) -> None:
|
|
"""User card renders the profile block for a user who never touched the API (006)."""
|
|
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=admin_email, password="adminpass-123", role="admin")
|
|
await _login_as(client, admin_email, "adminpass-123")
|
|
|
|
target_email = f"target-{uuid.uuid4().hex[:8]}@test.local"
|
|
target_id = await _seed_user(db_session, email=target_email, password="targetpass-123")
|
|
|
|
r = await client.get(f"/admin/users/{target_id}")
|
|
assert r.status_code == 200
|
|
assert "Профиль" in r.text
|
|
assert "не выбран" in r.text # language/timezone unset → defaults
|
|
|
|
# Lazy row materialized by the detail render.
|
|
row = (
|
|
await db_session.execute(
|
|
text("SELECT user_id FROM user_profiles WHERE user_id = :u"), {"u": target_id}
|
|
)
|
|
).first()
|
|
assert row is not None
|
|
|
|
|
|
async def test_admin_gift_subscription_then_upload_uses_quota(
|
|
client: httpx.AsyncClient,
|
|
db_session,
|
|
) -> None:
|
|
"""Admin gift subscription → uploads consume quota (ticket 010)."""
|
|
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=admin_email, password="adminpass-123", role="admin")
|
|
await _login_as(client, admin_email, "adminpass-123")
|
|
|
|
target_email = f"target-{uuid.uuid4().hex[:8]}@test.local"
|
|
target_id = await _seed_user(db_session, email=target_email, password="targetpass-123")
|
|
|
|
r = await client.post(
|
|
f"/admin/users/{target_id}/subscription",
|
|
data={"plan_code": "lite", "days": "30"},
|
|
headers={"hx-request": "true"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 200
|
|
assert "Lite" in r.text or "Подписка активирована" in r.text
|
|
|
|
# Verify subscription row exists.
|
|
row = (
|
|
await db_session.execute(
|
|
text("SELECT count(*) FROM subscriptions WHERE user_id = :u AND status = 'active'"),
|
|
{"u": target_id},
|
|
)
|
|
).scalar_one()
|
|
assert int(row) == 1
|
|
|
|
|
|
async def test_admin_refuses_second_active_subscription(
|
|
client: httpx.AsyncClient,
|
|
db_session,
|
|
) -> None:
|
|
"""Cannot grant a second active subscription (partial unique index)."""
|
|
admin_email = f"admin-{uuid.uuid4().hex[:8]}@test.local"
|
|
await _seed_user(db_session, email=admin_email, password="adminpass-123", role="admin")
|
|
await _login_as(client, admin_email, "adminpass-123")
|
|
|
|
target_email = f"target-{uuid.uuid4().hex[:8]}@test.local"
|
|
target_id = await _seed_user(db_session, email=target_email, password="targetpass-123")
|
|
|
|
r1 = await client.post(
|
|
f"/admin/users/{target_id}/subscription",
|
|
data={"plan_code": "lite"},
|
|
headers={"hx-request": "true"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r1.status_code == 200
|
|
|
|
r2 = await client.post(
|
|
f"/admin/users/{target_id}/subscription",
|
|
data={"plan_code": "pro"},
|
|
headers={"hx-request": "true"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r2.status_code == 200
|
|
# The card shows an error toast in the HTMX trigger header.
|
|
assert "already has an active subscription" in r2.headers.get("hx-trigger", "")
|