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

from __future__ import annotations

from pathlib import Path
from typing import Any

import pytest
from pydantic import SecretStr

from script_build_host.agents.model_manifest import resolve_script_role_model_manifest
from script_build_host.domain.records import Principal
from script_build_host.infrastructure.config import ScriptBuildSettings
from script_build_host.internal_runtime import (
    InternalBuildAuthorizer,
    InternalPrincipalProvider,
    create_internal_runtime,
)


def _settings(tmp_path: Path, **overrides: Any) -> ScriptBuildSettings:
    values: dict[str, Any] = {
        "environment": "test",
        "read_database_url": f"sqlite+aiosqlite:///{tmp_path / 'read.db'}",
        "write_database_url": f"sqlite+aiosqlite:///{tmp_path / 'write.db'}",
        "agent_data_root": tmp_path / "agent-data",
        "persona_root": tmp_path / "persona",
        "section_pattern_root": tmp_path / "sections",
        "decode_index_root": tmp_path / "decode",
        "knowledge_root": tmp_path / "knowledge.json",
        "prompt_fallback_root": tmp_path / "prompts",
        "role_model": "qwen3.7-max",
        "embedding_api_key": SecretStr("secret"),
    }
    values.update(overrides)
    return ScriptBuildSettings(**values)


def test_role_model_manifest_covers_all_sixteen_presets_and_keeps_overrides() -> None:
    result = resolve_script_role_model_manifest(
        {
            "embedding_model": "text-embedding-v4",
            "presets": {"script_root_validator": {"temperature": 0.1}},
        },
        model="qwen3.7-max",
        max_iterations=120,
    )
    presets = result["presets"]
    assert isinstance(presets, dict)
    assert len(presets) == 16
    assert {value["model"] for value in presets.values()} == {"qwen3.7-max"}
    assert presets["script_root_validator"]["temperature"] == 0.1
    assert presets["script_direction_worker"]["temperature"] == 0.2
    assert result["embedding_model"] == "text-embedding-v4"


@pytest.mark.asyncio
async def test_internal_auth_is_fixed_and_rejected_for_invalid_builds() -> None:
    provider = InternalPrincipalProvider(" internal-e2e ")
    principal = await provider.current(object())
    assert principal == Principal("internal-e2e")
    authorizer = InternalBuildAuthorizer()
    await authorizer.require_source_access(principal, execution_id=1, topic_build_id=2, topic_id=3)
    await authorizer.require_access(principal, 1)
    with pytest.raises(PermissionError):
        await authorizer.require_access(principal, 0)


def test_internal_runtime_uses_qwen_and_fails_closed_outside_test(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    received: dict[str, Any] = {}

    def fake_qwen(**kwargs: Any) -> Any:
        received.update(kwargs)

        async def call(**_call: Any) -> dict[str, Any]:
            return {"content": "unused", "finish_reason": "stop"}

        return call

    monkeypatch.setattr("script_build_host.internal_runtime.create_qwen_llm_call", fake_qwen)
    settings = _settings(tmp_path)
    runtime = create_internal_runtime(settings)
    assert runtime.runner.trace_store is not None
    assert received == {
        "model": "qwen3.7-max",
        "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
        "api_key": "secret",
        "enable_thinking": False,
        "request_timeout_seconds": 300.0,
        "max_retries": 2,
    }
    with pytest.raises(RuntimeError, match="only in test"):
        create_internal_runtime(
            _settings(
                tmp_path,
                environment="production",
                read_database_url="mysql+asyncmy://reader:r@db.example/app",
                read_database_socks_proxy=None,
                write_database_url="mysql+asyncmy://runtime:r@db.example/app",
                final_database_url="mysql+asyncmy://publisher:r@db.example/app",
            )
        )