Automate notification tests

Signed-off-by: ggediminass <[email protected]>

test/qa/lib/__init__.py

@@ -235,6 +235,13 @@ def set_killswitch(killswitch):
         print("WARNING:", ex)
 
 
+def set_notify(dns):
+    try:
+        print(sh.nordvpn.set.notify(dns))
+    except sh.ErrorReturnCode_1 as ex:
+        print("WARNING:", ex)
+
+
 def add_port_to_allowlist(port):
     try:
         print(sh.nordvpn.allowlist.add.port(port))

test/qa/lib/notify.py

@@ -0,0 +1,105 @@
+from lib import server
+import sh
+import subprocess
+from threading import Thread
+
+
+class NotificationCaptureThreadResult:
+    def __init__(self, icon_match: bool, summary_match: bool, body_match: bool):
+        self.icon_match = icon_match
+        self.summary_match = summary_match
+        self.body_match = body_match
+
+    def __eq__(self, other):
+        if isinstance(other, NotificationCaptureThreadResult):
+            return (self.icon_match == other.icon_match) and (self.summary_match == other.summary_match) and (self.body_match == other.body_match)
+        return False
+
+
+# Used for asserts in tests, [Icon match, Summary match, Body match]
+NOTIFICATION_DETECTED = NotificationCaptureThreadResult(True, True, True)
+NOTIFICATION_NOT_DETECTED = NotificationCaptureThreadResult(False, False, False)
+
+
+# Used to check if error messages are correct
+NOTIFY_MSG_ERROR_ALREADY_ENABLED = "Notifications are already set to 'enabled'."
+NOTIFY_MSG_ERROR_ALREADY_DISABLED = "Notifications are already set to 'disabled'."
+
+
+class NotificationCaptureThread(Thread):
+    def __init__(self, expected_msg):
+        Thread.__init__(self)
+        self.value: NotificationCaptureThreadResult = NotificationCaptureThreadResult(False, False, False)
+        self.expected_message = expected_msg
+
+    def capture_notifications(self, message):
+        """ 
+            returns `NotificationCaptureThreadResult`, and contains booleans - icon_match, summary_match, body_match - according to found notification contents
+        """
+
+        # Timeout is needed, in order for Thread not to hang, as we need to exit the process at some point
+        # Timeout can be altered according to how fast you connect to NordVPN server
+        command = ["timeout", "6", "dbus-monitor", "--session", "type=method_call,interface=org.freedesktop.Notifications"]
+        process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+
+        result = NotificationCaptureThreadResult(False, False, False)
+
+        for line in process.stdout:
+            if "/usr/share/icons/hicolor/scalable/apps/nordvpn.svg" in line:
+                result.icon_match = True
+
+            if "NordVPN" in line:
+                result.summary_match = True
+
+            if message in line:
+                result.body_match = True
+
+        return result
+
+    def run(self):
+        self.value = self.capture_notifications(self.expected_message)
+
+
+def connect_and_capture_notifications(tech, proto, obfuscated) -> NotificationCaptureThreadResult:
+    """ returns [True, True, True] if notification with all expected contents from NordVPN was captured while connecting to VPN server """
+
+    # Choose server for test, so we know the full expected message
+    name, hostname = server.get_hostname_by(tech, proto, obfuscated)
+    expected_msg = f"You are connected to {name} ({hostname})!"
+
+    # We try to capture notifications using other thread when connecting to NordVPN server
+    t_connect = NotificationCaptureThread(expected_msg)
+    t_connect.start()
+
+    sh.nordvpn.connect(hostname.split(".")[0])
+
+    t_connect.join()
+
+    return t_connect.value
+    # Return types, reikia koki structa pakurt ir returnint
+
+
+def disconnect_and_capture_notifications() -> NotificationCaptureThreadResult:
+    """ returns [True, True, True] if notification with all expected contents from NordVPN was captured while disconnecting from VPN server """
+
+    # We know what message we expect to appear in notification
+    expected_msg = "You are disconnected from NordVPN."
+
+    # We try to capture notifications using other thread when disconnecting from NordVPN server
+    t_disconnect = NotificationCaptureThread(expected_msg)
+    t_disconnect.start()
+
+    sh.nordvpn.disconnect()
+
+    t_disconnect.join()
+
+    return t_disconnect.value
+
+
+def print_tidy_exception(obj1: NotificationCaptureThreadResult, obj2: NotificationCaptureThreadResult) -> None:
+    """ Prints values of attributes from specified NotificationCaptureThreadResult type of objects """
+    return \
+        f"\n\n(icon, summary, body)\n" \
+        f"({obj1.icon_match}, {obj1.summary_match}, {obj1.body_match}) - connect_notification / disconnect_notification - found\n" \
+        f"({obj2.icon_match}, {obj2.summary_match}, {obj2.body_match}) - notify.NOTIFICATION_DETECTED / notify.NOTIFICATION_NOT_DETECTED - expected" \
+        f"\n\n"

