70 lines
3.0 KiB
Python
70 lines
3.0 KiB
Python
import queue
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from copienator_gui.batch_monitor import BatchMonitor, CHECK_INTERVAL_MS
|
|
from copienator_gui.notifications import notify_desktop
|
|
|
|
|
|
class BatchMonitorTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.scheduler = Mock()
|
|
self.scheduler.after.side_effect = lambda delay, callback: (delay, callback)
|
|
self.ready = Mock()
|
|
self.status = Mock()
|
|
self.monitor = BatchMonitor(self.scheduler, self.ready, self.status, Mock())
|
|
self.runner = Mock()
|
|
self.runner.events = queue.Queue()
|
|
self.factory = patch("copienator_gui.batch_monitor.ProcessRunner", return_value=self.runner)
|
|
self.factory.start()
|
|
self.addCleanup(self.factory.stop)
|
|
|
|
def start(self):
|
|
self.monitor.start(["check"], "/tmp", {}, None)
|
|
|
|
def test_checks_immediately_retries_in_five_minutes_and_notifies_once(self):
|
|
self.start()
|
|
self.runner.start.assert_called_once()
|
|
self.start()
|
|
self.runner.start.assert_called_once()
|
|
self.runner.events.put(("finished", (4, False)))
|
|
self.monitor._poll()
|
|
self.assertEqual(self.monitor.timer[0], CHECK_INTERVAL_MS)
|
|
self.assertEqual(CHECK_INTERVAL_MS, 300000)
|
|
self.ready.assert_not_called()
|
|
self.monitor.timer[1]()
|
|
self.assertEqual(self.runner.start.call_count, 2)
|
|
self.runner.events.put(("finished", (0, False)))
|
|
self.monitor._poll()
|
|
self.assertFalse(self.monitor.active)
|
|
self.assertIsNone(self.monitor.timer)
|
|
self.ready.assert_called_once()
|
|
self.monitor._poll()
|
|
self.ready.assert_called_once()
|
|
|
|
def test_stop_cancels_timer_and_running_check_without_notification(self):
|
|
self.start()
|
|
timer = self.monitor.timer
|
|
self.monitor.stop()
|
|
self.scheduler.after_cancel.assert_called_once_with(timer)
|
|
self.runner.force_stop.assert_called_once()
|
|
self.runner.events.put(("finished", (0, False)))
|
|
self.monitor._poll()
|
|
self.monitor._check()
|
|
self.ready.assert_not_called()
|
|
self.runner.start.assert_called_once()
|
|
|
|
def test_start_failure_retries_without_reporting_readiness(self):
|
|
self.runner.start.side_effect = OSError("unavailable")
|
|
self.start()
|
|
self.assertEqual(self.monitor.timer[0], CHECK_INTERVAL_MS)
|
|
self.ready.assert_not_called()
|
|
|
|
def test_linux_notification_passes_text_as_arguments(self):
|
|
with patch("copienator_gui.notifications.sys.platform", "linux"), patch(
|
|
"copienator_gui.notifications.find_executable", return_value="/usr/bin/notify-send"
|
|
), patch("copienator_gui.notifications.subprocess.Popen") as launch:
|
|
notify_desktop("Copienator", "Interro02 : résultats prêts")
|
|
self.assertEqual(launch.call_args.args[0], ["/usr/bin/notify-send", "--app-name=Copienator",
|
|
"--", "Copienator", "Interro02 : résultats prêts"])
|