72 lines
3.0 KiB
Python
72 lines
3.0 KiB
Python
import subprocess
|
||
import os
|
||
import tkinter as tk
|
||
import unittest
|
||
from unittest.mock import patch
|
||
|
||
from copienator_gui.syncthing import (
|
||
SERVICE,
|
||
SYSTEMCTL,
|
||
SyncthingPanel,
|
||
control_service,
|
||
service_status,
|
||
)
|
||
|
||
|
||
class SyncthingTests(unittest.TestCase):
|
||
def test_active_service_is_not_restarted(self):
|
||
with patch("copienator_gui.syncthing.service_status", return_value="active"), \
|
||
patch("copienator_gui.syncthing.subprocess.run") as run:
|
||
self.assertEqual(control_service("ensure"), "active")
|
||
run.assert_not_called()
|
||
|
||
def test_inactive_service_is_started_and_verified(self):
|
||
with patch("copienator_gui.syncthing.service_status", side_effect=["inactive", "active"]), \
|
||
patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run:
|
||
self.assertEqual(control_service("ensure"), "active")
|
||
self.assertEqual(run.call_args.args[0], [*SYSTEMCTL, "start", SERVICE])
|
||
|
||
def test_failed_start_reports_error(self):
|
||
with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \
|
||
patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 1, stderr="Access denied")):
|
||
with self.assertRaisesRegex(RuntimeError, "Access denied"):
|
||
control_service("ensure")
|
||
|
||
def test_stop_targets_system_instance(self):
|
||
with patch("copienator_gui.syncthing.service_status", return_value="inactive"), \
|
||
patch("copienator_gui.syncthing.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run:
|
||
self.assertEqual(control_service("stop"), "inactive")
|
||
self.assertEqual(run.call_args.args[0], [*SYSTEMCTL, "stop", SERVICE])
|
||
|
||
def test_status_uses_the_non_privileged_user_service(self):
|
||
with patch(
|
||
"copienator_gui.syncthing.subprocess.run",
|
||
return_value=subprocess.CompletedProcess([], 0, stdout="active\n"),
|
||
) as run:
|
||
self.assertEqual(service_status(), "active")
|
||
self.assertEqual(
|
||
run.call_args.args[0],
|
||
[*SYSTEMCTL, "show", SERVICE, "--property=ActiveState", "--value"],
|
||
)
|
||
|
||
|
||
@unittest.skipUnless(os.environ.get("DISPLAY"), "Tk requires a display")
|
||
class SyncthingPanelTests(unittest.TestCase):
|
||
def test_failed_automatic_start_warns_and_continues_export(self):
|
||
root = tk.Tk()
|
||
self.addCleanup(root.destroy)
|
||
with patch("copienator_gui.syncthing.threading.Thread.start"):
|
||
panel = SyncthingPanel(root)
|
||
panel.pack()
|
||
continued = []
|
||
panel.pending_export = lambda: continued.append(True)
|
||
panel.busy = True
|
||
panel.events.put(("ensure", None, "Access denied"))
|
||
|
||
with patch("copienator_gui.syncthing.messagebox.showwarning") as warning:
|
||
panel.poll()
|
||
|
||
warning.assert_called_once()
|
||
self.assertIn("L’export va continuer", warning.call_args.args[1])
|
||
self.assertEqual(continued, [True])
|