kg-agent/scripts/test_service_control.py

"""Real loopback service control without starting the business API."""
import json
from pathlib import Path
from types import SimpleNamespace
import socket
import tempfile
import threading
import time
import unittest
from unittest.mock import patch
from step4_web import service_control as control


class ControlTests(unittest.TestCase):
    def setUp(self):
        self.tmp=tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.root=Path(self.tmp.name)
        self.patch=patch.object(control,'ROOT',self.root)
        self.patch.start(); self.addCleanup(self.patch.stop)

    def test_stop_waits_for_exit_and_allows_new_instance(self):
        server=SimpleNamespace(should_exit=False)
        ready=threading.Event()
        def worker():
            with control.managed_service(server):
                ready.set()
                while not server.should_exit: time.sleep(.01)
        thread=threading.Thread(target=worker)
        thread.start()
        try:
            self.assertTrue(ready.wait(3))
            self.assertTrue(control.running())
            control.stop_for_restart(timeout=3)
            thread.join(3)
            self.assertFalse(thread.is_alive())
            with control.managed_service(SimpleNamespace(should_exit=False)):
                self.assertTrue(control.running())
            self.assertFalse((self.root/'.runtime/service-control.json').exists())
        finally:
            server.should_exit=True
            thread.join(3)

    def test_wrong_token_cannot_stop_service(self):
        server=SimpleNamespace(should_exit=False)
        with control.managed_service(server):
            data=json.loads((self.root/'.runtime/service-control.json').read_text())
            with socket.create_connection(('127.0.0.1',data['port'])) as conn:
                conn.sendall(b'wrong-token\n')
                self.assertEqual(conn.recv(32).strip(),b'DENIED')
            self.assertFalse(server.should_exit)
            with self.assertRaises(RuntimeError):
                with control.managed_service(server): pass

    def test_timeout_does_not_start_or_force_kill(self):
        server=SimpleNamespace(should_exit=False)
        with control.managed_service(server):
            with self.assertRaisesRegex(RuntimeError,'未启动新实例'):
                control.stop_for_restart(timeout=.01)
            self.assertTrue(server.should_exit)

    def test_stale_state_does_not_signal_any_process(self):
        folder=self.root/'.runtime'; folder.mkdir()
        (folder/'service-control.json').write_text('{}')
        control.stop_for_restart()
        self.assertFalse(control.running())


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