196 lines
11 KiB
Python
196 lines
11 KiB
Python
"""Merchant task parsing, confirmation, grounded fields and immutable scope."""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import copy
|
||
from dataclasses import replace
|
||
import json
|
||
import sqlite3
|
||
|
||
import httpx
|
||
import pytest
|
||
from fastapi import HTTPException
|
||
from pydantic import ValidationError
|
||
|
||
from apps.server.business_briefs import BusinessBriefConfirmInput, BusinessBriefInput, BusinessBriefService, checked_conditions
|
||
from apps.server.db import Store, StoreConflict, encoded
|
||
from apps.server.model_client import ModelCallError, ModelClient
|
||
from apps.server.tests.test_analysis import PRIMARY, FALLBACK, config, wire
|
||
from apps.server.tests.test_learning import client, initialize
|
||
from apps.server.tests.test_postgres import pg_client
|
||
|
||
TEXT = "在小红书帮我找杭州有装修需求的业主，我提供装修服务，搜索准备装修，不要服务推销。"
|
||
|
||
|
||
def fact(value, evidence=None):
|
||
return {"value": value, "evidence": evidence or value}
|
||
|
||
|
||
def answer():
|
||
return {"platform": fact("小红书"), "service": fact("装修服务"), "target_customer": fact("有装修需求的业主"),
|
||
"search_terms": [fact("准备装修")], "region": fact("杭州"), "region_scope": None,
|
||
"exclude_service_promotions": fact("不要服务推销")}
|
||
|
||
|
||
def confirmed(**changes):
|
||
return BusinessBriefConfirmInput(**({"platform": "xiaohongshu", "service": "装修服务", "target_customer": "有装修需求的业主",
|
||
"search_terms": ["准备装修"], "region": None, "region_scope": "unspecified", "exclude_service_promotions": True} | changes))
|
||
|
||
|
||
@pytest.fixture(params=["client", "pg_client"])
|
||
def brief_client(request):
|
||
return request.getfixturevalue(request.param)
|
||
|
||
|
||
def setup(client, handler):
|
||
initialize(client)
|
||
store = client.app_reference.state.store
|
||
owner = store.one("SELECT id FROM operators")["id"]
|
||
service = BusinessBriefService(store, config())
|
||
service.client = ModelClient(config(), http_transport=httpx.MockTransport(handler))
|
||
return service, owner
|
||
|
||
|
||
def test_business_brief_grounded_parse_immutable_confirm_reuse_and_ownership(brief_client):
|
||
requests = []
|
||
def handler(request):
|
||
payload = json.loads(request.content)
|
||
user = json.loads(payload["messages"][1]["content"])
|
||
assert user == {"request_text": TEXT}
|
||
assert "author" not in user and request.headers["authorization"] == "Bearer " + PRIMARY
|
||
requests.append(user)
|
||
return wire(answer())
|
||
service, owner = setup(brief_client, handler)
|
||
draft = asyncio.run(service.create(TEXT, owner))
|
||
assert draft["kind"] == "parse" and draft["status"] == "draft" and draft["parent_id"] is None and draft["route"] is None
|
||
assert draft["conditions"] == confirmed(region="杭州", region_scope="specified").model_dump()
|
||
assert draft["field_evidence"]["region"] == ["杭州"] and draft["condition_sources"]["service"] == "request_text"
|
||
assert draft["usage"]["total_tokens"] == 140 and draft["cost"]["status"] == "unknown"
|
||
service.client.config = replace(service.client.config, keys=(("primary", "rotated-fixture-key"),))
|
||
assert asyncio.run(service.create(TEXT, owner)) == draft and len(requests) == 1
|
||
original = service.get(draft["id"], owner)
|
||
final = service.confirm(draft["id"], confirmed(), owner)
|
||
assert final["status"] == final["kind"] == "confirmed" and final["parent_id"] == draft["id"] and final["parent_digest"] == draft["digest"]
|
||
assert final["conditions"]["region_scope"] == "unspecified" and final["conditions"]["region"] is None and "region" in final["missing_fields"]
|
||
assert set(final["condition_sources"].values()) == {"maintainer_edit"} and not any(final["field_evidence"].values())
|
||
assert final["attempts"] == [] and final["usage"] is None and final["route"]["execution_started"] is False
|
||
assert all(item["execution_available"] is False for item in final["route"]["follow_up"])
|
||
assert service.get(draft["id"], owner) == original and service.get_confirmed(final["id"], owner) == final
|
||
assert service.list(owner) == [final, draft] and len(requests) == 1
|
||
with pytest.raises(HTTPException) as error:
|
||
service.get_confirmed(draft["id"], owner)
|
||
assert error.value.detail["code"] == "BRIEF_NOT_CONFIRMED"
|
||
for record in (draft, final):
|
||
with pytest.raises(HTTPException) as error:
|
||
service.get_confirmed(record["id"], "other-owner")
|
||
assert error.value.status_code == 404
|
||
expected = StoreConflict if service.store.backend == "postgresql" else sqlite3.IntegrityError
|
||
for query in ("UPDATE business_briefs SET data_json='{}' WHERE id=$1", "DELETE FROM business_briefs WHERE id=$1"):
|
||
with pytest.raises(expected):
|
||
service.store.execute(query, (final["id"],))
|
||
if service.store.backend == "postgresql":
|
||
with pytest.raises(StoreConflict):
|
||
service.store.execute("TRUNCATE business_briefs CASCADE")
|
||
else:
|
||
assert BusinessBriefService(Store(service.store.home), config()).get(final["id"], owner) == final
|
||
assert service.store.one("SELECT version FROM schema_versions WHERE version=8")["version"] == 8
|
||
assert PRIMARY not in encoded(final) and FALLBACK not in encoded(draft)
|
||
|
||
|
||
def test_business_brief_missing_unsupported_and_failed_are_distinct(brief_client):
|
||
mode = {"answer": {field: [] if field == "search_terms" else None for field in answer()}}
|
||
service, owner = setup(brief_client, lambda request: wire(mode["answer"]))
|
||
incomplete = asyncio.run(service.create("帮我找客户", owner))
|
||
assert incomplete["status"] == "needs_clarification" and incomplete["conditions"]["service"] is None
|
||
assert set(incomplete["missing_fields"]) == {"platform", "service", "target_customer", "search_terms", "region", "exclude_service_promotions"}
|
||
assert service.confirm(incomplete["id"], confirmed(), owner)["status"] == "confirmed"
|
||
mode["answer"]["platform"] = fact("抖音")
|
||
mode["answer"]["service"] = fact("美甲服务")
|
||
unsupported = asyncio.run(service.create("抖音寻找美甲服务客户", owner))
|
||
assert unsupported["status"] == "unsupported" and unsupported["conditions"]["service"] == "美甲服务"
|
||
with pytest.raises(HTTPException) as error:
|
||
service.confirm(unsupported["id"], confirmed(), owner)
|
||
assert error.value.detail["code"] == "BRIEF_UNCONFIRMABLE"
|
||
mode["answer"]["service"] = fact("装修服务", "原文没有的依据")
|
||
failed = asyncio.run(service.create("抖音寻找美甲服务客户", owner))
|
||
# Same parsed unsupported input/config is cached; a failed sample needs new input.
|
||
assert failed == unsupported
|
||
failed = asyncio.run(service.create("抖音寻找美甲服务客户。", owner))
|
||
assert failed["status"] == "failed" and failed["error_codes"] == ["MODEL_OUTPUT_INVALID"]
|
||
assert failed["conditions"]["service"] is None and "原文没有" not in encoded(failed)
|
||
with pytest.raises(HTTPException):
|
||
service.confirm(failed["id"], confirmed(), owner)
|
||
|
||
|
||
def test_business_brief_grounding_strict_and_no_invented_defaults():
|
||
valid = answer()
|
||
result = checked_conditions(encoded(valid), TEXT)
|
||
assert result[-1] == "draft"
|
||
variants = [lambda v: v.update(extra="do not persist"), lambda v: v.update(platform=fact("抖音")),
|
||
lambda v: v.update(region=fact("北京", "杭州")), lambda v: v.update(region_scope=fact("全国")),
|
||
lambda v: v.update(search_terms=[fact("伪造搜索词")]), lambda v: v.update(exclude_service_promotions=fact("服务推销")),
|
||
lambda v: v.update(service={"value": "装修服务", "evidence": "装修服务", "extra": True})]
|
||
for mutate in variants:
|
||
changed = copy.deepcopy(valid)
|
||
mutate(changed)
|
||
with pytest.raises(ModelCallError) as error:
|
||
checked_conditions(encoded(changed), TEXT)
|
||
assert error.value.code == "MODEL_OUTPUT_INVALID"
|
||
with pytest.raises(ModelCallError):
|
||
checked_conditions('{"platform":null,"platform":null}', TEXT)
|
||
blank = {field: [] if field == "search_terms" else None for field in valid}
|
||
conditions, _, missing, _, status = checked_conditions(encoded(blank), "装修服务")
|
||
assert status == "needs_clarification" and conditions["platform"] is None and conditions["service"] is None and conditions["region_scope"] == "unspecified"
|
||
assert "region" in missing
|
||
|
||
|
||
def test_business_brief_confirmation_is_closed_explicit_and_region_not_nationwide():
|
||
valid = confirmed().model_dump()
|
||
invalid = [{key: value for key, value in valid.items() if key != field} for field in valid]
|
||
invalid += [valid | update for update in ({"platform": "douyin"}, {"service": "美甲服务"}, {"search_terms": ["装修"] * 4}, {"search_terms": ["装修", "装修"]},
|
||
{"exclude_service_promotions": 1}, {"region": "杭州"}, {"region_scope": "specified"}, {"region_scope": "unrestricted", "region": "全国"},
|
||
{"region_scope": "specified", "region": "全国"}, {"search_terms": ["13800138000"]}, {"unexpected": True})]
|
||
for value in invalid:
|
||
with pytest.raises((ValidationError, HTTPException)):
|
||
BusinessBriefConfirmInput(**value)
|
||
assert confirmed(region_scope="unrestricted").region is None
|
||
assert confirmed(region_scope="specified", region="杭州").region == "杭州"
|
||
|
||
|
||
def test_business_brief_sensitive_input_redacted_before_model_and_failed_fallback_retained(client):
|
||
calls = []
|
||
def handler(request):
|
||
calls.append(request)
|
||
assert "13800138000" not in request.content.decode() and "owner@example.test" not in request.content.decode()
|
||
if len(calls) == 1:
|
||
return httpx.Response(401, json={"error": {"message": "provider-secret-sentinel"}})
|
||
return wire({field: [] if field == "search_terms" else None for field in answer()}, usage=False)
|
||
service, owner = setup(client, handler)
|
||
parsed = asyncio.run(service.create("帮我找业主，我的手机13800138000，邮箱owner@example.test", owner))
|
||
assert len(calls) == 2 and [item["slot"] for item in parsed["attempts"]] == ["primary", "fallback"]
|
||
assert parsed["usage"] is None and parsed["cost"]["amount"] is None
|
||
assert "13800138000" not in encoded(parsed) and "owner@example.test" not in encoded(parsed) and "provider-secret" not in encoded(parsed)
|
||
for text in ("key api_key=fixture-not-real", "token authorization:fixture-not-real", "密码123456", "A" * 2001, " "):
|
||
with pytest.raises((HTTPException, ValidationError)):
|
||
asyncio.run(service.create(text, owner))
|
||
assert len(calls) == 2
|
||
service.client.config = replace(service.client.config, keys=())
|
||
with pytest.raises(HTTPException) as error:
|
||
asyncio.run(service.create(TEXT, owner))
|
||
assert error.value.status_code == 503 and len(calls) == 2
|
||
|
||
|
||
def test_business_brief_busy_is_bounded_no_model_call(client):
|
||
calls = []
|
||
service, owner = setup(client, lambda request: calls.append(request) or wire(answer()))
|
||
async def busy():
|
||
await service.lock.acquire()
|
||
try:
|
||
with pytest.raises(HTTPException) as error:
|
||
await service.create(TEXT, owner)
|
||
assert error.value.status_code == 409
|
||
finally:
|
||
service.lock.release()
|
||
asyncio.run(busy())
|
||
assert not calls
|