from __future__ import annotations import json from pathlib import Path import pytest from sqlalchemy import insert from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker from script_build_host.adapters.persona import FilePersonaSource from script_build_host.adapters.prompts import DatabaseFirstPromptSource from script_build_host.adapters.strategy import SqlAlchemyStrategySource from script_build_host.domain.errors import InputRelationMismatch from script_build_host.domain.ports import PromptRequest from script_build_host.infrastructure.legacy_tables import ( build_strategy, build_strategy_version, prompt, topic_build_composition_item, topic_build_item_relation, topic_build_item_source, topic_build_point, topic_build_point_item_relation, topic_build_record, topic_build_topic, topic_pattern_execution, ) from script_build_host.repositories.legacy_input import LegacySqlAlchemyInputReader async def _seed_topic(sessions: async_sessionmaker[AsyncSession], raw_upload_data: object) -> None: async with sessions() as session, session.begin(): await session.execute(insert(topic_pattern_execution).values(id=10, status="success")) await session.execute( insert(topic_build_record).values( id=20, execution_id=10, demand="make a useful script", demand_constraints={"audience": "reader"}, agent_type="AigcAgent", agent_config={"model_name": "test-model", "api_key": "must-redact-later"}, status="success", is_deleted=False, strategies_config={"always_on": [7], "on_demand": [8]}, personal_config={"account_name": "每天心理学"}, origin="upload", raw_upload_data=raw_upload_data, ) ) await session.execute( insert(topic_build_topic).values( id=30, build_id=20, execution_id=10, sort_order=0, topic_direction='{"title":"direction"}', result="topic result", status="mature", ) ) await session.execute( insert(topic_build_point), [ { "id": 40, "topic_id": 30, "build_id": 20, "point_type": "key", "point_result": "active", "is_active": True, }, { "id": 41, "topic_id": 30, "build_id": 20, "point_type": "key", "point_result": "inactive", "is_active": False, }, ], ) await session.execute( insert(topic_build_composition_item), [ { "id": 50, "topic_id": 30, "build_id": 20, "item_level": "element", "element_name": "first", "step": 1, "sort_order": 2, "is_active": True, "derivation_type": "user_demand", }, { "id": 51, "topic_id": 30, "build_id": 20, "item_level": "element", "element_name": "second", "step": 1, "sort_order": 1, "is_active": True, "derivation_type": "agent_reasoning", }, ], ) await session.execute( insert(topic_build_point_item_relation).values(id=60, point_id=40, item_id=50) ) await session.execute( insert(topic_build_item_relation).values( id=70, topic_id=30, source_item_id=51, target_item_id=50, reason="because" ) ) await session.execute( insert(topic_build_item_source).values( id=80, topic_id=30, target_item_id=50, derivation_type="post_extract", source_type="post", source_reference_id="post-1", source_detail={"title": "source"}, dataset_from="account", is_active=True, ) ) @pytest.mark.asyncio async def test_legacy_reader_freezes_relations_sources_and_arbitrary_raw_upload( database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], ) -> None: _, sessions = database await _seed_topic(sessions, ["uploaded", {"nested": True}]) graph = await LegacySqlAlchemyInputReader(sessions).read_topic_graph( execution_id=10, topic_build_id=20, topic_id=30 ) assert graph["build_record"]["raw_upload_data"] == ["uploaded", {"nested": True}] assert [point["id"] for point in graph["points"]] == [40] assert [item["id"] for item in graph["composition_items"]] == [51, 50] assert graph["points"][0]["item_ids"] == [50] assert graph["relations"][0]["source_item_id"] == 51 assert graph["sources"][0]["source_reference_id"] == "post-1" assert graph["sources"][0]["derivation_type"] == "post_extract" assert graph["sources"][0]["dataset_from"] == "account" @pytest.mark.asyncio async def test_legacy_reader_fails_closed_on_relation_mismatch( database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], ) -> None: _, sessions = database await _seed_topic(sessions, {}) reader = LegacySqlAlchemyInputReader(sessions) with pytest.raises(InputRelationMismatch): await reader.read_topic_graph(execution_id=11, topic_build_id=20, topic_id=30) async with sessions() as session, session.begin(): await session.execute( insert(topic_build_point_item_relation).values( id=61, point_id=999, item_id=50, ) ) with pytest.raises(InputRelationMismatch): await reader.read_topic_graph(execution_id=10, topic_build_id=20, topic_id=30) @pytest.mark.asyncio async def test_persona_adapter_applies_alias_filters_and_versions(tmp_path: Path) -> None: persona_root = tmp_path / "persona" section_root = tmp_path / "sections" alias = "每天一点心理学" (persona_root / alias).mkdir(parents=True) (section_root / alias).mkdir(parents=True) (persona_root / alias / "point_records.json").write_text( json.dumps( { "records": [ {"阶段": "创作", "点类型": "维度值", "名称": "keep", "帖子覆盖率": 0.2}, {"阶段": "创作", "点类型": "维度值", "名称": "drop", "帖子覆盖率": 0.19}, {"阶段": "选题", "点类型": "维度名", "名称": "wrong stage", "帖子覆盖率": 1}, ] }, ensure_ascii=False, ), encoding="utf-8", ) (section_root / alias / "section_main_dimension_template.json").write_text( json.dumps({"分段规律摘要": "summary", "ignored": "value"}, ensure_ascii=False), encoding="utf-8", ) loaded = await FilePersonaSource( persona_root=persona_root, section_pattern_root=section_root ).load("每天心理学") assert loaded.resolved_account_name == alias assert [item["点名称"] for item in loaded.persona_points] == ["keep"] assert loaded.section_patterns == ({"分段规律摘要": "summary"},) assert set(loaded.source_versions) == {"persona_points", "section_patterns"} @pytest.mark.asyncio async def test_strategy_adapter_uses_old_integer_ids_and_current_versions( database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], ) -> None: _, sessions = database async with sessions() as session, session.begin(): await session.execute( insert(build_strategy), [ { "id": 7, "strategy_type": "script", "name": "always", "description": "A", "is_active": True, "current_version": 2, }, { "id": 8, "strategy_type": "script", "name": "demand", "description": "B", "is_active": True, "current_version": 1, }, ], ) await session.execute( insert(build_strategy_version), [ {"id": 70, "strategy_id": 7, "version": 1, "content": "old"}, {"id": 71, "strategy_id": 7, "version": 2, "content": "always content"}, {"id": 80, "strategy_id": 8, "version": 1, "content": "demand content"}, ], ) loaded = await SqlAlchemyStrategySource(sessions).load(always_on=(7,), on_demand=(8,)) assert [(item["strategy_id"], item["mode"], item["version"]) for item in loaded] == [ (7, "always_on", 2), (8, "on_demand", 1), ] assert loaded[0]["content"] == "always content" assert loaded[1]["content"] == "demand content" assert loaded[1]["content_sha256"].startswith("sha256:") @pytest.mark.asyncio async def test_prompt_adapter_records_actual_db_and_file_sources( database: tuple[AsyncEngine, async_sessionmaker[AsyncSession]], tmp_path: Path ) -> None: _, sessions = database async with sessions() as session, session.begin(): await session.execute( insert(prompt).values( id=1, biz_type="script_planner", name="planner", prompt_content="from database", current_version=3, ) ) (tmp_path / "worker.md").write_text("from file", encoding="utf-8") source = DatabaseFirstPromptSource(sessions, fallback_root=tmp_path) manifests = await source.load( ( PromptRequest("script_planner", "planner.md", "script_planner", "planner"), PromptRequest("worker", "worker.md", "script_worker", "worker"), ) ) assert [(item["source"], item["version"]) for item in manifests] == [ ("database", 3), ("file", None), ]