iTi-System/tests/test_integration.py

from fastapi.testclient import TestClient

from iti import create_app
from iti.auth import create_access_token
from iti.config import BaseConfig
from iti.db import Base
from iti_system import create_system_module
from iti_system.models import Role, SysMenu, User
from iti_system.seeds import seed_system_data


def make_app():
    app = create_app(
        modules=[create_system_module()],
        config_mapping=BaseConfig(
            database_url="sqlite+pysqlite:///:memory:",
            testing=True,
            jwt_secret_key="test-secret",
            ratelimit_enabled=False,
            service_tokens={"erp": "svc-token"},
            audit_enabled=True,
            audit_flush_interval_seconds=999,
        ),
    )
    Base.metadata.create_all(app.state.db_engine)
    with app.state.db_sessionmaker() as db:
        seed_system_data(db, app.state.iti_modules)
    return app


def login(client: TestClient) -> dict:
    response = client.post(
        "/auth/loginByPassword",
        json={"username": "admin", "password": "123456"},
    )
    assert response.status_code == 200
    data = response.json()
    assert data["code"] == 200
    return {"Authorization": f"Bearer {data['data']['accessToken']}"}


def test_system_module_registers_routes_and_permissions():
    app = make_app()
    paths = {route.path for route in app.routes}
    registry = app.state.iti_modules

    assert app.state.iti_system is registry.get("iti_system")
    assert "/auth/loginByPassword" in paths
    assert "/sys/user/list" in paths
    assert "/sys/user-attributes/current" in paths
    assert "/sys/role/list" in paths
    assert "/sys/menu/list" in paths
    assert "/sys/dept/list" in paths
    assert "/sys/config/list" in paths
    assert "/sys/dict/type/page" in paths
    assert "/sys/log/page" in paths
    assert "/internal/audit/events" in paths
    assert "/sys/file/{file_id}" in paths
    assert "/upload" in paths
    assert "/file/{file_id}/download" in paths
    assert "system:user:list" in registry.permissions


def test_login_and_current_user_flow():
    app = make_app()
    client = TestClient(app)
    headers = login(client)

    response = client.get("/sys/user/current", headers=headers)

    assert response.status_code == 200
    data = response.json()["data"]
    assert data["username"] == "admin"
    assert data["roleCodes"] == ["ADMIN"]
    assert data["isSuper"] is True
    event = app.state.audit_dispatcher._queue.get_nowait()
    assert event.type == "login"
    assert event.actor_id == data["id"]
    assert event.success is True


def test_admin_can_list_system_routes():
    client = TestClient(make_app())
    headers = login(client)

    assert client.get("/sys/user/list", headers=headers).json()["code"] == 200
    assert client.get("/sys/menu/tree", headers=headers).json()["code"] == 200
    assert client.get("/sys/config/list", headers=headers).json()["code"] == 200


def test_menu_exists_supports_path_auth_code_and_excludes_current_id():
    client = TestClient(make_app())
    headers = login(client)

    by_path = client.get("/sys/menu/exists", params={"path": "/system/menu"}, headers=headers)
    same_path = client.get(
        "/sys/menu/exists",
        params={"id": "system-menu", "path": "/system/menu"},
        headers=headers,
    )
    by_auth_code = client.get(
        "/sys/menu/exists",
        params={"authCode": "system:menu:list"},
        headers=headers,
    )

    assert by_path.json()["data"] == {"exists": True}
    assert same_path.json()["data"] == {"exists": False}
    assert by_auth_code.json()["data"] == {"exists": True}


def test_menu_tree_returns_current_user_visible_routes_without_menu_admin_permission():
    app = make_app()
    with app.state.db_sessionmaker() as db:
        parent = SysMenu(
            id="orders",
            name="Orders",
            type="catalog",
            path="/orders",
            status="enabled",
        )
        child = SysMenu(
            id="orders-list",
            name="OrdersList",
            type="menu",
            path="/orders/list",
            component="/orders/list",
            auth_code="orders:list",
            parent=parent,
            status="enabled",
        )
        button = SysMenu(
            id="orders-create",
            name="OrdersCreate",
            type="button",
            auth_code="orders:create",
            parent=child,
            status="enabled",
        )
        role = Role(name="操作员", code="OPERATOR", menus=[child, button])
        user = User(username="operator", status="enabled", roles=[role])
        user.set_password("123456")
        db.add_all([parent, child, button, role, user])
        db.commit()
        user_id = user.id

    token = create_access_token(user_id, app.state.config)
    response = TestClient(app).get(
        "/sys/menu/tree",
        headers={"Authorization": f"Bearer {token}"},
    )

    assert response.status_code == 200
    data = response.json()["data"]
    assert [item["id"] for item in data] == ["orders"]
    assert [item["id"] for item in data[0]["children"]] == ["orders-list"]
    assert data[0]["children"][0]["children"] == []


def test_internal_audit_writes_sys_log():
    client = TestClient(make_app())

    response = client.post(
        "/internal/audit/events",
        headers={"Authorization": "Bearer svc-token"},
        json={
            "events": [
                {
                    "type": "OPERATION",
                    "title": "修改生产订单",
                    "actorId": "svc",
                    "actorType": "service",
                    "method": "POST",
                    "path": "/monitor/api/call",
                    "targetType": "order",
                    "targetId": "MO001",
                    "diff": {"qty": {"before": 1, "after": 2}},
                }
            ]
        },
    )

    assert response.json()["data"] == {"count": 1}
    headers = login(client)
    logs = client.get("/sys/log/page", headers=headers).json()["data"]["items"]
    assert logs[0]["name"] == "修改生产订单"
    assert logs[0]["userId"] == "svc"