xc_agent_platform_demo/tests/test_gate.py

"""Default live approval, kill switch, strategy and conservative hint gates."""
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from src.gate import evaluate_gate, attempt_live_submit, LIVE_CONFIRM_VALUE


class TestGate(unittest.TestCase):
    def setUp(self):
        env = patch.dict(os.environ, {'STRATEGY_ID': 'sid'}, clear=True)
        env.start()
        self.addCleanup(env.stop)

    def test_default_live_and_optional_confirmation(self):
        self.assertTrue(evaluate_gate().allowed)
        for value in ('', 'YES', LIVE_CONFIRM_VALUE):
            with patch.dict(os.environ, {'LIVE_CONFIRM': value}):
                self.assertTrue(evaluate_gate().allowed)

    def test_kill_switch_and_missing_strategy(self):
        for env in ({'ALLOW_LIVE_SUBMIT': '0'}, {'ALLOW_LIVE_SUBMIT': 'true'}, {'STRATEGY_ID': '   '}):
            with patch.dict(os.environ, env):
                self.assertFalse(evaluate_gate().allowed)

    def test_hints_and_threshold(self):
        for value in ('95', 'invalid', '-1'):
            with patch.dict(os.environ, {'ASYNC_QUEUE_DEPTH': value}):
                self.assertFalse(evaluate_gate().allowed)
        with patch.dict(os.environ, {'ASYNC_QUEUE_DEPTH': '94'}):
            self.assertTrue(evaluate_gate().allowed)
        with patch.dict(os.environ, {'LIVE_QUEUE_THRESHOLD': 'bad'}):
            self.assertFalse(evaluate_gate().allowed)
        with tempfile.TemporaryDirectory() as d:
            p = Path(d) / 'hint.json'
            p.write_text('{"depth":95}')
            with patch.dict(os.environ, {'QUEUE_SNAPSHOT_JSON': str(p), 'ASYNC_QUEUE_DEPTH': '1'}):
                self.assertFalse(evaluate_gate().allowed)

    def test_no_client_no_network(self):
        with patch('urllib.request.urlopen') as network:
            self.assertFalse(attempt_live_submit({})['submitted'])
            network.assert_not_called()