328 lines
10 KiB
Python
328 lines
10 KiB
Python
"""Admin billing panel integration tests (ticket 018).
|
|
|
|
Covers: user-card billing sections render, hold clear action, manual refund
|
|
(same service as the API refund), global /admin/invoices list + filters.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json as _json
|
|
import uuid
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
from sqlalchemy import text
|
|
|
|
from contract_check.core.config import get_settings
|
|
from contract_check.core.security.passwords import hash_password
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _toast(response: httpx.Response) -> str:
|
|
"""Decode the HX-Trigger toast (Cyrillic arrives as uXXXX JSON escapes)."""
|
|
return _json.loads(response.headers.get("hx-trigger", "{}")).get("showToast", "")
|
|
|
|
|
|
def _unique() -> str:
|
|
return uuid.uuid4().hex[:8]
|
|
|
|
|
|
async def _seed_admin(db_session) -> str:
|
|
email = f"admin-{_unique()}@test.local"
|
|
result = await db_session.execute(
|
|
text(
|
|
"INSERT INTO users (email, password_hash, role, is_active, credits_left) "
|
|
"VALUES (:e, :p, 'admin', TRUE, 0) RETURNING id"
|
|
),
|
|
{"e": email, "p": hash_password("adminpass-123")},
|
|
)
|
|
await db_session.commit()
|
|
return str(result.first()[0])
|
|
|
|
|
|
async def _seed_user(db_session, *, billing_hold: bool = False, credits: int = 0) -> str:
|
|
email = f"user-{_unique()}@test.local"
|
|
result = await db_session.execute(
|
|
text(
|
|
"INSERT INTO users (email, password_hash, role, is_active, credits_left, billing_hold) "
|
|
"VALUES (:e, :p, 'user', TRUE, :c, :h) RETURNING id"
|
|
),
|
|
{"e": email, "p": hash_password("userpass-123"), "c": credits, "h": billing_hold},
|
|
)
|
|
await db_session.commit()
|
|
return str(result.first()[0])
|
|
|
|
|
|
async def _seed_invoice(
|
|
db_session,
|
|
user_id: str,
|
|
*,
|
|
status: str = "succeeded",
|
|
kind: str = "topup",
|
|
amount: int = 19900,
|
|
credits: int | None = 1,
|
|
external_id: str | None = None,
|
|
) -> str:
|
|
result = await db_session.execute(
|
|
text(
|
|
"INSERT INTO invoices "
|
|
"(user_id, amount, status, kind, credits_purchased, provider, external_id, paid_at) "
|
|
"VALUES (:u, :a, :s, :k, :c, 'yookassa', :e, :p) RETURNING id"
|
|
),
|
|
{
|
|
"u": user_id,
|
|
"a": amount,
|
|
"s": status,
|
|
"k": kind,
|
|
"c": credits,
|
|
"e": external_id or f"pay-adm-{_unique()}",
|
|
"p": dt.datetime.now(tz=dt.UTC) if status == "succeeded" else None,
|
|
},
|
|
)
|
|
await db_session.commit()
|
|
return str(result.first()[0])
|
|
|
|
|
|
async def _login(client: httpx.AsyncClient, email: str) -> None:
|
|
r = await client.post(
|
|
"/admin/login", data={"email": email, "password": "adminpass-123"}, follow_redirects=False
|
|
)
|
|
assert r.status_code == 303
|
|
|
|
|
|
@pytest.fixture
|
|
async def admin_client(client: httpx.AsyncClient, db_session) -> httpx.AsyncClient:
|
|
await _seed_admin(db_session)
|
|
admin_email = (
|
|
await db_session.execute(
|
|
text("SELECT email FROM users WHERE role = 'admin' ORDER BY created_at DESC LIMIT 1")
|
|
)
|
|
).scalar_one()
|
|
await _login(client, admin_email)
|
|
return client
|
|
|
|
|
|
async def test_user_card_renders_billing_sections(
|
|
admin_client: httpx.AsyncClient, db_session
|
|
) -> None:
|
|
user_id = await _seed_user(db_session, credits=5)
|
|
await _seed_invoice(db_session, user_id)
|
|
await db_session.execute(
|
|
text(
|
|
"INSERT INTO credit_events (user_id, delta, kind, balance_after) "
|
|
"VALUES (:u, 5, 'admin_grant', 5)"
|
|
),
|
|
{"u": user_id},
|
|
)
|
|
await db_session.commit()
|
|
|
|
r = await admin_client.get(f"/admin/users/{user_id}")
|
|
assert r.status_code == 200
|
|
assert "Счета" in r.text
|
|
assert "История кредитов" in r.text
|
|
assert "topup" in r.text
|
|
assert "admin_grant" in r.text
|
|
assert "billing hold" not in r.text # no hold → no badge
|
|
|
|
|
|
async def test_billing_hold_badge_and_clear_action(
|
|
admin_client: httpx.AsyncClient, db_session
|
|
) -> None:
|
|
user_id = await _seed_user(db_session, billing_hold=True)
|
|
|
|
r = await admin_client.get(f"/admin/users/{user_id}")
|
|
assert r.status_code == 200
|
|
assert "billing hold" in r.text
|
|
assert "Снять hold" in r.text
|
|
|
|
r = await admin_client.post(
|
|
f"/admin/users/{user_id}/billing-hold/clear",
|
|
headers={"hx-request": "true"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 200
|
|
assert _toast(r) == "Hold снят"
|
|
|
|
hold = (
|
|
await db_session.execute(
|
|
text("SELECT billing_hold FROM users WHERE id = :u"), {"u": user_id}
|
|
)
|
|
).scalar_one()
|
|
assert hold is False
|
|
assert "billing hold" not in r.text # badge gone after refresh
|
|
|
|
|
|
@respx.mock
|
|
async def test_admin_refund_uses_same_service_as_api(
|
|
admin_client: httpx.AsyncClient, db_session
|
|
) -> None:
|
|
"""Admin refund runs execute_refund — invoice refunded, credits clawed back."""
|
|
import os
|
|
|
|
user_id = await _seed_user(db_session, credits=1)
|
|
invoice_id = await _seed_invoice(db_session, user_id, credits=1, amount=19900)
|
|
|
|
os.environ["YOOKASSA_ENABLED"] = "true"
|
|
os.environ["YOOKASSA_SHOP_ID"] = "123456"
|
|
os.environ["YOOKASSA_SECRET_KEY"] = "test-secret"
|
|
get_settings.cache_clear()
|
|
|
|
external_id = (
|
|
await db_session.execute(
|
|
text("SELECT external_id FROM invoices WHERE id = :i"), {"i": invoice_id}
|
|
)
|
|
).scalar_one()
|
|
|
|
respx.post("https://api.yookassa.ru/v3/refunds").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={"id": "ref-1", "payment_id": external_id, "status": "succeeded"},
|
|
)
|
|
)
|
|
|
|
r = await admin_client.post(
|
|
f"/admin/users/{user_id}/refund",
|
|
data={"invoice_id": invoice_id},
|
|
headers={"hx-request": "true"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 200
|
|
assert _toast(r).startswith("Возврат выполнен")
|
|
|
|
status_db = (
|
|
await db_session.execute(
|
|
text("SELECT status FROM invoices WHERE id = :i"), {"i": invoice_id}
|
|
)
|
|
).scalar_one()
|
|
assert status_db == "refunded"
|
|
|
|
# Clawback removed the purchased credit (balance 1 → 0), clawback event written.
|
|
balance = (
|
|
await db_session.execute(
|
|
text("SELECT credits_left FROM users WHERE id = :u"), {"u": user_id}
|
|
)
|
|
).scalar_one()
|
|
assert balance == 0
|
|
events = (
|
|
await db_session.execute(
|
|
text("SELECT count(*) FROM credit_events WHERE kind = 'clawback' AND invoice_id = :i"),
|
|
{"i": invoice_id},
|
|
)
|
|
).scalar_one()
|
|
assert int(events) == 1
|
|
|
|
os.environ.pop("YOOKASSA_ENABLED", None)
|
|
os.environ.pop("YOOKASSA_SHOP_ID", None)
|
|
os.environ.pop("YOOKASSA_SECRET_KEY", None)
|
|
get_settings.cache_clear()
|
|
|
|
|
|
@respx.mock
|
|
async def test_admin_refund_full_override_skips_calculator(
|
|
admin_client: httpx.AsyncClient, db_session
|
|
) -> None:
|
|
import os
|
|
|
|
user_id = await _seed_user(db_session, credits=0)
|
|
# Heavy usage would make the calculator refuse (used > purchased)…
|
|
invoice_id = await _seed_invoice(db_session, user_id, credits=1, amount=19900)
|
|
await db_session.execute(
|
|
text(
|
|
"INSERT INTO credit_events (user_id, delta, kind, balance_after) "
|
|
"VALUES (:u, -1, 'reserve', 0)"
|
|
),
|
|
{"u": user_id},
|
|
)
|
|
await db_session.commit()
|
|
|
|
os.environ["YOOKASSA_ENABLED"] = "true"
|
|
os.environ["YOOKASSA_SHOP_ID"] = "123456"
|
|
os.environ["YOOKASSA_SECRET_KEY"] = "test-secret"
|
|
get_settings.cache_clear()
|
|
|
|
external_id = (
|
|
await db_session.execute(
|
|
text("SELECT external_id FROM invoices WHERE id = :i"), {"i": invoice_id}
|
|
)
|
|
).scalar_one()
|
|
route = respx.post("https://api.yookassa.ru/v3/refunds").mock(
|
|
return_value=httpx.Response(
|
|
200, json={"id": "ref-2", "payment_id": external_id, "status": "succeeded"}
|
|
)
|
|
)
|
|
|
|
r = await admin_client.post(
|
|
f"/admin/users/{user_id}/refund",
|
|
data={"invoice_id": invoice_id, "full": "on"},
|
|
headers={"hx-request": "true"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 200
|
|
assert _toast(r).startswith("Возврат выполнен")
|
|
# Provider got the FULL amount despite 100% usage.
|
|
body = route.calls.last.request.read()
|
|
assert b"199.00" in body
|
|
|
|
status_db = (
|
|
await db_session.execute(
|
|
text("SELECT status FROM invoices WHERE id = :i"), {"i": invoice_id}
|
|
)
|
|
).scalar_one()
|
|
assert status_db == "refunded"
|
|
|
|
os.environ.pop("YOOKASSA_ENABLED", None)
|
|
os.environ.pop("YOOKASSA_SHOP_ID", None)
|
|
os.environ.pop("YOOKASSA_SECRET_KEY", None)
|
|
get_settings.cache_clear()
|
|
|
|
|
|
async def test_admin_refund_already_refunded_shows_error(
|
|
admin_client: httpx.AsyncClient, db_session
|
|
) -> None:
|
|
user_id = await _seed_user(db_session)
|
|
invoice_id = await _seed_invoice(db_session, user_id, status="refunded")
|
|
|
|
r = await admin_client.post(
|
|
f"/admin/users/{user_id}/refund",
|
|
data={"invoice_id": invoice_id},
|
|
headers={"hx-request": "true"},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 200
|
|
assert _toast(r) == "Счёт уже возвращён"
|
|
|
|
|
|
async def test_admin_invoices_list_filters_and_pagination(
|
|
admin_client: httpx.AsyncClient, db_session
|
|
) -> None:
|
|
user_id = await _seed_user(db_session)
|
|
await _seed_invoice(db_session, user_id, status="succeeded", kind="topup")
|
|
await _seed_invoice(db_session, user_id, status="pending", kind="subscription", credits=None)
|
|
|
|
r = await admin_client.get("/admin/invoices")
|
|
assert r.status_code == 200
|
|
assert "Счета" in r.text
|
|
assert "<td>topup</td>" in r.text
|
|
assert "<td>subscription</td>" in r.text
|
|
|
|
r = await admin_client.get("/admin/invoices?status_filter=succeeded")
|
|
assert r.status_code == 200
|
|
assert "<td>topup</td>" in r.text
|
|
assert "pill warn" not in r.text # pending rows filtered out
|
|
|
|
r = await admin_client.get("/admin/invoices?kind=subscription")
|
|
assert r.status_code == 200
|
|
assert "<td>subscription</td>" in r.text
|
|
assert "<td>topup</td>" not in r.text
|
|
|
|
r = await admin_client.get("/admin/invoices?status_filter=bogus")
|
|
assert r.status_code == 200 # unknown filter → ignored, all rows shown
|
|
|
|
|
|
async def test_admin_invoices_requires_admin(client: httpx.AsyncClient, db_session) -> None:
|
|
r = await client.get("/admin/invoices", follow_redirects=False)
|
|
assert r.status_code == 303
|
|
assert r.headers["location"].startswith("/admin/login")
|