test_v3_bitget_demo.py

import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path

import v3_bitget_demo as demo


def configured_credentials():
    return {
        "providers": {
            "bitget_demo": {
                "configured": True,
                "configured_field_count": 3,
                "required_field_count": 3,
                "masked": "****demo",
            }
        }
    }


class BitgetDemoControlPlaneTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        root = Path(self.temp.name)
        self.original_paths = demo.STATE_PATH, demo.AUDIT_PATH, demo.LOCK_PATH
        demo.STATE_PATH = root / "state.json"
        demo.AUDIT_PATH = root / "audit.jsonl"
        demo.LOCK_PATH = root / "state.lock"

    def tearDown(self):
        demo.STATE_PATH, demo.AUDIT_PATH, demo.LOCK_PATH = self.original_paths
        self.temp.cleanup()

    @staticmethod
    def intent(index=0):
        decision = datetime(2026, 7, 23, 13, 30, tzinfo=timezone.utc) + timedelta(minutes=index)
        return {
            "robot_id": "G1-US01",
            "group_id": "G1",
            "strategy_version": "frozen-v3-test",
            "ticker": "CL",
            "direction": "LONG",
            "quantity": 1,
            "entry_price": 79.0,
            "stop_loss": 75.84,
            "take_profit": 85.32,
            "order_notional_usd": 79.0,
            "leverage": 1.0,
            "available_time": (decision - timedelta(seconds=1)).isoformat(),
            "decision_time": decision.isoformat(),
            "execution_after": (decision + timedelta(seconds=1)).isoformat(),
            "paper_trading_only": True,
            "demo_only": True,
        }

    def test_preflight_is_local_only_and_requires_demo_credentials(self):
        with self.assertRaises(ValueError):
            demo.run_local_preflight({"providers": {}}, actor="test")
        status = demo.run_local_preflight(configured_credentials(), actor="test")
        self.assertEqual(status["phase"], "DEMO_DRY_RUN_READY")
        self.assertFalse(status["network_test_performed"])
        self.assertFalse(status["order_submission_enabled"])
        self.assertFalse(status["real_order_submission_enabled"])

    def test_intent_contract_has_demo_header_and_no_auth_or_network(self):
        demo.run_local_preflight(configured_credentials(), actor="test")
        row = demo.queue_intent(self.intent(), configured_credentials(), actor="test")
        preview = row["request_preview"]
        self.assertEqual(preview["required_demo_headers"], {"paptrading": "1"})
        self.assertEqual(preview["path"], "/api/v2/mix/order/place-order")
        self.assertFalse(preview["authentication_headers_included"])
        self.assertFalse(preview["secret_used"])
        self.assertFalse(preview["network_submission_enabled"])
        self.assertFalse(row["order_sent"])
        self.assertEqual(row["status"], "AWAITING_MANUAL_CONFIRMATION")

    def test_queue_is_idempotent_and_lifecycle_never_sends_order(self):
        demo.run_local_preflight(configured_credentials(), actor="test")
        first = demo.queue_intent(self.intent(), configured_credentials(), actor="test")
        second = demo.queue_intent(self.intent(), configured_credentials(), actor="test")
        self.assertEqual(first["intent_id"], second["intent_id"])
        approved = demo.decide_intent(first["intent_id"], "APPROVE", actor="test")
        self.assertEqual(approved["status"], "MANUALLY_CONFIRMED_DRY_RUN")
        lifecycle = demo.simulate_local_lifecycle(first["intent_id"], actor="test")
        self.assertEqual(lifecycle["status"], "LOCAL_LIFECYCLE_SIMULATED")
        self.assertIn("LOCAL_RECONCILIATION_PASSED", lifecycle["local_lifecycle"])
        self.assertFalse(lifecycle["order_sent"])
        self.assertFalse(lifecycle["real_order_sent"])

    def test_only_first_twenty_intents_require_manual_confirmation(self):
        demo.run_local_preflight(configured_credentials(), actor="test")
        for index in range(20):
            row = demo.queue_intent(self.intent(index), configured_credentials(), actor="test")
            self.assertEqual(row["status"], "AWAITING_MANUAL_CONFIRMATION")
            demo.decide_intent(row["intent_id"], "APPROVE", actor="test")
        twenty_first = demo.queue_intent(
            self.intent(20), configured_credentials(), actor="test"
        )
        self.assertEqual(twenty_first["status"], "DRY_RUN_AUTO_ELIGIBLE")
        status = demo.snapshot(configured_credentials())
        self.assertEqual(status["manual_confirmation_completed_count"], 20)
        self.assertEqual(status["manual_confirmation_remaining_count"], 0)

    def test_confirmation_count_is_capped_when_intents_are_prequeued(self):
        demo.run_local_preflight(configured_credentials(), actor="test")
        queued = [
            demo.queue_intent(self.intent(index), configured_credentials(), actor="test")
            for index in range(21)
        ]
        self.assertTrue(all(row["status"] == "AWAITING_MANUAL_CONFIRMATION" for row in queued))
        for row in queued[:20]:
            demo.decide_intent(row["intent_id"], "APPROVE", actor="test")
        status = demo.snapshot(configured_credentials())
        self.assertEqual(status["manual_confirmation_completed_count"], 20)
        self.assertEqual(status["manual_confirmation_remaining_count"], 0)
        released = next(
            row for row in status["recent_intents"]
            if row["intent_id"] == queued[20]["intent_id"]
        )
        self.assertEqual(released["status"], "DRY_RUN_AUTO_ELIGIBLE")

    def test_risk_and_time_contracts_fail_closed(self):
        demo.run_local_preflight(configured_credentials(), actor="test")
        excessive = self.intent()
        excessive["order_notional_usd"] = 101.0
        with self.assertRaisesRegex(ValueError, "order_notional_limit_exceeded"):
            demo.queue_intent(excessive, configured_credentials(), actor="test")
        invalid_time = self.intent(1)
        invalid_time["execution_after"] = invalid_time["decision_time"]
        with self.assertRaisesRegex(ValueError, "time_order_violation"):
            demo.queue_intent(invalid_time, configured_credentials(), actor="test")


if __name__ == "__main__":
    unittest.main()