Codex
38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
|
|
from scripts import run_live_adapter
|
|
|
|
|
|
class LiveAdapterScriptTests(unittest.TestCase):
|
|
def test_main_reports_database_unavailable_json(self) -> None:
|
|
output = io.StringIO()
|
|
|
|
with patch(
|
|
"scripts.run_live_adapter.LiveBrokerService.reconcile_orders",
|
|
side_effect=OperationalError("stmt", {}, Exception()),
|
|
), contextlib.redirect_stdout(output):
|
|
exit_code = run_live_adapter.main(
|
|
env={
|
|
"LIVE_ADAPTER_OUTPUT_JSON": "1",
|
|
"LIVE_ADAPTER_OPERATION": "reconcile",
|
|
}
|
|
)
|
|
|
|
payload = json.loads(output.getvalue())
|
|
self.assertEqual(exit_code, 2)
|
|
self.assertFalse(payload["passed"])
|
|
self.assertEqual(payload["error_type"], "OperationalError")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|