mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 20:21:58 +00:00
ThearchyHelios
ba28dee1f0
- api/copilot.py：event_stream 原样搬成 async generator answer_ops(payload, *, catalog_view=None)，交出的 op dict 不变；开流前的准备（数据源、目录检查器、助手模型、历史、查看服务）用自己的短会话，在第一条操作之前做完。SSE 路由只包一层：先取第一条，开流前的拒绝（模型没配好、限定的数据源都不在）照旧是 HTTP 错误
- catalog_view 一路传给查看服务和自查的 SQL 检查器（目录视图接缝的「生成过程从参数读」）
- providers/usage.py：Meter 是一个 LangChain 回调，和 get_usage_metadata_callback 同一个机制（登记过的上下文变量），按模型累计 tokens、次数，金额按价格目录；估不出的记 unpriced_calls。没直接用 get_usage_metadata_callback：with 块跨 yield 会把变量留在调用方上下文里，调用方随后发起的运行会继承它、被重复计一遍；metered 只在每一步 __anext__ 期间挂上，嵌套时并进外层
- 生成过程读到 done / look / ask 就关模型的流（后面的闲话不要），流末尾才报的用量拿不到：这一次照样计次数和已报的 token，记成估不出，不悄悄少算（遗留：真实接入上建图那几次多半是下限）
- providers/budget.py：BudgetFallback.astream 用 aclosing 包住里面那条流——以前调用方关流时里面那条要等垃圾回收才关，模型那次调用的收尾（用量回调）落在算完账之后；测试去掉它回跑会少计一次
- answer_ops 末尾多一条 {op: usage, by_model, input_tokens, output_tokens, calls, cost_usd, unpriced_calls}；/copilot/review 拆出 review_of_run（问数据页、评测共用），结果带 usage（没调模型时全 0）和 retry_instruction（要不要带着原图重来、说什么，按 chat/turn.py 的决定算；ReviewIn 多一个 attempt）
- 新增 app/chat/turn.py（问数据一轮的决定，复核这边先用上；final.next、对照测试、前端照做在下一个提交）；新增 tests/test_copilot_answer_ops.py（八种结局的 SSE 输出和抽出来之前录下的 fixtures/copilot_answer_ops_baseline.json 逐条相同，末尾多一条 usage；直接迭代和 SSE 同一串；开流前的拒绝仍是 HTTP 错误）、tests/test_usage_meter.py；读 SSE 的十一个测试文件的收流帮手跳过末尾那条 usage
388 lines
21 KiB
Python
388 lines
21 KiB
Python
"""对话里的纠正变成目录修改提案（阶段 4A）。
|
||
|
||
用户在助手（画布）或问数据页说出一条数据事实（「status=9 表示作废」），模型发一条 catalog_patch 操作；服务端
|
||
核对数据源、表、路径和取值格式，附上当前版本、改前和改后的值转给前端——**服务端绝不自动写入**，要人在卡片上点
|
||
「保存到数据目录」，带着读到的版本走 /catalog/{table}/patch（记为人工填写、已确认）。
|
||
|
||
- 纯函数：plan_patch 逐项核对并算出改前、改后；码值是补充，不整份替换；新增关联关系写 relations.new，编号按两端算；
|
||
计算公式不进目录。
|
||
- 流：合法的转出、带版本和改前值；不合法的丢弃并记日志；写在 reply 后面的提案也排到 reply 前面转出（前端收到
|
||
reply 就收尾）；只有提案、没有回答时补一句回答；全程不写库。
|
||
- 接口：预览只读；保存带 if_version，版本不符 409，X-Actor 记成署名。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import uuid
|
||
from types import SimpleNamespace
|
||
from urllib.parse import quote
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
|
||
from app.data import catalog
|
||
from app.data.engine import engines
|
||
from app.data.introspect import introspect
|
||
from tests.fixtures.catalog import scenic_notes
|
||
from tests.fixtures.sources import drop_source
|
||
|
||
_CACHE: dict[str, dict] = {}
|
||
|
||
|
||
async def _schema(path: str) -> dict:
|
||
if path not in _CACHE:
|
||
probe = SimpleNamespace(id=f"probe-{uuid.uuid4().hex[:6]}", name="scenic", kind="sqlite", database=path,
|
||
host=None, port=None, username=None, password=None, options={}, readonly=True,
|
||
description="", schema_cache={}, origin="manual")
|
||
try:
|
||
_CACHE[path] = await introspect(probe)
|
||
finally:
|
||
await engines.invalidate(probe.id)
|
||
return _CACHE[path]
|
||
|
||
|
||
def _visits() -> dict:
|
||
return scenic_notes.notes()["visits"]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# plan_patch / apply_patch：纯函数
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_plan_patch_reports_before_and_after(scenic_db):
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
plan = catalog.plan_patch(_visits(), [
|
||
{"path": "columns.visitor_count.measure", "value": "stock", "reason": "在园人数是存量，不能跨天相加"},
|
||
{"path": "kind", "value": "fact"},
|
||
], table="visits", tables=tables)
|
||
assert plan.problems == []
|
||
measure, kind = plan.changes
|
||
assert (measure.path, measure.before, measure.before_status, measure.after) == (
|
||
"columns.visitor_count.measure", "flow", "confirmed", "stock")
|
||
assert measure.state == "change" and measure.reason == "在园人数是存量，不能跨天相加"
|
||
assert (kind.path, kind.before, kind.before_status, kind.after, kind.state) == (
|
||
"kind", None, None, "fact", "change")
|
||
|
||
|
||
async def test_codes_are_supplemented_not_replaced(scenic_db):
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
plan = catalog.plan_patch(_visits(), [{"path": "columns.status.codes", "value": {"9": "作废"}}],
|
||
table="visits", tables=tables)
|
||
[change] = plan.changes
|
||
assert change.before == {"1": "有效", "0": "作废"}
|
||
assert change.after == {"1": "有效", "0": "作废", "9": "作废"}
|
||
# 交回保存的是原样的补充值：409 之后在最新的目录上重新补
|
||
assert change.value == {"9": "作废"}
|
||
|
||
|
||
async def test_codes_with_pending_meanings(scenic_db):
|
||
"""数据剖析写进来的码值候选含义为空（界面上写「含义待填写」）：提案可以补上含义；提案里含义为空的码只在原来
|
||
没有时加进去，不抹掉已有的含义；含义为空的码合法，保存不报错。"""
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
profiled = _visits()
|
||
profiled["columns"]["status"]["codes"] = catalog.make_item({"1": "有效", "0": "", "9": ""}, "profile", "proposed")
|
||
plan = catalog.plan_patch(profiled, [{"path": "columns.status.codes", "value": {"9": "作废", "1": ""}}],
|
||
table="visits", tables=tables)
|
||
assert plan.problems == []
|
||
[change] = plan.changes
|
||
assert change.after == {"1": "有效", "0": "", "9": "作废"} and change.state == "change"
|
||
# 只给了含义为空的新码：照样加进去，含义待人填
|
||
plan = catalog.plan_patch(profiled, [{"path": "columns.status.codes", "value": {"7": ""}}], table="visits",
|
||
tables=tables)
|
||
assert plan.changes[0].after == {"1": "有效", "0": "", "9": "", "7": ""}
|
||
# 只给了含义为空的已有码：什么都没变，这一项是「保存即确认」（原来是推断）
|
||
plan = catalog.plan_patch(profiled, [{"path": "columns.status.codes", "value": {"1": ""}}], table="visits",
|
||
tables=tables)
|
||
assert plan.changes[0].state == "confirm"
|
||
out = catalog.apply_patch(profiled, [{"path": "columns.status.codes", "value": {"9": "作废"}}], table="visits",
|
||
tables=tables)
|
||
codes = out["columns"]["status"]["codes"]
|
||
assert codes["value"] == {"1": "有效", "0": "", "9": "作废"} and (codes["source"], codes["status"]) == (
|
||
"human", "confirmed")
|
||
|
||
|
||
async def test_new_relation_gets_its_id_from_both_ends(scenic_db):
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
value = {"columns": ["visit_id"], "to_table": "visits", "to_columns": ["id"], "cardinality": "many_to_one"}
|
||
plan = catalog.plan_patch({}, [{"path": "relations.new", "value": value}], table="channel_visits", tables=tables)
|
||
[change] = plan.changes
|
||
rid = catalog.relation_id("channel_visits", ["visit_id"], "visits", ["id"])
|
||
assert change.path == f"relations.{rid}" and change.before is None
|
||
assert change.after == {**value, "coverage": None}
|
||
# 已有编号、两端对得上：改基数
|
||
notes = catalog.apply_patch({}, [{"path": "relations.new", "value": value}], table="channel_visits", tables=tables)
|
||
again = catalog.plan_patch(notes, [{"path": f"relations.{rid}", "value": {**value, "cardinality": "one_to_one"}}],
|
||
table="channel_visits", tables=tables)
|
||
assert again.changes[0].before["cardinality"] == "many_to_one" and again.problems == []
|
||
# 编号对不上两端：不认（改指向是另一条关系）
|
||
moved = catalog.plan_patch(notes, [{"path": f"relations.{rid}",
|
||
"value": {**value, "to_table": "channels", "to_columns": ["id"]}}],
|
||
table="channel_visits", tables=tables)
|
||
assert moved.changes == [] and len(moved.problems) == 1
|
||
|
||
|
||
async def test_invalid_changes_are_reported(scenic_db):
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
plan = catalog.plan_patch(_visits(), [
|
||
{"path": "formula", "value": "下单数 / 访问数"}, # 不是目录的项
|
||
{"path": "columns.visitor_count.measure", "value": "累计值"}, # 不在可选值内
|
||
{"path": "columns.ghost.meaning", "value": "不存在的列"}, # 表结构里没有这一列
|
||
{"path": "columns.visitor_count.meaning", "value": "转化率=下单数/访问数"}, # 计算公式不进目录
|
||
{"path": "keys", "value": ["ticket_no", "nope"]}, # 业务主键里有不存在的列
|
||
{"path": "relations.new", "value": {"columns": ["park_id"], "to_table": "ghost_table", "to_columns": ["id"]}},
|
||
"不是对象",
|
||
], table="visits", tables=tables)
|
||
assert plan.changes == []
|
||
assert len(plan.problems) == 7
|
||
assert any("口径卡" in p for p in plan.problems)
|
||
# 报错写界面上的叫法，不裸写键名
|
||
assert not any("measure" in p or "meaning" in p for p in plan.problems)
|
||
|
||
|
||
async def test_same_value_states(scenic_db):
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
proposed = scenic_notes.notes("proposed")["visits"]
|
||
plan = catalog.plan_patch(proposed, [{"path": "columns.visitor_count.measure", "value": "flow"}],
|
||
table="visits", tables=tables)
|
||
# 值没变、只是推断：保存即确认
|
||
assert plan.changes[0].state == "confirm"
|
||
plan = catalog.plan_patch(_visits(), [{"path": "columns.visitor_count.measure", "value": "flow"}],
|
||
table="visits", tables=tables)
|
||
assert plan.changes[0].state == "same"
|
||
|
||
|
||
async def test_apply_patch_records_human_confirmed(scenic_db):
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
proposed = scenic_notes.notes("proposed")["visits"]
|
||
out = catalog.apply_patch(proposed, [
|
||
{"path": "columns.visitor_count.measure", "value": "stock"},
|
||
{"path": "valid_filter", "value": "status = 1"}, # 值没变：只确认，来源不变
|
||
], table="visits", tables=tables, at="2026-10-03T00:00:00+00:00")
|
||
measure = out["columns"]["visitor_count"]["measure"]
|
||
assert (measure["value"], measure["source"], measure["status"]) == ("stock", "human", "confirmed")
|
||
assert (out["valid_filter"]["source"], out["valid_filter"]["status"]) == ("llm", "confirmed")
|
||
# 其余项原样
|
||
assert out["columns"]["status"] == proposed["columns"]["status"]
|
||
with pytest.raises(catalog.CatalogInvalid):
|
||
catalog.apply_patch(proposed, [{"path": "kind", "value": "nope"}], table="visits", tables=tables)
|
||
|
||
|
||
_PARKS = catalog.relation_id("visits", ["park_id"], "parks", ["id"])
|
||
|
||
|
||
async def test_rejected_items_show_their_value_before(scenic_db):
|
||
"""提案落在已驳回的项上：改前带上被驳回的原值、状态写已驳回（同导入预览「目标里这一项已经驳回」）。只改预览给出的
|
||
信息，保存的规矩不变：照旧换成人工确认的新值；码值不在驳回的码值上补；和驳回的值一样的也不算「值不变」，照样存成
|
||
人工确认。"""
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
rejected = scenic_notes.notes("rejected")["visits"]
|
||
rel_value = {"columns": ["park_id"], "to_table": "parks", "to_columns": ["id"], "cardinality": "many_to_one"}
|
||
changes = [
|
||
{"path": "columns.visitor_count.measure", "value": "stock"},
|
||
{"path": "columns.status.codes", "value": {"9": "作废"}},
|
||
{"path": "valid_filter", "value": "status = 1"}, # 和驳回的值一样
|
||
{"path": f"relations.{_PARKS}", "value": rel_value}, # 两端、基数都和驳回的那条一样
|
||
]
|
||
plan = catalog.plan_patch(rejected, changes, table="visits", tables=tables)
|
||
assert plan.problems == []
|
||
measure, codes, valid, rel = plan.changes
|
||
assert (measure.before, measure.before_status, measure.state) == ("flow", "rejected", "change")
|
||
assert (codes.before, codes.before_status, codes.state) == ({"1": "有效", "0": "作废"}, "rejected", "change")
|
||
assert codes.after == {"9": "作废"}, "码值不在驳回的码值上补"
|
||
assert (valid.before, valid.before_status, valid.state) == ("status = 1", "rejected", "change")
|
||
assert (rel.before, rel.before_status, rel.state) == ({**rel_value, "coverage": None}, "rejected", "change")
|
||
assert rel.as_dict()["before_status"] == "rejected" and rel.as_dict()["before"]["to_table"] == "parks"
|
||
out = catalog.apply_patch(rejected, changes, table="visits", tables=tables)
|
||
saved = [out["columns"]["visitor_count"]["measure"], out["columns"]["status"]["codes"], out["valid_filter"],
|
||
next(r for r in out["relations"] if r["id"] == _PARKS)]
|
||
assert all((item["source"], item["status"]) == ("human", "confirmed") for item in saved)
|
||
codes = out["columns"]["status"]["codes"]
|
||
assert codes["value"] == {"9": "作废"} and "complete" not in codes
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 协议
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def test_protocol_describes_catalog_patch():
|
||
from app.api.copilot import _STREAM_PROTOCOL
|
||
|
||
assert '"op":"catalog_patch"' in _STREAM_PROTOCOL
|
||
assert "relations.new" in _STREAM_PROTOCOL and "columns.<列名>.<字段>" in _STREAM_PROTOCOL
|
||
# 两条边界：只在用户明确陈述数据事实时提；计算公式进口径卡
|
||
assert "明确" in _STREAM_PROTOCOL and "口径卡" in _STREAM_PROTOCOL
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 流：校验后转出，不写库
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
class _Scripted:
|
||
def __init__(self, ops):
|
||
self.ops = ops
|
||
self.calls: list[str] = []
|
||
|
||
async def astream(self, messages):
|
||
from langchain_core.messages import AIMessageChunk
|
||
|
||
self.calls.append(next(text for role, text in messages if role == "system"))
|
||
for op in self.ops:
|
||
yield AIMessageChunk(content=json.dumps(op, ensure_ascii=False) + "\n")
|
||
|
||
|
||
@pytest.fixture
|
||
async def scenic_source(scenic_db):
|
||
"""库里一个带目录的景区数据源；用完删掉（级联删掉目录），不留在测试库里。"""
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import DataSource
|
||
|
||
name = f"scenic_{uuid.uuid4().hex[:8]}"
|
||
async with SessionLocal() as session:
|
||
row = DataSource(name=name, kind="sqlite", database=scenic_db, readonly=True)
|
||
row.schema_cache = await _schema(scenic_db)
|
||
session.add(row)
|
||
await session.commit()
|
||
for table, notes in scenic_notes.notes().items():
|
||
await catalog.write_entry(session, row.id, table, notes, if_version=0, actor="王敏")
|
||
source_id = row.id
|
||
yield SimpleNamespace(id=source_id, name=name)
|
||
await drop_source(source_id)
|
||
|
||
|
||
async def _stream(monkeypatch, model, source, base_graph=None):
|
||
import app.api.copilot as copilot
|
||
from app.main import app
|
||
|
||
async def _model(*_a, **_k):
|
||
return model, "mock-fast"
|
||
|
||
monkeypatch.setattr(copilot, "get_chat_model", _model)
|
||
events: list[dict] = []
|
||
body = {"instruction": "status=9 表示作废，统计时要排除", "intent": "answer", "datasource_ids": [source.id]}
|
||
if base_graph:
|
||
body["base_graph"] = base_graph
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
async with client.stream("POST", "/api/copilot/generate-stream", json=body) as r:
|
||
async for line in r.aiter_lines():
|
||
# 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作
|
||
if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'):
|
||
events.append(json.loads(line[6:]))
|
||
return events
|
||
|
||
|
||
async def _entry(source_id: str, table: str):
|
||
from app.db.base import SessionLocal
|
||
|
||
async with SessionLocal() as session:
|
||
return await catalog.read_entry(session, source_id, table)
|
||
|
||
|
||
async def test_stream_forwards_checked_patch_and_writes_nothing(scenic_source, monkeypatch, caplog):
|
||
before = await _entry(scenic_source.id, "visits")
|
||
patch = {"op": "catalog_patch", "source": scenic_source.name, "table": "visits", "changes": [
|
||
{"path": "columns.status.codes", "value": {"9": "作废"}, "reason": "用户说明 status=9 表示作废"},
|
||
{"path": "valid_filter", "value": "status = 1 AND status <> 9", "reason": "统计时排除作废"},
|
||
{"path": "columns.status.colour", "value": "红"}, # 不合法：丢弃，其余照转
|
||
]}
|
||
model = _Scripted([
|
||
{"op": "catalog_patch", "source": "不存在的库", "table": "visits",
|
||
"changes": [{"path": "kind", "value": "fact"}]}, # 数据源不在范围内：整条丢弃
|
||
{"op": "catalog_patch", "source": scenic_source.name, "table": "ghost",
|
||
"changes": [{"path": "kind", "value": "fact"}]}, # 表不存在：整条丢弃
|
||
{"op": "reply", "text": "好的，已整理成数据目录的修改建议。"},
|
||
patch, # 写在 reply 后面：照样转出，排在前面
|
||
])
|
||
with caplog.at_level(logging.WARNING, logger="app.api.copilot"):
|
||
events = await _stream(monkeypatch, model, scenic_source)
|
||
ops = [e["op"] for e in events]
|
||
assert ops.count("catalog_patch") == 1 and ops[-1] == "reply"
|
||
assert ops.index("catalog_patch") < ops.index("reply")
|
||
[out] = [e for e in events if e["op"] == "catalog_patch"]
|
||
assert out["source"] == scenic_source.name and out["source_id"] == scenic_source.id
|
||
assert out["table"] == "visits" and out["table_label"] == "入园记录" and out["version"] == before.version
|
||
codes, valid = out["changes"]
|
||
assert codes["path"] == "columns.status.codes" and codes["before"] == {"1": "有效", "0": "作废"}
|
||
assert codes["after"] == {"1": "有效", "0": "作废", "9": "作废"} and codes["value"] == {"9": "作废"}
|
||
assert codes["before_status"] == "confirmed" and codes["reason"] == "用户说明 status=9 表示作废"
|
||
assert valid["before"] == "status = 1" and valid["after"] == "status = 1 AND status <> 9"
|
||
# 丢掉的记了日志
|
||
assert sum("目录修改提案" in r.getMessage() for r in caplog.records) >= 3
|
||
# 服务端没写库
|
||
after = await _entry(scenic_source.id, "visits")
|
||
assert after.version == before.version and after.notes == before.notes
|
||
# 协议进了 system
|
||
assert '"op":"catalog_patch"' in model.calls[0]
|
||
|
||
|
||
async def test_patch_only_gets_a_reply(scenic_source, monkeypatch):
|
||
"""只发了提案、没回答也没改图：补一句回答收尾，不当成「模型没给出任何修改」。"""
|
||
model = _Scripted([{"op": "catalog_patch", "source": scenic_source.name, "table": "visits",
|
||
"changes": [{"path": "columns.visitor_count.measure", "value": "stock"}]}])
|
||
events = await _stream(monkeypatch, model, scenic_source)
|
||
assert [e["op"] for e in events if e["op"] not in ("model", "heartbeat", "context")] == ["catalog_patch", "reply"]
|
||
assert "数据目录" in events[-1]["text"]
|
||
|
||
|
||
async def test_patch_alongside_graph_ops(scenic_source, monkeypatch):
|
||
"""和图操作一起出现：提案照转，图照常收尾。"""
|
||
model = _Scripted([
|
||
{"op": "plan", "summary": "查入园人数"},
|
||
{"op": "catalog_patch", "source": scenic_source.name, "table": "visits",
|
||
"changes": [{"path": "columns.visitor_count.measure", "value": "stock"}]},
|
||
{"op": "add_node", "node": {"id": "start", "type": "input", "label": "输入", "config": {}}},
|
||
{"op": "add_node", "node": {"id": "out", "type": "output", "label": "成果", "config": {"fields": []}}},
|
||
{"op": "add_edge", "edge": {"source": "start", "target": "out"}},
|
||
{"op": "done", "explanation": "", "run": False},
|
||
])
|
||
events = await _stream(monkeypatch, model, scenic_source)
|
||
ops = [e["op"] for e in events]
|
||
assert "catalog_patch" in ops and ops[-1] == "final"
|
||
assert ops.index("catalog_patch") < ops.index("final")
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 接口：预览（只读）与保存
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_preview_and_save(scenic_source):
|
||
from app.main import app
|
||
|
||
base = f"/api/datasources/{scenic_source.id}/catalog/visits"
|
||
changes = [{"path": "columns.status.codes", "value": {"9": "作废"}, "reason": "作废单"}]
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||
r = await client.post(f"{base}/patch/preview", json={"changes": changes})
|
||
assert r.status_code == 200, r.text
|
||
preview = r.json()
|
||
version = preview["version"]
|
||
assert preview["changes"][0]["after"] == {"1": "有效", "0": "作废", "9": "作废"}
|
||
assert preview["problems"] == []
|
||
# 预览不写库
|
||
assert (await _entry(scenic_source.id, "visits")).version == version
|
||
|
||
r = await client.post(f"{base}/patch", json={"changes": changes, "if_version": version},
|
||
headers={"X-Actor": quote("王敏")})
|
||
assert r.status_code == 200, r.text
|
||
detail = r.json()
|
||
assert detail["version"] == version + 1 and detail["updated_by"] == "王敏"
|
||
codes = detail["notes"]["columns"]["status"]["codes"]
|
||
assert codes["value"]["9"] == "作废" and (codes["source"], codes["status"]) == ("human", "confirmed")
|
||
# 依据随项另存（确认时的依据，规则见 test_catalog_reason.py），不拼进备注
|
||
assert codes["reason"] == "作废单" and "note" not in codes
|
||
|
||
# 拿旧版本再存：409，什么都不写
|
||
r = await client.post(f"{base}/patch", json={"changes": changes, "if_version": version})
|
||
assert r.status_code == 409 and "刚被修改过" in r.json()["detail"]
|
||
# 重新预览：已经是这个值了
|
||
again = (await client.post(f"{base}/patch/preview", json={"changes": changes})).json()
|
||
assert again["version"] == version + 1 and again["changes"][0]["state"] == "same"
|
||
|
||
# 不合法的取值：422，原话说清
|
||
r = await client.post(f"{base}/patch", json={"changes": [{"path": "kind", "value": "nope"}],
|
||
"if_version": version + 1})
|
||
assert r.status_code == 422 and "表类型" in r.json()["detail"]
|