165 lines
6.5 KiB
Python
165 lines
6.5 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from tests.auth_provider import TestAuthProvider, install_test_auth
|
|
|
|
|
|
@pytest.fixture()
|
|
def secure_app(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "tenant.db"))
|
|
monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
|
|
from server.db.database import reset_engine
|
|
from server.state import store as world_store
|
|
|
|
install_test_auth(monkeypatch, "tenant-a-000000000000000000000001")
|
|
world_store._stores.clear()
|
|
reset_engine()
|
|
from server.gateway.app import create_app
|
|
app = create_app()
|
|
yield app
|
|
reset_engine()
|
|
world_store._stores.clear()
|
|
|
|
|
|
def login(client: TestClient, username: str) -> None:
|
|
response = client.post("/api/auth/login", json={
|
|
"method": "password", "username": username, "password": "test",
|
|
})
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_anonymous_business_routes_are_closed(secure_app):
|
|
client = TestClient(secure_app)
|
|
assert client.get("/api/health").status_code == 200
|
|
assert client.get("/api/workspace").status_code == 401
|
|
assert client.get("/api/world/summary").status_code == 401
|
|
assert client.post("/api/chat", json={"text": "hello"}).status_code == 401
|
|
|
|
|
|
def test_public_health_exposes_interface_handshake(secure_app):
|
|
from server.contracts import INTERFACE_VERSION
|
|
|
|
response = TestClient(secure_app).get("/api/health")
|
|
assert response.status_code == 200
|
|
assert response.json()["interfaceVersion"] == INTERFACE_VERSION
|
|
|
|
|
|
def test_shared_project_is_collaborative_but_personal_chat_is_private(secure_app):
|
|
owner = TestClient(secure_app)
|
|
editor = TestClient(secure_app)
|
|
login(owner, "planner")
|
|
login(editor, "collaborator")
|
|
|
|
owner_workspace = owner.get("/api/workspace").json()
|
|
personal_session = next(row for row in owner_workspace["sessions"] if row["scope"] == "personal")
|
|
assert owner.put(
|
|
f"/api/sessions/{personal_session['id']}/messages",
|
|
json={"messages": [{"role": "user", "text": "private"}]},
|
|
).status_code == 200
|
|
assert editor.get(f"/api/sessions/{personal_session['id']}/messages").status_code == 404
|
|
|
|
created = owner.post("/api/projects", json={"id": "proj_shared", "name": "协作项目"})
|
|
assert created.status_code == 200
|
|
project_session = created.json()["session"]
|
|
invited = owner.post(
|
|
"/api/projects/proj_shared/members",
|
|
json={"userId": 1002, "role": "editor"},
|
|
)
|
|
assert invited.status_code == 200
|
|
|
|
shared = editor.get("/api/workspace").json()
|
|
assert [(row["id"], row["role"]) for row in shared["projects"]] == [("proj_shared", "editor")]
|
|
assert any(row["id"] == project_session["id"] for row in shared["sessions"])
|
|
assert editor.put(
|
|
f"/api/sessions/{project_session['id']}/messages",
|
|
json={"messages": [{"role": "user", "text": "shared"}]},
|
|
).status_code == 200
|
|
messages = owner.get(f"/api/sessions/{project_session['id']}/messages").json()["messages"]
|
|
assert messages == [{"role": "user", "text": "shared"}]
|
|
|
|
|
|
def test_replacing_same_session_messages_twice_is_idempotent(secure_app):
|
|
client = TestClient(secure_app)
|
|
login(client, "planner")
|
|
workspace = client.get("/api/workspace").json()
|
|
session_id = next(row["id"] for row in workspace["sessions"] if row["scope"] == "personal")
|
|
first = [{"role": "user", "text": "你是谁？"}]
|
|
second = [*first, {"role": "assistant", "text": "我是工业智核 APS 助手。"}]
|
|
|
|
assert client.put(
|
|
f"/api/sessions/{session_id}/messages", json={"messages": first},
|
|
).status_code == 200
|
|
assert client.put(
|
|
f"/api/sessions/{session_id}/messages", json={"messages": second},
|
|
).status_code == 200
|
|
assert client.get(f"/api/sessions/{session_id}/messages").json()["messages"] == second
|
|
|
|
|
|
def test_viewer_cannot_modify_project(secure_app):
|
|
owner = TestClient(secure_app)
|
|
viewer = TestClient(secure_app)
|
|
login(owner, "planner")
|
|
login(viewer, "viewer")
|
|
owner.post("/api/projects", json={"id": "proj_readonly", "name": "只读项目"})
|
|
owner.post(
|
|
"/api/projects/proj_readonly/members",
|
|
json={"userId": 1003, "role": "viewer"},
|
|
)
|
|
assert viewer.get("/api/projects/proj_readonly/members").status_code == 200
|
|
assert viewer.patch("/api/projects/proj_readonly", json={"name": "越权修改"}).status_code == 403
|
|
assert viewer.post(
|
|
"/api/sessions", json={"projectId": "proj_readonly", "title": "越权会话"},
|
|
).status_code == 403
|
|
|
|
|
|
def test_cross_tenant_token_cannot_access_known_project_id(secure_app, monkeypatch):
|
|
owner = TestClient(secure_app)
|
|
login(owner, "planner")
|
|
owner.post("/api/projects", json={"id": "proj_tenant_a", "name": "租户 A"})
|
|
|
|
import asyncio
|
|
_, token = asyncio.run(TestAuthProvider(
|
|
"tenant-b-000000000000000000000001",
|
|
).login({"username": "collaborator"}))
|
|
tenant_b = TestClient(secure_app)
|
|
tenant_b.cookies.set("aps_session", token)
|
|
|
|
workspace = tenant_b.get("/api/workspace")
|
|
assert workspace.status_code == 200
|
|
assert workspace.json()["projects"] == []
|
|
assert tenant_b.get("/api/projects/proj_tenant_a/members").status_code == 404
|
|
assert tenant_b.patch("/api/projects/proj_tenant_a", json={"name": "越权"}).status_code == 403
|
|
|
|
|
|
def test_chat_worker_preserves_authenticated_project_context(secure_app, monkeypatch):
|
|
from server.auth.context import get_identity
|
|
from server.contracts import AgentReply
|
|
from server.state.projects import get_project_store
|
|
import server.gateway.app as gateway_app
|
|
|
|
async def assert_context(_store, _session_id, _intent, actor="planner"):
|
|
identity = get_identity(required=True)
|
|
snapshot = get_project_store().snapshot(include_messages=False)
|
|
assert identity.user_id == 1001
|
|
assert identity.tenant_uuid == "tenant-a-000000000000000000000001"
|
|
assert snapshot["activeProjectId"] == "proj_context_worker"
|
|
return AgentReply(text="context-ok")
|
|
|
|
monkeypatch.setattr(gateway_app, "handle_intent", assert_context)
|
|
client = TestClient(secure_app)
|
|
login(client, "planner")
|
|
created = client.post("/api/projects", json={
|
|
"id": "proj_context_worker", "name": "线程上下文项目",
|
|
})
|
|
assert created.status_code == 200
|
|
session_id = created.json()["session"]["id"]
|
|
response = client.post("/api/chat", json={
|
|
"sessionId": session_id,
|
|
"projectId": "proj_context_worker",
|
|
"text": "分析项目",
|
|
})
|
|
assert response.status_code == 200
|
|
assert "context-ok" in response.text
|