gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
|
|
SERVICE_WRAPPER = Path(__file__).parents[1] / "service" / "gti_agent_windows_service.py"
|
|
|
|
|
|
def load_service_wrapper():
|
|
spec = importlib.util.spec_from_file_location("gti_agent_windows_service_test", SERVICE_WRAPPER)
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_main_enters_scm_dispatcher_when_started_without_command(monkeypatch):
|
|
service = load_service_wrapper()
|
|
calls = []
|
|
|
|
class FakeServiceManager:
|
|
def Initialize(self):
|
|
calls.append("initialize")
|
|
|
|
def PrepareToHostSingle(self, service_class):
|
|
calls.append(("prepare", service_class))
|
|
|
|
def StartServiceCtrlDispatcher(self):
|
|
calls.append("dispatcher")
|
|
|
|
def forbidden_handle_command_line(service_class):
|
|
raise AssertionError("SCM start must not be parsed as a CLI command")
|
|
|
|
monkeypatch.setattr(service, "servicemanager", FakeServiceManager())
|
|
monkeypatch.setattr(service, "win32serviceutil", SimpleNamespace(HandleCommandLine=forbidden_handle_command_line))
|
|
monkeypatch.setattr(service.sys, "argv", [str(SERVICE_WRAPPER)])
|
|
|
|
service.main()
|
|
|
|
assert calls == ["initialize", ("prepare", service.GTIFiscalAgentService), "dispatcher"]
|
|
|
|
|
|
def test_main_preserves_pywin32_cli_commands(monkeypatch):
|
|
service = load_service_wrapper()
|
|
called = []
|
|
|
|
class ForbiddenServiceManager:
|
|
def __getattr__(self, name):
|
|
raise AssertionError("CLI commands must not enter the SCM dispatcher")
|
|
|
|
monkeypatch.setattr(service, "servicemanager", ForbiddenServiceManager())
|
|
monkeypatch.setattr(service, "win32serviceutil", SimpleNamespace(HandleCommandLine=lambda cls: called.append(cls)))
|
|
monkeypatch.setattr(service.sys, "argv", [str(SERVICE_WRAPPER), "start"])
|
|
|
|
service.main()
|
|
|
|
assert called == [service.GTIFiscalAgentService]
|
|
|
|
|
|
def test_a_fatal_error_in_the_agent_loop_stops_the_service_instead_of_leaving_a_dead_one_running():
|
|
text = SERVICE_WRAPPER.read_text(encoding="utf-8")
|
|
body = text.split("agent_service.run(config, os.environ, stopper=self.stop_flag)", 1)[1].split("def main", 1)[0]
|
|
assert "except BaseException" in body and "logger" not in body.split("except BaseException")[0][-40:] # wrapped in try/except
|
|
assert "LogErrorMsg" in body and "self.stop_flag.stop = True" in body and "win32event.SetEvent(self.stop_event)" in body
|