script-build-and-framework-tmp/script_build_host/tests/test_legacy_input_and_adapters.py

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, include_raw_upload=True).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=40,
                item_id=999,
            )
        )
    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),
    ]