"""watch_ops vol_breakout 분기.""" from __future__ import annotations import json from datetime import datetime from pathlib import Path from types import SimpleNamespace from bithumb.operations.watch_ops import ( WatchIssue, WatchReport, inspect_vol_watch, is_vol_breakout_ops, remediate_vol_watch, ) def test_is_vol_breakout_ops(tmp_path: Path) -> None: """vol state 파일이 있으면 vol 감시 모드.""" state_path = tmp_path / "vol_breakout_state.json" state_path.write_text( json.dumps({"strategy": "vol_breakout_15m_spot_long", "symbols": {}}), encoding="utf-8", ) settings = SimpleNamespace(vol_state_json=state_path) assert is_vol_breakout_ops(settings) is True def test_is_vol_breakout_ops_missing_file(tmp_path: Path) -> None: """state 없으면 fractal 감시.""" settings = SimpleNamespace(vol_state_json=tmp_path / "missing.json") assert is_vol_breakout_ops(settings) is False def test_inspect_vol_watch_recent_tick(tmp_path: Path) -> None: """최근 tick이면 이슈 없음.""" state_path = tmp_path / "vol.json" state_path.write_text( json.dumps({ "strategy": "vol_breakout_15m_spot_long", "symbols": {}, "last_run_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }), encoding="utf-8", ) settings = SimpleNamespace( vol_state_json=state_path, ops_watch_tick_stale_min=12, ) report = inspect_vol_watch(settings) assert report.issues == [] def test_remediate_vol_dry_run() -> None: """vol dry-run은 tick 없음.""" report = WatchReport( checked_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), issues=[ WatchIssue(kind="tick_stale", severity="critical", message="stale"), ], ) settings = SimpleNamespace( ops_telegram_enabled=False, telegram_bot_token="", telegram_chat_id="", ops_watch_auto_remediate=True, ops_mode="live", ) result = remediate_vol_watch(settings, report, dry_run=True) assert "dry_run" in result.actions assert result.tick_report is None