You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
114 lines
3.6 KiB
Python
114 lines
3.6 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from iti import create_app
|
|
from iti.config import BaseConfig
|
|
from iti.db import Base
|
|
from iti_system import create_system_module
|
|
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
|
|
assert response.json()["data"]["username"] == "admin"
|
|
event = app.state.audit_dispatcher._queue.get_nowait()
|
|
assert event.type == "login"
|
|
assert event.actor_id == response.json()["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_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"
|