note-zero/tests/test_api.py

from pathlib import Path

from argon2 import PasswordHasher
from fastapi.testclient import TestClient

from app import main
from app.auth import COOKIE_NAME
from app.config import Settings, UserSeed
from app.curator import Curator
from app.db import Database


WRITE_HEADERS = {"X-Note-Client": "xuxiang-web"}


def test_login_isolation_roles_and_debug_privacy(
    tmp_path: Path, monkeypatch
):
    hasher = PasswordHasher()
    admin_key = "key-admin-test"
    member_key = "key-member-test"
    users = (
        UserSeed(
            id="admin",
            label="管理员",
            role="admin",
            access_key_hash=hasher.hash(admin_key),
        ),
        UserSeed(
            id="member",
            label="用户二",
            role="member",
            access_key_hash=hasher.hash(member_key),
        ),
    )
    settings = Settings(
        data_dir=tmp_path,
        database_path=tmp_path / "api.sqlite3",
        users=users,
        session_secret="session-secret-for-tests",
        deepseek_api_key=None,
        deepseek_base_url="https://api.deepseek.com",
        deepseek_model="deepseek-v4-pro",
        cookie_secure=False,
        auth_disabled=False,
    )
    database = Database(settings.database_path)
    monkeypatch.setattr(main, "settings", settings)
    monkeypatch.setattr(main, "db", database)
    monkeypatch.setattr(main, "curator", Curator(database, settings))

    with TestClient(main.app) as client:
        assert client.post(
            "/api/login", json={"access_key": "wrong"}
        ).status_code == 401

        admin_login = client.post(
            "/api/login", json={"access_key": admin_key}
        )
        assert admin_login.status_code == 200
        assert admin_login.json()["user"]["role"] == "admin"
        admin_cookie = client.cookies.get(COOKIE_NAME)

        member_login = client.post(
            "/api/login", json={"access_key": member_key}
        )
        assert member_login.status_code == 200
        member_cookie = client.cookies.get(COOKIE_NAME)

        client.cookies.set(COOKIE_NAME, admin_cookie)
        admin_fragment = client.post(
            "/api/fragments",
            json={"content": "管理员的记录"},
            headers=WRITE_HEADERS,
        )
        assert admin_fragment.status_code == 201

        client.cookies.set(COOKIE_NAME, member_cookie)
        member_fragment = client.post(
            "/api/fragments",
            json={"content": "用户二的记录"},
            headers=WRITE_HEADERS,
        )
        assert member_fragment.status_code == 201
        assert [
            item["content"]
            for item in client.get("/api/fragments").json()["items"]
        ] == ["用户二的记录"]
        assert client.get("/api/admin/overview").status_code == 403

        run_id = database.start_agent_run(
            "member",
            member_fragment.json()["id"],
            "deepseek-v4-pro",
            "prompt.v1",
            "high",
        )
        database.add_agent_event(
            run_id,
            "member",
            "decision_committed",
            {
                "standalone": False,
                "assessments": [{"title": "私密产物", "maturity": 42}],
            },
        )

        client.cookies.set(COOKIE_NAME, admin_cookie)
        assert [
            item["content"]
            for item in client.get("/api/fragments").json()["items"]
        ] == ["管理员的记录"]
        redacted = client.get(f"/api/admin/runs/{run_id}").json()
        assert redacted["fragment"]["content_visible"] is False
        assert redacted["events"][-1]["payload"] == {
            "standalone": False,
            "assessment_count": 1,
        }

        client.cookies.set(COOKIE_NAME, member_cookie)
        shared = client.patch(
            "/api/account/debug-sharing",
            json={"enabled": True},
            headers=WRITE_HEADERS,
        )
        assert shared.status_code == 200

        client.cookies.set(COOKIE_NAME, admin_cookie)
        visible = client.get(f"/api/admin/runs/{run_id}").json()
        assert visible["fragment"]["content_visible"] is True
        assert visible["fragment"]["content"] == "用户二的记录"

        retry_error = RuntimeError("tool loop exceeded")
        database.finish_agent_run(
            run_id,
            status="error",
            duration_ms=100,
            attempt_count=1,
            model_rounds=4,
            tool_calls=4,
            search_calls=3,
            inspection_calls=1,
            input_tokens=400,
            output_tokens=80,
            reasoning_tokens=60,
            cached_tokens=0,
            error=retry_error,
        )
        database.set_fragment_status(
            member_fragment.json()["id"], "error", str(retry_error)
        )
        queued: list[str] = []
        monkeypatch.setattr(main.curator, "enqueue", queued.append)

        client.cookies.set(COOKIE_NAME, member_cookie)
        assert client.post(
            f"/api/admin/runs/{run_id}/retry",
            headers=WRITE_HEADERS,
        ).status_code == 403

        client.cookies.set(COOKIE_NAME, admin_cookie)
        retried = client.post(
            f"/api/admin/runs/{run_id}/retry",
            headers=WRITE_HEADERS,
        )
        assert retried.status_code == 202
        assert retried.json() == {"queued": True}
        assert queued == [member_fragment.json()["id"]]
        assert client.post(
            f"/api/admin/runs/{run_id}/retry",
            headers=WRITE_HEADERS,
        ).status_code == 409