xc_agent_platform_demo/tests/test_sigterm.py

"""SIGTERM should stop the server subprocess promptly (<< 30s platform window)."""

from __future__ import annotations

import json
import os
import signal
import subprocess
import sys
import time
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]


class TestSigterm(unittest.TestCase):
    def test_sigterm_exits_quickly(self) -> None:
        env = os.environ.copy()
        env["STRATEGY_ID"] = "sigterm-test"
        env["ALLOW_LIVE_SUBMIT"] = "0"
        env["HOST"] = "127.0.0.1"

        proc = subprocess.Popen(
            [sys.executable, '-c', """
from unittest.mock import patch
import src.server as server
class OfflineServer:
    def __init__(self, *args): pass
    def serve_forever(self, **kwargs): server._shutdown.wait(10)
    def shutdown(self): pass
    def server_close(self): pass
with patch.object(server, 'ThreadingHTTPServer', OfflineServer):
    server.main()
"""],
            cwd=str(ROOT),
            env=env,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.PIPE,
        )
        try:
            # Read readiness from stderr; no sockets or network calls.
            import selectors
            selector = selectors.DefaultSelector()
            self.addCleanup(selector.close)
            selector.register(proc.stderr, selectors.EVENT_READ)
            self.assertTrue(selector.select(timeout=5), 'server readiness timeout')
            self.assertIn(b'Listening on', proc.stderr.readline())

            t0 = time.time()
            proc.send_signal(signal.SIGTERM)
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
                self.fail("server did not exit within 5s after SIGTERM (platform allows ≤30s)")
            elapsed = time.time() - t0
            self.assertLess(elapsed, 5.0, f"exit took {elapsed:.2f}s")
            self.assertEqual(proc.returncode, 0, "SIGTERM must exit cleanly, not kill the process")
        finally:
            if proc.poll() is None:
                proc.kill()
                proc.wait(timeout=2)
            if proc.stderr:
                proc.stderr.close()


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