70 lines
4.2 KiB
Python
70 lines
4.2 KiB
Python
"""Confirmed business conditions flow through learned discovery and reading."""
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from apps.server.model_client import ModelClient
|
|
from apps.server.tests.test_analysis import config, wire
|
|
from apps.server.tests.test_business_briefs import TEXT, answer, confirmed
|
|
from apps.server.tests.test_discovery import BASE, item, search_candidate
|
|
from apps.server.tests.test_learning import client, initialize
|
|
from apps.server.tests.test_postgres import pg_client
|
|
from apps.server.tests.test_reading import NOTE_ID, TITLE, learned_detail
|
|
|
|
|
|
@pytest.fixture(params=["client", "pg_client"])
|
|
def bridge_client(request):
|
|
return request.getfixturevalue(request.param)
|
|
|
|
|
|
def test_business_api_confirmed_snapshot_drives_search_and_reading(bridge_client):
|
|
client = bridge_client
|
|
initialize(client)
|
|
briefs = client.app_reference.state.business_briefs
|
|
briefs.client = ModelClient(config(), http_transport=httpx.MockTransport(lambda request: wire(answer())))
|
|
parsed_response = client.post("/api/business/briefs", json={"request_text": TEXT})
|
|
assert parsed_response.status_code == 201, parsed_response.text
|
|
parsed = parsed_response.json()
|
|
assert parsed["status"] == "draft"
|
|
conditions = confirmed(search_terms=["准备装修", "装修求助"], region="杭州", region_scope="specified").model_dump()
|
|
response = client.post(f"/api/business/briefs/{parsed['id']}/confirm", json=conditions)
|
|
assert response.status_code == 201, response.text
|
|
final = response.json()
|
|
assert final["conditions"] == conditions and final["parent_id"] == parsed["id"]
|
|
assert client.get("/api/business/briefs").json()["total"] == 2
|
|
assert client.get(f"/api/business/briefs/{parsed['id']}").json() == parsed
|
|
_, candidate = search_candidate(client, [item(TITLE, "a")])
|
|
body = {"search_candidate_id": candidate["id"], "max_selected": 1}
|
|
before = len([call for call in client.worker_reference.calls if call[0] == "verify"])
|
|
for brief_id, invalid, status in ((parsed["id"], body, 422), ("missing", body, 404),
|
|
(final["id"], body | {"keyword": "覆盖条件"}, 422),
|
|
(final["id"], body | {"region": "北京"}, 422),
|
|
(final["id"], body | {"max_selected": True}, 422)):
|
|
rejected = client.post(f"/api/business/briefs/{brief_id}/discover", json=invalid)
|
|
assert rejected.status_code == status, rejected.text
|
|
assert len([call for call in client.worker_reference.calls if call[0] == "verify"]) == before
|
|
response = client.post(f"/api/business/briefs/{final['id']}/discover", json=body)
|
|
assert response.status_code == 201, response.text
|
|
run = response.json()
|
|
assert run["status"] == "completed" and run["keyword"] == "准备装修"
|
|
assert run["conditions"] == {"service": "装修服务", "target_customer": "有装修需求的业主", "region": "杭州"}
|
|
assert run["business_brief_id"] == final["id"] and run["business_brief_digest"] == final["digest"]
|
|
assert run["business_search_term_index"] == 0
|
|
assert len([call for call in client.worker_reference.calls if call[0] == "verify"]) == before + 1
|
|
_, detail, _ = learned_detail(client)
|
|
response = client.post(f"/api/discovery/runs/{run['id']}/readings", json={"detail_candidate_id": detail["id"], "note_id": NOTE_ID})
|
|
assert response.status_code == 201, response.text
|
|
reading = response.json()
|
|
assert reading["status"] == "completed"
|
|
assert reading["business_brief_id"] == final["id"] and reading["business_brief_digest"] == final["digest"]
|
|
assert client.get(f"/api/business/briefs/{final['id']}").json() == final
|
|
assert client.get(f"/api/business/briefs/{parsed['id']}").json() == parsed
|
|
# Old callers remain separate immutable rule runs, without fabricated intent.
|
|
old = client.post("/api/discovery/runs", json=body | {"keyword": "装修"}).json()
|
|
assert old["status"] == "completed" and "business_brief_id" not in old
|
|
with TestClient(client.app_reference, base_url=BASE) as anonymous:
|
|
assert anonymous.get("/api/business/briefs").status_code == 401
|
|
assert anonymous.post(f"/api/business/briefs/{final['id']}/discover", json=body).status_code in {401, 403}
|