Dominic Gundel
db6a85a703
Senden und Abbruch laufen über den Treiber, Orca-Doppelzeilen und tote Ideen-API fallen weg, Drucker folgen der Datei, Etikettenformat liegt in den Einstellungen.
116 lines
3.4 KiB
Python
116 lines
3.4 KiB
Python
"""Marlin-Dienst starten, ohne ihn zu beenden. Kein echtes USB-Gerät."""
|
|
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from grotti import marlin_supervisor as sup
|
|
|
|
|
|
class SupervisorTests(unittest.TestCase):
|
|
def tearDown(self):
|
|
sup.disable()
|
|
|
|
def test_running_service_is_not_started_again(self):
|
|
starts = []
|
|
service = sup.MarlinSupervisor(
|
|
Path("."),
|
|
healthy=lambda: True,
|
|
start=lambda folder: starts.append(folder),
|
|
wait_s=0.1,
|
|
)
|
|
result = service.ensure()
|
|
self.assertTrue(result["running"])
|
|
self.assertFalse(result["started"])
|
|
self.assertEqual(starts, [])
|
|
|
|
def test_starts_once_when_down(self):
|
|
state = {"up": False}
|
|
starts = []
|
|
|
|
def healthy():
|
|
return state["up"]
|
|
|
|
def start(folder):
|
|
starts.append(folder)
|
|
state["up"] = True
|
|
|
|
service = sup.MarlinSupervisor(Path("."), healthy=healthy, start=start, wait_s=0.2)
|
|
first = service.ensure()
|
|
second = service.ensure()
|
|
self.assertTrue(first["started"])
|
|
self.assertEqual(len(starts), 1)
|
|
self.assertFalse(second["started"])
|
|
self.assertTrue(second["running"])
|
|
|
|
def test_failed_start_does_not_repeat_immediately(self):
|
|
starts = []
|
|
service = sup.MarlinSupervisor(
|
|
Path("."),
|
|
healthy=lambda: False,
|
|
start=lambda folder: starts.append(1),
|
|
wait_s=0.05,
|
|
)
|
|
service.ensure()
|
|
service.ensure()
|
|
self.assertEqual(len(starts), 1)
|
|
|
|
def test_parallel_ensure_starts_once(self):
|
|
gate = threading.Event()
|
|
starts = []
|
|
|
|
def start(folder):
|
|
starts.append(1)
|
|
gate.wait(1)
|
|
|
|
service = sup.MarlinSupervisor(
|
|
Path("."),
|
|
healthy=lambda: False,
|
|
start=start,
|
|
wait_s=0.05,
|
|
)
|
|
threads = [threading.Thread(target=service.ensure) for _ in range(4)]
|
|
for thread in threads:
|
|
thread.start()
|
|
gate.set()
|
|
for thread in threads:
|
|
thread.join(2)
|
|
self.assertEqual(len(starts), 1)
|
|
|
|
def test_spawn_is_detached_and_runs_the_module(self):
|
|
seen = {}
|
|
|
|
class Proc:
|
|
pid = 42
|
|
|
|
def popen(**kwargs):
|
|
seen.update(kwargs)
|
|
if kwargs.get("creationflags", 0) & sup._CREATE_BREAKAWAY:
|
|
raise OSError("breakaway")
|
|
return Proc()
|
|
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
pid = sup.spawn_detached(folder, popen=popen)
|
|
self.assertEqual(pid, 42)
|
|
self.assertEqual(seen["args"][1:], ["-m", "grotti.marlin_daemon"])
|
|
self.assertIs(seen["stdin"], sup.subprocess.DEVNULL)
|
|
if sup.os.name == "nt":
|
|
flags = seen["creationflags"]
|
|
self.assertTrue(flags & sup._DETACHED)
|
|
self.assertFalse(flags & sup._CREATE_BREAKAWAY)
|
|
else:
|
|
self.assertTrue(seen["start_new_session"])
|
|
|
|
def test_enable_reuses_a_live_daemon(self):
|
|
sup.enable(Path("."))
|
|
sup._supervisor.healthy = lambda: True
|
|
sup._supervisor.start = lambda folder: (_ for _ in ()).throw(AssertionError("gestartet"))
|
|
result = sup.ensure_if_enabled()
|
|
self.assertTrue(result["running"])
|
|
self.assertFalse(result["started"])
|