AdrianoDev
104 lines
2.8 KiB
Python
104 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class FakeReq(BaseModel):
|
|
instrument_name: str
|
|
qty: float
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_call_logs_success(monkeypatch):
|
|
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 {"order_id": "abc123", "state": "filled"}
|
|
|
|
result = 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 result == {"order_id": "abc123", "state": "filled"}
|
|
assert len(logged) == 1
|
|
rec = logged[0]
|
|
assert rec["actor"] == "testnet"
|
|
assert rec["exchange"] == "deribit"
|
|
assert rec["action"] == "place_order"
|
|
assert rec["target"] == "BTC-PERPETUAL"
|
|
assert rec["payload"]["qty"] == 0.1
|
|
assert rec["result"]["order_id"] == "abc123"
|
|
assert "error" not in rec or rec.get("error") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_call_logs_error_and_reraises(monkeypatch):
|
|
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 = "mainnet"
|
|
state = _State()
|
|
|
|
async def tool_fn():
|
|
raise RuntimeError("upstream timeout")
|
|
|
|
with pytest.raises(RuntimeError, match="upstream timeout"):
|
|
await audit_call(
|
|
request=FakeRequest(), # type: ignore[arg-type]
|
|
exchange="deribit",
|
|
action="cancel_order",
|
|
target_field="instrument_name",
|
|
params=FakeReq(instrument_name="BTC-PERPETUAL", qty=0.0),
|
|
tool_fn=tool_fn,
|
|
)
|
|
assert len(logged) == 1
|
|
rec = logged[0]
|
|
assert rec["actor"] == "mainnet"
|
|
assert "RuntimeError: upstream timeout" in rec["error"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_call_no_params_no_target():
|
|
from cerbero_mcp.common.audit_helpers import audit_call
|
|
|
|
class FakeRequest:
|
|
class _State:
|
|
environment = "testnet"
|
|
state = _State()
|
|
|
|
async def tool_fn():
|
|
return {"ok": True}
|
|
|
|
result = await audit_call(
|
|
request=FakeRequest(), # type: ignore[arg-type]
|
|
exchange="bybit",
|
|
action="cancel_all_orders",
|
|
tool_fn=tool_fn,
|
|
)
|
|
assert result == {"ok": True}
|