ai-kefu-admin/tests/test_site_import_api.py

"""site_import API（Task 7）——路由/鉴权/异常映射；管线全打桩（spec §4/§5）。

夹具：conftest 的 db_source_app + super_client（超管）；租户管理员照
test_tenant_admin_api 建法（创建向导带管理员 → 登录 + 首登改密过 409 门）。
task.start / task.get 打桩——本文件只测路由层（鉴权/校验/异常映射/响应
形状），管线本身归 test_site_import_task.py。
"""
from app.platform.site_import import task

TENANTS = "/api/platform/tenants"
LOGIN = "/api/platform/auth/login"
CHANGE_PW = "/api/platform/auth/change-password"

# spec §4 草稿包样例（generator.build 的输出形状——六键整包）
DRAFT = {
    "brand": {"title": "Acme", "color": "#123456",
              "whatsapp": "+8613800138000", "email": "sales@acme.com",
              "welcome": "Hi! What project?", "topics": ["guardrail"],
              "placeholder": "Type your message...", "privacy_url": ""},
    "avatar": None,
    "prompts": {"system": "s...", "company_facts": {"主营": "guardrail"},
                "extractor": "e...", "evaluator": "v..."},
    "wall_tails": {"en": "tail", "zh": "尾巴", "es": None},
    "contact": {"cc_completion": True, "email_light_ask": True},
    "warnings": ["规格判定：不公开（依据：产品页参数表判定）"],
}


def _auth(client):
    return {"Authorization": f"Bearer {client.platform_token}"}


def _bearer(token):
    return {"Authorization": f"Bearer {token}"}


def _create_tenant(client, slug, *, profile="jiancai", admin=None,
                   password=None):
    """创建租户便捷 POST（test_tenant_admin_api._create 同款收敛）。"""
    r = client.post(TENANTS, headers=_auth(client), json={
        "name": f"{slug} 租户", "slug": slug, "profile": profile,
        "admin_username": admin or f"{slug}-admin",
        "admin_password": password or f"{slug}-pass-1",
        "domains": []})
    assert r.status_code == 201, r.text
    return r


def _tenant_admin_token(client, username, old_password,
                        new_password="tn-pass-1"):
    """租户管理员登录 + 首登改密（过 must_change 409 门），返回可用 token。"""
    r = client.post(LOGIN, json={"username": username,
                                 "password": old_password})
    assert r.status_code == 200
    token = r.json()["token"]
    r = client.post(CHANGE_PW, headers=_bearer(token), json={
        "old_password": old_password, "new_password": new_password})
    assert r.status_code == 200
    return token


class _FakeStart:
    """task.start 桩：记录调用 (slug, url, retry_task_id)；error 非空即抛。"""

    def __init__(self, error=None, result="tid-stub"):
        self.error = error
        self.result = result
        self.calls = []

    async def __call__(self, slug, url, *, retry_task_id=None,
                       config_data=None):
        self.calls.append((slug, url, retry_task_id))
        if self.error is not None:
            raise self.error
        return self.result


def _stub_start(monkeypatch, error=None, result="tid-stub"):
    stub = _FakeStart(error, result)
    monkeypatch.setattr(task, "start", stub)
    return stub


def _stub_get(monkeypatch, t):
    """task.get 桩：(slug, task_id) 联合索引语义（他人 slug → None）。"""
    monkeypatch.setattr(
        task, "get",
        lambda slug, task_id: t if (slug, task_id) == (t.slug, t.task_id)
        else None)


