From a78c168705189169a0ceb5e94a395a5d129a38da Mon Sep 17 00:00:00 2001 From: Paolo Gentili Date: Mon, 25 Nov 2024 11:40:36 +0100 Subject: [PATCH] Move jobs to dedicated external provider (New) (#1613) * Change: so long zapper * Change: moved desktop_session to resource provider * Change: remove explicit Zapper support from run_watcher * Add: unit tests for removable_storage_watcher::main --- .../checkbox_support/scripts/run_watcher.py | 92 +++--- .../scripts/tests/test_zapper_proxy.py | 194 ------------ .../checkbox_support/scripts/zapper_proxy.py | 112 ------- .../tests/test_run_watcher.py | 92 +----- checkbox-support/pyproject.toml | 1 - checkbox-support/setup.cfg | 1 - .../using-checkbox/advanced-commands.rst | 1 - providers/base/README.rst | 2 +- providers/base/bin/bt_a2dp.py | 78 ----- providers/base/bin/edid_cycle.py | 212 ------------- .../base/bin/removable_storage_watcher.py | 53 +--- providers/base/bin/zapper_keyboard_test.py | 180 ----------- providers/base/data/edids/1280x1024.edid | Bin 256 -> 0 bytes providers/base/data/edids/1920x1080.edid | Bin 256 -> 0 bytes providers/base/data/edids/2560x1440.edid | Bin 256 -> 0 bytes providers/base/tests/test_edid_cycle.py | 232 -------------- .../tests/test_removable_storage_watcher.py | 87 +++++ providers/base/tests/test_zapper_keyboard.py | 296 ------------------ providers/base/units/canary/test-plan.pxu | 11 - providers/base/units/desktop/resource.pxu | 6 - providers/base/units/zapper/jobs.pxu | 68 ---- providers/base/units/zapper/packaging.pxu | 5 - providers/base/units/zapper/resource.pxu | 18 -- providers/base/units/zapper/test-plan.pxu | 23 -- .../units/client-cert-iot-desktop-24-04.pxu | 1 - .../units/client-cert-iot-server-24-04.pxu | 1 - .../units/client-cert-iot-ubuntucore-22.pxu | 1 - .../units/client-cert-iot-ubuntucore-24.pxu | 1 - .../bin/desktop_session_resource.py} | 0 providers/resource/jobs/resource.pxu | 6 + .../tests/test_desktop_session_resource.py} | 14 +- providers/sru/units/sru.pxu | 1 - 32 files changed, 167 insertions(+), 1622 deletions(-) delete mode 100644 checkbox-support/checkbox_support/scripts/tests/test_zapper_proxy.py delete mode 100644 checkbox-support/checkbox_support/scripts/zapper_proxy.py delete mode 100755 providers/base/bin/bt_a2dp.py delete mode 100755 providers/base/bin/edid_cycle.py delete mode 100755 providers/base/bin/zapper_keyboard_test.py delete mode 100644 providers/base/data/edids/1280x1024.edid delete mode 100644 providers/base/data/edids/1920x1080.edid delete mode 100644 providers/base/data/edids/2560x1440.edid delete mode 100644 providers/base/tests/test_edid_cycle.py create mode 100644 providers/base/tests/test_removable_storage_watcher.py delete mode 100644 providers/base/tests/test_zapper_keyboard.py delete mode 100644 providers/base/units/desktop/resource.pxu delete mode 100644 providers/base/units/zapper/jobs.pxu delete mode 100644 providers/base/units/zapper/packaging.pxu delete mode 100644 providers/base/units/zapper/resource.pxu delete mode 100644 providers/base/units/zapper/test-plan.pxu rename providers/{base/bin/desktop_session.py => resource/bin/desktop_session_resource.py} (100%) rename providers/{base/tests/test_desktop_session.py => resource/tests/test_desktop_session_resource.py} (77%) diff --git a/checkbox-support/checkbox_support/scripts/run_watcher.py b/checkbox-support/checkbox_support/scripts/run_watcher.py index 2124d8ed21..93b36cb8be 100644 --- a/checkbox-support/checkbox_support/scripts/run_watcher.py +++ b/checkbox-support/checkbox_support/scripts/run_watcher.py @@ -10,7 +10,6 @@ """ import argparse import logging -import os import re import select import sys @@ -19,7 +18,6 @@ from abc import ABC, abstractmethod from checkbox_support.helpers.timeout import timeout -from checkbox_support.scripts.zapper_proxy import zapper_run from checkbox_support.scripts.usb_read_write import ( mount_usb_storage, gen_random_file, @@ -35,6 +33,33 @@ ACTION_TIMEOUT = 30 +class ControllerInterface(ABC): + """ + Perform actions on the device user test. + """ + + @abstractmethod + def action(self, action_type): + """Perform action of type `action_type` on the DUT.""" + + +class ManualController(ControllerInterface): + """ + Interact with the device under test manually. + """ + + def action(self, action_type): + """Perform action of type `action_type` on the DUT.""" + + if action_type == "insertion": + print("\n\nINSERT NOW\n\n", flush=True) + elif action_type == "removal": + print("\n\nREMOVE NOW\n\n", flush=True) + else: + raise SystemExit("Invalid test case") + print("Timeout: {} seconds".format(ACTION_TIMEOUT), flush=True) + + class StorageInterface(ABC): """ StorageInterface makes sure each type of storage class should implement @@ -72,41 +97,21 @@ class StorageWatcher(StorageInterface): function to detect the insertion and removal of storage. """ - def __init__(self, storage_type, zapper_usb_address): + def __init__(self, storage_type, controller=ManualController()): self.storage_type = storage_type - self.zapper_usb_address = zapper_usb_address self.testcase = None self.test_passed = False self.mounted_partition = None + self._controller = controller def run(self): j = journal.Reader() j.seek_realtime(time.time()) p = select.poll() p.register(j, j.get_events()) - if self.zapper_usb_address: - zapper_host = os.environ.get("ZAPPER_HOST") - if not zapper_host: - raise SystemExit("ZAPPER_HOST environment variable not found!") - usb_address = self.zapper_usb_address - if self.testcase == "insertion": - print("Calling zapper to connect the USB device") - zapper_run( - zapper_host, "typecmux_set_state", usb_address, "DUT" - ) - elif self.testcase == "removal": - print("Calling zapper to disconnect the USB device") - zapper_run( - zapper_host, "typecmux_set_state", usb_address, "OFF" - ) - else: - if self.testcase == "insertion": - print("\n\nINSERT NOW\n\n", flush=True) - elif self.testcase == "removal": - print("\n\nREMOVE NOW\n\n", flush=True) - else: - raise SystemExit("Invalid test case") - print("Timeout: {} seconds".format(ACTION_TIMEOUT), flush=True) + + self._controller.action(self.testcase) + while p.poll(): if j.process() != journal.APPEND: continue @@ -168,8 +173,8 @@ class USBStorage(StorageWatcher): USBStorage handles the insertion and removal of usb2 and usb3. """ - def __init__(self, *args): - super().__init__(*args) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) self.mounted_partition = None self.device = None self.number = None @@ -257,8 +262,8 @@ class MediacardStorage(StorageWatcher): MediacardStorage handles the insertion and removal of sd, sdhc, mmc etc... """ - def __init__(self, *args): - super().__init__(*args) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) self.mounted_partition = None self.action = None self.device = None @@ -316,8 +321,8 @@ class MediacardComboStorage(StorageWatcher): etc., for devices that combine mediacard and usb storage. """ - def __init__(self, *args): - super().__init__(*args) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) self.mounted_partition = None self.action = None self.device = None @@ -358,8 +363,8 @@ class ThunderboltStorage(StorageWatcher): storage. """ - def __init__(self, *args): - super().__init__(*args) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) self.mounted_partition = None self.action = None @@ -419,11 +424,6 @@ def parse_args(): ], help=("usb2, usb3, mediacard, mediacard_combo or thunderbolt"), ) - parser.add_argument( - "--zapper-usb-address", - type=str, - help="Zapper's USB switch address to use", - ) return parser.parse_args() @@ -432,17 +432,13 @@ def main(): watcher = None if args.storage_type == "thunderbolt": - watcher = ThunderboltStorage( - args.storage_type, args.zapper_usb_address - ) + watcher = ThunderboltStorage(args.storage_type) elif args.storage_type == "mediacard": - watcher = MediacardStorage(args.storage_type, args.zapper_usb_address) + watcher = MediacardStorage(args.storage_type) elif args.storage_type == "mediacard_combo": - watcher = MediacardComboStorage( - args.storage_type, args.zapper_usb_address - ) + watcher = MediacardComboStorage(args.storage_type) else: - watcher = USBStorage(args.storage_type, args.zapper_usb_address) + watcher = USBStorage(args.storage_type) if args.testcase == "insertion": mounted_partition = watcher.run_insertion() diff --git a/checkbox-support/checkbox_support/scripts/tests/test_zapper_proxy.py b/checkbox-support/checkbox_support/scripts/tests/test_zapper_proxy.py deleted file mode 100644 index a0dd72491f..0000000000 --- a/checkbox-support/checkbox_support/scripts/tests/test_zapper_proxy.py +++ /dev/null @@ -1,194 +0,0 @@ -# Copyright 2022 Canonical Ltd. -# Written by: -# Maciej Kisielewski -# Paolo Gentili -# -# This is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 3, -# as published by the Free Software Foundation. -# -# This file is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this file. If not, see . -"""Tests for zapper_proxy module.""" - -from unittest import TestCase -from unittest.mock import Mock, patch - -from checkbox_support.scripts.zapper_proxy import ( - get_capabilities, - zapper_run, - main, -) - - -class ZapperProxyV1Tests(TestCase): - """Unit tests for ZapperProxy module.""" - - def setUp(self): - self._rpyc_mock = Mock() - self._mocked_conn = Mock() - self._rpyc_mock.connect.return_value = self._mocked_conn - - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_zapper_run_smoke(self, import_mock): - """ - Check if zapper_run calls the appropriate function on the rpyc client. - """ - import_mock.return_value = self._rpyc_mock - self._mocked_conn.root.command.return_value = "test" - - args = ["a", "b"] - kwargs = {"k1": "v1", "k2": "v2"} - result = zapper_run("0.0.0.0", "command", *args, **kwargs) - self._mocked_conn.root.command.assert_called_once_with(*args, **kwargs) - assert result == "test" - - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_zapper_run_wrong_cmd(self, import_mock): - """ - Check if SystemExit is raised when an unavailable command is requested. - """ - import_mock.return_value = self._rpyc_mock - self._mocked_conn.root.command.side_effect = AttributeError() - with self.assertRaises(SystemExit): - zapper_run("0.0.0.0", "command") - - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_zapper_run_missing_rpyc(self, import_mock): - """ - Check if SystemExit is raised when RPyC cannot be imported. - """ - import_mock.side_effect = ImportError - with self.assertRaises(SystemExit): - zapper_run("0.0.0.0", "command") - - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_zapper_run_service_error(self, import_mock): - """ - Check if SystemExit is raised when an error occurs on Zapper service. - """ - import_mock.return_value = self._rpyc_mock - - class TestException(Exception): - pass - - self._rpyc_mock.core.vinegar.GenericException = TestException - self._mocked_conn.root.command.side_effect = TestException() - - with self.assertRaises(SystemExit): - zapper_run("0.0.0.0", "command") - - @patch("time.sleep", Mock()) - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_zapper_run_connection_error(self, import_mock): - """ - Check if SystemExit is raised when the connections cannot be established - after two tentatives. - """ - import_mock.return_value = self._rpyc_mock - self._rpyc_mock.connect.side_effect = ConnectionRefusedError() - - with self.assertRaises(SystemExit): - zapper_run("0.0.0.0", "command") - assert self._rpyc_mock.connect.call_count == 2 - - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_get_capabilities_one_cap(self, import_mock): - """ - Check if get_capabilities properly prints one record. - - The record should be in Checkbox resource syntax and should not be - surrounded by any newlines. - """ - import_mock.return_value = self._rpyc_mock - - ret_val = [{"foo": "bar"}] - self._mocked_conn.root.get_capabilities = Mock(return_value=ret_val) - - with patch("builtins.print") as mocked_print: - get_capabilities("0.0.0.0") - mocked_print.assert_called_once_with("foo: bar\n\navailable: True") - - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_get_capabilities_error(self, import_mock): - """ - Check if get_capabilities prints nothing on error while - fetching capabilities. - """ - import_mock.return_value = self._rpyc_mock - - self._mocked_conn.root.get_capabilities.side_effect = AttributeError - - with patch("builtins.print") as mocked_print: - get_capabilities("0.0.0.0") - mocked_print.assert_called_once_with("available: False") - - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_get_capabilities_empty(self, import_mock): - """Check if get_capabilities prints nothing on no capabilities.""" - import_mock.return_value = self._rpyc_mock - - ret_val = [] - self._mocked_conn.root.get_capabilities = Mock(return_value=ret_val) - with patch("builtins.print") as mocked_print: - get_capabilities("0.0.0.0") - mocked_print.assert_called_once_with("available: True") - - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_get_capabilities_multiple_caps(self, import_mock): - """ - Check if get_capabilities properly prints multiple records. - - The records should be in Checkbox resource syntax. Records should be - separated by an empty line. - """ - import_mock.return_value = self._rpyc_mock - - ret_val = [{"foo": "bar"}, {"baz": "biz"}] - self._mocked_conn.root.get_capabilities = Mock(return_value=ret_val) - - with patch("builtins.print") as mocked_print: - get_capabilities("0.0.0.0") - mocked_print.assert_called_once_with( - "foo: bar\n\nbaz: biz\n\navailable: True" - ) - - @patch("checkbox_support.scripts.zapper_proxy.import_module") - def test_get_capabilities_one_cap_multi_rows(self, import_mock): - """ - Check if get_capabilities properly prints a record with multiple caps. - - Each capability should be printed in a separate line. - No additional newlines should be printed. - """ - import_mock.return_value = self._rpyc_mock - - ret_val = [{"foo": "bar", "foo2": "bar2"}] - self._mocked_conn.root.get_capabilities = Mock(return_value=ret_val) - - with patch("builtins.print") as mocked_print: - get_capabilities("0.0.0.0") - mocked_print.assert_called_once_with( - "foo: bar\nfoo2: bar2\n\navailable: True" - ) - - @patch("checkbox_support.scripts.zapper_proxy.zapper_run") - def test_main_run(self, mock_run): - """ - Check if main calls zapper_run with proper parameters. - """ - main(["command", "arg1", "arg2", "--host", "myhost"]) - mock_run.assert_called_once_with("myhost", "command", "arg1", "arg2") - - @patch("checkbox_support.scripts.zapper_proxy.get_capabilities") - def test_main_capabilities(self, mock_cap): - """ - Check if main calls get_capabilities with zapper host. - """ - main(["get_capabilities", "arg1", "arg2", "--host", "myhost"]) - mock_cap.assert_called_once_with("myhost") diff --git a/checkbox-support/checkbox_support/scripts/zapper_proxy.py b/checkbox-support/checkbox_support/scripts/zapper_proxy.py deleted file mode 100644 index 4307ccd022..0000000000 --- a/checkbox-support/checkbox_support/scripts/zapper_proxy.py +++ /dev/null @@ -1,112 +0,0 @@ -# Copyright 2022 Canonical Ltd. -# Written by: -# Maciej Kisielewski -# Paolo Gentili -# -# This is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 3, -# as published by the Free Software Foundation. -# -# This file is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this file. If not, see . -""" -This program acts as a proxy to Zapper Hardware. - -It uses internal Zapper Control RPyC API. -""" -import argparse -import os -import time - -from importlib import import_module - - -def zapper_run(host, cmd, *args, **kwargs): - """ - Run command on Zapper. - - :param host: Zapper IP address - :param cmd: command to be executed - :param args: command arguments - :param kwargs: command keyword arguments - :returns: whatever is returned by Zapper service - :raises SystemExit: if the connection cannot be established - or the command is unknown - or a service error occurs - """ - try: - _rpyc = import_module("rpyc") - except ImportError: - try: - _rpyc = import_module("plainbox.vendor.rpyc") - except ImportError as exc: - msg = "RPyC not found. Neither from sys nor from Checkbox" - raise SystemExit(msg) from exc - - for _ in range(2): - try: - conn = _rpyc.connect(host, 60000, config={"allow_all_attrs": True}) - break - except ConnectionRefusedError: - time.sleep(1) - else: - raise SystemExit("Cannot connect to Zapper Host.") - - try: - return getattr(conn.root, cmd)(*args, **kwargs) - except AttributeError: - raise SystemExit( - "Zapper host does not provide a '{}' command.".format(cmd) - ) - except _rpyc.core.vinegar.GenericException as exc: - raise SystemExit( - "Zapper host failed to process the requested command." - ) from exc - - -def get_capabilities(host): - """Get Zapper capabilities.""" - try: - capabilities = zapper_run(host, "get_capabilities") - capabilities.append({"available": True}) - except SystemExit: - capabilities = [{"available": False}] - - def stringify_cap(cap): - return "\n".join( - "{}: {}".format(key, val) for key, val in sorted(cap.items()) - ) - - print("\n\n".join(stringify_cap(cap) for cap in capabilities)) - - -def main(arguments=None): - """Entry point.""" - - parser = argparse.ArgumentParser() - parser.add_argument( - "--host", - default=os.environ.get("ZAPPER_HOST"), - help=( - "Address of Zapper to connect to. If not supplied, " - "ZAPPER_HOST environment variable will be used." - ), - ) - parser.add_argument("cmd") - parser.add_argument("args", nargs="*") - args = parser.parse_args(arguments) - - if args.cmd == "get_capabilities": - get_capabilities(args.host) - else: - result = zapper_run(args.host, args.cmd, *args.args) - print(result) - - -if __name__ == "__main__": - main() diff --git a/checkbox-support/checkbox_support/tests/test_run_watcher.py b/checkbox-support/checkbox_support/tests/test_run_watcher.py index 865963d4fd..68d6066fb5 100644 --- a/checkbox-support/checkbox_support/tests/test_run_watcher.py +++ b/checkbox-support/checkbox_support/tests/test_run_watcher.py @@ -18,7 +18,7 @@ # along with Checkbox. If not, see . import unittest -from unittest.mock import patch, call, MagicMock, mock_open +from unittest.mock import patch, call, MagicMock import argparse from systemd import journal @@ -26,6 +26,7 @@ from checkbox_support.scripts.run_watcher import ( StorageWatcher, USBStorage, + ManualController, MediacardStorage, MediacardComboStorage, ThunderboltStorage, @@ -47,7 +48,7 @@ def test_storage_watcher_run_with_insertion(self, mock_poll, mock_journal): mock_poll.return_value.poll.side_effect = [True, True, False] mock_storage_watcher = MagicMock() - mock_storage_watcher.zapper_usb_address = "" + mock_storage_watcher._controller = ManualController() # Test insertion mock_storage_watcher.testcase = "insertion" @@ -72,7 +73,7 @@ def test_storage_watcher_run_with_removal(self, mock_poll, mock_journal): mock_poll.return_value.poll.side_effect = [True, False] mock_storage_watcher = MagicMock() - mock_storage_watcher.zapper_usb_address = "" + mock_storage_watcher._controller = ManualController() # Test removal mock_storage_watcher.testcase = "removal" @@ -89,62 +90,12 @@ def test_storage_watcher_run_with_removal(self, mock_poll, mock_journal): def test_storage_watcher_run_invalid_testcase(self): mock_storage_watcher = MagicMock() mock_storage_watcher.testcase = "invalid" - mock_storage_watcher.zapper_usb_address = "" + mock_storage_watcher._controller = ManualController() with self.assertRaises(SystemExit) as cm: StorageWatcher.run(mock_storage_watcher) self.assertEqual(cm.exception.args[0], "Invalid test case") - @patch("systemd.journal.Reader") - @patch("select.poll") - @patch("os.environ.get") - @patch("checkbox_support.scripts.run_watcher.zapper_run") - def test_storage_watcher_run_with_insertion_with_zapper( - self, mock_zapper_run, mock_get, mock_poll, mock_journal - ): - mock_journal.return_value.process.return_value = journal.APPEND - mock_journal.return_value.__iter__.return_value = [ - {"MESSAGE": "line1"} - ] - mock_poll.return_value.poll.side_effect = [True, False] - mock_get.return_value = "zapper_addr" - - mock_storage_watcher = MagicMock() - mock_storage_watcher.zapper_usb_address = "usb_address" - - # Test insertion with zapper - mock_storage_watcher.testcase = "insertion" - StorageWatcher.run(mock_storage_watcher) - mock_zapper_run.assert_called_with( - "zapper_addr", "typecmux_set_state", "usb_address", "DUT" - ) - mock_storage_watcher._process_lines.assert_called_with(["line1"]) - - @patch("systemd.journal.Reader") - @patch("select.poll") - @patch("os.environ.get") - @patch("checkbox_support.scripts.run_watcher.zapper_run") - def test_storage_watcher_run_with_removal_with_zapper( - self, mock_zapper_run, mock_get, mock_poll, mock_journal - ): - mock_journal.return_value.process.return_value = journal.APPEND - mock_journal.return_value.__iter__.return_value = [ - {"MESSAGE": "line1"} - ] - mock_poll.return_value.poll.side_effect = [True, False] - mock_get.return_value = "zapper_addr" - - mock_storage_watcher = MagicMock() - mock_storage_watcher.zapper_usb_address = "usb_address" - - # Test removal with zapper - mock_storage_watcher.testcase = "removal" - StorageWatcher.run(mock_storage_watcher) - mock_zapper_run.assert_called_with( - "zapper_addr", "typecmux_set_state", "usb_address", "OFF" - ) - mock_storage_watcher._process_lines.assert_called_with(["line1"]) - @patch("systemd.journal.Reader") @patch("select.poll") def test_storage_watcher_run_not_passed(self, mock_poll, mock_journal): @@ -155,7 +106,6 @@ def test_storage_watcher_run_not_passed(self, mock_poll, mock_journal): mock_poll.return_value.poll.side_effect = [True, False] mock_storage_watcher = MagicMock() - mock_storage_watcher.zapper_usb_address = "" mock_storage_watcher.test_passed = False # Test not passed @@ -241,9 +191,8 @@ def test_storage_watcher_run_storage( mock_read_test.assert_called_with("file") def test_usb_storage_init(self): - usb_storage = USBStorage("usb2", "zapper_addr") + usb_storage = USBStorage("usb2") self.assertEqual(usb_storage.storage_type, "usb2") - self.assertEqual(usb_storage.zapper_usb_address, "zapper_addr") self.assertIsNone(usb_storage.mounted_partition) self.assertIsNone(usb_storage.device) self.assertIsNone(usb_storage.number) @@ -367,9 +316,8 @@ def test_usb_storage_parse_journal_line(self): self.assertEqual(mock_usb_storage.action, None) def test_mediacard_storage_init(self): - mediacard_storage = MediacardStorage("mediacard", "zapper_addr") + mediacard_storage = MediacardStorage("mediacard") self.assertEqual(mediacard_storage.storage_type, "mediacard") - self.assertEqual(mediacard_storage.zapper_usb_address, "zapper_addr") self.assertIsNone(mediacard_storage.mounted_partition) def test_mediacard_storage_validate_insertion(self): @@ -425,13 +373,8 @@ def test_mediacard_storage_parse_journal_line(self): self.assertEqual(mock_mediacard_storage.action, None) def test_mediacard_combo_storage_init(self): - mediacard_combo_storage = MediacardComboStorage( - "mediacard", "zapper_addr" - ) + mediacard_combo_storage = MediacardComboStorage("mediacard") self.assertEqual(mediacard_combo_storage.storage_type, "mediacard") - self.assertEqual( - mediacard_combo_storage.zapper_usb_address, "zapper_addr" - ) self.assertIsNone(mediacard_combo_storage.mounted_partition) def test_mediacard_combo_storage_validate_insertion(self): @@ -510,9 +453,8 @@ def test_mediacard_combo_storage_parse_journal_line(self): self.assertEqual(mock_mediacard_combo_storage.action, None) def test_thunderbolt_storage_init(self): - thunderbolt_storage = ThunderboltStorage("thunderbolt", "zapper_addr") + thunderbolt_storage = ThunderboltStorage("thunderbolt") self.assertEqual(thunderbolt_storage.storage_type, "thunderbolt") - self.assertEqual(thunderbolt_storage.zapper_usb_address, "zapper_addr") self.assertIsNone(thunderbolt_storage.mounted_partition) self.assertIsNone(thunderbolt_storage.action) @@ -577,21 +519,20 @@ def test_thunderbolt_storage_parse_journal_line(self): def test_parse_args(self): with patch( "sys.argv", - ["script.py", "insertion", "usb2", "--zapper-usb-address", "addr"], + ["script.py", "insertion", "usb2"], ): args = parse_args() self.assertEqual(args.testcase, "insertion") self.assertEqual(args.storage_type, "usb2") - self.assertEqual(args.zapper_usb_address, "addr") @patch("checkbox_support.scripts.run_watcher.USBStorage", spec=USBStorage) @patch("checkbox_support.scripts.run_watcher.parse_args") def test_main_usb_insertion(self, mock_parse_args, mock_usb): mock_parse_args.return_value = argparse.Namespace( - testcase="insertion", storage_type="usb2", zapper_usb_address=None + testcase="insertion", storage_type="usb2" ) main() - mock_usb.assert_called_with("usb2", None) + mock_usb.assert_called_with("usb2") self.assertEqual(mock_usb.return_value.run_insertion.call_count, 1) self.assertEqual(mock_usb.return_value.run_removal.call_count, 1) # get the watcher object from main @@ -604,10 +545,10 @@ def test_main_usb_insertion(self, mock_parse_args, mock_usb): @patch("checkbox_support.scripts.run_watcher.parse_args") def test_main_usb_storage(self, mock_parse_args, mock_usb): mock_parse_args.return_value = argparse.Namespace( - testcase="storage", storage_type="usb2", zapper_usb_address=None + testcase="storage", storage_type="usb2" ) main() - mock_usb.assert_called_with("usb2", None) + mock_usb.assert_called_with("usb2") self.assertEqual(mock_usb.return_value.run_insertion.call_count, 1) self.assertEqual(mock_usb.return_value.run_storage.call_count, 1) self.assertEqual(mock_usb.return_value.run_removal.call_count, 1) @@ -616,7 +557,7 @@ def test_main_usb_storage(self, mock_parse_args, mock_usb): @patch("checkbox_support.scripts.run_watcher.parse_args") def test_main_usb_invalid(self, mock_parse_args, mock_usb): mock_parse_args.return_value = argparse.Namespace( - testcase="invalid", storage_type="usb2", zapper_usb_address=None + testcase="invalid", storage_type="usb2" ) with self.assertRaises(SystemExit) as cm: main() @@ -631,7 +572,6 @@ def test_main_mediacard(self, mock_parse_args, mock_mediacard): mock_parse_args.return_value = argparse.Namespace( testcase="insertion", storage_type="mediacard", - zapper_usb_address=None, ) main() self.assertEqual(mock_mediacard.call_count, 1) @@ -649,7 +589,6 @@ def test_main_mediacard_combo(self, mock_parse_args, mock_mediacard): mock_parse_args.return_value = argparse.Namespace( testcase="insertion", storage_type="mediacard_combo", - zapper_usb_address=None, ) main() self.assertEqual(mock_mediacard.call_count, 1) @@ -667,7 +606,6 @@ def test_main_thunderbolt(self, mock_parse_args, mock_thunderbolt): mock_parse_args.return_value = argparse.Namespace( testcase="insertion", storage_type="thunderbolt", - zapper_usb_address=None, ) main() self.assertEqual(mock_thunderbolt.call_count, 1) diff --git a/checkbox-support/pyproject.toml b/checkbox-support/pyproject.toml index 1e167ca827..e416a448c3 100644 --- a/checkbox-support/pyproject.toml +++ b/checkbox-support/pyproject.toml @@ -41,7 +41,6 @@ checkbox-support-eddystone_scanner = "checkbox_support.scripts.eddystone_scanner:main" checkbox-support-lsusb = "checkbox_support.scripts.lsusb:main" checkbox-support-parse = "checkbox_support.parsers:main" - checkbox-support-zapper-proxy = "checkbox_support.scripts.zapper_proxy:main" checkbox-support-image_checker = "checkbox_support.scripts.image_checker:main" [project.entry-points."plainbox.parsers"] pactl-list="checkbox_support.parsers.pactl:parse_pactl_output" diff --git a/checkbox-support/setup.cfg b/checkbox-support/setup.cfg index c8907ea385..dd1873ea07 100644 --- a/checkbox-support/setup.cfg +++ b/checkbox-support/setup.cfg @@ -37,4 +37,3 @@ console_scripts= checkbox-support-eddystone_scanner=checkbox_support.scripts.eddystone_scanner:main checkbox-support-lsusb=checkbox_support.scripts.lsusb:main checkbox-support-parse=checkbox_support.parsers:main - checkbox-support-zapper-proxy=checkbox_support.scripts.zapper_proxy:main diff --git a/docs/tutorial/using-checkbox/advanced-commands.rst b/docs/tutorial/using-checkbox/advanced-commands.rst index 84953834ff..2091335073 100644 --- a/docs/tutorial/using-checkbox/advanced-commands.rst +++ b/docs/tutorial/using-checkbox/advanced-commands.rst @@ -135,7 +135,6 @@ To create a table listing each job id and their summary, run: acpi_sleep_attachment | (...) xinput | Creates resource info from xinput output. - zapper_capabilities | Get Zapper's setup capabilities collect-manifest | Collect the hardware manifest (interactively) manifest | Hardware Manifest diff --git a/providers/base/README.rst b/providers/base/README.rst index 68383d7ea3..68b2ee8105 100644 --- a/providers/base/README.rst +++ b/providers/base/README.rst @@ -25,7 +25,7 @@ Base Provider Units +------------+-------------+-------------+------------------+-------------+----------------+ | camera_ | firewire | kernel-snap | oob-management | submission | wwan | +------------+-------------+-------------+------------------+-------------+----------------+ -| canary | firmware | keys | optical | suspend | zapper | +| canary | firmware | keys | optical | suspend | | +------------+-------------+-------------+------------------+-------------+----------------+ | codecs | gadget | led | power-management | thunderbolt | | +------------+-------------+-------------+------------------+-------------+----------------+ diff --git a/providers/base/bin/bt_a2dp.py b/providers/base/bin/bt_a2dp.py deleted file mode 100755 index 981bae91a7..0000000000 --- a/providers/base/bin/bt_a2dp.py +++ /dev/null @@ -1,78 +0,0 @@ -#!/usr/bin/env python3 -# Copyright 2023 Canonical Ltd. -# All rights reserved. -# -# Written by: -# Authors: Maciej Kisielewski -""" -This program checks if the host can connect to a Zapper using Bluetooth. -""" -import sys -import time - -from contextlib import contextmanager - -import bluetooth - -from checkbox_support.scripts.zapper_proxy import zapper_run # noqa: E402 - - -@contextmanager -def zapper_as_a_speaker(host): - """Ensure that the service is turned off after using it.""" - addr = zapper_run(host, "bluetooth_start", "a2dp") - try: - yield addr - finally: - zapper_run(host, "bluetooth_stop") - - -def main(): - """Entry point to the test.""" - if len(sys.argv) != 2: - raise SystemExit("Usage: {} ZAPPER_HOST".format(sys.argv[0])) - print("Asking Zapper to become a Bluetooth speaker") - with zapper_as_a_speaker(sys.argv[1]) as zapper_addr: - print("Zapper Bluetooth address is {}".format(zapper_addr)) - retry_count = 5 - for i in range(1, retry_count + 1): - print( - "Discovering Bluetooth devices (try {}/{})".format( - i, retry_count - ) - ) - devices = bluetooth.discover_devices() - print("Devices found: {}".format(devices)) - if zapper_addr in devices: - break - print("Zapper not found in the device list... Retrying") - else: - raise SystemExit("Zapper not found") - - for i in range(1, retry_count + 1): - print( - "Trying to connect to Zapper (try {}/{})".format( - i, retry_count - ) - ) - socket = bluetooth.BluetoothSocket(bluetooth.L2CAP) - try: - socket.connect((zapper_addr, 25)) # 25 - A2DP port - # the sleep below is there to be able to see the icon change - # when observing the test with human eyes - # change it to a longer delay to enable easy debugging - time.sleep(1) - socket.close() - break - except bluetooth.btcommon.BluetoothError as exc: - print("Failed to connect. {}".format(exc)) - # the sleep below lets the Bluetooth stack react to the - # changing environment. Without it the retries may happen - # too quickly - time.sleep(10) - else: - raise SystemExit("Failed to connect to Zapper via BT") - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/providers/base/bin/edid_cycle.py b/providers/base/bin/edid_cycle.py deleted file mode 100755 index a251396839..0000000000 --- a/providers/base/bin/edid_cycle.py +++ /dev/null @@ -1,212 +0,0 @@ -#!/usr/bin/env python3 -""" -This program tests whether the system changes the resolution automatically -when supplied with a new EDID information. - -To run the test you need Zapper board connected and set up. -""" - -import argparse -import pathlib -import os -import time -from contextlib import contextmanager - -from checkbox_support.monitor_config import ( - MonitorConfig, -) # noqa: E402 -from checkbox_support.helpers import display_info # noqa: E402 -from checkbox_support.scripts.zapper_proxy import zapper_run # noqa: E402 - -EDID_FILES = list( - pathlib.Path(os.getenv("PLAINBOX_PROVIDER_DATA", ".")).glob("edids/*.edid") -) - - -@contextmanager -def zapper_monitor(zapper_host: str): - """Unplug the Zapper monitor at the end of the test.""" - try: - yield - finally: - _clear_edid(zapper_host) - - -def discover_video_output_device( - zapper_host: str, monitor_config: MonitorConfig -) -> str: - """ - Try to discover the output device connected to Zapper - checking the difference in randr when a monitor gets - plugged in. - - :param zapper_host: Target Zapper IP - :return: Video Output Port, i.e. `HDMI-1` - :raises IOError: cannot discover the video port under test - """ - - _clear_edid(zapper_host) - - # Not knowing the target I can't use a busy loop here - # and I'm waiting for the DUT to react to the EDID change. - time.sleep(5) - - devices = monitor_config.get_connected_monitors() - - # It doesn't really matter which EDID file we set in this function: - # we just want to recognize the port type and index, not the resolution. - _set_edid(zapper_host, EDID_FILES[0]) - - # Wait until the list of active devices changes. - # From manual testing, 5 seconds seem more than - # enough for such changes to happen. - targets = [] - for _ in range(5): - targets = monitor_config.get_connected_monitors() - devices - if targets: - break - - time.sleep(1) - - if len(targets) != 1: - raise IOError( - "Can't identify the video port under test, " - "got {} new devices.".format(len(targets)) - ) - - return next(iter(targets)) - - -def test_edid( - zapper_host: str, - monitor_config: MonitorConfig, - edid_file: pathlib.Path, - video_device: str, -): - """ - Set a EDID file and check whether the resolution - is recognized and selected on DUT. - - :param zapper_host: Target Zapper IP - :param edid_file: path to the EDID file to test - :param video_device: video output port under test - - :raises AssertionError: in case of mismatch between set - and read resolution - """ - resolution = edid_file.stem - print("switching EDID to {}".format(resolution)) - - try: - _switch_edid(zapper_host, monitor_config, edid_file, video_device) - configuration = monitor_config.set_extended_mode() - except TimeoutError as exc: - raise AssertionError("Timed out switching EDID") from exc - - # A mismatch between requested and applied resolution might - # indicated an incompatibility at HW level, most of the time - # due to HDMI < 1.3. - applied_res = configuration[video_device] - actual_res = monitor_config.get_current_resolutions()[video_device] - - if applied_res != resolution: - print("SKIP, max available was {}".format(applied_res)) - return - - if actual_res != resolution: - raise AssertionError( - "FAIL, got {} but {} expected".format(actual_res, resolution) - ) - - print("PASS") - - -def _switch_edid(zapper_host, monitor_config, edid_file, video_device): - """Clear EDID and then 'plug' back a new monitor.""" - - _clear_edid(zapper_host) - _wait_edid_change(monitor_config, video_device, False) - - _set_edid(zapper_host, edid_file) - _wait_edid_change(monitor_config, video_device, True) - - -def _set_edid(zapper_host, edid_file): - """Request EDID change to Zapper.""" - with open(str(edid_file), "rb") as f: - zapper_run(zapper_host, "change_edid", f.read()) - - -def _clear_edid(zapper_host): - """Request EDID clear to Zapper.""" - zapper_run(zapper_host, "change_edid", None) - - -def _check_connected(monitor_config, device): - """Check if the video input device is recognized and active.""" - - return device in monitor_config.get_current_resolutions().keys() - - -def _wait_edid_change(monitor_config, video_device, expected): - """ - Wait until `expected` connection state is reached. - Times out after 5 seconds. - """ - iteration = 0 - max_iter = 5 - sleep = 1 - while ( - _check_connected(monitor_config, video_device) != expected - and iteration < max_iter - ): - time.sleep(sleep) - iteration += 1 - - if iteration == max_iter: - raise TimeoutError( - "Reacting to the EDID change took more than {}s.".format( - max_iter * sleep - ) - ) - - -def main(args=None): - """ - Test for different EDID files whether the resolution - is recognized and selected on DUT. - """ - parser = argparse.ArgumentParser() - parser.add_argument("host", help="Zapper IP address") - args = parser.parse_args(args) - - try: - monitor_config = display_info.get_monitor_config() - print("Running with `{}`.".format(monitor_config.__class__.__name__)) - except ValueError as exc: - raise SystemExit("Current host is not supported.") from exc - - with zapper_monitor(args.host): - try: - video_device = discover_video_output_device( - args.host, monitor_config - ) - print("Testing EDID cycling on {}".format(video_device)) - except IOError as exc: - raise SystemExit( - "Cannot detect the target video output device." - ) from exc - - failed = False - for edid_file in EDID_FILES: - try: - test_edid(args.host, monitor_config, edid_file, video_device) - except AssertionError as exc: - print(exc.args[0]) - failed = True - - return failed - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/providers/base/bin/removable_storage_watcher.py b/providers/base/bin/removable_storage_watcher.py index d10999c598..66b9932a67 100755 --- a/providers/base/bin/removable_storage_watcher.py +++ b/providers/base/bin/removable_storage_watcher.py @@ -5,9 +5,7 @@ import copy import dbus import logging -import os import sys -import threading import gi @@ -26,7 +24,6 @@ from checkbox_support.parsers.udevadm import CARD_READER_RE # noqa: E402 from checkbox_support.parsers.udevadm import GENERIC_RE # noqa: E402 from checkbox_support.parsers.udevadm import FLASH_RE # noqa: E402 -from checkbox_support.scripts.zapper_proxy import zapper_run # noqa: E402 from checkbox_support.udev import get_interconnect_speed # noqa: E402 from checkbox_support.udev import get_udev_block_devices # noqa: E402 @@ -1004,7 +1001,7 @@ def _summarize_changes(self, delta_records): ) -def main(): +def main(argv=sys.argv[1:]): description = "Wait for the specified device to be inserted or removed." parser = argparse.ArgumentParser(description=description) parser.add_argument("action", choices=["insert", "remove"]) @@ -1048,13 +1045,8 @@ def main(): action="store_true", help="Don't require drive being automounted", ) - parser.add_argument( - "--zapper-usb-address", - type=str, - help="Zapper's USB switch address to use", - ) parser.set_defaults(logging_level=logging.WARNING) - args = parser.parse_args() + args = parser.parse_args(argv) # Configure logging as requested # XXX: This may be incorrect as logging.basicConfig() fails after any other @@ -1097,42 +1089,11 @@ def main(): ) # Run the actual listener and wait till it either times out of discovers # the appropriate media changes - if args.zapper_usb_address: - zapper_host = os.environ.get("ZAPPER_HOST") - if not zapper_host: - raise SystemExit("ZAPPER_HOST environment variable not found!") - usb_address = args.zapper_usb_address - delay = 5 # in seconds - - def do_the_insert(): - logging.info("Calling zapper to connect the USB device") - zapper_run(zapper_host, "typecmux_set_state", usb_address, "DUT") - - insert_timer = threading.Timer(delay, do_the_insert) - - def do_the_remove(): - logging.info("Calling zapper to disconnect the USB device") - zapper_run(zapper_host, "typecmux_set_state", usb_address, "OFF") - - remove_timer = threading.Timer(delay, do_the_remove) - if args.action == "insert": - logging.info("Starting timer for delayed insertion") - insert_timer.start() - elif args.action == "remove": - logging.info("Starting timer for delayed removal") - remove_timer.start() - try: - res = listener.check(args.timeout) - return res - except KeyboardInterrupt: - return 1 - - else: - print("\n\n{} NOW\n\n".format(args.action.upper()), flush=True) - try: - return listener.check(args.timeout) - except KeyboardInterrupt: - return 1 + print("\n\n{} NOW\n\n".format(args.action.upper()), flush=True) + try: + return listener.check(args.timeout) + except KeyboardInterrupt: + return 1 if __name__ == "__main__": diff --git a/providers/base/bin/zapper_keyboard_test.py b/providers/base/bin/zapper_keyboard_test.py deleted file mode 100755 index a9290762c9..0000000000 --- a/providers/base/bin/zapper_keyboard_test.py +++ /dev/null @@ -1,180 +0,0 @@ -#!/usr/bin/env python3 -""" -This program tests whether the system responds correctly to an external -USB keyboard. - -To run the test you need Zapper board connected and set up. -""" -import os -import struct -import sys -import threading -import time -from enum import Enum -from pathlib import Path - -from checkbox_support.scripts.zapper_proxy import zapper_run # noqa: E402 - -ROBOT_INIT = """ -*** Settings *** -Library libraries/ZapperHid.py - -*** Test Cases *** -Do nothing - Log Re-configure HID device -""" - -ROBOT_TESTCASE_COMBO = """ -*** Settings *** -Library libraries/ZapperHid.py - -*** Test Cases *** -Press and Release a combination of keys. - ${keys}= Create List A B - Keys Combo ${keys} -""" - -ROBOT_TESTCASE_TYPE = """ -*** Settings *** -Library libraries/ZapperHid.py - -*** Test Cases *** -Type the requested string - Type String hello -""" - - -class KeyEvent(Enum): - """Key events of interest.""" - - UP = 0 - DOWN = 1 - - -class KeyboardListener(threading.Thread): - """Listen for keyboard events.""" - - EVENT_BIN_FORMAT = "llHHI" # expected data layout - EVENT_BIN_SIZE = struct.calcsize(EVENT_BIN_FORMAT) - - def __init__(self, event_file, callback): - super().__init__() - self._keep_running = True - self._event_no = os.open(event_file, os.O_NONBLOCK | os.O_RDONLY) - self._callback = callback - - def run(self): - """Start polling keyboard events.""" - while self._keep_running: - self._read_keyboard_events() - - def stop(self): - """Stop loop and close the file.""" - self._keep_running = False - os.close(self._event_no) - - def _read_keyboard_events(self): - """Read keyboard events from the given file and run the callback.""" - - try: - data = os.read(self._event_no, self.EVENT_BIN_SIZE) - except BlockingIOError: - return - - _, _, event_type, code, value = struct.unpack( - self.EVENT_BIN_FORMAT, data - ) - if event_type == 1: # 0x01 is for _kbd_ events - self._callback((KeyEvent(value), code)) - - -def assert_key_combo(host, events): - """ - Request to press a key combination and expect the corresponding events. - """ - - events.clear() - zapper_run(host, "robot_run", ROBOT_TESTCASE_COMBO.encode(), {}, {}) - - assert events == [ - (KeyEvent.DOWN, 30), - (KeyEvent.DOWN, 48), - (KeyEvent.UP, 48), - (KeyEvent.UP, 30), - ] - - -def assert_type_string(host, events): - """Request to type a string and expect the corresponding events.""" - - events.clear() - zapper_run(host, "robot_run", ROBOT_TESTCASE_TYPE.encode(), {}, {}) - - assert events == [ - (KeyEvent.DOWN, 35), - (KeyEvent.UP, 35), - (KeyEvent.DOWN, 18), - (KeyEvent.UP, 18), - (KeyEvent.DOWN, 38), - (KeyEvent.UP, 38), - (KeyEvent.DOWN, 38), - (KeyEvent.UP, 38), - (KeyEvent.DOWN, 24), - (KeyEvent.UP, 24), - ] - - -def get_zapper_kbd_device(): - """ - Get Zapper keyboard device by ID. - - Being a composite device, interface number might change depending on - Zapper HID configuration. - """ - zapper_kbd = "usb-Canonical_Zapper_main_board_123456*-event-kbd" - - start = time.time() - for _ in range(5): - for file_path in Path("/dev/input/by-id/").glob(zapper_kbd): - print(time.time() - start) - return str(file_path) - time.sleep(1) - raise FileNotFoundError("Cannot find Zapper Keyboard.") - - -def main(argv): - """ - Request Zapper to type on keyboard and assert the received events - are like expected. - """ - - if len(argv) != 2: - raise SystemExit("Usage: {} ".format(argv[0])) - - # A simple robot-run to initialize the Zapper HID device - zapper_run(argv[1], "robot_run", ROBOT_INIT.encode(), {}, {}) - - try: - zapper_kbd = get_zapper_kbd_device() - except FileNotFoundError as exc: - raise SystemExit("Cannot find Zapper Keyboard.") from exc - - if not os.access(zapper_kbd, os.R_OK): - raise SystemExit("Cannot read from Zapper Keyboard.") - - events = [] - listener = KeyboardListener(zapper_kbd, events.append) - listener.start() - - try: - assert_key_combo(argv[1], events) - assert_type_string(argv[1], events) - except AssertionError as exc: - raise SystemExit("Mismatch in received keyboard events.") from exc - finally: - listener.stop() - listener.join() - - -if __name__ == "__main__": - raise SystemExit(main(sys.argv)) diff --git a/providers/base/data/edids/1280x1024.edid b/providers/base/data/edids/1280x1024.edid deleted file mode 100644 index 057bfecf696479ccce4ba6748644c7a1a004bfb7..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 256 zcmZSh4+abZYs3r!1Dzb1_!*fSl%y-(y_>o?#Al`&e?Z7mZ-zqu#s)?tV4%e?BalJM z!Jt5xA&5_qfkBRef#E-cuc?u#si~ofqf=lY7f|F6gNv(=kAk7Ofr(o{7)azVgN2V= RARhyl0vIreL98D*5&$4ao`C=W diff --git a/providers/base/data/edids/1920x1080.edid b/providers/base/data/edids/1920x1080.edid deleted file mode 100644 index 567437de4df91a46db28737b1a984ec151affd18..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 256 zcmZSh4+abZYZ@Gaoh=;|Bp6wi+9*|6d2d}F=Dk>zDIlh5joY8XFja zAS{ky!GZ;v0R{@D450yv3=C2X3=IDn%)R^~O-(E;94(!Lxqu>n7+hR^d=x^B%ngk_ zd_f|A84SZ@3*{KO6u^LSArrIg$AAf}EX+*oVvKBpLZZSV@(RkF_3X`z3=B+G4HAVG zx(*RKst{}B7-fMXGGYpFk$H@=3`Kz#N{7BKOkhOH^XLziwuSSD;*je${QGg zAS{ky!GZ;v0R{@D450yv3=C2X3=IDn+&wHzUCf>R3_SvZxqu>n7+hR^d=x^B%neOF zxD#Ev7Cun>7%~@D+@CdyBMR0ypXVhqLMNvJ9{%D14Ak^4}$^& zlU0L6p@ptPgpMl2E;&Y7pool^0$gMsqbx&F-~~B_JdIGOnjWqTT?!9$1qAG-K|~}x U>={lqG#EHY1qegc0Nv99044K10RR91 diff --git a/providers/base/tests/test_edid_cycle.py b/providers/base/tests/test_edid_cycle.py deleted file mode 100644 index dcd289ebf5..0000000000 --- a/providers/base/tests/test_edid_cycle.py +++ /dev/null @@ -1,232 +0,0 @@ -"""This module provides test cases for the edid_cycle module.""" - -import sys -import unittest -from pathlib import Path -from unittest.mock import patch, call, Mock, MagicMock - -sys.modules["dbus"] = MagicMock() -sys.modules["dbus.mainloop.glib"] = MagicMock() -sys.modules["gi"] = MagicMock() -sys.modules["gi.repository"] = MagicMock() - -import edid_cycle - - -class ZapperEdidCycleTests(unittest.TestCase): - """This class provides test cases for the edid_cycle module.""" - - @patch("edid_cycle.zapper_run", new=Mock) - @patch("time.sleep", new=Mock) - @patch("builtins.open") - def test_discover_video_output_device(self, mock_open): - """ - Check if the function automatically discover the port - under test hot-plugging the Zapper monitor. - """ - edid_cycle.EDID_FILES = [Path("1920x1080.edid")] - - mock_monitor = Mock() - mock_monitor.get_connected_monitors.side_effect = [ - set(), - {"HDMI-1"}, - ] - - port = edid_cycle.discover_video_output_device( - "zapper-ip", mock_monitor - ) - self.assertEqual(port, "HDMI-1") - - @patch("time.sleep", new=Mock) - @patch("edid_cycle.zapper_run", new=Mock) - @patch("builtins.open") - def test_discover_video_output_device_error(self, mock_open): - """ - Check if the function raises an exception when - if fails to discover the video port. - """ - edid_cycle.EDID_FILES = [Path("1920x1080.edid")] - - mock_monitor = Mock() - mock_monitor.get_connected_monitors.return_value = {"HDMI-1"} - - with self.assertRaises(IOError): - edid_cycle.discover_video_output_device("zapper-ip", mock_monitor) - - @patch("time.sleep", new=Mock) - @patch("edid_cycle.zapper_run", new=Mock) - @patch("builtins.open") - def test_test_edid(self, mock_open): - """ - Check the function set an EDID and assert the actual - resolution matches the request. - """ - - mock_monitor = Mock() - mock_monitor.get_current_resolutions.side_effect = [ - { - "eDP-1": "1280x1024", - }, - { - "eDP-1": "1280x1024", - "HDMI-1": "1280x1024", # when connected it's in mirror mode - }, - { - "eDP-1": "1280x1024", - "HDMI-1": "1920x1080", # and then we set extended mode - }, - ] - mock_monitor.set_extended_mode.return_value = { - "eDP-1": "1280x1024", - "HDMI-1": "1920x1080", - } - - edid_cycle.test_edid( - "zapper-ip", mock_monitor, Path("1920x1080.edid"), "HDMI-1" - ) - mock_open.assert_called_with("1920x1080.edid", "rb") - mock_monitor.set_extended_mode.assert_called_once_with() - - @patch("time.sleep", new=Mock) - @patch("edid_cycle.zapper_run", new=Mock) - @patch("builtins.open") - def test_test_edid_skip(self, mock_open): - """ - Check the function returns w/o errors if - the resolution doesn't match because of - HW incompatibility. - """ - - mock_monitor = Mock() - mock_monitor.get_current_resolutions.side_effect = [ - { - "eDP-1": "1280x1024", - }, - { - "eDP-1": "1280x1024", - "HDMI-1": "1280x1024", # when connected it's in mirror mode - }, - { - "eDP-1": "1280x1024", - "HDMI-1": "2048x1440", # and then we set extended mode - }, - ] - mock_monitor.set_extended_mode.return_value = { - "eDP-1": "1280x1024", - "HDMI-1": "2048x1440", - } - - edid_cycle.test_edid( - "zapper-ip", mock_monitor, Path("2560x1440.edid"), "HDMI-1" - ) - mock_open.assert_called_with("2560x1440.edid", "rb") - mock_monitor.set_extended_mode.assert_called_once_with() - - @patch("time.sleep", new=Mock) - @patch("edid_cycle.zapper_run", new=Mock) - @patch("builtins.open") - def test_test_edid_error(self, mock_open): - """ - Check the function raises an exception when the assertion - on resolution fails. - """ - mock_monitor = Mock() - mock_monitor.get_current_resolutions.side_effect = [ - { - "eDP-1": "1280x1024", - }, - { - "eDP-1": "1280x1024", - "HDMI-1": "1280x1024", - }, - { - "eDP-1": "1280x1024", - "HDMI-1": "1280x1024", # still not at requested resolution - }, - ] - mock_monitor.set_extended_mode.return_value = { - "eDP-1": "1280x1024", - "HDMI-1": "1920x1080", - } - - with self.assertRaises(AssertionError): - edid_cycle.test_edid( - "zapper-ip", mock_monitor, Path("1920x1080.edid"), "HDMI-1" - ) - - @patch("edid_cycle.zapper_monitor", MagicMock()) - @patch("edid_cycle.display_info", Mock()) - @patch("edid_cycle.discover_video_output_device") - def test_main_no_device(self, mock_discover): - """Test if main function exits when no device is detected.""" - args = ["zapper-ip"] - mock_discover.side_effect = IOError - - with self.assertRaises(SystemExit): - edid_cycle.main(args) - - @patch("edid_cycle.display_info") - def test_main_no_monitor_config(self, mock_display_info): - """Test if main function exits when no monitor config is available.""" - mock_display_info.get_monitor_config.side_effect = ValueError - with self.assertRaises(SystemExit): - edid_cycle.main([]) - - @patch("edid_cycle.display_info") - @patch("edid_cycle.zapper_monitor") - @patch("edid_cycle.test_edid") - @patch("edid_cycle.discover_video_output_device") - def test_main( - self, mock_discover, mock_test_edid, mock_monitor, mock_display_info - ): - """ - Test if main function run the EDID test for every available EDID file. - """ - args = ["zapper-ip"] - edid_cycle.EDID_FILES = [ - Path("file1"), - Path("file2"), - Path("file3"), - ] - monitor_config = mock_display_info.get_monitor_config.return_value - - self.assertFalse(edid_cycle.main(args)) - mock_test_edid.assert_has_calls( - [ - call( - "zapper-ip", - monitor_config, - Path("file1"), - mock_discover.return_value, - ), - call( - "zapper-ip", - monitor_config, - Path("file2"), - mock_discover.return_value, - ), - call( - "zapper-ip", - monitor_config, - Path("file3"), - mock_discover.return_value, - ), - ] - ) - - mock_test_edid.side_effect = AssertionError("Mismatch") - self.assertTrue(edid_cycle.main(args)) - - mock_monitor.assert_called_with("zapper-ip") - - @patch("edid_cycle._clear_edid") - def test_zapper_monitor(self, mock_clear): - """ - Test whether this context manager unplugs the Zapper monitor - at exit. - """ - - with edid_cycle.zapper_monitor("zapper-ip"): - pass - - mock_clear.assert_called_with("zapper-ip") diff --git a/providers/base/tests/test_removable_storage_watcher.py b/providers/base/tests/test_removable_storage_watcher.py new file mode 100644 index 0000000000..73c613b32a --- /dev/null +++ b/providers/base/tests/test_removable_storage_watcher.py @@ -0,0 +1,87 @@ +import unittest +import sys +from unittest.mock import patch, MagicMock + +sys.modules["dbus"] = MagicMock() +sys.modules["dbus.exceptions"] = MagicMock() +sys.modules["dbus.mainloop.glib"] = MagicMock() +sys.modules["gi"] = MagicMock() +sys.modules["gi.repository"] = MagicMock() + +import removable_storage_watcher + + +class RemovableStorageWatcherTests(unittest.TestCase): + + @patch("removable_storage_watcher.connect_to_system_bus") + @patch("removable_storage_watcher.is_udisks2_supported") + @patch("removable_storage_watcher.UDisks1StorageDeviceListener") + def test_main_udisk1(self, mock_udisk, mock_check, mock_connect): + """ + Test the main function connects to the system bus, + opens a UDisks1 listener and check the action is performed + until timeout. + """ + + mock_connect.return_value = ("bus", "loop") + mock_check.return_value = False + + argv = ["insert", "usb"] + value = removable_storage_watcher.main(argv) + + mock_connect.assert_called_once_with() + mock_udisk.assert_called_once_with( + "bus", + "loop", + "insert", + ["usb"], + 0, + False, + ) + + mock_udisk.return_value.check.assert_called_once_with(20) + + self.assertEqual(value, mock_udisk.return_value.check.return_value) + + @patch("removable_storage_watcher.connect_to_system_bus") + @patch("removable_storage_watcher.is_udisks2_supported") + @patch("removable_storage_watcher.UDisks2StorageDeviceListener") + def test_main_udisk2(self, mock_udisk, mock_check, mock_connect): + """ + Test the main function uses a UDisks2 listener when + supported. + """ + + mock_connect.return_value = ("bus", "loop") + mock_check.return_value = True + + argv = ["insert", "usb"] + removable_storage_watcher.main(argv) + + mock_udisk.assert_called_once_with( + "bus", + "loop", + "insert", + ["usb"], + 0, + False, + False, + ) + + @patch("removable_storage_watcher.connect_to_system_bus") + @patch("removable_storage_watcher.is_udisks2_supported") + @patch("removable_storage_watcher.UDisks1StorageDeviceListener") + def test_main_interrupted(self, mock_udisk, mock_check, mock_connect): + """ + Test the main function returns 1 when interrupted + via CTRL-C + """ + + mock_connect.return_value = ("bus", "loop") + mock_check.return_value = False + + argv = ["insert", "usb"] + mock_udisk.return_value.check.side_effect = KeyboardInterrupt + + value = removable_storage_watcher.main(argv) + self.assertEqual(value, 1) diff --git a/providers/base/tests/test_zapper_keyboard.py b/providers/base/tests/test_zapper_keyboard.py deleted file mode 100644 index 65a2c1bdba..0000000000 --- a/providers/base/tests/test_zapper_keyboard.py +++ /dev/null @@ -1,296 +0,0 @@ -"""This module provides test cases for the zapper_keyboard_test module.""" - -import os -import struct -import threading -import unittest -from pathlib import Path -from unittest.mock import patch, Mock - -import zapper_keyboard_test - - -class ZapperKeyboardTests(unittest.TestCase): - """This class provides test cases for the zapper_keyboard_test module.""" - - @patch("zapper_keyboard_test.zapper_run") - def test_assert_key_combo(self, mock_run): - """ - Check if the function properly clear the events list and - run the right zapper_run method for key combos. - """ - events = [1, 2, 3] - host = "127.0.0.1" - - with self.assertRaises(AssertionError): - zapper_keyboard_test.assert_key_combo(host, events) - - assert events == [] - mock_run.assert_called_with( - host, - "robot_run", - zapper_keyboard_test.ROBOT_TESTCASE_COMBO.encode(), - {}, - {}, - ) - - @patch("zapper_keyboard_test.zapper_run") - def test_assert_type_string(self, mock_run): - """ - Check if the function properly clear the events list and - run the right zapper_run method for string typing. - """ - events = [1, 2, 3] - host = "127.0.0.1" - - with self.assertRaises(AssertionError): - zapper_keyboard_test.assert_type_string(host, events) - - assert events == [] - mock_run.assert_called_with( - host, - "robot_run", - zapper_keyboard_test.ROBOT_TESTCASE_TYPE.encode(), - {}, - {}, - ) - - def test_main_no_args(self): - """Check main exits with failure if input is missing.""" - with self.assertRaises(SystemExit): - zapper_keyboard_test.main([1]) - - @patch("time.sleep", Mock()) - @patch("zapper_keyboard_test.Path") - def test_get_zapper_kbd_device(self, mock_path): - """ - Test whether the function returns a path to the Zapper - keyboard device when it's the only Zapper HID device. - """ - - mock_path.return_value.glob.side_effect = [ - [], - [], - [], - [], - [ - Path( - "/dev/input/by-id/" - "usb-Canonical_Zapper_main_board_123456-event-kbd", - ) - ], - ] - device = zapper_keyboard_test.get_zapper_kbd_device() - self.assertEqual( - device, - "/dev/input/by-id/" - "usb-Canonical_Zapper_main_board_123456-event-kbd", - ) - - @patch("time.sleep", Mock()) - @patch("zapper_keyboard_test.Path") - def test_get_zapper_kbd_device_if01(self, mock_path): - """ - Test whether the function returns a path to the Zapper - keyboard device when it's the second Zapper HID device. - """ - - mock_path.return_value.glob.return_value = [ - Path( - "/dev/input/by-id/" - "usb-Canonical_Zapper_main_board_123456-if01-event-kbd" - ), - ] - device = zapper_keyboard_test.get_zapper_kbd_device() - self.assertEqual( - device, - "/dev/input/by-id/" - "usb-Canonical_Zapper_main_board_123456-if01-event-kbd", - ) - - @patch("time.sleep", Mock()) - @patch("zapper_keyboard_test.Path") - def test_get_zapper_kbd_device_not_found(self, mock_path): - """ - Test whether the function raises an exception if no - Zapper keyboard can be found. - """ - - mock_path.return_value.glob.return_value = [] - with self.assertRaises(FileNotFoundError): - zapper_keyboard_test.get_zapper_kbd_device() - - @patch("zapper_keyboard_test.zapper_run", Mock()) - @patch("zapper_keyboard_test.get_zapper_kbd_device") - def test_main_no_keyboard(self, mock_get_dev): - """Check main exits with failure if Zapper keyboard is missing.""" - mock_get_dev.side_effect = FileNotFoundError - with self.assertRaises(SystemExit): - zapper_keyboard_test.main([1, 2]) - - @patch("zapper_keyboard_test.zapper_run", Mock()) - @patch("zapper_keyboard_test.get_zapper_kbd_device") - @patch("os.access") - def test_main_no_file_or_permission(self, mock_access, mock_get_dev): - """Check main exits with failure if Zapper keyboard is missing.""" - mock_get_dev.return_value = "/dev/input/by-id/keyboard" - mock_access.return_value = False - with self.assertRaises(SystemExit): - zapper_keyboard_test.main([1, 2]) - - @patch("zapper_keyboard_test.zapper_run") - @patch("zapper_keyboard_test.get_zapper_kbd_device") - @patch("zapper_keyboard_test.assert_type_string") - @patch("zapper_keyboard_test.assert_key_combo") - @patch("zapper_keyboard_test.KeyboardListener") - @patch("os.access") - def test_main( - self, - mock_access, - mock_key, - mock_combo, - mock_type, - mock_get_dev, - mock_run, - ): - """Check main exits with failure if any of the test fails.""" - - mock_get_dev.return_value = "/dev/input/by-id/keyboard" - mock_access.return_value = True - - mock_combo.side_effect = AssertionError - mock_type.side_effect = None - with self.assertRaises(SystemExit): - zapper_keyboard_test.main([1, "127.0.0.1"]) - mock_key.return_value.start.assert_called_once_with() - mock_key.return_value.stop.assert_called_once_with() - mock_key.return_value.join.assert_called_once_with() - mock_run.assert_called_once_with( - "127.0.0.1", - "robot_run", - zapper_keyboard_test.ROBOT_INIT.encode(), - {}, - {}, - ) - - mock_combo.side_effect = None - mock_type.side_effect = AssertionError - with self.assertRaises(SystemExit): - zapper_keyboard_test.main([1, 2]) - - # Happy case - mock_combo.side_effect = None - mock_type.side_effect = None - zapper_keyboard_test.main([1, 2]) - - -class KeyboardListenerTests(unittest.TestCase): - """This class provides test cases for the KeyboardListener class.""" - - @patch("os.open") - def test_init(self, mock_open): - """Test init function open and event file with the right parameters.""" - event_file_path = "/dev/input/event0" - listener = zapper_keyboard_test.KeyboardListener(event_file_path, None) - - assert listener._keep_running - mock_open.assert_called_with( - event_file_path, os.O_RDONLY | os.O_NONBLOCK - ) - - @patch("os.read") - @patch("os.open") - def test_read_keyboard_events(self, mock_open, mock_read): - """Test the function unpack events and call the callback if needed.""" - - callback = Mock() - listener = zapper_keyboard_test.KeyboardListener(None, callback) - - value = zapper_keyboard_test.KeyEvent.DOWN - code = 30 - mock_read.return_value = struct.pack( - "llHHI", - 0, - 0, - 1, - code, - value.value, - ) - - listener._read_keyboard_events() - - mock_read.assert_called_with( - mock_open.return_value, listener.EVENT_BIN_SIZE - ) - callback.assert_called_with((value, code)) - - @patch("os.open", Mock()) - @patch("os.read") - def test_read_keyboard_events_not_ready(self, mock_read): - """ - Test the function can handle a BlockingIOError and do not call - the callback when that happens. - """ - - callback = Mock() - listener = zapper_keyboard_test.KeyboardListener(None, callback) - - mock_read.side_effect = BlockingIOError - - listener._read_keyboard_events() - callback.assert_not_called() - - @patch("os.open", Mock()) - @patch("os.read") - def test_read_keyboard_events_not_kbd(self, mock_read): - """Test the function doesn't call the callback for non-kbd events.""" - - callback = Mock() - listener = zapper_keyboard_test.KeyboardListener(None, callback) - - value = zapper_keyboard_test.KeyEvent.DOWN - code = 30 - mock_read.return_value = struct.pack( - "llHHI", - 0, - 0, - 2, - code, - value.value, - ) - - listener._read_keyboard_events() - callback.assert_not_called() - - @patch("os.open", Mock()) - def test_run(self): - """Test run function polls the event file until stopped.""" - - listener = zapper_keyboard_test.KeyboardListener(None, None) - called_event = threading.Event() - listener._read_keyboard_events = called_event.set - - runner = threading.Thread(target=listener.run) - runner.start() - - called_event.wait(1) - if not called_event.is_set(): - raise AssertionError("Reader function not called") - - listener._keep_running = False - runner.join(1) - - if runner.is_alive(): - raise AssertionError("Runner didn't stop.") - - @patch("os.close") - @patch("os.open") - def test_stop(self, mock_open, mock_close): - """ - Test stop function stop the running loop and close the event files. - """ - listener = zapper_keyboard_test.KeyboardListener(None, None) - listener.stop() - - assert not listener._keep_running - mock_close.assert_called_with(mock_open.return_value) diff --git a/providers/base/units/canary/test-plan.pxu b/providers/base/units/canary/test-plan.pxu index d84c885144..2fee500a7b 100644 --- a/providers/base/units/canary/test-plan.pxu +++ b/providers/base/units/canary/test-plan.pxu @@ -97,14 +97,3 @@ include: com.canonical.certification::after-suspend-ethernet/detect com.canonical.certification::after-suspend-ethernet/ping_.* com.canonical.certification::after-suspend-usb/storage-detect - -id: canary-zapper -unit: test plan -_name: Zapper release self-test (canary test plan) -_description: - This test plan is meant to run on Zapper versions that are candidates for - a release against the latest beta Checkbox. -estimated_duration: 5m -nested_part: - com.canonical.certification::zapper-enabled-automated -include: diff --git a/providers/base/units/desktop/resource.pxu b/providers/base/units/desktop/resource.pxu deleted file mode 100644 index f89d70155e..0000000000 --- a/providers/base/units/desktop/resource.pxu +++ /dev/null @@ -1,6 +0,0 @@ -id: desktop_session -plugin: resource -category_id: com.canonical.plainbox::info -_summary: Check whether a desktop session is available and of which type. -environ: XDG_SESSION_TYPE XDG_CURRENT_DESKTOP -command: desktop_session.py resources diff --git a/providers/base/units/zapper/jobs.pxu b/providers/base/units/zapper/jobs.pxu deleted file mode 100644 index e4657142ad..0000000000 --- a/providers/base/units/zapper/jobs.pxu +++ /dev/null @@ -1,68 +0,0 @@ -id: monitor/zapper-edid -requires: - zapper_capabilities.capability == 'hdmi-capture' - zapper_capabilities.edid_cycling == 'True' - desktop_session.desktop_session == 'True' - desktop_session.session_type in ['x11', 'wayland'] -category_id: com.canonical.plainbox::monitor -plugin: shell -estimated_duration: 60 -_summary: Check if the system automatically changes the resolution based on EDID -environ: ZAPPER_HOST -command: edid_cycle.py "$ZAPPER_HOST" - -unit: template -template-resource: zapper_capabilities -template-filter: zapper_capabilities.capability == 'USB hotplug' and zapper_capabilities.usb_version == '2' -category_id: com.canonical.plainbox::usb -id: usb/zapper-usb-insert-{{ port_alias }} -template-id: usb/zapper-usb-insert-port_alias -template-engine: jinja2 -_summary: Zapper-driven USB insertion + removal test on port {{ port_alias }} -_purpose: - Check the insertion and removal of a USB 2.0 storage device using zapper. -requires: usb.usb2 == 'supported' -plugin: shell -user: root -environ: ZAPPER_HOST -command: - checkbox-support-run_watcher --zapper-usb-address={{ port_alias }} insertion usb2 -estimated_duration: 30 - -unit: template -template-resource: zapper_capabilities -template-filter: zapper_capabilities.capability == 'USB hotplug' and zapper_capabilities.usb_version == '3' -category_id: com.canonical.plainbox::usb -id: usb3/zapper-usb-insert-{{ port_alias }} -template-id: usb3/zapper-usb-insert-port_alias -template-engine: jinja2 -_summary: Zapper-driven USB 3 insertion + removal test test on port {{ port_alias }} -_purpose: - Check the insertion and removal of a USB 3.0 storage device using zapper. -requires: usb.usb3 == 'supported' -plugin: shell -user: root -environ: ZAPPER_HOST -command: - checkbox-support-run_watcher --zapper-usb-address={{ port_alias }} insertion usb3 -estimated_duration: 30 - -id: bluetooth/zapper-a2dp -requires: zapper_capabilities.capability == 'bluetooth' and 'a2dp' in zapper_capabilities.profiles -category_id: com.canonical.plainbox::bluetooth -plugin: shell -estimated_duration: 60 -_summary: Check if the system can connect to a Bluetooth speaker using A2DP profile -environ: ZAPPER_HOST -command: bt_a2dp.py "$ZAPPER_HOST" -depends: bluetooth/detect-output - -id: input/zapper-keyboard -requires: zapper_capabilities.capability == 'hid' -category_id: com.canonical.plainbox::input -plugin: shell -user: root -estimated_duration: 10 -_summary: Check if the system recognizes keyboard events from Zapper -environ: ZAPPER_HOST -command: zapper_keyboard_test.py "$ZAPPER_HOST" diff --git a/providers/base/units/zapper/packaging.pxu b/providers/base/units/zapper/packaging.pxu deleted file mode 100644 index fad3696847..0000000000 --- a/providers/base/units/zapper/packaging.pxu +++ /dev/null @@ -1,5 +0,0 @@ -# This is for Zapper-enabled Bluetooth tests -unit: packaging meta-data -os-id: debian -Depends: python3-bluez - diff --git a/providers/base/units/zapper/resource.pxu b/providers/base/units/zapper/resource.pxu deleted file mode 100644 index 99c0b77c8f..0000000000 --- a/providers/base/units/zapper/resource.pxu +++ /dev/null @@ -1,18 +0,0 @@ -id: zapper_capabilities -plugin: resource -category_id: com.canonical.plainbox::info -_summary: Get Zapper's setup capabilities -_description: - Connect to Zapper and list functions that the current setup (DUT + Zapper) are - capable off. -environ: ZAPPER_HOST -command: checkbox-support-zapper-proxy get_capabilities - -id: zapper_reset_setup -plugin: shell -requires: zapper_capabilities.available == 'True' -category_id: com.canonical.plainbox::info -_summary: Reset the Zapper environment to prevent it interfering with other test cases. -environ: ZAPPER_HOST -command: checkbox-support-zapper-proxy reset_testing_setup || { echo "Some jobs might be affected..."; true; } - diff --git a/providers/base/units/zapper/test-plan.pxu b/providers/base/units/zapper/test-plan.pxu deleted file mode 100644 index 03892f1b15..0000000000 --- a/providers/base/units/zapper/test-plan.pxu +++ /dev/null @@ -1,23 +0,0 @@ -id: zapper-enabled-full -unit: test plan -_name: Tests using Zapper -_description: Tests using Zapper -include: -nested_part: - zapper-enabled-automated - -id: zapper-enabled-automated -unit: test plan -_name: Tests using Zapper (automated) -_description: Tests using Zapper -include: - bluetooth/zapper-a2dp - input/zapper-keyboard - monitor/zapper-edid - usb/zapper-usb-insert-.* - usb3/zapper-usb-insert-.* -bootstrap_include: - desktop_session - graphics_card - zapper_capabilities - zapper_reset_setup diff --git a/providers/certification-client/units/client-cert-iot-desktop-24-04.pxu b/providers/certification-client/units/client-cert-iot-desktop-24-04.pxu index 40a704c8ac..3aa463105e 100644 --- a/providers/certification-client/units/client-cert-iot-desktop-24-04.pxu +++ b/providers/certification-client/units/client-cert-iot-desktop-24-04.pxu @@ -169,7 +169,6 @@ nested_part: wireless-wifi-master-mode-auto wwan-automated va-api-automated - zapper-enabled-automated after-suspend-audio-automated after-suspend-bluez-automated after-suspend-eeprom-automated diff --git a/providers/certification-client/units/client-cert-iot-server-24-04.pxu b/providers/certification-client/units/client-cert-iot-server-24-04.pxu index 0ca7d38096..ccd3967ca5 100644 --- a/providers/certification-client/units/client-cert-iot-server-24-04.pxu +++ b/providers/certification-client/units/client-cert-iot-server-24-04.pxu @@ -157,7 +157,6 @@ nested_part: wireless-automated wireless-wifi-master-mode-auto wwan-automated - zapper-enabled-automated after-suspend-audio-automated after-suspend-bluez-automated after-suspend-eeprom-automated diff --git a/providers/certification-client/units/client-cert-iot-ubuntucore-22.pxu b/providers/certification-client/units/client-cert-iot-ubuntucore-22.pxu index f05e8750b3..4a5d84fe41 100644 --- a/providers/certification-client/units/client-cert-iot-ubuntucore-22.pxu +++ b/providers/certification-client/units/client-cert-iot-ubuntucore-22.pxu @@ -142,7 +142,6 @@ nested_part: wireless-automated wireless-wifi-master-mode-auto wwan-automated - zapper-enabled-automated after-suspend-audio-automated after-suspend-bluez-automated after-suspend-ethernet-automated diff --git a/providers/certification-client/units/client-cert-iot-ubuntucore-24.pxu b/providers/certification-client/units/client-cert-iot-ubuntucore-24.pxu index d2b032c533..0143783269 100644 --- a/providers/certification-client/units/client-cert-iot-ubuntucore-24.pxu +++ b/providers/certification-client/units/client-cert-iot-ubuntucore-24.pxu @@ -158,7 +158,6 @@ nested_part: wireless-automated wireless-wifi-master-mode-auto wwan-automated - zapper-enabled-automated after-suspend-audio-automated after-suspend-bluez-automated after-suspend-eeprom-automated diff --git a/providers/base/bin/desktop_session.py b/providers/resource/bin/desktop_session_resource.py similarity index 100% rename from providers/base/bin/desktop_session.py rename to providers/resource/bin/desktop_session_resource.py diff --git a/providers/resource/jobs/resource.pxu b/providers/resource/jobs/resource.pxu index 2bae08bd4f..24d72a683d 100644 --- a/providers/resource/jobs/resource.pxu +++ b/providers/resource/jobs/resource.pxu @@ -532,3 +532,9 @@ plugin: resource _summary: Discover if the system supports power modes via acpi _description: Discover resource info from /sys/firmware/apci command: platform_profile_resource.py + +id: desktop_session +plugin: resource +_summary: Check whether a desktop session is available and of which type. +environ: XDG_SESSION_TYPE XDG_CURRENT_DESKTOP +command: desktop_session_resource.py resources \ No newline at end of file diff --git a/providers/base/tests/test_desktop_session.py b/providers/resource/tests/test_desktop_session_resource.py similarity index 77% rename from providers/base/tests/test_desktop_session.py rename to providers/resource/tests/test_desktop_session_resource.py index dd396f296f..ea98abc74a 100644 --- a/providers/base/tests/test_desktop_session.py +++ b/providers/resource/tests/test_desktop_session_resource.py @@ -1,22 +1,22 @@ -"""This module provides test cases for the desktop_session module.""" +"""This module provides test cases for the desktop_session_resource module.""" import os import unittest from unittest.mock import call, patch -import desktop_session +import desktop_session_resource class DesktopSessionTests(unittest.TestCase): - """Tests for the desktop_session module.""" + """Tests for the desktop_session_resource module.""" - @patch("desktop_session.resources") + @patch("desktop_session_resource.resources") def test_main(self, mock_resources): """ Test whether the main function calls the resources function when requested via CLI. """ - desktop_session.main(["resources"]) + desktop_session_resource.main(["resources"]) mock_resources.assert_called_once_with() @patch("builtins.print") @@ -27,7 +27,7 @@ def test_resources_server(self, mock_print): "XDG_SESSION_TYPE": "tty", } with patch.dict(os.environ, server_session, clear=True): - desktop_session.resources() + desktop_session_resource.resources() mock_print.assert_has_calls( [ @@ -45,7 +45,7 @@ def test_resources_desktop(self, mock_print): "XDG_CURRENT_DESKTOP": "hyprland", } with patch.dict(os.environ, server_session, clear=True): - desktop_session.resources() + desktop_session_resource.resources() mock_print.assert_has_calls( [ diff --git a/providers/sru/units/sru.pxu b/providers/sru/units/sru.pxu index 827fa5e98f..c93ae52ae5 100644 --- a/providers/sru/units/sru.pxu +++ b/providers/sru/units/sru.pxu @@ -128,7 +128,6 @@ nested_part: monitor-discrete-gpu-cert-automated graphics-integrated-gpu-cert-automated graphics-discrete-gpu-cert-automated - zapper-enabled-automated # start of suspend related tests before-suspend-reference-cert-full # suspend point