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.
94 lines
2.6 KiB
Python
94 lines
2.6 KiB
Python
import httpx
|
|
import json
|
|
from fastapi import Request
|
|
from fastapi.testclient import TestClient
|
|
|
|
from iti import create_app
|
|
from iti.audit import audit_operation, build_diff
|
|
from iti.config import BaseConfig
|
|
from iti.service_client import register_service_client
|
|
|
|
|
|
class AuditModule:
|
|
name = "audit"
|
|
|
|
def register_routes(self, app):
|
|
@app.post("/change")
|
|
def change(request: Request):
|
|
audit_operation(
|
|
request,
|
|
title="修改",
|
|
target_type="demo",
|
|
target_id="1",
|
|
before={"name": "old", "password": "a"},
|
|
after={"name": "new", "password": "b"},
|
|
)
|
|
return {"ok": True}
|
|
|
|
|
|
def make_app():
|
|
return create_app(
|
|
modules=[AuditModule()],
|
|
config_mapping=BaseConfig(
|
|
database_url="sqlite+pysqlite:///:memory:",
|
|
testing=True,
|
|
audit_enabled=True,
|
|
audit_flush_interval_seconds=999,
|
|
),
|
|
)
|
|
|
|
|
|
def test_build_diff_masks_sensitive_fields():
|
|
assert build_diff(
|
|
{"name": "old", "password": "a"},
|
|
{"name": "new", "password": "b"},
|
|
) == {
|
|
"name": {"before": "old", "after": "new"},
|
|
"password": {"before": "***", "after": "***"},
|
|
}
|
|
|
|
|
|
def test_audit_operation_queues_event_without_blocking():
|
|
app = make_app()
|
|
client = TestClient(app)
|
|
|
|
response = client.post("/change")
|
|
|
|
assert response.json()["data"] == {"ok": True}
|
|
event = app.state.audit_dispatcher._queue.get_nowait()
|
|
assert event.title == "修改"
|
|
assert event.diff == {
|
|
"name": {"before": "old", "after": "new"},
|
|
"password": {"before": "***", "after": "***"},
|
|
}
|
|
|
|
|
|
def test_audit_dispatcher_sends_events_to_configured_service():
|
|
requests: list[dict] = []
|
|
app = make_app()
|
|
client = TestClient(app)
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append(json.loads(request.content))
|
|
return httpx.Response(200, json={"data": {"count": 1}, "code": 200, "message": "成功"})
|
|
|
|
register_service_client(
|
|
app,
|
|
"audit",
|
|
{"base_url": "https://audit.local", "token": "svc-token"},
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
client.post("/change")
|
|
event = app.state.audit_dispatcher._queue.get_nowait()
|
|
|
|
app.state.audit_dispatcher._send([event])
|
|
|
|
sent = requests[0]["events"][0]
|
|
assert sent["type"] == "operation"
|
|
assert sent["title"] == "修改"
|
|
assert sent["path"] == "/change"
|
|
assert sent["diff"] == {
|
|
"name": {"before": "old", "after": "new"},
|
|
"password": {"before": "***", "after": "***"},
|
|
}
|