muse-agent-example/tests/集成/test_结构接入.py

"""安装入口发布种子，HTTP 从真实库读取相同的固定结构；不激活实例绑定。"""

import json
import subprocess
import sys
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from muse.共享.调用身份 import 用途
from muse.接入.http.应用 import 创建应用
from muse.配置 import 应用配置, 服务配置

pytestmark = pytest.mark.数据库


@pytest.mark.case_id(
    "NC-meta-cli-http",
    environment="隔离 PostgreSQL、真实 CLI 与 ASGI",
    given="空隔离库和受控维护、作者配置",
    when="任意 cwd CLI 两次发布种子，再用 HTTP 和 CLI 读取固定版本",
    then=["24种子幂等发布，动态嵌套和公共字段可见", "CLI 与 HTTP 合同相同且没有激活实例绑定"],
    contract="docs/系统架构/新版设计/接口契约/元数据投影与版本.md",
)
def test_结构CLI发布与HTTP读取共用库内合同__a43001(应用测试库, tmp_path: Path) -> None:
    配置文件 = tmp_path / "维护.toml"
    配置文件.write_text(
        '["数据库"]\n"取值方式" = "受控存储"\n"位置" = '
        + json.dumps(应用测试库[用途.维护].引用.位置)
        + """
["资源"]
"发布身份" = "test"
["运行"]
"用途" = "maintenance"
"""
    )
    命令 = [sys.executable, "-m", "muse", "结构", str(配置文件), "发布内置"]
    首次 = subprocess.run(命令, cwd="/", capture_output=True, text=True, timeout=20)
    assert 首次.returncode == 0, 首次.stderr
    重复 = subprocess.run(命令, cwd="/", capture_output=True, text=True, timeout=20)
    assert 重复.returncode == 0, 重复.stderr
    assert json.loads(首次.stdout) == json.loads(重复.stdout)
    assert len(json.loads(首次.stdout)["published"]) == 24
    口令 = tmp_path / "口令.txt"
    口令.write_text("synthetic-schema-session")
    配置 = 应用配置(
        应用测试库[用途.生产].引用,
        "test",
        HTTP=服务配置(
            str(口令),
            公开地址="http://testserver",
            允许来源=("http://testserver",),
        ),
    )
    with TestClient(创建应用(配置)) as 客户端:
        客户端.headers["Origin"] = "http://testserver"
        assert (
            客户端.post("/api/v1/session", json={"password": 口令.read_text()}).status_code == 200
        )
        响应 = 客户端.get("/api/v1/schemas/character/versions/1")
        assert 响应.status_code == 200, 响应.text
        内容 = 响应.json()
        assert 内容["schema_version"] == 1
        assert 内容["contract"]["properties"]["名称"]["type"] == "string"
        assert 内容["contract"]["properties"]["演变历程"]["type"] == "array"
        读取 = subprocess.run(
            命令[:-1] + ["查看", "character", "--版本", "1"],
            cwd="/",
            capture_output=True,
            text=True,
            timeout=20,
        )
        assert 读取.returncode == 0, 读取.stderr
        assert json.loads(读取.stdout) == 内容
    with 应用测试库[用途.维护].连接() as 连:
        assert 连.execute("SELECT count(*) FROM metadata.schema_binding").fetchone()[0] == 0


@pytest.mark.case_id(
    "NC-schema-author-commands",
    environment="隔离 PostgreSQL 与 ASGI/CLI适配",
    given="新类型及两版结构，第二版新增必填",
    when="HTTP登记候选、CLI适配发布、HTTP兼容预览",
    then=["固定版本可读，未发布不可读，新增必填报告不兼容，不启用实例"],
    contract="docs/系统架构/新版设计/接口契约/元数据投影与版本.md",
)
def test_新类型结构候选发布与兼容预览不启用__a43002(应用测试库, tmp_path: Path) -> None:
    from dataclasses import asdict

    from muse.元数据.接口 import 值规则, 字段定义, 类型定义, 结构定义
    from muse.共享.错误 import Muse错误
    from muse.启动 import 构建
    from muse.接入.cli.结构命令 import 运行结构命令

    口令 = tmp_path / "口令.txt"
    口令.write_text("synthetic-author")
    配置 = 应用配置(
        应用测试库[用途.生产].引用,
        "test",
        HTTP=服务配置(
            str(口令),
            公开地址="http://testserver",
            允许来源=("http://testserver",),
        ),
    )
    定义 = 结构定义(
        "new-type", "new-schema", 1, (字段定义("name", "名称", "名称", 值规则("text")),)
    )
    with TestClient(创建应用(配置)) as client:
        client.headers["Origin"] = "http://testserver"
        assert (
            client.post("/api/v1/session", json={"password": "synthetic-author"}).status_code == 200
        )
        body = {
            "action": "登记候选",
            "类型": asdict(类型定义("new-type", "测试新型", "world", "entity", ("entity",))),
            "结构": asdict(定义),
        }
        response = client.post("/api/v1/schemas/commands", json=body)
        assert response.status_code == 200, response.text
        assert response.json()["state"] == "candidate"
        未发布 = client.get("/api/v1/schemas/new-schema/versions/1")
        assert 未发布.status_code == 422
        assert 未发布.json() == {
            "code": "SCHEMA_UNKNOWN",
            "message": "指定结构版本尚未发布",
            "target_ref": None,
            "retryable": False,
            "details": {},
            "actions": [],
        }
        请求文件 = tmp_path / "结构请求.json"
        请求文件.write_text(
            json.dumps(
                {
                    "action": "发布版本",
                    "schema_id": "new-schema",
                    "version": 1,
                    "content_hash": 定义.内容哈希,
                }
            )
        )
        维护装配 = 构建(配置)
        with 维护装配.生命周期():
            published = 运行结构命令(维护装配, "维护", str(请求文件), 1)
        assert published["state"] == "published"
        assert client.get("/api/v1/schemas/new-schema/versions/1").status_code == 200
        新定义 = 结构定义(
            "new-type",
            "new-schema",
            2,
            (*定义.字段, 字段定义("age", "年岁", "年岁", 值规则("integer"), 必填=True)),
            父版本=1,
        )
        diff = client.post(
            "/api/v1/schemas/commands",
            json={
                "action": "兼容检查",
                "schema_id": "new-schema",
                "base_version": 1,
                "结构": asdict(新定义),
            },
        )
        assert diff.status_code == 200 and diff.json()["分类"] == "incompatible"
        assert diff.json()["转换字段"] == ["age"]
        非法请求 = {"action": "登记候选"}
        请求文件.write_text(json.dumps(非法请求, ensure_ascii=False))
        非法响应 = client.post("/api/v1/schemas/commands", json=非法请求)
        assert 非法响应.status_code == 422
        非法装配 = 构建(配置)
        with 非法装配.生命周期(), pytest.raises(Muse错误) as 捕获:
            运行结构命令(非法装配, "维护", str(请求文件), 1)
        assert 捕获.value.呈现() == 非法响应.json()