🚧 Implement service stub
This commit is contained in:
Jonas Bengtsson
@@ -0,0 +1,317 @@
|
||||
"""Phase 3 Windows service tests — pure logic, no SCM needed.
|
||||
|
||||
The actual SCM calls (install/start/stop against the real Service Control
|
||||
Manager) are necessarily elevated-Windows-only; they are verified live in a
|
||||
documented manual pass. What is testable anywhere:
|
||||
|
||||
- the SCM dispatch detection (``_scm_started_us``) — the argv/CWD shape the
|
||||
SCM starts the exe with, and the shapes a human CLI run produces;
|
||||
- the failure-actions dict passed to ``ChangeServiceConfig2`` — its exact
|
||||
pywin32 parse requirements (verified against the SWIG parser);
|
||||
- the service class identity (name, display name, description) and the
|
||||
registry class string install records;
|
||||
- the CLI service verbs against injected failures: actionable messages,
|
||||
exit codes (2 for environment errors), install's config validation;
|
||||
- the service verbs' non-Windows guard.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
import opcda_diagnozer.main as main_mod
|
||||
import opcda_diagnozer.service as svc
|
||||
from opcda_diagnozer.config import ConfigError
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# SCM dispatch detection
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def _scm_environment(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The environment the SCM gives a service process: CWD = System32."""
|
||||
monkeypatch.setattr(main_mod.sys, "platform", "win32")
|
||||
monkeypatch.chdir("C:/Windows/System32")
|
||||
monkeypatch.setenv("SystemRoot", "C:/Windows")
|
||||
|
||||
|
||||
def test_scm_detection_the_registered_shape(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""argv == [exe, 'service'] with CWD System32 => SCM dispatch."""
|
||||
_scm_environment(monkeypatch)
|
||||
monkeypatch.setattr(main_mod.sys, "argv", [r"C:\diag\opcda-diagnozer.exe", "service"])
|
||||
assert main_mod._scm_started_us() is True
|
||||
|
||||
|
||||
def test_scm_detection_rejects_plain_cli(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A human verb from System32 with one arg must NOT dispatch."""
|
||||
_scm_environment(monkeypatch)
|
||||
monkeypatch.setattr(main_mod.sys, "argv", [r"C:\diag\opcda-diagnozer.exe", "status"])
|
||||
assert main_mod._scm_started_us() is False
|
||||
|
||||
|
||||
def test_scm_detection_rejects_normal_cwd(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""'service' as the verb from any other CWD is a CLI call (unknown verb)."""
|
||||
monkeypatch.setattr(main_mod.sys, "platform", "win32")
|
||||
monkeypatch.chdir(tmp_path)
|
||||
monkeypatch.setattr(main_mod.sys, "argv", [r"C:\diag\opcda-diagnozer.exe", "service"])
|
||||
assert main_mod._scm_started_us() is False
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform != "win32", reason="SCM detection is Windows-only")
|
||||
def test_scm_detection_real_environment() -> None:
|
||||
"""The check never fires under pytest (a normal CWD + rich argv)."""
|
||||
monkeypatch = pytest.MonkeyPatch()
|
||||
monkeypatch.setattr(sys, "argv", [sys.executable, "pytest"])
|
||||
assert main_mod._scm_started_us() is False
|
||||
|
||||
|
||||
def test_main_dispatches_to_service_host(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""When the SCM started us, main hands over without parsing the CLI."""
|
||||
_scm_environment(monkeypatch)
|
||||
monkeypatch.setattr(main_mod.sys, "argv", [r"C:\diag\opcda-diagnozer.exe", "service"])
|
||||
dispatched: list[list[str]] = []
|
||||
|
||||
def _fake_host(argv):
|
||||
dispatched.append(list(argv))
|
||||
return 0
|
||||
|
||||
monkeypatch.setattr(
|
||||
"opcda_diagnozer.service.handle_service_command_line", _fake_host
|
||||
)
|
||||
assert main_mod.main() == 0
|
||||
assert dispatched == [[r"C:\diag\opcda-diagnozer.exe", "service"]]
|
||||
|
||||
|
||||
def test_main_with_explicit_argv_is_always_cli(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""main(argv=[...]) (tests, console_script plumbing) never SCM-dispatches."""
|
||||
_scm_environment(monkeypatch)
|
||||
good = tmp_path / "config.yaml"
|
||||
good.write_text(
|
||||
"opc: {server: Fake.Server.1}\ntags: [{name: T.Counter}]\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
called = []
|
||||
monkeypatch.setattr(
|
||||
"opcda_diagnozer.service.handle_service_command_line",
|
||||
lambda argv: called.append(argv) or 1,
|
||||
)
|
||||
# an explicit argv bypasses the SCM check entirely
|
||||
assert main_mod.main(["check-config", "--config", str(good)]) == 0
|
||||
assert called == [] # never dispatched
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# failure actions (SCM recovery, plan §6.3)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
@pytest.mark.skipif(sys.platform != "win32", reason="pywin32 SC_ACTION constants")
|
||||
class TestFailureActions:
|
||||
def test_shape_matches_pywin32_parser(self) -> None:
|
||||
"""The dict must carry every key the SWIG parser requires.
|
||||
|
||||
Verified live: the parser rejects partial dicts with
|
||||
"must be a dictionary containing {'ResetPeriod':int,'RebootMsg':
|
||||
unicode,'Command':unicode,'Actions':sequence of 2 tuples(int,int)".
|
||||
"""
|
||||
import win32service
|
||||
|
||||
fa = svc.failure_actions()
|
||||
assert set(fa) == {"ResetPeriod", "RebootMsg", "Command", "Actions"}
|
||||
assert isinstance(fa["ResetPeriod"], int)
|
||||
assert isinstance(fa["RebootMsg"], str)
|
||||
assert isinstance(fa["Command"], str)
|
||||
actions = fa["Actions"]
|
||||
for action in actions:
|
||||
assert isinstance(action, tuple)
|
||||
assert len(action) == 2
|
||||
assert all(isinstance(v, int) for v in action)
|
||||
# first AND second failure restart the service (spec §10.1)
|
||||
assert [a[0] for a in actions] == [
|
||||
win32service.SC_ACTION_RESTART,
|
||||
win32service.SC_ACTION_RESTART,
|
||||
]
|
||||
|
||||
def test_restart_delay_is_configurable(self) -> None:
|
||||
fa = svc.failure_actions(restart_delay_ms=30_000)
|
||||
assert all(delay == 30_000 for _, delay in fa["Actions"])
|
||||
|
||||
def test_parses_through_change_service_config2(self, tmp_path: Path) -> None:
|
||||
"""End-to-end: the exact dict install passes must survive the parser.
|
||||
|
||||
Parse success is proven by the handle error (invalid handle — a
|
||||
parse failure would raise TypeError naming the expected dict).
|
||||
"""
|
||||
import pywintypes
|
||||
import win32service
|
||||
|
||||
try:
|
||||
win32service.ChangeServiceConfig2(
|
||||
None,
|
||||
win32service.SERVICE_CONFIG_FAILURE_ACTIONS,
|
||||
svc.failure_actions(),
|
||||
)
|
||||
except TypeError as exc:
|
||||
pytest.fail(f"failure-actions dict does not parse: {exc}")
|
||||
except pywintypes.error as exc:
|
||||
assert exc.winerror == 6 # ERROR_INVALID_HANDLE: parse OK
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# service class identity + install payload
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
@pytest.mark.skipif(sys.platform != "win32", reason="ServiceFramework is Windows-only")
|
||||
class TestServiceClass:
|
||||
def test_identity(self) -> None:
|
||||
cls = svc.get_service_class()
|
||||
assert cls is not None
|
||||
assert cls._svc_name_ == svc.SERVICE_NAME == "OPCDA-Diagnozer"
|
||||
assert cls._svc_display_name_ == "OPC DA Diagnozer"
|
||||
assert "read-only" in cls._svc_description_.lower()
|
||||
|
||||
def test_class_string_is_importable_shape(self) -> None:
|
||||
"""The registry class string must be module.Class of THIS module."""
|
||||
assert svc._SERVICE_CLASS_STRING == (
|
||||
"opcda_diagnozer.service.OpcdaDiagnozerService"
|
||||
)
|
||||
|
||||
def test_stop_timeout_exceeds_poll_slice(self) -> None:
|
||||
"""The SCM stop wait must be generous vs. the poller's stop latency."""
|
||||
from opcda_diagnozer.monitor.poller import _WATCH_SLICE_S
|
||||
|
||||
assert svc._STOP_TIMEOUT_S > _WATCH_SLICE_S * 10
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# CLI service verbs (injected SCM functions)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
class _ScmError(Exception):
|
||||
"""win32service.error stand-in with the attributes the CLI reads."""
|
||||
|
||||
def __init__(self, winerror: int, strerror: str) -> None:
|
||||
super().__init__(strerror)
|
||||
self.winerror = winerror
|
||||
self.strerror = strerror
|
||||
|
||||
|
||||
class TestServiceVerbs:
|
||||
def test_status_prints_state(self, capsys) -> None:
|
||||
monkeypatch = pytest.MonkeyPatch()
|
||||
monkeypatch.setattr(svc, "service_status", lambda: "STOPPED")
|
||||
try:
|
||||
code = main_mod._cmd_service_verb("status")
|
||||
finally:
|
||||
monkeypatch.undo()
|
||||
out = capsys.readouterr().out
|
||||
assert code == 0
|
||||
assert "OPCDA-Diagnozer" in out and "STOPPED" in out
|
||||
|
||||
def test_install_validates_config_first(self, tmp_path: Path, capsys) -> None:
|
||||
"""A broken config must abort install BEFORE touching the SCM."""
|
||||
bad = tmp_path / "bad.yaml"
|
||||
bad.write_text("opc: {server: S}\ntags: []\ndevices: []\n", encoding="utf-8")
|
||||
installed = []
|
||||
|
||||
def _fail(*args, **kwargs): # pragma: no cover — must not be reached
|
||||
installed.append(args)
|
||||
|
||||
monkeypatch = pytest.MonkeyPatch()
|
||||
monkeypatch.setattr(svc, "install_service", _fail)
|
||||
try:
|
||||
code = main_mod._cmd_service_verb("install", config_path=str(bad))
|
||||
finally:
|
||||
monkeypatch.undo()
|
||||
assert code == 2
|
||||
assert installed == [] # SCM never touched
|
||||
assert "invalid" in capsys.readouterr().err
|
||||
|
||||
def test_install_happy_path(self, tmp_path: Path, capsys) -> None:
|
||||
good = tmp_path / "config.yaml"
|
||||
good.write_text(
|
||||
"opc: {server: Fake.Server.1}\ntags: [{name: T.Counter}]\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
calls = {}
|
||||
|
||||
def _install(config_path, account=None, password=None):
|
||||
calls["config"] = config_path
|
||||
calls["account"] = account
|
||||
|
||||
monkeypatch = pytest.MonkeyPatch()
|
||||
monkeypatch.setattr(svc, "install_service", _install)
|
||||
try:
|
||||
code = main_mod._cmd_service_verb("install", config_path=str(good))
|
||||
finally:
|
||||
monkeypatch.undo()
|
||||
out = capsys.readouterr().out
|
||||
assert code == 0
|
||||
assert calls["config"] == str(good)
|
||||
assert "auto-start" in out
|
||||
|
||||
def test_access_denied_gets_the_elevation_hint(self, capsys) -> None:
|
||||
exc = _ScmError(5, "Access is denied.")
|
||||
out = main_mod._service_verb_error("install", exc)
|
||||
err = capsys.readouterr().err
|
||||
assert out == 2
|
||||
assert "elevated" in err
|
||||
assert "Access is denied." in err
|
||||
|
||||
def test_not_installed_gets_the_install_hint(self, capsys) -> None:
|
||||
exc = _ScmError(1060, "The specified service does not exist as an installed service.")
|
||||
out = main_mod._service_verb_error("start", exc)
|
||||
err = capsys.readouterr().err
|
||||
assert out == 2
|
||||
assert "not installed" in err
|
||||
|
||||
def test_not_active_stop_hint(self, capsys) -> None:
|
||||
exc = _ScmError(1062, "The service has not been started.")
|
||||
out = main_mod._service_verb_error("stop", exc)
|
||||
err = capsys.readouterr().err
|
||||
assert out == 2
|
||||
assert "not running" in err
|
||||
|
||||
def test_unknown_winerror_still_names_the_code(self, capsys) -> None:
|
||||
exc = _ScmError(1053, "The service did not respond to the start request.")
|
||||
out = main_mod._service_verb_error("start", exc)
|
||||
err = capsys.readouterr().err
|
||||
assert out == 2
|
||||
assert "1053" in err
|
||||
assert "net helpmsg" in err
|
||||
|
||||
def test_plain_exception_without_winerror(self, capsys) -> None:
|
||||
out = main_mod._service_verb_error("uninstall", ConfigError("registry offline"))
|
||||
err = capsys.readouterr().err
|
||||
assert out == 2
|
||||
assert "registry offline" in err
|
||||
|
||||
def test_verb_dispatch_via_main(self, monkeypatch: pytest.MonkeyPatch, capsys) -> None:
|
||||
"""The verbs must be reachable through the real parser."""
|
||||
monkeypatch.setattr(svc, "service_status", lambda: "RUNNING")
|
||||
monkeypatch.setattr(svc, "install_service", lambda *a, **k: None)
|
||||
monkeypatch.setattr(svc, "uninstall_service", lambda: None)
|
||||
monkeypatch.setattr(svc, "start_service", lambda: None)
|
||||
monkeypatch.setattr(svc, "stop_service", lambda: None)
|
||||
assert main_mod.main(["status"]) == 0
|
||||
assert "RUNNING" in capsys.readouterr().out
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# non-Windows guard
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def test_service_module_imports_everywhere() -> None:
|
||||
"""The module is importable without pywin32 (guarded imports, plan §9)."""
|
||||
assert svc.SERVICE_NAME == "OPCDA-Diagnozer"
|
||||
assert callable(svc.handle_service_command_line)
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform == "win32", reason="guard test for other platforms")
|
||||
def test_get_service_class_none_off_windows() -> None:
|
||||
assert svc.get_service_class() is None
|
||||
Reference in New Issue
Block a user