diff --git a/.github/actions/checkbox_source_deb/action.yaml b/.github/actions/checkbox_source_deb/action.yaml new file mode 100644 index 0000000000..aa59898a5e --- /dev/null +++ b/.github/actions/checkbox_source_deb/action.yaml @@ -0,0 +1,83 @@ +name: Submit a Checkbox Test plan (or subset of it) to the lab +inputs: + data_source: + description: "Target image and provisioning data (ex. `url:` or `distro:`)" + required: false + default: null + queue: + description: "Queue that will run the testing (ex. 202012-28526)" + required: true + test_plan: + description: "Test plan to run (ex. com.canonical.certification::sru)" + required: true + match: + description: "Subset of jobs to run (ex. .*wireless.*)" + required: false + default: ".*" + launcher_override: + description: "Launcher with additional values that will take priority over the defaults" + default: "" + required: false + checkbox_revision: + description: "Revision of checkbox that has to be provisioned (ex. commit_hash, branch name, can be `beta`)" + required: true + zapper_channel: + description: "Zapper channel to be used, will be ignored if no Zapper (ex. edge, beta, stable)" + required: false + default: "beta" +runs: + using: composite + steps: + - uses: actions/checkout@v4 + - name: Install dependencies + shell: bash + run: | + sudo apt install -y -qq gettext + - name: Build test resource + shell: bash + env: + INPUT_DATA_SOURCE: ${{ inputs.data_source }} + INPUT_QUEUE: ${{ inputs.queue }} + INPUT_MATCH: ${{ inputs.match || '.*' }} + INPUT_TEST_PLAN: ${{ inputs.test_plan }} + INPUT_LAUNCHER_OVERRIDE: ${{ inputs.launcher_override }} + INPUT_CHECKBOX_REVISION: ${{ inputs.checkbox_revision }} + INPUT_ZAPPER_CHANNEL: ${{ inputs.zapper_channel || 'beta' }} + working-directory: ${{ github.action_path }}/../../../tools/lab_dispatch + run: | + echo "::group::Building the testflinger job" + if [ -n "$INPUT_DATA_SOURCE" ]; then + INPUT_DATA_SOURCE="provision_data: $INPUT_DATA_SOURCE" + fi + envsubst '$INPUT_CHECKBOX_REVISION $INPUT_DATA_SOURCE $INPUT_QUEUE $INPUT_ZAPPER_CHANNEL' < generic_source.yaml | tee job.yaml + echo "::endgroup::" + + echo "::group::Building the Checkbox launcher" + # this goes from .template. (missing secret, testplan, match etc. to .partial.) + # this is partial as some values are filled in on the agent (like wireless access points names) + envsubst '$INPUT_TEST_PLAN $INPUT_MATCH' < resources/checkbox.no-manifest.template.conf | tee resources/checkbox.no-manifest.partial.conf + echo "::endgroup::" + + echo "::group::Dumping launcher overrides" + echo "$INPUT_LAUNCHER_OVERRIDE" | tee launcher_override.conf + echo "::endgroup::" + - name: Workaroud cwd + shell: bash + run: | + # this allows us to dispatch the action and the attachments with relative + # paths even when called form outside the Checkbox repo + action_path=$(realpath ${{ github.action_path }}/../../../tools/) + workdir_path=$(realpath tools/) + if [ ! -e "$workdir_path" ]; then + cp -rT "$action_path" "$workdir_path" + fi + if [ "$action_path" = "$workdir_path" ]; then + echo "Skipping copy as the action is already running in workdir" + else + cp -rT "$action_path" "$workdir_path" + fi + - name: Submit and monitor job + uses: canonical/testflinger/.github/actions/submit@main + with: + poll: true + job-path: tools/lab_dispatch/job.yaml diff --git a/.github/workflows/dispatch_lab_job.yaml b/.github/workflows/dispatch_lab_job.yaml index 1832bcecd7..d01818f5d6 100644 --- a/.github/workflows/dispatch_lab_job.yaml +++ b/.github/workflows/dispatch_lab_job.yaml @@ -9,6 +9,7 @@ on: # - queue: machine that will run the job (ex. 202012-28526) # - test_plan: Checkbox test plan to run (ex. com.canonical.certification::sru) # - match: subset of jobs to run (ex. .*wireless.*) + # - zapper_channel: refreshes the zapper snap to the channel if provided, default is beta (ex. "beta") # # One possible matrix_to_create would therefore look like this: # matrix_to_create=[{ data_source: "distro: desktop-22-04-2-uefi", queue: "202012-28526", match: ".*wireless.*", test_plan: "com.canonical.certification::sru" }]' @@ -24,39 +25,34 @@ jobs: run-matrix: runs-on: [self-hosted, testflinger] strategy: + fail-fast: false matrix: spec: ${{ fromJson(inputs.matrix_to_create) }} defaults: run: working-directory: tools/lab_dispatch steps: - - uses: actions/checkout@v4 - with: - fetch-depth: 0 - - name: Install dependencies - run: | - sudo apt install gettext - - name: Build test resource - env: - INPUT_DATA_SOURCE: ${{ matrix.spec.data_source }} - INPUT_QUEUE: ${{ matrix.spec.queue }} - INPUT_MATCH: ${{ matrix.spec.match }} - INPUT_TEST_PLAN: ${{ matrix.spec.test_plan }} - INPUT_PASSWORD_SECRET: ${{ secrets.INPUT_PASSWORD_SECRET }} - run: | - echo "::group::Building the testflinger job" - export INPUT_CHECKBOX_REVISION="$(git rev-parse HEAD)" - envsubst '$INPUT_CHECKBOX_REVISION $INPUT_DATA_SOURCE $INPUT_QUEUE' < generic_source.yaml | tee job.yaml - echo "::endgroup::" - echo "::group::Building the Checkbox launcher" + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Get current commit SHA + id: get_sha + run: echo "sha=$(git rev-parse HEAD)" >> $GITHUB_OUTPUT - # this goes from .template. (missing secret, testplan, match etc. to .partial.) - # this is partial as some values are filled in on the agent (like wireless access points names) - envsubst '$INPUT_TEST_PLAN $INPUT_MATCH $INPUT_PASSWORD_SECRET' < resources/checkbox.no-manifest.template.conf | tee resources/checkbox.no-manifest.partial.conf - echo "::endgroup::" - - name: Submit and monitor job - uses: canonical/testflinger/.github/actions/submit@main + - name: Run the spec + uses: canonical/checkbox/.github/actions/checkbox_source_deb@main with: - poll: true - job-path: tools/lab_dispatch/job.yaml + data_source: ${{ matrix.spec.data_source }} + queue: ${{ matrix.spec.queue }} + test_plan: ${{ matrix.spec.test_plan }} + match: ${{ matrix.spec.match }} + zapper_channel: ${{ matrix.spec.zapper_channel }} + launcher_override: | + [environment] + WPA_BG_PSK = ${{ secrets.INPUT_PASSWORD_SECRET }} + WPA_N_PSK = ${{ secrets.INPUT_PASSWORD_SECRET }} + WPA_AC_PSK = ${{ secrets.INPUT_PASSWORD_SECRET }} + WPA_AX_PSK = ${{ secrets.INPUT_PASSWORD_SECRET }} + WPA3_AX_PSK = ${{ secrets.INPUT_PASSWORD_SECRET }} + checkbox_revision: ${{ steps.get_sha.outputs.sha }} diff --git a/.github/workflows/validate_workflows.yaml b/.github/workflows/validate_workflows.yaml index f8f02eabaa..df40cdcb78 100644 --- a/.github/workflows/validate_workflows.yaml +++ b/.github/workflows/validate_workflows.yaml @@ -16,7 +16,7 @@ jobs: uses: asdf-vm/actions/install@v3 with: tool_versions: | - action-validator 0.5.1 + action-validator 0.6.0 - name: Lint Actions run: | find .github/workflows -type f \( -iname \*.yaml -o -iname \*.yml \) \ diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 5bcdfbc316..6636cdcbf3 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -27,7 +27,7 @@ install everything you need in a Python virtual environment. Install the required tools: - $ sudo apt install git python3-virtualenv + $ sudo apt install git python3-virtualenv libasound2-dev Prepare the development environment. If you are an external contributor and plan on submitting some changes, you will have to [fork the Checkbox repository diff --git a/checkbox-ng/checkbox_ng/launcher/subcommands.py b/checkbox-ng/checkbox_ng/launcher/subcommands.py index 55d4666f80..a5d3ef559b 100644 --- a/checkbox-ng/checkbox_ng/launcher/subcommands.py +++ b/checkbox-ng/checkbox_ng/launcher/subcommands.py @@ -24,8 +24,8 @@ from collections import defaultdict from string import Formatter from tempfile import TemporaryDirectory -import textwrap import fnmatch +import itertools import contextlib import gettext import json @@ -1324,6 +1324,37 @@ def register_arguments(self, parser): help=_("output format: 'text' or 'json' (default: %(default)s)"), ) + def _get_relevant_manifest_units(self, jobs_and_templates_list): + """ + Get all manifest units that are cited in the jobs_and_templates_list + resource expressions + """ + # get all manifest units + manifest_units = filter( + lambda unit: unit.unit == "manifest entry", + self.sa._context.unit_list, + ) + # get all jobs/templates that have a requires and do require a manifest + # entry + job_requires = [ + requires + for requires in map( + lambda x: x.get_record_value("requires"), + jobs_and_templates_list, + ) + if requires and "manifest" in requires + ] + + # only return manifest entries that are actually required by any job in + # the list + return filter( + lambda manifest_unit: any( + "manifest.{}".format(manifest_unit.partial_id) in require + for require in job_requires + ), + manifest_units, + ) + def invoked(self, ctx): self.ctx = ctx session_title = "checkbox-expand-{}".format(ctx.args.TEST_PLAN) @@ -1340,31 +1371,48 @@ def invoked(self, ctx): tp = self.sa._context._test_plan_list[0] tp_us = TestPlanUnitSupport(tp) self.override_list = tp_us.override_list + jobs_and_templates_list = select_units( all_jobs_and_templates, [tp.get_mandatory_qualifier()] + [tp.get_qualifier()], ) + relevant_manifest_units = self._get_relevant_manifest_units( + jobs_and_templates_list + ) + units_to_print = itertools.chain( + relevant_manifest_units, iter(jobs_and_templates_list) + ) obj_list = [] - for unit in jobs_and_templates_list: + for unit in units_to_print: obj = unit._raw_data.copy() obj["unit"] = unit.unit obj["id"] = unit.id # To get the fully qualified id - obj["certification-status"] = ( - self.get_effective_certification_status(unit) - ) - if unit.template_id: - obj["template-id"] = unit.template_id + # these two don't make sense for manifest units + if unit.unit != "manifest entry": + obj["certification-status"] = ( + self.get_effective_certification_status(unit) + ) + if unit.template_id: + obj["template-id"] = unit.template_id obj_list.append(obj) - obj_list.sort(key=lambda x: x.get("template-id", x["id"])) + + obj_list.sort(key=lambda x: x.get("template-id", x["id"]) or x["id"]) + if ctx.args.format == "json": - print(json.dumps(obj_list, sort_keys=True)) + json.dump(obj_list, sys.stdout, sort_keys=True) else: for obj in obj_list: if obj["unit"] == "template": print("Template '{}'".format(obj["template-id"])) - else: + elif obj["unit"] == "manifest entry": + print("Manifest '{}'".format(obj["id"])) + elif obj["unit"] == "job": print("Job '{}'".format(obj["id"])) + else: + raise AssertionError( + "Unknown unit type {}".format(obj["unit"]) + ) def get_effective_certification_status(self, unit): if unit.unit == "template": diff --git a/checkbox-ng/checkbox_ng/launcher/test_subcommands.py b/checkbox-ng/checkbox_ng/launcher/test_subcommands.py index 949a5b9004..466411a766 100644 --- a/checkbox-ng/checkbox_ng/launcher/test_subcommands.py +++ b/checkbox-ng/checkbox_ng/launcher/test_subcommands.py @@ -813,6 +813,16 @@ def setUp(self): self.launcher = Expand() self.ctx = Mock() self.ctx.args = Mock(TEST_PLAN="", format="") + + selected_1 = Mock(unit="manifest entry", id="some", partial_id="some") + selected_1._raw_data.copy.return_value = {} + selected_2 = Mock( + unit="manifest entry", id="other", partial_id="other" + ) + selected_2._raw_data.copy.return_value = {} + not_selected = Mock(unit="manifest entry", partial_id="not_selected") + not_selected._raw_data.copy.return_value = {} + self.ctx.sa = Mock( start_new_session=Mock(), get_test_plans=Mock(return_value=["test-plan1", "test-plan2"]), @@ -821,6 +831,7 @@ def setUp(self): _context=Mock( state=Mock(unit_list=[]), _test_plan_list=[Mock()], + unit_list=[selected_1, selected_2, not_selected], ), ) @@ -844,17 +855,22 @@ def test_invoke__text(self, mock_select_units, mock_tpus, stdout): "template-id": "test-template", "id": "test-{res}", "template-summary": "Test Template Summary", + "requires": "manifest.some == 'True'", } ) job1 = JobDefinition( { "id": "job1", + "requires": "manifest.other == 'Other'", } ) mock_select_units.return_value = [job1, template1] self.ctx.args.TEST_PLAN = "test-plan1" self.launcher.invoked(self.ctx) self.assertIn("Template 'test-template'", stdout.getvalue()) + self.assertIn("Manifest 'some'", stdout.getvalue()) + self.assertIn("Manifest 'other'", stdout.getvalue()) + self.assertNotIn("Manifest 'not_selected'", stdout.getvalue()) @patch("sys.stdout", new_callable=StringIO) @patch("checkbox_ng.launcher.subcommands.TestPlanUnitSupport") @@ -865,18 +881,24 @@ def test_invoke__json(self, mock_select_units, mock_tpus, stdout): "template-id": "test-template", "id": "test-{res}", "template-summary": "Test Template Summary", + "requires": "manifest.some == 'True'", } ) job1 = JobDefinition( { "id": "job1", + "requires": "manifest.other == 'Other'", } ) + mock_select_units.return_value = [job1, template1] self.ctx.args.TEST_PLAN = "test-plan1" self.ctx.args.format = "json" self.launcher.invoked(self.ctx) self.assertIn('"template-id": "test-template"', stdout.getvalue()) + self.assertIn('"id": "some"', stdout.getvalue()) + self.assertIn('"id": "other"', stdout.getvalue()) + self.assertNotIn('"id": "not_selected"', stdout.getvalue()) def test_get_effective_certificate_status(self): job1 = JobDefinition( diff --git a/docs/how-to/index.rst b/docs/how-to/index.rst index d8a8c49fdf..f1175fca8d 100644 --- a/docs/how-to/index.rst +++ b/docs/how-to/index.rst @@ -13,4 +13,5 @@ These how-to guides cover key operations and processes in Checkbox. side-loading agent-service freeze-checkbox-version - launcher/* \ No newline at end of file + launcher/* + using-match diff --git a/docs/how-to/using-match.rst b/docs/how-to/using-match.rst new file mode 100644 index 0000000000..aee053cc60 --- /dev/null +++ b/docs/how-to/using-match.rst @@ -0,0 +1,52 @@ +Using ``match`` +^^^^^^^^^^^^^^^ + +When a subset of a test plan run fails, it can be expensive to re-run it all. +To help with this, the ``match`` keyword was introduced. It allows you to +re-run only a subset of a test plan. + +Key features of ``match``: + +* All tests in the bootstrap section will always be included +* Test Selection screen is still shown and functional, but only matching tests are shown +* Matched tests pull their dependencies automatically +* ``exclude`` has the priority over ``match`` + +To only re-run the ``wireless`` portion of the ``sru`` test plan, use the +following launcher: + +.. code-block:: ini + + [test plan] + unit = com.canonical.certification::sru + forced = yes + + [test selection] + match = .*wireless.* + +To only re-run the WiFi ``bg_np`` and ``ac_np`` tests for ``wlan0``: + +.. code-block:: ini + :emphasize-lines: 7-8 + + [test plan] + unit = com.canonical.certification::sru + forced = yes + + [test selection] + match = + com.canonical.certification::wireless/wireless_connection_open_ac_np_wlan0 + com.canonical.certification::wireless/wireless_connection_open_bg_np_wlan0 + +To re-run all wireless tests but ``bg_np``: + +.. code-block:: ini + :emphasize-lines: 6-7 + + [test plan] + unit = com.canonical.certification::sru + forced = yes + + [test selection] + exclude = .*wireless.*bg_np.* + match = .*wireless.* diff --git a/providers/base/bin/camera_test.py b/providers/base/bin/camera_test.py index e4b104d648..8128513089 100755 --- a/providers/base/bin/camera_test.py +++ b/providers/base/bin/camera_test.py @@ -413,39 +413,30 @@ def image(self): """ pixelformat = self._get_default_format()["pixelformat"] if self.output: - self._still_image_helper( + self._capture_image( self.output, self._width, self._height, pixelformat ) else: with NamedTemporaryFile( prefix="camera_test_", suffix=".jpg", delete=False ) as f: - self._still_image_helper( + self._capture_image( f.name, self._width, self._height, pixelformat ) - def _still_image_helper(self, filename, width, height, pixelformat): + def _capture_image(self, filename, width, height, pixelformat): """ Captures an image to a given filename. If the image capture fails with fswebcam, it will try to capture the image with gstreamer. """ - # Try to take a picture with fswebcam - use_fswebcam = True - use_gstreamer = False - - if use_fswebcam: - result = self._capture_image_fswebcam( - filename, width, height, pixelformat - ) - if not result: - print("Failed to capture image with fswebcam, using gstreamer") - use_gstreamer = True - - # If fswebcam fails, try with gstreamer - if use_gstreamer: + if not self._capture_image_fswebcam( + filename, width, height, pixelformat + ): + print("Failed to capture image with fswebcam, using gstreamer") + # If fswebcam fails, try with gstreamer self._capture_image_gstreamer(filename, width, height, pixelformat) - print("Image saved to %s" % filename) + print("Image saved to %s" % filename) if not self.headless: self._display_image(filename, width, height) @@ -465,16 +456,15 @@ def _capture_image_fswebcam(self, filename, width, height, pixelformat): filename, ] if pixelformat: - if "MJPG" == pixelformat: # special tweak for fswebcam - pixelformat = "MJPEG" - command.extend(["-p", pixelformat]) + # special tweak for fswebcam + command.extend( + ["-p", pixelformat if pixelformat != "MJPG" else "MJPEG"] + ) try: check_call(command, stdout=open(os.devnull, "w"), stderr=STDOUT) - if os.path.getsize(filename) == 0: - return False + return os.path.getsize(filename) != 0 except (CalledProcessError, OSError): return False - return True def _capture_image_gstreamer(self, filename, width, height, pixelformat): """ @@ -635,7 +625,7 @@ def resolutions(self): ) print("Taking a picture at %sx%s" % (w, h)) - self._still_image_helper( + self._capture_image( f.name, w, h, pixelformat=format["pixelformat"] ) if self._validate_image(f.name, w, h): @@ -671,7 +661,7 @@ def _save_debug_image(self, format, device, output): ) print("Saving debug image to %s" % filepath) with open(filepath, "w") as f: - self._still_image_helper(f.name, w, h, format["pixelformat"]) + self._capture_image(f.name, w, h, format["pixelformat"]) def _get_supported_pixel_formats(self, device, maxformats=5): """ diff --git a/providers/base/bin/cpuid.py b/providers/base/bin/cpuid.py index 6fa39842b5..1af0e68679 100755 --- a/providers/base/bin/cpuid.py +++ b/providers/base/bin/cpuid.py @@ -178,7 +178,7 @@ def cpuid_to_human_friendly(cpuid: str) -> str: "AMD Bergamo": ['0xaa0f01'], "AMD Siena SP6": ['0xaa0f02'], "AMD Raphael": ['0xa60f12'], - "AMD Turin": ['0xb00f21'], + "AMD Turin": ['0xb00f21', '0xb10f10'], "Broadwell": ['0x4067', '0x306d4', '0x5066', '0x406f'], "Canon Lake": ['0x6066'], "Cascade Lake": ['0x50655', '0x50656', '0x50657'], diff --git a/providers/base/bin/reboot_check_test.py b/providers/base/bin/reboot_check_test.py new file mode 100755 index 0000000000..17527b3447 --- /dev/null +++ b/providers/base/bin/reboot_check_test.py @@ -0,0 +1,450 @@ +#!/usr/bin/env python3 + +import argparse +import os +import subprocess as sp +import re +import shutil +import filecmp +import sys +import typing as T +from checkbox_support.scripts.image_checker import has_desktop_environment + + +# Checkbox could run in a snap container, so we need to prepend this root path +RUNTIME_ROOT = os.getenv("CHECKBOX_RUNTIME", default="") +# Snap mount point, see +# https://snapcraft.io/docs/environment-variables#heading--snap +SNAP = os.getenv("SNAP", default="") + + +class DeviceInfoCollector: + + class Device: + PCI = "pci" + WIRELESS = "wireless" + USB = "usb" + DRM = "drm" + + DEFAULT_DEVICES = { + "required": [ + Device.WIRELESS, + Device.PCI, + Device.USB, + ], # these can fail the test case + "optional": [Device.DRM], # these only produce warnings + } # type: dict[str, list[str]] + # to modify, add more values in the enum + # and reference them in required/optional respectively + + def get_drm_info(self) -> str: + return str(os.listdir("/sys/class/drm")) + + def get_wireless_info(self) -> str: + iw_out = sp.run( + ["iw", "dev"], stdout=sp.PIPE, stderr=sp.PIPE, check=True + ) + lines = iw_out.stdout.decode().splitlines() + lines_to_write = list( + filter( + lambda line: "addr" in line + or "Interface" in line + or "ssid" in line, + sorted(lines), + ) + ) + return "\n".join(map(lambda line: line.strip(), lines_to_write)) + + def get_usb_info(self) -> str: + return sp.run( + [ + "checkbox-support-lsusb", + "-f", + '"{}"/var/lib/usbutils/usb.ids'.format(RUNTIME_ROOT), + "-s", + ], + check=True, + ).stdout.decode() + + def get_pci_info(self) -> str: + return sp.run( + ["lspci", "-i", "{}/usr/share/misc/pci.ids".format(SNAP)], + check=True, + ).stdout.decode() + + def compare_device_lists( + self, + expected_dir: str, + actual_dir: str, + devices: T.Dict[str, T.List[str]] = DEFAULT_DEVICES, + ) -> bool: + """Compares the list of devices in expected_dir against actual_dir + + :param expected_dir: files containing the expected device list + :param actual_dir: files containing the actual device list + :param devices: what devices do we want to compare, see DEFAULT_DEVICES + :return: whether the device list matches + """ + print( + "Comparing devices in (expected) {} against (actual) {}...".format( + expected_dir, actual_dir + ) + ) + for device in devices["required"]: + # file paths of the expected and actual device lists + expected = "{}/{}_log".format(expected_dir, device) + actual = "{}/{}_log".format(actual_dir, device) + if not filecmp.cmp(expected, actual): + print( + "[ ERR ] The output of {} differs!".format(device), + file=sys.stderr, + ) + return False + + for device in devices["optional"]: + expected = "{}/{}_log".format(expected_dir, device) + actual = "{}/{}_log".format(actual_dir, device) + if not filecmp.cmp(expected, actual): + print( + "[ WARN ] Items under {} have changed.".format(actual), + file=sys.stderr, + ) + + return True + + def dump( + self, + output_directory: str, + devices: T.Dict[str, T.List[str]] = DEFAULT_DEVICES, + ) -> None: + os.makedirs(output_directory, exist_ok=True) + # add extra behavior if necessary + for device in devices["required"]: + with open( + "{}/{}_log".format(output_directory, device), "w" + ) as file: + file.write(self.dump_function[device]()) + + for device in devices["optional"]: + with open( + "{}/{}_log".format(output_directory, device), "w" + ) as file: + file.write(self.dump_function[device]()) + + os.sync() + + def __init__(self) -> None: + self.dump_function = { + self.Device.PCI: self.get_pci_info, + self.Device.DRM: self.get_drm_info, + self.Device.USB: self.get_usb_info, + self.Device.WIRELESS: self.get_wireless_info, + } + + +class FwtsTester: + def is_fwts_supported(self) -> bool: + return shutil.which("fwts") is not None + + def fwts_log_check_passed( + self, output_directory: str, fwts_arguments=["klog", "oops"] + ) -> bool: + """ + Check if fwts logs passes the checks specified in sleep_test_log_check + This script live in the same directory + + :param output_directory: where the output of fwts should be written to + :type output_directory: str + :return: whether sleep_test_log_check.py returned 0 (success) + :rtype: bool + """ + log_file_path = "{}/fwts_{}.log".format( + output_directory, "_".join(fwts_arguments) + ) + sp.run(["fwts", "-r", log_file_path, *fwts_arguments]) + result = sp.run( + [ + "sleep_test_log_check.py", + "-v", + "--ignore-warning", + "-t", + "all", + log_file_path, + ], + ) + + return result.returncode == 0 + + +class HardwareRendererTester: + + def has_display_connection(self) -> bool: + """ + Checks if a display is connected by searching /sys/class/drm + + :return: True if there's at least 1 node that is connected + """ + + # look for GPU file nodes first + DRM_PATH = "/sys/class/drm" + possible_gpu_nodes = [ + directory + for directory in os.listdir(DRM_PATH) + if directory != "version" + ] + if len(possible_gpu_nodes) == 0: + # kernel doesn't see any GPU nodes + print( + "There's nothing under {}".format(DRM_PATH), + "if an external GPU is connected," + "check if the connection is loose", + ) + return False + + print("These nodes", possible_gpu_nodes, "exist") + + for gpu in possible_gpu_nodes: + # for each gpu, check for connection + # return true if anything is connected + try: + with open("{}/{}/status".format(DRM_PATH, gpu)) as status_file: + if status_file.read().strip().lower() == "connected": + print("{} is connected to display!".format(gpu)) + return True + except FileNotFoundError: + # this just means we don't have a status file + # => no connection, continue to the next + pass + except Exception as e: + print("Unexpected error: ", e, file=sys.stderr) + + print( + "No display is connected. This case will be skipped.", + "Maybe the display cable is not connected?", + "If the device is not supposed to have a display," + "then skipping is expected", + ) + return False + + def is_hardware_renderer_available(self) -> bool: + """ + Checks if hardware rendering is being used. + THIS ASSUMES A DRM CONNECTION EXISTS + - self.has_display_connection() should be called first if unsure + + :return: True if a hardware renderer is active, otherwise return False + :rtype: bool + """ + + DISPLAY = os.getenv("DISPLAY", "") + + if DISPLAY == "": + print("$DISPLAY is not set, we will let unity_support infer this") + else: + print("Checking $DISPLAY={}".format(DISPLAY)) + + unity_support_output = sp.run( + ["{}/usr/lib/nux/unity_support_test".format(RUNTIME_ROOT), "-p"] + ) + if unity_support_output.returncode != 0: + print( + "[ ERR ] unity support test returned {}".format( + unity_support_output.returncode + ), + file=sys.stderr, + ) + return False + + is_hardware_rendered = ( + self.parse_unity_support_output( + unity_support_output.stdout.decode() + ).get("Not software rendered") + == "yes" + ) + if is_hardware_rendered: + print("[ OK ] This machine is using a hardware renderer!") + return True + + print("[ ERR ] Software rendering detected", file=sys.stderr) + return False + + def parse_unity_support_output( + self, unity_output_string: str + ) -> T.Dict[str, str]: + """ + Parses the output of `unity_support_test` into a dictionary + + :param output_string: the raw output from running unity_support_test -p + :type output_string: str + :return: string key-value pairs that mirror the output of unity_support + Left hand side of the first colon are the keys; + right hand side are the values. + :rtype: dict[str, str] + """ + + output = {} # type: dict[str, str] + for line in unity_output_string.split("\n"): + # max_split=1 to prevent splitting the string after the 1st colon + words = line.split(":", maxsplit=1) + if len(words) == 2: + key = words[0].strip() + value = remove_color_code(words[1].strip()) + output[key] = value + + return output + + +def get_failed_services() -> T.List[str]: + """ + Counts the number of failed services listed in systemctl + + :return: a list of failed services as they appear in systemctl + """ + command = [ + "systemctl", + "list-units", + "--system", + "--no-ask-password", + "--no-pager", + "--no-legend", + "--state=failed", + "--plain", # plaintext, otherwise it includes color codes + ] + + return sp.run(command).stdout.decode().splitlines() + + +def create_parser(): + parser = argparse.ArgumentParser( + prog="Reboot tests", + description="Collects device info and compares them across reboots", + ) + parser.add_argument( + "-d", + "--dump-to", + required=False, + dest="output_directory", + help="Device info-dumps will be written here", + ) + parser.add_argument( + "-c", + "--compare-to", + dest="comparison_directory", + help="Directory of ground-truth for device info comparison", + ) + parser.add_argument( + "-s", + "--service-check", + default=False, + dest="do_service_check", + action="store_true", + help="If specified, check if all system services are running", + ) + parser.add_argument( + "-f", + "--fwts-check", + default=False, + dest="do_fwts_check", + action="store_true", + help="If specified, look for fwts log errors", + ) + parser.add_argument( + "-g", + "--graphics", + default=False, + dest="do_renderer_check", + action="store_true", + help="If specified, check if hardware rendering is being used", + ) + + return parser + + +def remove_color_code(string: str) -> str: + """ + Removes ANSI color escape sequences from string + + :param string: the string that you would like to remove color code + credit: Hanhsuan Lee + """ + return re.sub(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])", "", string) + + +def main() -> int: + """Main routine + + :return: an return code for checkbox to consume, 1 = failed, 0 = success + :rtype: int + """ + + args = create_parser().parse_args() + + # all 4 tests pass by default + # they only fail if their respective flags are specified + # if no flags are specified, calling this script is a no-op + fwts_passed = True + device_comparison_passed = True + renderer_test_passed = True + service_check_passed = True + + if args.comparison_directory is not None: + if args.output_directory is None: + print( + "[ ERR ] Please specify an output directory with the -d flag.", + file=sys.stderr, + ) + raise ValueError( + "Cmoparison directory is specified, but output directory isn't" + ) + else: + collector = DeviceInfoCollector() + collector.dump(args.output_directory) + if collector.compare_device_lists( + args.comparison_directory, args.output_directory + ): + print("[ OK ] Devices match!") + + # dump (no checks) if only output_directory is specified + if args.output_directory is not None and args.comparison_directory is None: + DeviceInfoCollector().dump(args.output_directory) + + if args.do_fwts_check: + tester = FwtsTester() + if tester.is_fwts_supported() and not tester.fwts_log_check_passed( + args.output_directory + ): + fwts_passed = False + else: + print("[ OK ] fwts checks passed!") + + if args.do_service_check: + failed_services = get_failed_services() + if len(failed_services) > 0: + print( + "These services failed: {}".format("\n".join(failed_services)), + file=sys.stderr, + ) + service_check_passed = False + else: + print("[ OK ] Didn't find any failed system services!") + + if args.do_renderer_check: + tester = HardwareRendererTester() + if has_desktop_environment() and tester.has_display_connection(): + # skip renderer test if there's no display + renderer_test_passed = tester.is_hardware_renderer_available() + + if ( + fwts_passed + and device_comparison_passed + and renderer_test_passed + and service_check_passed + ): + return 0 + else: + return 1 + + +if __name__ == "__main__": + return_code = main() + exit(return_code) diff --git a/providers/base/bin/wwan_tests.py b/providers/base/bin/wwan_tests.py index 4e33f09e00..92536ae6db 100755 --- a/providers/base/bin/wwan_tests.py +++ b/providers/base/bin/wwan_tests.py @@ -334,7 +334,7 @@ def _ping_test(if_name): class ThreeGppConnection: - def invoked(self): + def register_argument(self): parser = argparse.ArgumentParser() parser.add_argument( "hw_id", type=str, help="The hardware ID of the modem" @@ -353,22 +353,25 @@ def invoked(self): default=30, help="delay before ping test", ) - args = parser.parse_args(sys.argv[2:]) + return parser.parse_args(sys.argv[2:]) - mm = MMCLI() - mm_id = mm.equipment_id_to_mm_id(args.hw_id) - wwan_control_if = mm.get_primary_port(mm_id) + def invoked(self): + + args = self.register_argument() ret_code = 1 try: - _create_3gpp_connection(wwan_control_if, args.apn) - _wwan_radio_on() - time.sleep(args.wwan_setup_time) - ret_code = _ping_test(args.wwan_net_if) + with WWANTestCtx(args.hw_id, True, True) as ctx: + wwan_control_if = ctx.mm_obj.get_primary_port( + str(ctx.modem_idx) + ) + _create_3gpp_connection(wwan_control_if, args.apn) + time.sleep(args.wwan_setup_time) + ret_code = _ping_test(args.wwan_net_if) except subprocess.SubprocessError: pass _destroy_3gpp_connection() - _wwan_radio_off() + sys.exit(ret_code) @@ -453,25 +456,33 @@ def invoked(self): class Resources: - def invoked(self): + def register_arguments(self): parser = argparse.ArgumentParser() parser.add_argument( "--use-cli", action="store_true", help="Use mmcli for all calls rather than dbus", ) - args = parser.parse_args(sys.argv[2:]) + return parser.parse_args(sys.argv[2:]) + + def invoked(self): + args = self.register_arguments() if args.use_cli: mm = MMCLI() else: mm = MMDbus() + for m in mm.get_modem_ids(): print("mm_id: {}".format(m)) print("hw_id: {}".format(mm.get_equipment_id(m))) print("manufacturer: {}".format(mm.get_manufacturer(m))) print("model: {}".format(mm.get_model_name(m))) - print("firmware_revision: {}".format(mm.get_firmware_revision(m))) - print("hardware_revision: {}".format(mm.get_hardware_revision(m))) + print( + "firmware_revision: {}".format(mm.get_firmware_revision(m)) + ) + print( + "hardware_revision: {}".format(mm.get_hardware_revision(m)) + ) print() diff --git a/providers/base/tests/test_camera_test.py b/providers/base/tests/test_camera_test.py index c303bc5e24..87c645131a 100644 --- a/providers/base/tests/test_camera_test.py +++ b/providers/base/tests/test_camera_test.py @@ -353,7 +353,7 @@ def test_image(self): } CameraTest.image(mock_camera) - self.assertEqual(mock_camera._still_image_helper.call_count, 1) + self.assertEqual(mock_camera._capture_image.call_count, 1) def test_image_without_output(self): mock_camera = MagicMock() @@ -366,32 +366,32 @@ def test_image_without_output(self): with patch("tempfile.NamedTemporaryFile"): CameraTest.image(mock_camera) - self.assertEqual(mock_camera._still_image_helper.call_count, 1) + self.assertEqual(mock_camera._capture_image.call_count, 1) - def test_still_image_helper(self): + def test_capture_image_helper(self): mock_camera = MagicMock() mock_camera._capture_image_fswebcam.return_value = True mock_camera._display_image.return_value = True mock_camera.headless = False - CameraTest._still_image_helper( + CameraTest._capture_image( mock_camera, "/tmp/test.jpg", 640, 480, "YUYV" ) self.assertEqual(mock_camera._capture_image_fswebcam.call_count, 1) self.assertEqual(mock_camera._display_image.call_count, 1) - def test_still_image_headless(self): + def test_capture_image_headless(self): mock_camera = MagicMock() mock_camera._capture_image_fswebcam.return_value = True mock_camera.headless = True - CameraTest._still_image_helper( + CameraTest._capture_image( mock_camera, "/tmp/test.jpg", 640, 480, "YUYV" ) self.assertEqual(mock_camera._display_image.call_count, 0) - def test_still_image_helper_fswebcam_fails(self): + def test_capture_image_helper_fswebcam_fails(self): mock_camera = MagicMock() mock_camera._capture_image_fswebcam.return_value = False - CameraTest._still_image_helper( + CameraTest._capture_image( mock_camera, "/tmp/test.jpg", 640, 480, "YUYV" ) self.assertEqual(mock_camera._capture_image_gstreamer.call_count, 1) @@ -514,7 +514,7 @@ def test_resolutions(self): self.assertEqual(mock_camera._get_default_format.call_count, 1) self.assertEqual(mock_camera._save_debug_image.call_count, 1) - self.assertEqual(mock_camera._still_image_helper.call_count, 2) + self.assertEqual(mock_camera._capture_image.call_count, 2) self.assertEqual(mock_camera._validate_image.call_count, 2) # Test that the function also works with no output @@ -545,7 +545,7 @@ def test_save_debug_image(self, mock_exists): CameraTest._save_debug_image( mock_camera, format, "/dev/video0", "/tmp" ) - self.assertEqual(mock_camera._still_image_helper.call_count, 1) + self.assertEqual(mock_camera._capture_image.call_count, 1) @patch("camera_test.os.path.exists") def test_save_debug_image_fails_if_path_not_exists(self, mock_exists): diff --git a/providers/base/tests/test_reboot_check_test.py b/providers/base/tests/test_reboot_check_test.py new file mode 100644 index 0000000000..f179da53e4 --- /dev/null +++ b/providers/base/tests/test_reboot_check_test.py @@ -0,0 +1,371 @@ +import shutil +from shlex import split as sh_split +from unittest.mock import MagicMock, mock_open, patch +import reboot_check_test as RCT +import unittest +import os +import typing as T +import subprocess as sp + + +def do_nothing(args: T.List[str], **kwargs): + return sp.CompletedProcess(args, 0, "".encode(), "".encode()) + + +class UnitySupportParserTests(unittest.TestCase): + def setUp(self): + self.tester = RCT.HardwareRendererTester() + + def test_parse_ok_unity_support_string(self): + OK_UNITY_STRING = """\ + OpenGL vendor string: Intel + OpenGL renderer string: Mesa Intel(R) UHD Graphics (ICL GT1) + OpenGL version string: 4.6 (Compatibility Profile) Mesa 23.2.1 + + Not software rendered: \x1B[033myes\x1B[0m + Not blacklisted: \x1B[033myes\x1B[0m + GLX fbconfig: \x1B[033myes\x1B[0m + GLX texture from pixmap: \x1B[033myes\x1B[0m + GL npot or rect textures: \x1B[033myes\x1B[0m + GL vertex program: \x1B[033myes\x1B[0m + GL fragment program: \x1B[033myes\x1B[0m + GL vertex buffer object: \x1B[033mno\x1B[0m + GL framebuffer object: \x1B[033myes\x1B[0m + GL version is 1.4+: \x1B[033myes\x1B[0m + + Unity 3D supported: \x1B[033myes\x1B[0m + """ + + expected = { + "OpenGL vendor string": "Intel", + "OpenGL renderer string": "Mesa Intel(R) UHD Graphics (ICL GT1)", + "OpenGL version string": "4.6 (Compatibility Profile) Mesa 23.2.1", + "Not software rendered": "yes", + "Not blacklisted": "yes", + "GLX fbconfig": "yes", + "GLX texture from pixmap": "yes", + "GL npot or rect textures": "yes", + "GL vertex program": "yes", + "GL fragment program": "yes", + "GL vertex buffer object": "no", + "GL framebuffer object": "yes", + "GL version is 1.4+": "yes", + "Unity 3D supported": "yes", + } + + actual = self.tester.parse_unity_support_output(OK_UNITY_STRING) + self.assertDictEqual(expected, actual) + + def test_parse_bad_unity_support_string(self): + BAD_UNITY_STRING = """ + OpenGL vendor string Intel + OpenGL renderer string: Mesa Intel(R) UHD Graphics (ICL GT1) + OpenGL version string 4.6 (Compatibility Profile) Mesa 23.2.1-1ubuntu + GL version is 1.4+% \x1B[033myes\x1B[0m + """ + actual = self.tester.parse_unity_support_output(BAD_UNITY_STRING) + + expected = { + "OpenGL renderer string": "Mesa Intel(R) UHD Graphics (ICL GT1)", + } + + self.assertDictEqual(expected, actual) + + ARBITRARY_STRING = "askjaskdnasdn" + # should return empty dict if input string literally doesn't make sense + self.assertEqual( + self.tester.parse_unity_support_output(ARBITRARY_STRING), + {}, + ) + + +class DisplayConnectionTests(unittest.TestCase): + + def setUp(self) -> None: + self.tester = RCT.HardwareRendererTester() + + def test_display_check_happy_path(self): + with patch( + "os.listdir", return_value=["fakeCard0", "fakeCard1"] + ), patch( + "builtins.open", + new_callable=mock_open, + read_data="connected", + ): + self.assertTrue(self.tester.has_display_connection()) + + def test_display_check_no_display_path(self): + with patch("os.listdir", return_value=["version"]): + self.assertFalse(self.tester.has_display_connection()) + with patch( + "os.listdir", return_value=["fakeCard0", "fakeCard1"] + ), patch( + "builtins.open", + new_callable=mock_open, + read_data="not connected", + ): + self.assertFalse(self.tester.has_display_connection()) + + @patch( + "reboot_check_test.HardwareRendererTester.parse_unity_support_output" + ) + @patch("subprocess.run") + def test_is_hardware_renderer_available( + self, + mock_run: MagicMock, + mock_parse: MagicMock, + ): + mock_run.side_effect = do_nothing + mock_parse.return_value = { + "Not software rendered": "yes", + } + tester = RCT.HardwareRendererTester() + self.assertTrue(tester.is_hardware_renderer_available()) + + @patch( + "reboot_check_test.HardwareRendererTester.parse_unity_support_output" + ) + @patch("subprocess.run") + def test_is_hardware_renderer_available_fail( + self, + mock_run: MagicMock, + mock_parse: MagicMock, + ): + + mock_run.side_effect = lambda _: sp.CompletedProcess( + [], 1, "".encode(), "".encode() + ) + tester = RCT.HardwareRendererTester() + self.assertFalse(tester.is_hardware_renderer_available()) + + mock_run.reset_mock() + mock_run.side_effect = do_nothing + mock_parse.return_value = { + "Not software rendered": "no", + } + tester = RCT.HardwareRendererTester() + self.assertFalse(tester.is_hardware_renderer_available()) + + +class InfoDumpTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.temp_output_dir = "{}/temp_output_dir".format(os.getcwd()) + cls.temp_comparison_dir = "{}/temp_comparison_dir".format(os.getcwd()) + + def tearDown(self): + shutil.rmtree(self.temp_output_dir, ignore_errors=True) + shutil.rmtree(self.temp_comparison_dir, ignore_errors=True) + + def mock_run(self, args: T.List[str], **_) -> sp.CompletedProcess: + stdout = "" + if args[0] == "iw": + stdout = """\ + addr some address + Interface some interface + ssid some ssid + """ + elif args[0] == "checkbox-support-lsusb": + stdout = """\ + usb1 + usb2 + usb3 + """ + elif args[0] == "lspci": + stdout = """\ + pci1 + pci2 + pci3 + """ + else: + raise Exception("Unexpected use of this mock") + + return sp.CompletedProcess(args, 0, stdout.encode(), "".encode()) + + @patch("subprocess.run") + def test_info_dump_only_happy_path(self, mock_run: MagicMock): + # wrap over run's return value + mock_run.side_effect = self.mock_run + RCT.DeviceInfoCollector().dump(self.temp_output_dir) + + @patch("subprocess.run") + def test_info_dump_and_comparison_happy_path(self, mock_run: MagicMock): + mock_run.side_effect = self.mock_run + + collector = RCT.DeviceInfoCollector() + + collector.dump(self.temp_comparison_dir) + collector.dump(self.temp_output_dir) + + self.assertTrue( + collector.compare_device_lists( + self.temp_comparison_dir, self.temp_output_dir + ) + ) + + # required + with open( + "{}/{}_log".format( + self.temp_comparison_dir, + RCT.DeviceInfoCollector.Device.WIRELESS, + ), + "w", + ) as f: + f.write("extra text that shouldn't be there") + + self.assertFalse( + collector.compare_device_lists( + self.temp_comparison_dir, self.temp_output_dir + ) + ) + + collector.dump(self.temp_comparison_dir) + + # optional + with open( + "{}/{}_log".format( + self.temp_comparison_dir, + RCT.DeviceInfoCollector.Device.DRM, + ), + "w", + ) as f: + f.write("extra text that shouldn't be there") + + self.assertTrue( + collector.compare_device_lists( + self.temp_comparison_dir, self.temp_output_dir + ) + ) + + +class FailedServiceCheckerTests(unittest.TestCase): + + @patch("subprocess.run") + def test_get_failed_services_happy_path(self, mock_run: MagicMock): + mock_run.return_value = sp.CompletedProcess( + [], 0, "".encode(), "".encode() + ) + self.assertEqual(RCT.get_failed_services(), []) + + @patch("subprocess.run") + def test_get_failed_services_with_failed_services( + self, mock_run: MagicMock + ): + mock_run.return_value = sp.CompletedProcess( + [], + 0, + "snap.checkbox.agent.service loaded failed failed Service\ + for snap applictaion checkbox.agent".encode(), + "", + ) + self.assertEqual( + RCT.get_failed_services(), [mock_run.return_value.stdout.decode()] + ) + + +class MainFunctionTests(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.tmp_output_dir = "{}/temp_output_dir".format(os.getcwd()) + cls.tmp_comparison_dir = "{}/temp_comparison_dir".format(os.getcwd()) + + def tearDown(self): + shutil.rmtree(self.tmp_output_dir, ignore_errors=True) + shutil.rmtree(self.tmp_comparison_dir, ignore_errors=True) + + @patch("subprocess.run") + def test_partial_main(self, mock_run: MagicMock): + # this test only validates the main function logic + # (if it picks out the correct tests to run) + mock_run.side_effect = do_nothing + + with patch( + "sys.argv", + sh_split("reboot_check_test.py -d {}".format(self.tmp_output_dir)), + ): + RCT.main() + self.assertEqual( + mock_run.call_count, + len(RCT.DeviceInfoCollector.DEFAULT_DEVICES["required"]), + ) + + mock_run.reset_mock() + + with patch( + "sys.argv", + sh_split( + 'reboot_check_test.py -d "{}" -c "{}"'.format( + self.tmp_output_dir, self.tmp_comparison_dir + ) + ), + ), patch( + "reboot_check_test.DeviceInfoCollector.compare_device_lists" + ) as mock_compare: + RCT.main() + + self.assertEqual( + mock_run.call_count, + len(RCT.DeviceInfoCollector.DEFAULT_DEVICES["required"]), + ) # only lspci, lsusb, iw calls + self.assertEqual(mock_compare.call_count, 1) + + @patch("subprocess.run") + def test_main_function_full(self, mock_run: MagicMock): + mock_run.side_effect = do_nothing + # Full suite + with patch( + "sys.argv", + sh_split( + 'reboot_check_test.py -d "{}" -c "{}" -f -s -g'.format( + self.tmp_output_dir, self.tmp_comparison_dir + ) + ), + ), patch( + "reboot_check_test.DeviceInfoCollector.compare_device_lists" + ) as mock_compare, patch( + "reboot_check_test.FwtsTester.is_fwts_supported" + ) as mock_is_fwts_supported: + mock_is_fwts_supported.return_value = True + mock_compare.return_value = True + + RCT.main() + + self.assertTrue(mock_is_fwts_supported.called) + + expected_commands = { + "systemctl", + "sleep_test_log_check.py", + "fwts", + } + + actual = set() + for call in mock_run.call_args_list: + # [0] takes the 1st from (args, kwargs, ) = call, + # then take tha actual list from args + # then take the 1st element, which is the command name + actual.add(call[0][0][0]) + + # <= is an overloaded operator for sets + # that checks the isSubset relation + self.assertLessEqual( + expected_commands, actual, "should be a subset" + ) + + with patch( + "reboot_check_test.get_failed_services" + ) as mock_get_failed_services: + mock_get_failed_services.return_value = [ + "failed service1", + "failed service2", + ] + self.assertEqual(RCT.main(), 1) + + def test_only_comparison_is_specified(self): + with patch( + "sys.argv", + sh_split( + 'reboot_check_test.py -c "{}"'.format(self.tmp_output_dir) + ), + ), self.assertRaises(ValueError): + RCT.main() diff --git a/providers/base/tests/test_wwan_tests.py b/providers/base/tests/test_wwan_tests.py index 06ae054c2c..a40ba0100f 100644 --- a/providers/base/tests/test_wwan_tests.py +++ b/providers/base/tests/test_wwan_tests.py @@ -88,12 +88,16 @@ def test_invoked_with_mmcli(self, mock_mmcli): with redirect_stdout(StringIO()): wwan_tests.Resources().invoked() - self.assertTrue(mock_mmcli.called) - self.assertTrue(mmcli_instance.get_equipment_id.called) - self.assertTrue(mmcli_instance.get_manufacturer.called) - self.assertTrue(mmcli_instance.get_model_name.called) - self.assertTrue(mmcli_instance.get_firmware_revision.called) - self.assertTrue(mmcli_instance.get_hardware_revision.called) + self.assertEqual(mock_mmcli.call_count, 1) + self.assertEqual(mmcli_instance.get_equipment_id.call_count, 1) + self.assertEqual(mmcli_instance.get_manufacturer.call_count, 1) + self.assertEqual(mmcli_instance.get_model_name.call_count, 1) + self.assertEqual( + mmcli_instance.get_firmware_revision.call_count, 1 + ) + self.assertEqual( + mmcli_instance.get_hardware_revision.call_count, 1 + ) @patch("wwan_tests.MMDbus") def test_invoked_with_mmdbus(self, mock_mmdbus): @@ -105,12 +109,16 @@ def test_invoked_with_mmdbus(self, mock_mmdbus): with redirect_stdout(StringIO()): wwan_tests.Resources().invoked() - self.assertTrue(mock_mmdbus.called) - self.assertTrue(mmdbus_instance.get_equipment_id.called) - self.assertTrue(mmdbus_instance.get_manufacturer.called) - self.assertTrue(mmdbus_instance.get_model_name.called) - self.assertTrue(mmdbus_instance.get_firmware_revision.called) - self.assertTrue(mmdbus_instance.get_hardware_revision.called) + self.assertEqual(mock_mmdbus.call_count, 1) + self.assertEqual(mmdbus_instance.get_equipment_id.call_count, 1) + self.assertEqual(mmdbus_instance.get_manufacturer.call_count, 1) + self.assertEqual(mmdbus_instance.get_model_name.call_count, 1) + self.assertEqual( + mmdbus_instance.get_firmware_revision.call_count, 1 + ) + self.assertEqual( + mmdbus_instance.get_hardware_revision.call_count, 1 + ) class TestCommonFunctions(unittest.TestCase): @@ -264,3 +272,84 @@ def test_invoked_call_error(self, mock_arg, mock_mmctx, mock_run): stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) + + +class TestThreeGppConnectionTest(unittest.TestCase): + + def test_register_argument(self): + + sys.argv = [ + "wwan_tests.py", + "3gpp-connection", + "hw_id", + "wwan_net_if", + "apn", + "30", + ] + obj_3gppscan = wwan_tests.ThreeGppConnection() + ret_args = obj_3gppscan.register_argument() + self.assertEqual(ret_args.hw_id, "hw_id") + self.assertEqual(ret_args.wwan_net_if, "wwan_net_if") + self.assertEqual(ret_args.apn, "apn") + self.assertEqual(ret_args.wwan_setup_time, 30) + + @patch("wwan_tests._ping_test") + @patch("wwan_tests._destroy_3gpp_connection") + @patch("wwan_tests._create_3gpp_connection") + @patch("wwan_tests.WWANTestCtx") + @patch("wwan_tests.ThreeGppConnection.register_argument") + def test_invoked_successfully( + self, mock_arg, mock_mmctx, mock_create_conn, mock_rm_conn, mock_ping + ): + mock_arg.return_value = argparse.Namespace( + hw_id="2", wwan_net_if="wwan0", apn="internet", wwan_setup_time=0.1 + ) + mock_get_primary_port = Mock() + mmcli_instance = Mock() + mmcli_instance.modem_idx = "0" + mmcli_instance.mm_obj = Mock(get_primary_port=mock_get_primary_port) + mock_mmctx.return_value.__enter__.return_value = mmcli_instance + mock_ping.return_value = 0 + + with redirect_stdout(StringIO()): + with self.assertRaises(SystemExit) as context: + obj_3gppscan = wwan_tests.ThreeGppConnection() + obj_3gppscan.invoked() + + mock_mmctx.assert_called_with("2", True, True) + self.assertEqual(mock_arg.call_count, 1) + self.assertEqual(mock_get_primary_port.call_count, 1) + self.assertEqual(mock_ping.call_count, 1) + self.assertEqual(mock_create_conn.call_count, 1) + self.assertEqual(mock_rm_conn.call_count, 1) + self.assertEqual(context.exception.code, 0) + + @patch("wwan_tests._ping_test") + @patch("wwan_tests._destroy_3gpp_connection") + @patch("wwan_tests._create_3gpp_connection") + @patch("wwan_tests.WWANTestCtx") + @patch("wwan_tests.ThreeGppConnection.register_argument") + def test_invoked_failed_exit_code( + self, mock_arg, mock_mmctx, mock_create_conn, mock_rm_conn, mock_ping + ): + mock_arg.return_value = argparse.Namespace( + hw_id="2", wwan_net_if="wwan0", apn="internet", wwan_setup_time=0.1 + ) + mock_get_primary_port = Mock() + mmcli_instance = Mock() + mmcli_instance.modem_idx = "0" + mmcli_instance.mm_obj = Mock(get_primary_port=mock_get_primary_port) + mock_mmctx.return_value.__enter__.return_value = mmcli_instance + mock_ping.return_value = 1 + + with redirect_stdout(StringIO()): + with self.assertRaises(SystemExit) as context: + obj_3gppscan = wwan_tests.ThreeGppConnection() + obj_3gppscan.invoked() + + mock_mmctx.assert_called_with("2", True, True) + self.assertEqual(mock_arg.call_count, 1) + self.assertEqual(mock_ping.call_count, 1) + self.assertEqual(mock_create_conn.call_count, 1) + self.assertEqual(mock_rm_conn.call_count, 1) + self.assertEqual(context.exception.code, 1) diff --git a/providers/base/units/camera/jobs.pxu b/providers/base/units/camera/jobs.pxu index 279697972a..1845a67412 100644 --- a/providers/base/units/camera/jobs.pxu +++ b/providers/base/units/camera/jobs.pxu @@ -35,7 +35,7 @@ _summary: Webcam video display test for {product_slug} estimated_duration: 120.0 depends: camera/detect command: - camera_test.py display -d /dev/{name} + camera_test.py video -d /dev/{name} _purpose: This test will check that the {product_slug} camera works _steps: @@ -81,7 +81,7 @@ _summary: Webcam still image capture test for {product_slug} estimated_duration: 120.0 depends: camera/detect command: - camera_test.py still -d /dev/{name} + camera_test.py image -d /dev/{name} _purpose: This test will check that the {product_slug} works _steps: diff --git a/providers/base/units/snapd/jobs.pxu b/providers/base/units/snapd/jobs.pxu index c172dda487..2b7e99ffc4 100644 --- a/providers/base/units/snapd/jobs.pxu +++ b/providers/base/units/snapd/jobs.pxu @@ -63,7 +63,7 @@ plugin: shell estimated_duration: 30s category_id: snapd user: root -depends: snapd/reboot-after-snap-refresh-{type}-{name}-to-stable-rev +depends: snapd/snap-refresh-{type}-{name}-to-stable-rev command: path="$PLAINBOX_SESSION_SHARE/{name}_snap_revision_info" snap_update_test.py --verify-refresh --info-path "$path" {name} diff --git a/providers/base/units/stress/boot.pxu b/providers/base/units/stress/boot.pxu index 9f37184c31..72922b6d60 100644 --- a/providers/base/units/stress/boot.pxu +++ b/providers/base/units/stress/boot.pxu @@ -41,7 +41,7 @@ _purpose: This creates baseline data sets which will be considered the master unit: job plugin: shell command: - reboot_check_test.sh -d "$PLAINBOX_SESSION_SHARE/before_reboot" + reboot_check_test.py -d "$PLAINBOX_SESSION_SHARE/before_reboot" environ: LD_LIBRARY_PATH user: root estimated_duration: 1s @@ -90,7 +90,7 @@ unit: job plugin: shell environ: LD_LIBRARY_PATH command: - reboot_check_test.sh -c "$PLAINBOX_SESSION_SHARE/before_reboot" -d "$PLAINBOX_SESSION_SHARE/cold_reboot_cycle1" -s -f + reboot_check_test.py -g -c "$PLAINBOX_SESSION_SHARE/before_reboot" -d "$PLAINBOX_SESSION_SHARE/cold_reboot_cycle1" -s -f user: root flags: preserve-locale estimated_duration: 1.0 @@ -107,7 +107,7 @@ template-unit: job plugin: shell environ: LD_LIBRARY_PATH command: - reboot_check_test.sh -c "$PLAINBOX_SESSION_SHARE/before_reboot" -d "$PLAINBOX_SESSION_SHARE/cold_reboot_cycle{reboot_id}" -s -f + reboot_check_test.py -g -c "$PLAINBOX_SESSION_SHARE/before_reboot" -d "$PLAINBOX_SESSION_SHARE/cold_reboot_cycle{reboot_id}" -s -f user: root flags: preserve-locale estimated_duration: 1.0 @@ -154,7 +154,7 @@ unit: job plugin: shell environ: LD_LIBRARY_PATH command: - reboot_check_test.sh -c "$PLAINBOX_SESSION_SHARE/before_reboot" -d "$PLAINBOX_SESSION_SHARE/warm_reboot_cycle1" -s -f + reboot_check_test.py -g -c "$PLAINBOX_SESSION_SHARE/before_reboot" -d "$PLAINBOX_SESSION_SHARE/warm_reboot_cycle1" -s -f user: root flags: preserve-locale estimated_duration: 1.0 @@ -171,7 +171,7 @@ template-unit: job plugin: shell environ: LD_LIBRARY_PATH command: - reboot_check_test.sh -c "$PLAINBOX_SESSION_SHARE/before_reboot" -d "$PLAINBOX_SESSION_SHARE/warm_reboot_cycle{reboot_id}" -s -f + reboot_check_test.py -g -c "$PLAINBOX_SESSION_SHARE/before_reboot" -d "$PLAINBOX_SESSION_SHARE/warm_reboot_cycle{reboot_id}" -s -f user: root flags: preserve-locale estimated_duration: 1.0 diff --git a/providers/base/units/wwan/jobs.pxu b/providers/base/units/wwan/jobs.pxu index ca269e82e1..6cb68b49ef 100644 --- a/providers/base/units/wwan/jobs.pxu +++ b/providers/base/units/wwan/jobs.pxu @@ -32,9 +32,10 @@ template-unit: job id: wwan/gsm-connection-{manufacturer}-{model}-{hw_id}-auto template-id: wwan/gsm-connection-manufacturer-model-hw_id-auto _summary: Verify a GSM broadband modem can create a data connection +_template-summary: Verify a GSM broadband modem can create a data connection multiple times _purpose: - Any modems discovered by the resource job that list GSM support - will be tested to ensure a data connection can be made. + Any modems discovered by the resource job that list GSM support + will be tested to ensure a data connection can be made. plugin: shell command: BEGIN_CONNECTION_TEST_TS=$(date '+%Y-%m-%d %H:%M:%S') @@ -54,7 +55,14 @@ imports: from com.canonical.plainbox import manifest depends: wwan/check-sim-present-{manufacturer}-{model}-{hw_id}-auto requires: manifest.has_wwan_module == 'True' + manifest.has_sim_card == 'True' snap.name == 'modem-manager' or package.name == 'modemmanager' +_siblings: + [{{ + "id": "wwan/gsm-reconnection-{manufacturer}-{model}-{hw_id}-auto", + "_summary": "Verify a GSM broadband modem can recreate a data connection", + "depends": "wwan/gsm-connection-{manufacturer}-{model}-{hw_id}-auto" + }}] unit: template template-resource: wwan_resource @@ -74,6 +82,7 @@ flags: preserve-locale also-after-suspend preserve-cwd imports: from com.canonical.plainbox import manifest requires: manifest.has_wwan_module == 'True' + manifest.has_sim_card == 'True' snap.name == 'modem-manager' or package.name == 'modemmanager' unit: template @@ -100,6 +109,7 @@ flags: preserve-locale also-after-suspend preserve-cwd imports: from com.canonical.plainbox import manifest requires: manifest.has_wwan_module == 'True' + manifest.has_sim_card == 'True' snap.name == 'modem-manager' or package.name == 'modemmanager' unit: template @@ -122,6 +132,7 @@ depends: wwan/check-sim-present-{manufacturer}-{model}-{hw_id}-auto imports: from com.canonical.plainbox import manifest requires: manifest.has_wwan_module == 'True' + manifest.has_sim_card == 'True' snap.name == 'modem-manager' or package.name == 'modemmanager' id: wwan/detect-manual @@ -140,7 +151,8 @@ flags: also-after-suspend imports: from com.canonical.plainbox import manifest category_id: wwan requires: - manifest.has_wwan_module == 'True' + manifest.has_wwan_module == 'True' + manifest.has_sim_card == 'True' id: wwan/check-sim-present-manual plugin: manual @@ -158,9 +170,10 @@ flags: also-after-suspend imports: from com.canonical.plainbox import manifest category_id: wwan requires: - manifest.has_wwan_module == 'True' + manifest.has_wwan_module == 'True' + manifest.has_sim_card == 'True' depends: - wwan/detect-manual + wwan/detect-manual id: wwan/gsm-connection-manual plugin: manual @@ -190,9 +203,10 @@ flags: also-after-suspend category_id: wwan imports: from com.canonical.plainbox import manifest requires: - manifest.has_wwan_module == 'True' + manifest.has_wwan_module == 'True' + manifest.has_sim_card == 'True' depends: - wwan/check-sim-present-manual + wwan/check-sim-present-manual id: wwan/gsm-connection-interrupted-manual plugin: manual @@ -230,6 +244,7 @@ flags: also-after-suspend category_id: wwan imports: from com.canonical.plainbox import manifest requires: - manifest.has_wwan_module == 'True' + manifest.has_wwan_module == 'True' + manifest.has_sim_card == 'True' depends: - wwan/check-sim-present-manual + wwan/check-sim-present-manual diff --git a/providers/base/units/wwan/manifest.pxu b/providers/base/units/wwan/manifest.pxu index 8420e2e794..42e4cd9512 100644 --- a/providers/base/units/wwan/manifest.pxu +++ b/providers/base/units/wwan/manifest.pxu @@ -8,3 +8,8 @@ unit: manifest entry id: has_wwan_module _name: A WWAN Module value-type: bool + +unit: manifest entry +id: has_sim_card +_name: A working SIM card inserted +value-type: bool diff --git a/providers/base/units/wwan/test-plan.pxu b/providers/base/units/wwan/test-plan.pxu index 0e2959c076..17b7579698 100644 --- a/providers/base/units/wwan/test-plan.pxu +++ b/providers/base/units/wwan/test-plan.pxu @@ -17,8 +17,8 @@ include: # Note these tests require snap calling snap support wwan/detect certification-status=blocker wwan/3gpp-scan-manufacturer-model-hw_id-auto - wwan/gsm-connection-.*-auto certification-status=blocker - wwan/check-sim-present-.*-auto + wwan/gsm-connection-manufacturer-model-hw_id-auto + wwan/check-sim-present-manufacturer-model-hw_id-auto bootstrap_include: wwan_resource @@ -37,9 +37,9 @@ _name: Automated wwan tests (after suspend) _description: Automated wwan tests for Snappy Ubuntu Core devices include: after-suspend-wwan/detect - after-suspend-wwan/check-sim-present-manufacturer-model-hw_id-auto - after-suspend-wwan/3gpp-scan-manufacturer-model-hw_id-auto - after-suspend-wwan/gsm-connection-.*-auto + wwan/check-sim-present-manufacturer-model-hw_id-auto + wwan/3gpp-scan-manufacturer-model-hw_id-auto + wwan/gsm-connection-manufacturer-model-hw_id-auto bootstrap_include: wwan_resource diff --git a/providers/sru/units/sru.pxu b/providers/sru/units/sru.pxu index 8bc344c600..7b314eab76 100644 --- a/providers/sru/units/sru.pxu +++ b/providers/sru/units/sru.pxu @@ -54,7 +54,7 @@ nested_part: mediacard-cert-automated mediacard-automated memory-automated - mobilebroadband-cert-automated + wwan-automated ethernet-cert-automated networking-cert-automated optical-cert-automated @@ -68,6 +68,7 @@ nested_part: before-suspend-reference-cert-full # suspend point after-suspend-reference-cert-full + after-suspend-wwan-automated after-suspend-touchscreen-cert-automated after-suspend-wireless-cert-automated # The following tests should run BEFORE the automated tests. The reboot and diff --git a/tools/lab_dispatch/build_install_deb.py b/tools/lab_dispatch/build_install_deb.py index d580ac3ba2..6f57b52601 100644 --- a/tools/lab_dispatch/build_install_deb.py +++ b/tools/lab_dispatch/build_install_deb.py @@ -77,6 +77,7 @@ def install_local_package(repo_root, deb_name_glob): "DEBIAN_FRONTEND=noninteractive", "apt-get", "--fix-broken", + "--allow-downgrades", "-y", "install", ] diff --git a/tools/lab_dispatch/generic_source.yaml b/tools/lab_dispatch/generic_source.yaml index c8c143d645..35ea95dbc4 100644 --- a/tools/lab_dispatch/generic_source.yaml +++ b/tools/lab_dispatch/generic_source.yaml @@ -1,8 +1,7 @@ job_queue: $INPUT_QUEUE global_timeout: 3600 output_timeout: 1800 -provision_data: - $INPUT_DATA_SOURCE +$INPUT_DATA_SOURCE test_data: attachments: - local: "tools/lab_dispatch/resources/manifest.conf" @@ -11,6 +10,8 @@ test_data: agent: "checkbox.no-manifest.partial.conf" - local: "tools/lab_dispatch/build_install_deb.py" agent: "build_install_deb.py" + - local: "tools/lab_dispatch/launcher_override.conf" + agent: "launcher_override.conf" test_cmds: | #!/usr/bin/env bash @@ -28,23 +29,28 @@ test_data: source install_tools.sh $TOOLS_PATH # ensure device is available before continuing - wait_for_ssh + wait_for_ssh --allow-degraded - _run sudo add-apt-repository ppa:checkbox-dev/edge _run install_packages git python3 python3-pip dpkg-dev + refresh_zapper_if_needed --channel "$INPUT_ZAPPER_CHANNEL" - wait_for_ssh + wait_for_ssh --allow-degraded + _run clean_machine --im-sure + _run sudo add-apt-repository ppa:checkbox-dev/edge _put $RESOURCES_PATH/build_install_deb.py : - _run git clone https://github.com/canonical/checkbox.git - _run git -C checkbox checkout $CHECKBOX_REVISION + + # clone the Checkbox revision without history (easier for slow systems, preserves RAM/storage/networking) + _run git_get_shallow https://github.com/canonical/checkbox.git commit $CHECKBOX_REVISION + + _run wait_for_packages_complete _run python3 build_install_deb.py --clean checkbox/checkbox-ng \ checkbox/checkbox-support checkbox/providers/resource \ checkbox/providers/base checkbox/providers/sru _run sudo systemctl restart checkbox-ng + _run wait_for_packages_complete - git clone https://github.com/canonical/checkbox.git - git -C checkbox checkout $CHECKBOX_REVISION + git_get_shallow https://github.com/canonical/checkbox.git commit $CHECKBOX_REVISION pipx install --spec checkbox/checkbox-ng checkbox-ng # retrieve manifest @@ -62,9 +68,9 @@ test_data: which envsubst || install_packages gettext envsubst < $RESOURCES_PATH/checkbox.no-manifest.partial.conf > checkbox.no-manifest.conf # then insert the manifest entries via the stacker - stacker --output checkbox.conf checkbox.no-manifest.conf $MANIFEST_FILE + stacker --output checkbox.conf checkbox.no-manifest.conf $MANIFEST_FILE $RESOURCES_PATH/launcher_override.conf - wait_for_ssh + wait_for_ssh --allow-degraded check_for_checkbox_service diff --git a/tools/lab_dispatch/resources/checkbox.no-manifest.template.conf b/tools/lab_dispatch/resources/checkbox.no-manifest.template.conf index f5c95c536b..3bb66095f2 100644 --- a/tools/lab_dispatch/resources/checkbox.no-manifest.template.conf +++ b/tools/lab_dispatch/resources/checkbox.no-manifest.template.conf @@ -16,15 +16,10 @@ type = silent [environment] ROUTERS = multiple WPA_BG_SSID = $WPA_BG_SSID -WPA_BG_PSK = $INPUT_PASSWORD_SECRET WPA_N_SSID = $WPA_N_SSID -WPA_N_PSK = $INPUT_PASSWORD_SECRET WPA_AC_SSID = $WPA_AC_SSID -WPA_AC_PSK = $INPUT_PASSWORD_SECRET WPA_AX_SSID = $WPA_AX_SSID -WPA_AX_PSK = $INPUT_PASSWORD_SECRET WPA3_AX_SSID = $WPA3_AX_SSID -WPA3_AX_PSK = $INPUT_PASSWORD_SECRET OPEN_BG_SSID = $OPEN_BG_SSID OPEN_N_SSID = $OPEN_N_SSID OPEN_AC_SSID = $OPEN_AC_SSID