iTi-Flask/tests/test_exchange.py

from __future__ import annotations

from dataclasses import dataclass

import httpx
from fastapi.testclient import TestClient

from iti import create_app
from iti.config import BaseConfig
from iti.db import Base, reset_db
from iti.exchange import (
    ExchangeBusinessSpec,
    ExchangeOperation,
    ExchangeScope,
    ExchangeTaskContext,
    ExchangeTaskResult,
    ExchangeTemplateLayout,
    ExchangeTemplatePlan,
    ExchangeVariable,
    MappingExchangeSource,
    RemoteExchangeSource,
    get_exchange_registry,
    register_exchange_source,
    register_exchange_spec,
)
from iti.exchange.excel import ExcelTemplateCodec, ExcelWorkbookCodec
from iti.exchange.service import ExchangeService
from iti.service_client import register_service_client


def make_app(*, exchange_enabled: bool = True):
    config = BaseConfig(
        database_url="sqlite+pysqlite:///:memory:",
        testing=True,
        exchange_enabled=exchange_enabled,
    )
    app = create_app(config_mapping=config)
    Base.metadata.create_all(app.state.db_engine)
    return app


def user_import_spec() -> ExchangeBusinessSpec:
    return ExchangeBusinessSpec(
        scope=ExchangeScope("system", "user", ExchangeOperation.IMPORT),
        name="用户导入",
        description="导入系统用户",
        layout=ExchangeTemplateLayout(title="用户导入", sheet_name="用户", header_row=2),
        variables=(
            ExchangeVariable(key="username", label="用户名", required=True, example="alice"),
            ExchangeVariable(key="mobile", label="手机号"),
        ),
    )


def test_exchange_module_is_auto_registered():
    app = make_app()

    assert app.state.iti_modules.get("exchange") is not None
    assert "exchange:template:list" in app.state.iti_modules.permissions


def test_business_spec_registry_builds_catalog_and_handler():
    app = make_app()

    def handler(context: ExchangeTaskContext) -> ExchangeTaskResult:
        return ExchangeTaskResult(success_count=1, result_payload={"task": context.task_id})

    spec = register_exchange_spec(app, user_import_spec(), handler=handler)
    registry = get_exchange_registry(app)

    assert spec.generated_code() == "system.user.import"
    assert registry.get_scope_handler(
        biz_domain="system",
        biz_obj="user",
        operation="import",
    ) is handler
    assert registry.catalog()[0]["objects"][0]["operations"][0]["variables"][0]["key"] == "username"


def test_template_workbook_roundtrip_uses_business_variables():
    plan = user_import_spec().to_plan(template_id="tpl1", version_id="v1", version="1.0.0")
    codec = ExcelTemplateCodec()

    content = codec.dump(plan)
    parsed = codec.load(content)

    assert parsed["title"] == "用户导入"
    assert parsed["sheet_name"] == "用户"
    assert parsed["variables"][0]["key"] == "username"
    assert parsed["variables"][0]["label"] == "用户名"


def test_service_create_publish_resolve_and_task_flow():
    reset_db()
    app = make_app()
    register_exchange_spec(app, user_import_spec())
    service = ExchangeService(app, app.state.db_sessionmaker())

    template = service.create_template(
        name="用户导入模板",
        biz_domain="system",
        biz_obj="user",
        operation="import",
    )
    assert template.code == "system.user.import"

    version = service.publish_version(template_id=template.id, version="1.0.0")
    assert version.variables[0]["key"] == "username"

    plan = service.resolve_plan(
        biz_domain="system",
        biz_obj="user",
        operation="import",
        template_id=template.id,
    )
    assert plan.version_id == version.id
    assert plan.variables[0].key == "username"

    task = service.create_task(
        biz_domain="system",
        biz_obj="user",
        operation="import",
        template_id=template.id,
        version_id=version.id,
        input_payload={"source": "upload"},
    )
    assert task.operation == "import"
    assert task.template_version_id == version.id


def test_excel_workbook_maps_headers_to_business_variables():
    variables = [
        ExchangeVariable(key="username", label="用户名"),
        ExchangeVariable(key="mobile", label="手机号"),
    ]
    codec = ExcelWorkbookCodec()
    content = codec.export_rows_with_variables(
        variables=variables,
        rows=[{"username": "alice", "mobile": None}],
        sheet_name="用户",
    )

    rows = codec.import_rows_with_variables(content, variables=variables)

    assert rows == [{"username": "alice", "mobile": None}]


def test_mapping_source_supports_template_less_plan():
    source = MappingExchangeSource()
    plan = source.resolve_plan(
        biz_domain="system",
        biz_obj="user",
        operation="import",
        variables=[ExchangeVariable(key="username", label="用户名")],
        layout={"title": "用户导入", "sheet_name": "用户"},
    )

    assert plan.generated_code() == "system.user.import"
    assert plan.layout.sheet_name == "用户"
    assert plan.variables[0].key == "username"


