224 lines
8.0 KiB
Python
224 lines
8.0 KiB
Python
import re
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import gi
|
|
|
|
gi.require_version("Gdk", "3.0")
|
|
gi.require_version("Gtk", "3.0")
|
|
gi.require_version("Atspi", "2.0")
|
|
|
|
from gi.repository import Atspi
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
|
|
|
soundGeneratorModule = sys.modules.get("cthulhu.sound_generator")
|
|
if soundGeneratorModule is not None and not hasattr(soundGeneratorModule, "SoundGenerator"):
|
|
class _StubSoundGenerator:
|
|
pass
|
|
|
|
soundGeneratorModule.SoundGenerator = _StubSoundGenerator
|
|
|
|
from cthulhu.ax_object import AXObject
|
|
from cthulhu.ax_utilities import AXUtilities
|
|
from cthulhu import cthulhu_state
|
|
from cthulhu.scripts.apps.steamwebhelper import script as steam_script
|
|
|
|
|
|
class SteamInactiveNotificationContractTests(unittest.TestCase):
|
|
@staticmethod
|
|
def _prepare_event_script(testScript, presentIfInactive):
|
|
handler = mock.Mock()
|
|
testScript.presentIfInactive = presentIfInactive
|
|
testScript.listeners = {"object:state-changed:showing": handler}
|
|
testScript.generatorCache = {"stale": object()}
|
|
testScript.skipObjectEvent = mock.Mock(return_value=False)
|
|
return handler
|
|
|
|
def test_steam_notification_is_processed_while_script_is_inactive(self):
|
|
app = object()
|
|
|
|
def initialize_chromium(testScript, initializedApp):
|
|
testScript.app = initializedApp
|
|
testScript.presentIfInactive = False
|
|
|
|
with mock.patch.object(
|
|
steam_script.Chromium.Script,
|
|
"__init__",
|
|
new=initialize_chromium,
|
|
):
|
|
testScript = steam_script.Script(app)
|
|
|
|
handler = self._prepare_event_script(testScript, testScript.presentIfInactive)
|
|
event = mock.Mock(type="object:state-changed:showing", source=object())
|
|
|
|
with (
|
|
mock.patch.object(cthulhu_state, "activeScript", object()),
|
|
mock.patch.object(AXObject, "get_role", return_value=Atspi.Role.NOTIFICATION),
|
|
):
|
|
testScript.processObjectEvent(event)
|
|
|
|
self.assertTrue(testScript.presentIfInactive)
|
|
handler.assert_called_once_with(event)
|
|
|
|
def test_generic_chromium_notification_is_deferred_while_script_is_inactive(self):
|
|
testScript = steam_script.Chromium.Script.__new__(
|
|
steam_script.Chromium.Script
|
|
)
|
|
handler = self._prepare_event_script(testScript, presentIfInactive=False)
|
|
event = mock.Mock(type="object:state-changed:showing", source=object())
|
|
|
|
with (
|
|
mock.patch.object(cthulhu_state, "activeScript", object()),
|
|
mock.patch.object(AXObject, "get_role", return_value=Atspi.Role.NOTIFICATION),
|
|
):
|
|
testScript.processObjectEvent(event)
|
|
|
|
handler.assert_not_called()
|
|
|
|
|
|
class SteamNotificationRootTests(unittest.TestCase):
|
|
def test_prefers_notification_ancestor_over_live_region_descendant(self):
|
|
testScript = steam_script.Script.__new__(steam_script.Script)
|
|
fragment = object()
|
|
notification = object()
|
|
|
|
def get_attribute(obj, name):
|
|
if obj is fragment and name == "container-live":
|
|
return "assertive"
|
|
return None
|
|
|
|
def find_ancestor(obj, predicate):
|
|
if obj is fragment and predicate(notification):
|
|
return notification
|
|
return None
|
|
|
|
with (
|
|
mock.patch.object(AXUtilities, "is_notification", side_effect=lambda obj: obj is notification),
|
|
mock.patch.object(AXUtilities, "is_alert", return_value=False),
|
|
mock.patch.object(AXObject, "get_attribute", side_effect=get_attribute),
|
|
mock.patch.object(AXObject, "find_ancestor", side_effect=find_ancestor),
|
|
):
|
|
result = testScript._findSteamNotificationRoot(fragment)
|
|
|
|
self.assertIs(result, notification)
|
|
|
|
|
|
class SteamNotificationRoutingContractTests(unittest.TestCase):
|
|
def test_steam_notification_override_handles_showing_before_chromium(self):
|
|
testScript = steam_script.Script.__new__(steam_script.Script)
|
|
notification = object()
|
|
event = mock.Mock(source=notification, detail1=1)
|
|
testScript._presentSteamNotification = mock.Mock()
|
|
|
|
with (
|
|
mock.patch.object(
|
|
testScript,
|
|
"_isSteamNotification",
|
|
return_value=True,
|
|
),
|
|
mock.patch.object(
|
|
steam_script.Chromium.Script,
|
|
"onShowingChanged",
|
|
) as chromiumHandler,
|
|
):
|
|
testScript.onShowingChanged(event)
|
|
|
|
testScript._presentSteamNotification.assert_called_once_with(notification)
|
|
chromiumHandler.assert_not_called()
|
|
|
|
def test_nonsteam_showing_event_defers_to_chromium(self):
|
|
testScript = steam_script.Script.__new__(steam_script.Script)
|
|
event = mock.Mock(source=object(), detail1=1)
|
|
testScript._presentSteamNotification = mock.Mock()
|
|
|
|
with (
|
|
mock.patch.object(
|
|
testScript,
|
|
"_isSteamNotification",
|
|
return_value=False,
|
|
),
|
|
mock.patch.object(
|
|
steam_script.Chromium.Script,
|
|
"onShowingChanged",
|
|
) as chromiumHandler,
|
|
):
|
|
testScript.onShowingChanged(event)
|
|
|
|
testScript._presentSteamNotification.assert_not_called()
|
|
chromiumHandler.assert_called_once_with(event)
|
|
|
|
|
|
class SteamNotificationQueueTests(unittest.TestCase):
|
|
@staticmethod
|
|
def _make_script():
|
|
testScript = steam_script.Script.__new__(steam_script.Script)
|
|
testScript._lastSteamNotification = ("", 0.0)
|
|
testScript._steamPendingNotification = None
|
|
testScript._steamRelativeTimePattern = re.compile(
|
|
r"^(?:just now|now|\d+\s+(?:second|minute|hour|day|week|month|year)s?\s+ago)$",
|
|
re.IGNORECASE,
|
|
)
|
|
testScript._resetSteamPendingTimer = mock.Mock()
|
|
testScript._presentSteamNotificationTextNow = mock.Mock()
|
|
return testScript
|
|
|
|
def test_merges_non_overlapping_fragments_for_same_notification(self):
|
|
testScript = self._make_script()
|
|
notification = object()
|
|
|
|
testScript._queueSteamNotification("username", notification)
|
|
testScript._queueSteamNotification("Playing: Borderlands 2", notification)
|
|
|
|
testScript._presentSteamNotificationTextNow.assert_not_called()
|
|
self.assertEqual(
|
|
testScript._steamPendingNotification["text"],
|
|
"username Playing: Borderlands 2",
|
|
)
|
|
|
|
testScript._flushSteamPendingNotification(fromTimer=True)
|
|
|
|
testScript._presentSteamNotificationTextNow.assert_called_once_with(
|
|
"username Playing: Borderlands 2",
|
|
notification,
|
|
)
|
|
|
|
def test_merges_relative_timestamp_without_repeated_presentation(self):
|
|
testScript = self._make_script()
|
|
notification = object()
|
|
|
|
testScript._queueSteamNotification("Username is now online", notification)
|
|
testScript._queueSteamNotification("2 minutes ago", notification)
|
|
|
|
testScript._presentSteamNotificationTextNow.assert_not_called()
|
|
self.assertEqual(
|
|
testScript._steamPendingNotification["text"],
|
|
"Username is now online. 2 minutes ago",
|
|
)
|
|
|
|
testScript._flushSteamPendingNotification(fromTimer=True)
|
|
|
|
testScript._presentSteamNotificationTextNow.assert_called_once_with(
|
|
"Username is now online. 2 minutes ago",
|
|
notification,
|
|
)
|
|
|
|
def test_duplicate_complete_notification_is_presented_once(self):
|
|
testScript = self._make_script()
|
|
notification = object()
|
|
|
|
testScript._queueSteamNotification("Username is now online", notification)
|
|
testScript._queueSteamNotification("Username is now online", notification)
|
|
testScript._flushSteamPendingNotification(fromTimer=True)
|
|
|
|
testScript._presentSteamNotificationTextNow.assert_called_once_with(
|
|
"Username is now online",
|
|
notification,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|