Codex
249 lines
9.2 KiB
Python
249 lines
9.2 KiB
Python
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
|
|
from src.config.settings import settings
|
|
from src.services.broker import LiveBrokerService, validate_live_submit_request
|
|
|
|
|
|
class LiveBrokerServiceTests(unittest.TestCase):
|
|
def test_missing_ticket_is_rejected_and_audited(self) -> None:
|
|
repo = _LiveRepo()
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", True):
|
|
result = service.submit_confirmed_ticket(
|
|
404,
|
|
requested_by="alice",
|
|
explicit_live_confirmation=True,
|
|
)
|
|
|
|
self.assertEqual(result.status, "rejected")
|
|
self.assertIn("execution_ticket_missing", result.reason_codes)
|
|
self.assertEqual(repo.live_audits[-1]["execution_ticket_id"], 404)
|
|
|
|
def test_submit_confirmed_ticket_rejects_when_live_disabled(self) -> None:
|
|
repo = _LiveRepo()
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", False):
|
|
result = service.submit_confirmed_ticket(
|
|
10,
|
|
requested_by="alice",
|
|
explicit_live_confirmation=True,
|
|
)
|
|
|
|
self.assertIsNotNone(result)
|
|
assert result is not None
|
|
self.assertEqual(result.status, "rejected")
|
|
self.assertIn("live_trading_disabled", result.reason_codes)
|
|
self.assertEqual(repo.live_audits[0]["operation_type"], "submit_bracket_order")
|
|
|
|
def test_submit_confirmed_ticket_requires_explicit_confirmation(self) -> None:
|
|
repo = _LiveRepo()
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", True):
|
|
result = service.submit_confirmed_ticket(
|
|
10,
|
|
requested_by="alice",
|
|
explicit_live_confirmation=False,
|
|
)
|
|
|
|
self.assertEqual(result.status, "rejected")
|
|
self.assertIn("explicit_live_confirmation_required", result.reason_codes)
|
|
|
|
def test_submit_confirmed_ticket_calls_adapter_when_enabled_and_confirmed(self) -> None:
|
|
repo = _LiveRepo()
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", True):
|
|
result = service.submit_confirmed_ticket(
|
|
10,
|
|
requested_by="alice",
|
|
explicit_live_confirmation=True,
|
|
)
|
|
|
|
self.assertEqual(result.status, "placeholder_not_implemented")
|
|
self.assertEqual(repo.live_audits[-1]["adapter_name"], "okx")
|
|
self.assertEqual(repo.live_audits[-1]["execution_ticket_id"], 10)
|
|
self.assertIn("live_order_envelope", repo.live_audits[-1]["payload"])
|
|
|
|
def test_submit_confirmed_ticket_rejects_paper_ticket(self) -> None:
|
|
repo = _LiveRepo()
|
|
repo.tickets[10]["execution_mode"] = "paper"
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", True):
|
|
result = service.submit_confirmed_ticket(
|
|
10,
|
|
requested_by="alice",
|
|
explicit_live_confirmation=True,
|
|
)
|
|
|
|
self.assertEqual(result.status, "rejected")
|
|
self.assertIn("execution_ticket_mode_not_live", result.reason_codes)
|
|
|
|
def test_submit_confirmed_ticket_rejects_missing_operator_protocol(self) -> None:
|
|
repo = _LiveRepo()
|
|
repo.tickets[10]["ticket_snapshot"] = dict(repo.tickets[10]["ticket_snapshot"])
|
|
repo.tickets[10]["ticket_snapshot"].pop("confirmation")
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", True):
|
|
result = service.submit_confirmed_ticket(
|
|
10,
|
|
requested_by="alice",
|
|
explicit_live_confirmation=True,
|
|
)
|
|
|
|
self.assertEqual(result.status, "rejected")
|
|
self.assertIn("operator_confirmation_protocol_missing", result.reason_codes)
|
|
|
|
def test_unsupported_adapter_is_audited(self) -> None:
|
|
repo = _LiveRepo()
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", True):
|
|
result = service.cancel_all(
|
|
requested_by="ops",
|
|
instrument_id=1,
|
|
explicit_live_confirmation=True,
|
|
adapter_name="unknown",
|
|
)
|
|
|
|
self.assertEqual(result.status, "rejected")
|
|
self.assertIn("unsupported_broker_adapter", result.reason_codes)
|
|
|
|
def test_reconcile_is_audited(self) -> None:
|
|
repo = _LiveRepo()
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", True):
|
|
result = service.reconcile_orders(requested_by="ops", instrument_id=1)
|
|
|
|
self.assertEqual(result.status, "placeholder_reconciliation")
|
|
self.assertEqual(repo.live_audits[-1]["operation_type"], "reconcile_orders")
|
|
|
|
def test_cancel_all_requires_explicit_confirmation(self) -> None:
|
|
repo = _LiveRepo()
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", True):
|
|
result = service.cancel_all(requested_by="ops", explicit_live_confirmation=False)
|
|
|
|
self.assertEqual(result.status, "rejected")
|
|
self.assertIn("explicit_live_confirmation_required", result.reason_codes)
|
|
|
|
def test_kill_switch_is_audited(self) -> None:
|
|
repo = _LiveRepo()
|
|
service = LiveBrokerService(repository=repo)
|
|
|
|
with patch.object(settings, "live_trading_enabled", True):
|
|
result = service.activate_kill_switch(
|
|
requested_by="ops",
|
|
reason="manual_test",
|
|
explicit_live_confirmation=True,
|
|
)
|
|
|
|
self.assertEqual(result.status, "placeholder_kill_switch_active")
|
|
self.assertEqual(repo.live_audits[-1]["operation_type"], "activate_kill_switch")
|
|
|
|
|
|
class LiveValidationTests(unittest.TestCase):
|
|
def test_validate_live_submit_request_requires_confirmed_ticket(self) -> None:
|
|
reasons = validate_live_submit_request(
|
|
ticket={"status": "awaiting_confirmation", "execution_mode": "paper", "confirmed_by": None, "confirmed_at": None, "execution_id": None},
|
|
live_trading_enabled=False,
|
|
explicit_live_confirmation=False,
|
|
)
|
|
|
|
self.assertEqual(
|
|
reasons,
|
|
[
|
|
"live_trading_disabled",
|
|
"explicit_live_confirmation_required",
|
|
"execution_ticket_mode_not_live",
|
|
"execution_ticket_not_confirmed",
|
|
"execution_ticket_missing_confirmation_metadata",
|
|
"operator_confirmation_protocol_missing",
|
|
"execution_ticket_missing_execution_record",
|
|
],
|
|
)
|
|
|
|
def test_validate_live_submit_request_accepts_protocolized_ticket(self) -> None:
|
|
reasons = validate_live_submit_request(
|
|
ticket={
|
|
"status": "confirmed",
|
|
"execution_mode": "semi_auto",
|
|
"confirmed_by": "bob",
|
|
"confirmed_at": "2026-05-06T00:00:00+00:00",
|
|
"execution_id": 20,
|
|
"ticket_snapshot": {
|
|
"confirmation": {
|
|
"protocol": {"protocol_version": "live-confirmation-v1"}
|
|
}
|
|
},
|
|
},
|
|
live_trading_enabled=True,
|
|
explicit_live_confirmation=True,
|
|
)
|
|
|
|
self.assertEqual(reasons, [])
|
|
|
|
|
|
class _LiveRepo:
|
|
def __init__(self) -> None:
|
|
self.live_audits = []
|
|
self.tickets = {
|
|
10: {
|
|
"id": 10,
|
|
"signal_id": 1,
|
|
"execution_mode": "semi_auto",
|
|
"status": "confirmed",
|
|
"confirmed_by": "bob",
|
|
"confirmed_at": "2026-05-06T00:00:00+00:00",
|
|
"execution_id": 20,
|
|
"ticket_snapshot": {
|
|
"signal": {
|
|
"side": "buy",
|
|
"signal_id": 1,
|
|
},
|
|
"confirmation": {
|
|
"confirmed_by": "bob",
|
|
"confirmed_at": "2026-05-06T00:00:00+00:00",
|
|
"protocol": {"protocol_version": "live-confirmation-v1"},
|
|
},
|
|
"decision": {
|
|
"entry_price": 100.5,
|
|
"size": 1.0,
|
|
"meta": {
|
|
"stop_loss": 99.0,
|
|
"tp1": 102.0,
|
|
"tp2": 104.0,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
|
|
def fetch_execution_ticket_by_id(self, ticket_id: int):
|
|
return dict(self.tickets.get(ticket_id)) if ticket_id in self.tickets else None
|
|
|
|
def fetch_signal_for_execution(self, signal_id: int):
|
|
return {
|
|
"id": signal_id,
|
|
"instrument_id": 1,
|
|
"side": "buy",
|
|
}
|
|
|
|
def insert_live_operation_audit(self, **kwargs):
|
|
self.live_audits.append(kwargs)
|
|
return len(self.live_audits)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|