class TestPostStart:

    def test_超管新建任务200(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        stub = _stub_start(monkeypatch, result="tid-1")
        r = super_client.post(f"{TENANTS}/demo/site-import",
                              headers=_auth(super_client),
                              json={"url": "https://acme.com/"})
        assert r.status_code == 200, r.text
        assert r.json() == {"task_id": "tid-1"}
        assert stub.calls == [("demo", "https://acme.com/", None)]

    def test_租户管理员自己租户200(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo", admin="demo-admin",
                       password="demo-pass-1")
        token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
        stub = _stub_start(monkeypatch)
        r = super_client.post(f"{TENANTS}/demo/site-import",
                              headers=_bearer(token),
                              json={"url": "https://acme.com/"})
        assert r.status_code == 200, r.text
        assert r.json()["task_id"] == "tid-stub"
        assert stub.calls == [("demo", "https://acme.com/", None)]

    def test_租户管理员打别的租户403(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo", admin="demo-admin",
                       password="demo-pass-1")
        _create_tenant(super_client, "other", profile="blank")
        token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
        stub = _stub_start(monkeypatch)
        # 已存在租户与不存在 slug 都是 403——_tenant_guard 依赖先于处理器
        # （勿断言 404，spec §5）
        assert super_client.post(
            f"{TENANTS}/other/site-import", headers=_bearer(token),
            json={"url": "https://a.com/"}).status_code == 403
        assert super_client.post(
            f"{TENANTS}/ghost/site-import", headers=_bearer(token),
            json={"url": "https://a.com/"}).status_code == 403
        assert stub.calls == []                    # 处理器未触达
        # 对照：超管打不存在 slug → 处理器内租户存在性检查 404
        assert super_client.post(
            f"{TENANTS}/ghost/site-import", headers=_auth(super_client),
            json={"url": "https://a.com/"}).status_code == 404

    def test_未认证401(self, super_client):
        _create_tenant(super_client, "demo")
        assert super_client.post(
            f"{TENANTS}/demo/site-import",
            json={"url": "https://a.com/"}).status_code == 401

    def test_停用租户403(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        r = super_client.patch(f"{TENANTS}/demo", headers=_auth(super_client),
                               json={"status": "disabled"})
        assert r.status_code == 200
        stub = _stub_start(monkeypatch)
        r = super_client.post(f"{TENANTS}/demo/site-import",
                              headers=_auth(super_client),
                              json={"url": "https://a.com/"})
        assert r.status_code == 403
        assert "停用" in r.json()["detail"]
        assert stub.calls == []

    def test_非http协议400(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        stub = _stub_start(monkeypatch)
        r = super_client.post(f"{TENANTS}/demo/site-import",
                              headers=_auth(super_client),
                              json={"url": "ftp://x.com/"})
        assert r.status_code == 400
        assert r.json()["detail"] == "仅支持公网网站地址"
        # spec §5：内网 IP 字面量 POST 即拒（连接层只是二次兜底）
        assert super_client.post(
            f"{TENANTS}/demo/site-import", headers=_auth(super_client),
            json={"url": "http://127.0.0.1/x"}).status_code == 400
        # 空 URL（无 retry 形态）同 400
        assert super_client.post(
            f"{TENANTS}/demo/site-import", headers=_auth(super_client),
            json={}).status_code == 400
        assert stub.calls == []

    def test_频控429(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        _stub_start(monkeypatch, error=task.ImportRateLimited(
            "导入太频繁，15 分钟内最多 3 次，稍后再试"))
        r = super_client.post(f"{TENANTS}/demo/site-import",
                              headers=_auth(super_client),
                              json={"url": "https://a.com/"})
        assert r.status_code == 429
        assert "导入太频繁" in r.json()["detail"]

    def test_并发409(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        _stub_start(monkeypatch, error=task.ConcurrentLimit(
            "已有导入任务进行中，稍后再试"))
        r = super_client.post(f"{TENANTS}/demo/site-import",
                              headers=_auth(super_client),
                              json={"url": "https://a.com/"})
        assert r.status_code == 409
        assert "已有导入任务进行中" in r.json()["detail"]

    def test_retry不存在task404(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        stub = _stub_start(monkeypatch, error=LookupError(
            "task 'nope' not found for tenant 'demo'"))
        # retry 形态预检：task_id 空 / 非法 retry 值 → 400
        assert super_client.post(
            f"{TENANTS}/demo/site-import", headers=_auth(super_client),
            json={"retry": "generate", "task_id": ""}).status_code == 400
        assert super_client.post(
            f"{TENANTS}/demo/site-import", headers=_auth(super_client),
            json={"retry": "crawl", "task_id": "x"}).status_code == 400
        # 任务不存在 → 404（str(exc) 作 detail）
        r = super_client.post(f"{TENANTS}/demo/site-import",
                              headers=_auth(super_client),
                              json={"retry": "generate", "task_id": "nope"})
        assert r.status_code == 404
        assert "nope" in r.json()["detail"]
        assert len(stub.calls) == 1        # 形态预检 400 的两次未触达桩

    def test_retry非generatefailed态409(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        _stub_start(monkeypatch, error=ValueError(
            "任务当前状态不可重试（仅 generate 阶段失败可复活）"))
        r = super_client.post(f"{TENANTS}/demo/site-import",
                              headers=_auth(super_client),
                              json={"retry": "generate", "task_id": "tid-x"})
        assert r.status_code == 409
        assert "不可重试" in r.json()["detail"]


class TestGetTask:

    def test_进行中返回state与pages(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        t = task.ImportTask("demo", "tid-1", "https://a.com/")
        t.state = task.TaskState.CRAWLING
        t.phase = "crawling"
        t.pages_fetched = 3
        _stub_get(monkeypatch, t)
        r = super_client.get(f"{TENANTS}/demo/site-import/tid-1",
                             headers=_auth(super_client))
        assert r.status_code == 200
        assert r.json() == {"status": "crawling", "phase": "crawling",
                            "pages": 3, "draft": None, "error": None}
        # 未认证 401（依赖链先于处理器）
        assert super_client.get(
            f"{TENANTS}/demo/site-import/tid-1").status_code == 401

    def test_完成返回draft草稿包(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo", admin="demo-admin",
                       password="demo-pass-1")
        token = _tenant_admin_token(super_client, "demo-admin", "demo-pass-1")
        t = task.ImportTask("demo", "tid-2", "https://a.com/")
        t.state = task.TaskState.DONE
        t.pages_fetched = 5
        t.draft = DRAFT
        t.finished_at = 1234.5
        _stub_get(monkeypatch, t)
        # 租户管理员角色走 GET（两角色 200 的另一半；超管在其余用例覆盖）
        r = super_client.get(f"{TENANTS}/demo/site-import/tid-2",
                             headers=_bearer(token))
        assert r.status_code == 200
        body = r.json()
        assert body["status"] == "done"
        assert body["draft"] == DRAFT          # spec §4 草稿包整包透传
        assert body["pages"] == 5
        assert body["phase"] is None
        assert body["error"] is None

    def test_失败返回白话error(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        t = task.ImportTask("demo", "tid-4", "https://a.com/")
        t.fail("网站打不开或响应太慢——检查 URL", phase="crawl")
        _stub_get(monkeypatch, t)
        r = super_client.get(f"{TENANTS}/demo/site-import/tid-4",
                             headers=_auth(super_client))
        assert r.status_code == 200
        assert r.json() == {"status": "failed", "phase": None, "pages": 0,
                            "draft": None,
                            "error": "网站打不开或响应太慢——检查 URL"}

    def test_他人taskid404(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        _create_tenant(super_client, "other", profile="blank")
        t = task.ImportTask("demo", "tid-3", "https://a.com/")
        t.state = task.TaskState.GENERATING
        t.phase = "generating"
        _stub_get(monkeypatch, t)
        # (slug, task_id) 联合索引：他人 slug 同 task_id → miss → 404
        assert super_client.get(
            f"{TENANTS}/other/site-import/tid-3",
            headers=_auth(super_client)).status_code == 404
        # 本租户同 task_id → 200
        assert super_client.get(
            f"{TENANTS}/demo/site-import/tid-3",
            headers=_auth(super_client)).status_code == 200

    def test_过期后404(self, super_client, monkeypatch):
        _create_tenant(super_client, "demo")
        monkeypatch.setattr(task, "get", lambda slug, task_id: None)
        r = super_client.get(f"{TENANTS}/demo/site-import/gone",
                             headers=_auth(super_client))
        assert r.status_code == 404
        assert r.json()["detail"] == "任务不存在或已过期"


class TestTenantAiSlot:
    def test_POST传租户配置快照给task_start(self, super_client, monkeypatch):
        """终审 Important #2：租户 ai 组槽位——routes 取最新版 config_json 传入。"""
        _create_tenant(super_client, "demo")
        stub = _stub_start(monkeypatch, result="tid-slot")
        r = super_client.post(f"{TENANTS}/demo/site-import",
                              headers=_auth(super_client),
                              json={"url": "https://acme.com/"})
        assert r.status_code == 200, r.text
        # config_data 透传由签名接受即可证（不炸 = 路由确实传了该参）
        assert stub.calls and stub.calls[0][0] == "demo"