def test_remote_source_resolves_plan_and_template_bytes():
    app = create_app(config_mapping=BaseConfig(database_url="sqlite+pysqlite:///:memory:", testing=True))

    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/exchange/template-versions/v1":
            return httpx.Response(
                200,
                json={
                    "data": {
                        "scope": {
                            "bizDomain": "system",
                            "bizObj": "user",
                            "operation": "import",
                        },
                        "templateId": "tpl1",
                        "versionId": "v1",
                        "version": "1.0.0",
                        "code": "system.user.import",
                        "name": "用户导入",
                        "layout": {"title": "用户导入", "sheetName": "用户"},
                        "variables": [{"key": "username", "label": "用户名"}],
                    },
                    "code": 200,
                    "message": "成功",
                },
            )
        if request.url.path == "/exchange/template-versions/v1/download":
            content = ExcelTemplateCodec().dump(
                ExchangeTemplatePlan.from_mapping(
                    biz_domain="system",
                    biz_obj="user",
                    operation="import",
                    version_id="v1",
                    variables=[ExchangeVariable(key="username", label="用户名")],
                )
            )
            return httpx.Response(
                200,
                content=content,
                headers={
                    "Content-Type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
                },
            )
        raise AssertionError(f"unexpected path: {request.url.path}")

    register_service_client(
        app,
        "template_center",
        {"base_url": "https://template-center.local"},
        transport=httpx.MockTransport(handler),
    )
    source = RemoteExchangeSource(app, service_name="template_center")

    plan = source.resolve_plan(
        biz_domain="system",
        biz_obj="user",
        operation="import",
        version_id="v1",
    )
    assert plan.template_id == "tpl1"
    assert plan.name == "用户导入"
    assert plan.variables[0].key == "username"

    content = source.load_template_file(plan)
    assert content is not None
    assert len(content) > 0


def test_custom_registered_source_can_drive_plan_and_file():
    @dataclass
    class CustomSource:
        plan: ExchangeTemplatePlan
        content: bytes

        def resolve_plan(self, **kwargs):
            return self.plan

        def load_template_file(self, plan: ExchangeTemplatePlan) -> bytes | None:
            return self.content

    app = make_app()
    custom_plan = ExchangeTemplatePlan.from_mapping(
        biz_domain="system",
        biz_obj="user",
        operation="import",
        template_id="tpl-custom",
        version_id="v-custom",
        version="1.0.0",
        variables=[ExchangeVariable(key="username", label="用户名")],
    )
    register_exchange_source(
        app,
        "custom-center",
        CustomSource(plan=custom_plan, content=b"custom-template"),
    )

    client = TestClient(app)
    response = client.post(
        "/exchange/plans/resolve",
        json={
            "bizDomain": "system",
            "bizObj": "user",
            "operation": "import",
            "sourceName": "custom-center",
        },
    )
    assert response.status_code == 200
    assert response.json()["data"]["template_id"] == "tpl-custom"

    file_resp = client.post(
        "/exchange/plans/template-file",
        json={
            "bizDomain": "system",
            "bizObj": "user",
            "operation": "import",
            "sourceName": "custom-center",
        },
    )
    assert file_resp.status_code == 200
    assert file_resp.content == b"custom-template"


def test_exchange_routes_support_template_and_catalog_flow():
    app = make_app()
    register_exchange_spec(app, user_import_spec())
    client = TestClient(app)

    catalog = client.get("/exchange/catalog")
    assert catalog.status_code == 200
    assert catalog.json()["data"][0]["biz_domain"] == "system"

    code = client.get(
        "/exchange/templates/code",
        params={"biz_domain": "system", "biz_obj": "user", "operation": "import"},
    )
    assert code.status_code == 200
    assert code.json()["data"]["code"] == "system.user.import"

    created = client.post(
        "/exchange/templates",
        json={
            "name": "用户导入模板",
            "bizDomain": "system",
            "bizObj": "user",
            "operation": "import",
        },
    )
    assert created.status_code == 200
    template_id = created.json()["data"]["id"]

    version = client.post(
        f"/exchange/templates/{template_id}/versions",
        json={"version": "1.0.0"},
    )
    assert version.status_code == 200
    assert version.json()["data"]["variables"][0]["key"] == "username"

    plan = client.post(
        "/exchange/plans/resolve",
        json={
            "bizDomain": "system",
            "bizObj": "user",
            "operation": "import",
            "templateId": template_id,
        },
    )
    assert plan.status_code == 200
    assert plan.json()["data"]["version_id"] == version.json()["data"]["id"]


def test_task_run_uses_registered_business_handler():
    reset_db()
    app = make_app()

    def handler(context: ExchangeTaskContext) -> ExchangeTaskResult:
        assert context.plan.scope.biz_domain == "system"
        return ExchangeTaskResult(success_count=2, result_payload={"handled": True})

    register_exchange_spec(app, user_import_spec(), handler=handler)
    service = ExchangeService(app, app.state.db_sessionmaker())
    template = service.create_template(
        name="用户导入模板",
        biz_domain="system",
        biz_obj="user",
        operation="import",
    )
    version = service.publish_version(template_id=template.id, version="1.0.0")
    task = service.create_task(
        biz_domain="system",
        biz_obj="user",
        operation="import",
        template_id=template.id,
        version_id=version.id,
    )

    finished = service.run_task(task.id)

    assert finished.status == "success"
    assert finished.success_count == 2
    assert finished.result_payload == {"handled": True}