test/qa/lib/settings.py

@@ -33,3 +33,8 @@ def dns_visible_in_settings(dns: list) -> bool:
 def get_is_tpl_enabled():
     """ returns True, if Threat Protection Lite is enabled in application settings """
     return "Threat Protection Lite: enabled" in sh.nordvpn.settings()
+
+
+def get_is_notify_enabled():
+    """ returns True, if Threat Protection Lite is enabled in application settings """
+    return "Notify: enabled" in sh.nordvpn.settings()

test/qa/test_notify.py

@@ -0,0 +1,179 @@
+from lib import (
+    daemon,
+    info,
+    logging,
+    login,
+    notify,
+    settings
+)
+import lib
+import pytest
+import sh
+import timeout_decorator
+
+
+def setup_module(module):
+    daemon.start()
+    login.login_as("default")
+
+
+def teardown_module(module):
+    sh.nordvpn.logout("--persist-token")
+    daemon.stop()
+
+
+def setup_function(function):
+    logging.log()
+
+    # Make sure that Notifications are disabled before we execute each test
+    lib.set_notify("off")
+
+
+def teardown_function(function):
+    logging.log(data=info.collect())
+    logging.log()
+
+
+@pytest.mark.parametrize("tech,proto,obfuscated", lib.TECHNOLOGIES)
+@pytest.mark.flaky(reruns=2, reruns_delay=90)
+@timeout_decorator.timeout(40)
+def test_notifications_disabled_connect(tech, proto, obfuscated):
+    lib.set_technology_and_protocol(tech, proto, obfuscated)
+
+    assert not settings.get_is_notify_enabled()
+
+    connect_notification = notify.connect_and_capture_notifications(tech, proto, obfuscated)
+
+    assert connect_notification == notify.NOTIFICATION_NOT_DETECTED, \
+        notify.print_tidy_exception(connect_notification, notify.NOTIFICATION_NOT_DETECTED)
+
+    disconnect_notification = notify.disconnect_and_capture_notifications()
+
+    assert disconnect_notification == notify.NOTIFICATION_NOT_DETECTED, \
+        notify.print_tidy_exception(connect_notification, notify.NOTIFICATION_NOT_DETECTED)
+
+
+@pytest.mark.parametrize("tech,proto,obfuscated", lib.TECHNOLOGIES)
+@pytest.mark.flaky(reruns=2, reruns_delay=90)
+@timeout_decorator.timeout(40)
+def test_notifications_enabled_connect(tech, proto, obfuscated):
+    lib.set_technology_and_protocol(tech, proto, obfuscated)
+
+    sh.nordvpn.set.notify.on()
+    assert settings.get_is_notify_enabled()
+
+    connect_notification = notify.connect_and_capture_notifications(tech, proto, obfuscated)
+
+    # Should fail here, if tested with 3.16.6, since notification icon is missing
+    assert connect_notification == notify.NOTIFICATION_DETECTED, \
+        notify.print_tidy_exception(connect_notification, notify.NOTIFICATION_DETECTED)
+
+    disconnect_notification = notify.disconnect_and_capture_notifications()
+
+    assert disconnect_notification == notify.NOTIFICATION_DETECTED, \
+        notify.print_tidy_exception(disconnect_notification, notify.NOTIFICATION_DETECTED)
+
+
+@pytest.mark.parametrize("tech,proto,obfuscated", lib.TECHNOLOGIES)
+@pytest.mark.flaky(reruns=2, reruns_delay=90)
+@timeout_decorator.timeout(40)
+def test_notifications_enabled_connected_disable(tech, proto, obfuscated):
+    lib.set_technology_and_protocol(tech, proto, obfuscated)
+
+    sh.nordvpn.set.notify.on()
+    assert settings.get_is_notify_enabled()
+
+    connect_notification = notify.connect_and_capture_notifications(tech, proto, obfuscated)
+
+    # Should fail here, if tested with 3.16.6, since notification icon is missing
+    assert connect_notification == notify.NOTIFICATION_DETECTED, \
+        notify.print_tidy_exception(connect_notification, notify.NOTIFICATION_DETECTED)
+
+    sh.nordvpn.set.notify.off()
+    assert not settings.get_is_notify_enabled()
+
+    disconnect_notification = notify.disconnect_and_capture_notifications()
+    assert disconnect_notification == notify.NOTIFICATION_NOT_DETECTED, \
+        notify.print_tidy_exception(disconnect_notification, notify.NOTIFICATION_NOT_DETECTED)
+
+
+@pytest.mark.parametrize("tech,proto,obfuscated", lib.TECHNOLOGIES)
+@pytest.mark.flaky(reruns=2, reruns_delay=90)
+@timeout_decorator.timeout(40)
+def test_notifications_disabled_connected_enable(tech, proto, obfuscated):
+    lib.set_technology_and_protocol(tech, proto, obfuscated)
+
+    assert not settings.get_is_notify_enabled()
+
+    connect_notification = notify.connect_and_capture_notifications(tech, proto, obfuscated)
+
+    assert connect_notification == notify.NOTIFICATION_NOT_DETECTED, \
+        notify.print_tidy_exception(connect_notification, notify.NOTIFICATION_NOT_DETECTED)
+
+    sh.nordvpn.set.notify.on()
+    assert settings.get_is_notify_enabled()
+
+    # Should fail here, if tested with 3.16.6, since notification icon is missing
+    disconnect_notification = notify.disconnect_and_capture_notifications()
+    assert disconnect_notification == notify.NOTIFICATION_DETECTED, \
+        notify.print_tidy_exception(disconnect_notification, notify.NOTIFICATION_DETECTED)
+
+
+@pytest.mark.parametrize("tech,proto,obfuscated", lib.TECHNOLOGIES)
+@pytest.mark.flaky(reruns=2, reruns_delay=90)
+@timeout_decorator.timeout(40)
+def test_notify_already_enabled_disconnected(tech, proto, obfuscated):
+    lib.set_technology_and_protocol(tech, proto, obfuscated)
+
+    sh.nordvpn.set.notify.on()
+    assert settings.get_is_notify_enabled()
+
+    output = sh.nordvpn.set.notify.on()
+    assert notify.NOTIFY_MSG_ERROR_ALREADY_ENABLED in str(output)
+    assert settings.get_is_notify_enabled()
+
+
+@pytest.mark.parametrize("tech,proto,obfuscated", lib.TECHNOLOGIES)
+@pytest.mark.flaky(reruns=2, reruns_delay=90)
+@timeout_decorator.timeout(40)
+def test_notify_already_enabled_connected(tech, proto, obfuscated):
+    lib.set_technology_and_protocol(tech, proto, obfuscated)
+
+    with lib.Defer(sh.nordvpn.disconnect):
+        sh.nordvpn.connect()
+
+        sh.nordvpn.set.notify.on()
+        assert settings.get_is_notify_enabled()
+
+        output = sh.nordvpn.set.notify.on()
+        assert notify.NOTIFY_MSG_ERROR_ALREADY_ENABLED in str(output)
+        assert settings.get_is_notify_enabled()
+
+
+@pytest.mark.parametrize("tech,proto,obfuscated", lib.TECHNOLOGIES)
+@pytest.mark.flaky(reruns=2, reruns_delay=90)
+@timeout_decorator.timeout(40)
+def test_notify_already_disabled_disconnected(tech, proto, obfuscated):
+    lib.set_technology_and_protocol(tech, proto, obfuscated)
+
+    assert not settings.get_is_notify_enabled()
+
+    output = sh.nordvpn.set.notify.off()
+    assert notify.NOTIFY_MSG_ERROR_ALREADY_DISABLED in str(output)
+    assert not settings.get_is_notify_enabled()
+
+
+@pytest.mark.parametrize("tech,proto,obfuscated", lib.TECHNOLOGIES)
+@pytest.mark.flaky(reruns=2, reruns_delay=90)
+@timeout_decorator.timeout(40)
+def test_notify_already_disabled_connected(tech, proto, obfuscated):
+    lib.set_technology_and_protocol(tech, proto, obfuscated)
+
+    with lib.Defer(sh.nordvpn.disconnect):
+        sh.nordvpn.connect()
+
+        assert not settings.get_is_notify_enabled()
+
+        output = sh.nordvpn.set.notify.off()
+        assert notify.NOTIFY_MSG_ERROR_ALREADY_DISABLED in str(output)
+        assert not settings.get_is_notify_enabled()