Codex
87 lines
3.4 KiB
Python
87 lines
3.4 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from scripts import run_okx_sandbox_smoke
|
|
|
|
|
|
class OKXSandboxSmokeScriptTests(unittest.TestCase):
|
|
def test_main_writes_passed_json_artifact(self) -> None:
|
|
payload = {
|
|
"status": "passed",
|
|
"execute_requested": True,
|
|
"blockers": [],
|
|
"steps": [{"operation": "submit_order", "ok": True}],
|
|
}
|
|
service = Mock()
|
|
service.run_isolated_smoke.return_value = payload
|
|
output = io.StringIO()
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
artifact = Path(tmpdir) / "smoke.json"
|
|
with patch("scripts.run_okx_sandbox_smoke.OKXSandboxPreflightService", return_value=service):
|
|
with contextlib.redirect_stdout(output):
|
|
exit_code = run_okx_sandbox_smoke.main(
|
|
env={
|
|
"OKX_SANDBOX_SMOKE_OUTPUT_JSON": "1",
|
|
"OKX_SANDBOX_SMOKE_EXECUTE": "1",
|
|
"OKX_SANDBOX_SMOKE_INSTRUMENT_ID": "2",
|
|
"OKX_SANDBOX_SMOKE_TIMEFRAME": "5m",
|
|
"OKX_SANDBOX_SMOKE_PRICE": "1.2",
|
|
"OKX_SANDBOX_SMOKE_SIZE": "3",
|
|
"OKX_SANDBOX_SMOKE_SIDE": "sell",
|
|
"OKX_SANDBOX_SMOKE_ORDER_TYPE": "limit",
|
|
"OKX_SANDBOX_SMOKE_TD_MODE": "cross",
|
|
"OKX_SANDBOX_SMOKE_TIMEOUT_SECONDS": "4.5",
|
|
"OKX_SANDBOX_SMOKE_ARTIFACT_PATH": str(artifact),
|
|
}
|
|
)
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertEqual(json.loads(output.getvalue())["status"], "passed")
|
|
self.assertEqual(json.loads(artifact.read_text(encoding="utf-8"))["status"], "passed")
|
|
service.run_isolated_smoke.assert_called_once_with(
|
|
instrument_id=2,
|
|
timeframe="5m",
|
|
execute=True,
|
|
order_price="1.2",
|
|
order_size="3",
|
|
order_side="sell",
|
|
order_type="limit",
|
|
td_mode="cross",
|
|
timeout_seconds=4.5,
|
|
)
|
|
|
|
def test_main_returns_nonzero_for_blocked_smoke(self) -> None:
|
|
payload = {
|
|
"status": "blocked",
|
|
"execute_requested": False,
|
|
"blockers": ["okx_sandbox_smoke_execute_not_enabled"],
|
|
"steps": [],
|
|
}
|
|
service = Mock()
|
|
service.run_isolated_smoke.return_value = payload
|
|
output = io.StringIO()
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
artifact = Path(tmpdir) / "smoke.json"
|
|
with patch("scripts.run_okx_sandbox_smoke.OKXSandboxPreflightService", return_value=service):
|
|
with contextlib.redirect_stdout(output):
|
|
exit_code = run_okx_sandbox_smoke.main(
|
|
env={
|
|
"OKX_SANDBOX_SMOKE_ARTIFACT_PATH": str(artifact),
|
|
}
|
|
)
|
|
|
|
self.assertEqual(exit_code, 2)
|
|
self.assertIn("okx_sandbox_smoke_execute_not_enabled", output.getvalue())
|
|
self.assertEqual(json.loads(artifact.read_text(encoding="utf-8"))["status"], "blocked")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|