"""运营后台（FR-M6-09，m6-plan §2，决策 M6-2 / M6-4 / M6-7）：独立进程、用户名 + 密码 + TOTP、IP 白名单； 跨企业只读走 bd_ops，对某个企业的操作走它自己的租户事务并写审计；不显示业务内容。""" from __future__ import annotations import base64 import itertools from collections.abc import AsyncIterator from urllib.parse import parse_qs, urlsplit import httpx import pytest from sqlalchemy import text from sqlalchemy.exc import ProgrammingError from bid_doctor.core.crypto import totp from bid_doctor.core.db import TenantContext, ops_tx, tenant_tx from tests.conftest import User, project_with_tender, register, run_parse from tests.test_billing_flow import INVOICE, entries _names = itertools.count(1) async def ops_client(ip: str = "127.0.0.1") -> httpx.AsyncClient: from bid_doctor.ops.app import create_ops_app transport = httpx.ASGITransport(app=create_ops_app(), client=(ip, 51000)) return httpx.AsyncClient(transport=transport, base_url="http://ops/ops/v1") async def new_admin() -> tuple[str, bytes]: from bid_doctor.ops.auth import create_admin username = f"ops{next(_names)}" uri = await create_admin(username, "运营小王", "correct horse battery") b32 = parse_qs(urlsplit(uri).query)["secret"][0] return username, base64.b32decode(b32 + "=" * (-len(b32) % 8)) @pytest.fixture async def ops() -> AsyncIterator[httpx.AsyncClient]: username, secret = await new_admin() async with await ops_client() as c: r = await c.post( "/auth/login", json={"username": username, "password": "correct horse battery", "totp": totp(secret)} ) assert r.status_code == 200, r.text c.headers["X-CSRF-Token"] = c.cookies["bd_ops_csrf"] yield c async def test_login_needs_password_totp_and_allowed_ip() -> None: username, secret = await new_admin() async with await ops_client() as c: bad = await c.post( "/auth/login", json={"username": username, "password": "correct horse battery", "totp": "000000"} ) assert bad.status_code == 401 and bad.json()["code"] == "ops_login_failed" assert (await c.get("/me")).status_code == 401 async with await ops_client("10.1.2.3") as far: denied = await far.post("/auth/login", json={"username": username, "password": "x", "totp": totp(secret)}) assert denied.status_code == 403 and denied.json()["code"] == "ops_ip_denied" # 用户端的 bd_app 读不到运营账号表 async with tenant_tx( TenantContext(tenant_id=__import__("uuid").uuid4(), user_id=__import__("uuid").UUID(int=0)) ) as s: with pytest.raises(ProgrammingError): await s.execute(text("SELECT 1 FROM ops_admin")) async def test_tenants_activation_and_adjust(ops: httpx.AsyncClient, app_client: httpx.AsyncClient) -> None: u: User = await register(app_client, "运营测试甲公司") project_id, doc_id = await project_with_tender(u) assert (await run_parse(u, doc_id))["status"] == "succeeded" found = (await ops.get("/tenants", params={"q": "运营测试甲"})).json()["items"] assert [t["name"] for t in found] == ["运营测试甲公司"] and found[0]["balance"] == 2 assert (await ops.get("/tenants", params={"q": u.phone})).json()["items"][0]["id"] == u.tenant_id detail = (await ops.get(f"/tenants/{u.tenant_id}")).json() assert detail["projects"] == 1 and detail["jobs"]["parse_job"] == {"succeeded": 1} assert "未命名项目" not in str(detail) and project_id not in str(detail) # 不显示项目名称和编号 # 开通：用户提交 → 运营标记到账 → 开通 → 登记发票；全程写审计（操作人是运营） req = (await u.post("/activation-requests", json={"invoice_info": INVOICE})).json() listed = (await ops.get("/activation-requests", params={"status": "submitted"})).json()["items"] assert req["id"] in [x["id"] for x in listed] assert (await ops.get("/tenants", params={"q": req["request_no"]})).json()["items"][0]["id"] == u.tenant_id assert (await ops.post(f"/activation-requests/{req['id']}/invoice", json={"invoice_no": "No.1"})).status_code == 409 assert (await ops.post(f"/activation-requests/{req['id']}/paid")).status_code == 200 r = await ops.post(f"/activation-requests/{req['id']}/activate", json={}) assert r.status_code == 200 and r.json()["subscription"]["status"] == "active", r.text assert (await ops.post(f"/activation-requests/{req['id']}/invoice", json={"invoice_no": "No.1"})).status_code == 200 assert (await u.get("/me")).json()["plan"]["code"] == "pro" # 再申请一份，运营退回（要写原因），用户看得到原因 r2 = (await u.post("/activation-requests", json={"invoice_info": INVOICE})).json() assert (await ops.post(f"/activation-requests/{r2['id']}/reject", json={"note": ""})).status_code == 422 assert ( await ops.post(f"/activation-requests/{r2['id']}/reject", json={"note": "抬头和转账单位不一致"}) ).status_code == 200 mine = (await u.get("/activation-requests")).json()["items"] assert mine[0]["status"] == "rejected" and mine[0]["ops_note"] == "抬头和转账单位不一致" # 调额度：原因必填；不能扣成负数；流水里看得到 assert ( await ops.post(f"/tenants/{u.tenant_id}/quota-adjustments", json={"amount": 20, "reason": ""}) ).status_code == 422 too_much = await ops.post(f"/tenants/{u.tenant_id}/quota-adjustments", json={"amount": -9999, "reason": "测试扣减"}) assert too_much.status_code == 422 assert ( await ops.post(f"/tenants/{u.tenant_id}/quota-adjustments", json={"amount": 20, "reason": "试用补偿"}) ).status_code == 201 adj = (await entries(u))[0] assert (adj["kind"], adj["amount"], adj["reason"]) == ("adjust", 20, "试用补偿") events = (await ops.get("/audit-events", params={"tenant_id": u.tenant_id, "action": "ops."})).json()["items"] actions = {e["action"] for e in events} assert { "ops.tenant.view", "ops.activation.paid", "ops.activation.activate", "ops.activation.reject", "ops.activation.invoice", "ops.quota.adjust", } <= actions assert all(e["actor_kind"] == "ops" and e["actor_name"] == "运营小王" for e in events) async def test_feedback_and_failed_jobs_without_content(ops: httpx.AsyncClient, user: User) -> None: from tests.test_check_flow import checked_project project_id, _, run = await checked_project(user) finding = (await user.get(f"/check-runs/{run['id']}/findings")).json()["items"][0] r = await user.post( "/feedback", json={ "project_id": project_id, "object_type": "finding", "object_id": finding["id"], "kind": "false_positive", "note": "这不是问题（误报）", }, ) assert r.status_code == 201 items = (await ops.get("/feedback", params={"kind": "false_positive"})).json()["items"] mine = next(x for x in items if x["note"] == "这不是问题（误报）") assert mine["title"] == finding["title"] and mine["rule_set_version"] and mine["pages"] assert "quote" not in str(mine) and "text" not in mine # 不带原文 async with ops_tx() as s: # bd_ops 只读：改不了业务表 with pytest.raises(ProgrammingError): await s.execute(text("UPDATE project SET name = 'x'")) assert isinstance((await ops.get("/failed-jobs")).json()["items"], list) async def test_every_endpoint_checks_permission() -> None: """角色只留结构：现在只有管理员（全部权限）；没定义的角色什么权限都没有，接口一律 403，/me 照常给出权限点。""" username, secret = await new_admin() async with ops_tx() as s: await s.execute(text("UPDATE ops_admin SET role = 'sales' WHERE username = :u"), {"u": username}) async with await ops_client() as c: r = await c.post( "/auth/login", json={"username": username, "password": "correct horse battery", "totp": totp(secret)} ) assert r.status_code == 200 and r.json()["permissions"] == [] and r.json()["role"] == "sales" c.headers["X-CSRF-Token"] = c.cookies["bd_ops_csrf"] for path in ("/tenants", "/activation-requests", "/feedback", "/failed-jobs", "/audit-events"): assert (await c.get(path)).status_code == 403, path assert (await c.get("/summary")).json() == {"open_requests": 0, "failed_jobs_7d": 0, "feedback_7d": 0} async with await ops_client() as c: # 管理员 username, secret = await new_admin() me = ( await c.post( "/auth/login", json={"username": username, "password": "correct horse battery", "totp": totp(secret)} ) ).json() assert ( me["role"] == "admin" and me["role_label"] == "管理员" and "audit.view" in me["permissions"] and me["env"] == "test" )