test_legacy_input_and_adapters.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277
  1. from __future__ import annotations
  2. import json
  3. from pathlib import Path
  4. import pytest
  5. from sqlalchemy import insert
  6. from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
  7. from script_build_host.adapters.persona import FilePersonaSource
  8. from script_build_host.adapters.prompts import DatabaseFirstPromptSource
  9. from script_build_host.adapters.strategy import SqlAlchemyStrategySource
  10. from script_build_host.domain.errors import InputRelationMismatch
  11. from script_build_host.domain.ports import PromptRequest
  12. from script_build_host.infrastructure.legacy_tables import (
  13. build_strategy,
  14. build_strategy_version,
  15. prompt,
  16. topic_build_composition_item,
  17. topic_build_item_relation,
  18. topic_build_item_source,
  19. topic_build_point,
  20. topic_build_point_item_relation,
  21. topic_build_record,
  22. topic_build_topic,
  23. topic_pattern_execution,
  24. )
  25. from script_build_host.repositories.legacy_input import LegacySqlAlchemyInputReader
  26. async def _seed_topic(sessions: async_sessionmaker[AsyncSession], raw_upload_data: object) -> None:
  27. async with sessions() as session, session.begin():
  28. await session.execute(insert(topic_pattern_execution).values(id=10, status="success"))
  29. await session.execute(
  30. insert(topic_build_record).values(
  31. id=20,
  32. execution_id=10,
  33. demand="make a useful script",
  34. demand_constraints={"audience": "reader"},
  35. agent_type="AigcAgent",
  36. agent_config={"model_name": "test-model", "api_key": "must-redact-later"},
  37. status="success",
  38. is_deleted=False,
  39. strategies_config={"always_on": [7], "on_demand": [8]},
  40. personal_config={"account_name": "每天心理学"},
  41. origin="upload",
  42. raw_upload_data=raw_upload_data,
  43. )
  44. )
  45. await session.execute(
  46. insert(topic_build_topic).values(
  47. id=30,
  48. build_id=20,
  49. execution_id=10,
  50. sort_order=0,
  51. topic_direction='{"title":"direction"}',
  52. result="topic result",
  53. status="mature",
  54. )
  55. )
  56. await session.execute(
  57. insert(topic_build_point),
  58. [
  59. {
  60. "id": 40,
  61. "topic_id": 30,
  62. "build_id": 20,
  63. "point_type": "key",
  64. "point_result": "active",
  65. "is_active": True,
  66. },
  67. {
  68. "id": 41,
  69. "topic_id": 30,
  70. "build_id": 20,
  71. "point_type": "key",
  72. "point_result": "inactive",
  73. "is_active": False,
  74. },
  75. ],
  76. )
  77. await session.execute(
  78. insert(topic_build_composition_item),
  79. [
  80. {
  81. "id": 50,
  82. "topic_id": 30,
  83. "build_id": 20,
  84. "item_level": "element",
  85. "element_name": "first",
  86. "step": 1,
  87. "sort_order": 2,
  88. "is_active": True,
  89. "derivation_type": "user_demand",
  90. },
  91. {
  92. "id": 51,
  93. "topic_id": 30,
  94. "build_id": 20,
  95. "item_level": "element",
  96. "element_name": "second",
  97. "step": 1,
  98. "sort_order": 1,
  99. "is_active": True,
  100. "derivation_type": "agent_reasoning",
  101. },
  102. ],
  103. )
  104. await session.execute(
  105. insert(topic_build_point_item_relation).values(id=60, point_id=40, item_id=50)
  106. )
  107. await session.execute(
  108. insert(topic_build_item_relation).values(
  109. id=70, topic_id=30, source_item_id=51, target_item_id=50, reason="because"
  110. )
  111. )
  112. await session.execute(
  113. insert(topic_build_item_source).values(
  114. id=80,
  115. topic_id=30,
  116. target_item_id=50,
  117. derivation_type="post_extract",
  118. source_type="post",
  119. source_reference_id="post-1",
  120. source_detail={"title": "source"},
  121. dataset_from="account",
  122. is_active=True,
  123. )
  124. )
  125. @pytest.mark.asyncio
  126. async def test_legacy_reader_freezes_relations_sources_and_arbitrary_raw_upload(
  127. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
  128. ) -> None:
  129. _, sessions = database
  130. await _seed_topic(sessions, ["uploaded", {"nested": True}])
  131. graph = await LegacySqlAlchemyInputReader(sessions).read_topic_graph(
  132. execution_id=10, topic_build_id=20, topic_id=30
  133. )
  134. assert graph["build_record"]["raw_upload_data"] == ["uploaded", {"nested": True}]
  135. assert [point["id"] for point in graph["points"]] == [40]
  136. assert [item["id"] for item in graph["composition_items"]] == [51, 50]
  137. assert graph["points"][0]["item_ids"] == [50]
  138. assert graph["relations"][0]["source_item_id"] == 51
  139. assert graph["sources"][0]["source_reference_id"] == "post-1"
  140. assert graph["sources"][0]["derivation_type"] == "post_extract"
  141. assert graph["sources"][0]["dataset_from"] == "account"
  142. @pytest.mark.asyncio
  143. async def test_legacy_reader_fails_closed_on_relation_mismatch(
  144. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
  145. ) -> None:
  146. _, sessions = database
  147. await _seed_topic(sessions, {})
  148. reader = LegacySqlAlchemyInputReader(sessions)
  149. with pytest.raises(InputRelationMismatch):
  150. await reader.read_topic_graph(execution_id=11, topic_build_id=20, topic_id=30)
  151. async with sessions() as session, session.begin():
  152. await session.execute(
  153. insert(topic_build_point_item_relation).values(
  154. id=61,
  155. point_id=999,
  156. item_id=50,
  157. )
  158. )
  159. with pytest.raises(InputRelationMismatch):
  160. await reader.read_topic_graph(execution_id=10, topic_build_id=20, topic_id=30)
  161. @pytest.mark.asyncio
  162. async def test_persona_adapter_applies_alias_filters_and_versions(tmp_path: Path) -> None:
  163. persona_root = tmp_path / "persona"
  164. section_root = tmp_path / "sections"
  165. alias = "每天一点心理学"
  166. (persona_root / alias).mkdir(parents=True)
  167. (section_root / alias).mkdir(parents=True)
  168. (persona_root / alias / "point_records.json").write_text(
  169. json.dumps(
  170. {
  171. "records": [
  172. {"阶段": "创作", "点类型": "维度值", "名称": "keep", "帖子覆盖率": 0.2},
  173. {"阶段": "创作", "点类型": "维度值", "名称": "drop", "帖子覆盖率": 0.19},
  174. {"阶段": "选题", "点类型": "维度名", "名称": "wrong stage", "帖子覆盖率": 1},
  175. ]
  176. },
  177. ensure_ascii=False,
  178. ),
  179. encoding="utf-8",
  180. )
  181. (section_root / alias / "section_main_dimension_template.json").write_text(
  182. json.dumps({"分段规律摘要": "summary", "ignored": "value"}, ensure_ascii=False),
  183. encoding="utf-8",
  184. )
  185. loaded = await FilePersonaSource(
  186. persona_root=persona_root, section_pattern_root=section_root
  187. ).load("每天心理学")
  188. assert loaded.resolved_account_name == alias
  189. assert [item["点名称"] for item in loaded.persona_points] == ["keep"]
  190. assert loaded.section_patterns == ({"分段规律摘要": "summary"},)
  191. assert set(loaded.source_versions) == {"persona_points", "section_patterns"}
  192. @pytest.mark.asyncio
  193. async def test_strategy_adapter_uses_old_integer_ids_and_current_versions(
  194. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]],
  195. ) -> None:
  196. _, sessions = database
  197. async with sessions() as session, session.begin():
  198. await session.execute(
  199. insert(build_strategy),
  200. [
  201. {
  202. "id": 7,
  203. "strategy_type": "script",
  204. "name": "always",
  205. "description": "A",
  206. "is_active": True,
  207. "current_version": 2,
  208. },
  209. {
  210. "id": 8,
  211. "strategy_type": "script",
  212. "name": "demand",
  213. "description": "B",
  214. "is_active": True,
  215. "current_version": 1,
  216. },
  217. ],
  218. )
  219. await session.execute(
  220. insert(build_strategy_version),
  221. [
  222. {"id": 70, "strategy_id": 7, "version": 1, "content": "old"},
  223. {"id": 71, "strategy_id": 7, "version": 2, "content": "always content"},
  224. {"id": 80, "strategy_id": 8, "version": 1, "content": "demand content"},
  225. ],
  226. )
  227. loaded = await SqlAlchemyStrategySource(sessions).load(always_on=(7,), on_demand=(8,))
  228. assert [(item["strategy_id"], item["mode"], item["version"]) for item in loaded] == [
  229. (7, "always_on", 2),
  230. (8, "on_demand", 1),
  231. ]
  232. assert loaded[0]["content"] == "always content"
  233. assert loaded[1]["content"] == "demand content"
  234. assert loaded[1]["content_sha256"].startswith("sha256:")
  235. @pytest.mark.asyncio
  236. async def test_prompt_adapter_records_actual_db_and_file_sources(
  237. database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], tmp_path: Path
  238. ) -> None:
  239. _, sessions = database
  240. async with sessions() as session, session.begin():
  241. await session.execute(
  242. insert(prompt).values(
  243. id=1,
  244. biz_type="script_planner",
  245. name="planner",
  246. prompt_content="from database",
  247. current_version=3,
  248. )
  249. )
  250. (tmp_path / "worker.md").write_text("from file", encoding="utf-8")
  251. source = DatabaseFirstPromptSource(sessions, fallback_root=tmp_path)
  252. manifests = await source.load(
  253. (
  254. PromptRequest("script_planner", "planner.md", "script_planner", "planner"),
  255. PromptRequest("worker", "worker.md", "script_worker", "worker"),
  256. )
  257. )
  258. assert [(item["source"], item["version"]) for item in manifests] == [
  259. ("database", 3),
  260. ("file", None),
  261. ]