feat(V2): X-Bot-Tag header obbligatorio + endpoint /admin/audit con filtri

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
AdrianoDev
2026-05-01 08:51:40 +02:00
parent bd6b03ce43
commit 69ac878893
10 changed files with 549 additions and 8 deletions
+64
View File
@@ -101,3 +101,67 @@ async def test_audit_call_no_params_no_target():
tool_fn=tool_fn,
)
assert result == {"ok": True}
@pytest.mark.asyncio
async def test_audit_call_propagates_bot_tag(monkeypatch):
"""bot_tag letto da request.state e propagato a audit_write_op."""
from cerbero_mcp.common.audit_helpers import audit_call
logged = []
def fake_audit(**kw):
logged.append(kw)
monkeypatch.setattr("cerbero_mcp.common.audit_helpers.audit_write_op", fake_audit)
class FakeRequest:
class _State:
environment = "testnet"
bot_tag = "scanner-alpha"
state = _State()
async def tool_fn():
return {"order_id": "abc"}
await audit_call(
request=FakeRequest(), # type: ignore[arg-type]
exchange="deribit",
action="place_order",
target_field="instrument_name",
params=FakeReq(instrument_name="BTC-PERPETUAL", qty=0.1),
tool_fn=tool_fn,
)
assert len(logged) == 1
assert logged[0]["bot_tag"] == "scanner-alpha"
assert logged[0]["actor"] == "testnet"
@pytest.mark.asyncio
async def test_audit_call_bot_tag_none_when_missing(monkeypatch):
"""Se request.state.bot_tag non esiste, audit riceve None senza errore."""
from cerbero_mcp.common.audit_helpers import audit_call
logged = []
def fake_audit(**kw):
logged.append(kw)
monkeypatch.setattr("cerbero_mcp.common.audit_helpers.audit_write_op", fake_audit)
class FakeRequest:
class _State:
environment = "testnet"
state = _State()
async def tool_fn():
return {"ok": True}
await audit_call(
request=FakeRequest(), # type: ignore[arg-type]
exchange="bybit",
action="cancel_all_orders",
tool_fn=tool_fn,
)
assert len(logged) == 1
assert logged[0]["bot_tag"] is None