From be54e33833603db841f9cbdaa497a0adcd3e5fbf Mon Sep 17 00:00:00 2001 From: rickwu4444 Date: Fri, 12 Apr 2024 09:57:00 +0800 Subject: [PATCH] Add connect and disconnect as context manager --- .../bin/check_gpio.py | 131 +++++++++++----- .../tests/test_check_gpio.py | 143 ++++++++++++------ .../units/gpio/jobs.pxu | 11 +- .../units/gpio/test-plan.pxu | 1 - 4 files changed, 186 insertions(+), 100 deletions(-) diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/check_gpio.py b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/check_gpio.py index 3dcf91bf1..4d153c221 100755 --- a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/check_gpio.py +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/check_gpio.py @@ -1,6 +1,8 @@ #!/usr/bin/env python3 import argparse import os +from contextlib import contextmanager +# from checkbox_support.snap_utils.snapd import Snapd from checkbox_support.snap_utils.snapd import Snapd from checkbox_support.snap_utils.system import get_gadget_snap import requests @@ -25,10 +27,7 @@ def list_gpio_slots(snapd, gadget_name): for slots in snapd.interfaces()["slots"]: if slots["interface"] == "gpio" and slots["snap"] == gadget_name: gpio_slot[slots["slot"]] = {"number": slots["attrs"]["number"]} - if gpio_slot: - return gpio_slot - else: - raise SystemExit("Error: Can not find any GPIO slot") + return gpio_slot def parse_config(config): @@ -88,51 +87,99 @@ def check_gpio_list(gpio_list, config): SystemExit: If any expected GPIO slot is not defined in the gadget snap. """ - expect_port = parse_config(config) - for gpio in gpio_list.values(): - if gpio["number"] in expect_port: - expect_port.remove(gpio["number"]) - if expect_port: - for gpio_slot in expect_port: - print( - "Error: Slot of GPIO {} is not defined in gadget snap".format( - gpio_slot) + if gpio_list: + expect_port = parse_config(config) + for gpio in gpio_list.values(): + if gpio["number"] in expect_port: + expect_port.remove(gpio["number"]) + if expect_port: + for gpio_slot in expect_port: + print( + "Error: Slot of GPIO {} is not defined in gadget snap". + format(gpio_slot) + ) + raise SystemExit(1) + else: + print("All expected GPIO slots have been defined in gadget snap.") + else: + raise SystemExit("Error: No any GPIO slots existed!") + + +@contextmanager +def interface_test(gpio_slot, gadget_name, timeout=60): + snap = os.environ['SNAP_NAME'] + timeout = int(os.environ.get('SNAPD_TASK_TIMEOUT', timeout)) + try: + connect_interface(gadget_name, + gpio_slot, + snap, + timeout) + yield + finally: + disconnect_interface(gadget_name, + gpio_slot, + snap, + timeout) + + +def connect_interface(gadget_name, + gpio_slot, + snap, + timeout): + """ + Connect GPIO plugs of checkbox to GPIO slots of gadget snap. + + Args: + gpio_slot: A GPIO slot information. + gadget_name: The name of the gadget snap. + + Raises: + SystemExit: If failed to connect any GPIO. + """ + + # Get the snap name of checkbox + print("Attempting connect GPIO to {}:{}".format(gadget_name, gpio_slot)) + try: + Snapd(task_timeout=timeout).connect( + gadget_name, + gpio_slot, + snap, + "gpio" ) + print("Success") + except requests.HTTPError: + print("Failed to connect {}".format(gpio_slot)) raise SystemExit(1) - else: - print("All expected GPIO slots have been defined in gadget snap.") -def connect_gpio(gpio_slots, gadget_name): +def disconnect_interface(gadget_name, + gpio_slot, + snap, + timeout): """ Connect GPIO plugs of checkbox to GPIO slots of gadget snap. Args: - gpio_slots: A dictionary containing GPIO slot information. + gpio_slot: A GPIO slot information. gadget_name: The name of the gadget snap. Raises: SystemExit: If failed to connect any GPIO. """ - exit_code = 0 # Get the snap name of checkbox - snap = os.environ['SNAP_NAME'] - timeout = int(os.environ.get('SNAPD_TASK_TIMEOUT', 60)) - for gpio_num in gpio_slots.keys(): - print("Attempting connect gpio to {}:{}".format(gadget_name, gpio_num)) - try: - Snapd(task_timeout=timeout).connect( - gadget_name, - gpio_num, - snap, - "gpio" - ) - print("Success") - except requests.HTTPError: - print("Failed to connect {}".format(gpio_num)) - exit_code = 1 - if exit_code == 1: + print("Attempting disconnect GPIO slot {}:{}". + format(gadget_name, gpio_slot)) + try: + Snapd(task_timeout=timeout).disconnect( + gadget_name, + gpio_slot, + snap, + "gpio" + ) + print("Success") + except requests.HTTPError: + print("Failed to disconnect {}".format(gpio_slot)) raise SystemExit(1) @@ -158,7 +205,7 @@ def main(): parser = argparse.ArgumentParser() subparsers = parser.add_subparsers( dest='action', - help="Action in check-gpio, check-node, connect and dump", + help="Action in check-gpio, check-node and dump", ) check_gpio_subparser = subparsers.add_parser("check-gpio") check_gpio_subparser.add_argument( @@ -176,9 +223,12 @@ def main(): required=True, help="GPIO number to check if node exported", ) - subparsers.add_parser( - "connect", - help="Connect checkbox GPIO plug and gadget GPIO slots " + check_node_subparser.add_argument( + "-s", + "--slot", + type=str, + required=True, + help="GPIO slot to connect.", ) subparsers.add_parser( "dump", @@ -190,8 +240,6 @@ def main(): gpio_slots = list_gpio_slots(snapd, gadget_name) if args.action == "check-gpio": check_gpio_list(gpio_slots, args.config) - if args.action == "connect": - connect_gpio(gpio_slots, gadget_name) if args.action == "dump": for x in gpio_slots: print( @@ -200,7 +248,8 @@ def main(): ) ) if args.action == "check-node": - check_node(args.num) + with interface_test(args.slot, gadget_name): + check_node(args.num) if __name__ == "__main__": diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_check_gpio.py b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_check_gpio.py index be4dd8f3a..95fd5c96c 100755 --- a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_check_gpio.py +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/tests/test_check_gpio.py @@ -3,6 +3,7 @@ from unittest.mock import patch, MagicMock, call import requests import check_gpio +import os class TestCheckGpio(unittest.TestCase): @@ -65,12 +66,11 @@ def test_list_gpio_slots_not_exist(self): "attrs": {}}, ] } - expected_result = "Error: Can not find any GPIO slot" - with self.assertRaises(SystemExit) as err: - check_gpio.list_gpio_slots(snapd_mock, gadget_name) - self.assertEqual(err.exception.args[0], expected_result) + expected_result = {} + self.assertEqual(check_gpio.list_gpio_slots(snapd_mock, gadget_name), + expected_result) - @patch('builtins.print') # Mock print function to prevent actual printing + @patch('builtins.print') def test_check_gpio_list_all_defined(self, mock_print): gpio_list = { 1: {"number": 499}, @@ -85,7 +85,7 @@ def test_check_gpio_list_all_defined(self, mock_print): mock_print.assert_called_with( "All expected GPIO slots have been defined in gadget snap.") - @patch('builtins.print') # Mock print function to prevent actual printing + @patch('builtins.print') def test_check_gpio_list_missing(self, mock_print): gpio_list = { 1: {"number": 499}, @@ -105,65 +105,91 @@ def test_check_gpio_list_missing(self, mock_print): # Assert that SystemExit is raised with exit code 1 self.assertEqual(context.exception.code, 1) - @patch('check_gpio.os.environ') + @patch('builtins.print') @patch('check_gpio.Snapd') - def test_connect_gpio_success(self, mock_snapd, mock_environ): - mock_environ.__getitem__.side_effect = lambda x: { - 'SNAP_NAME': 'checkbox_snap', - 'SNAPD_TASK_TIMEOUT': '30' - }[x] + def test_connect_interface_success(self, mock_snapd, mock_print): mock_snapd.return_value.connect.side_effect = None - - gpio_slots = { - "gpio-499": {"number": 499}, - "gpio-500": {"number": 500} - } + gpio_slot = "gpio-499" gadget_name = "gadget_snap" + snap = "checkbox_snap" + timeout = 30 + check_gpio.connect_interface(gadget_name, + gpio_slot, + snap, + timeout) - check_gpio.connect_gpio(gpio_slots, gadget_name) - - # Assert that connect is called for each GPIO slot expected_calls = [call(gadget_name, - 'gpio-499', - 'checkbox_snap', - 'gpio'), - call(gadget_name, - 'gpio-500', - 'checkbox_snap', - 'gpio') - ] + gpio_slot, + snap, + 'gpio')] mock_snapd.return_value.connect.assert_has_calls(expected_calls) + mock_print.assert_called_with("Success") - @patch('check_gpio.os.environ') + @patch('builtins.print') @patch('check_gpio.Snapd') - def test_connect_gpio_fail(self, mock_snapd, mock_environ): - mock_environ.__getitem__.side_effect = lambda x: { - 'SNAP_NAME': 'checkbox_snap', - 'SNAPD_TASK_TIMEOUT': '30' - }[x] + def test_connect_interface_fail(self, mock_snapd, mock_print): mock_snapd.return_value.connect.side_effect = requests.HTTPError - - gpio_slots = { - "gpio-499": {"number": 499}, - "gpio-500": {"number": 500} - } + gpio_slot = "gpio-499" gadget_name = "gadget_snap" + snap = "checkbox_snap" + timeout = 30 with self.assertRaises(SystemExit) as err: - check_gpio.connect_gpio(gpio_slots, gadget_name) + check_gpio.connect_interface(gadget_name, + gpio_slot, + snap, + timeout) - # Assert that connect is called for each GPIO slot expected_calls = [call(gadget_name, - 'gpio-499', - 'checkbox_snap', - 'gpio'), - call(gadget_name, - 'gpio-500', - 'checkbox_snap', + gpio_slot, + snap, 'gpio')] mock_snapd.return_value.connect.assert_has_calls(expected_calls) + mock_print.assert_called_with("Failed to connect gpio-499") self.assertEqual(err.exception.code, 1) - @patch('builtins.print') # Mock print function to prevent actual printing + @patch('builtins.print') + @patch('check_gpio.Snapd') + def test_disconnect_interface_success(self, mock_snapd, mock_print): + mock_snapd.return_value.disconnect.side_effect = None + gpio_slot = "gpio-499" + gadget_name = "gadget_snap" + snap = "checkbox_snap" + timeout = 30 + check_gpio.disconnect_interface(gadget_name, + gpio_slot, + snap, + timeout) + + expected_calls = [call(gadget_name, + gpio_slot, + snap, + 'gpio')] + mock_snapd.return_value.disconnect.assert_has_calls(expected_calls) + mock_print.assert_called_with("Success") + + @patch('builtins.print') + @patch('check_gpio.Snapd') + def test_disconnect_interface_fail(self, mock_snapd, mock_print): + mock_snapd.return_value.disconnect.side_effect = requests.HTTPError + gpio_slot = "gpio-499" + gadget_name = "gadget_snap" + snap = "checkbox_snap" + timeout = 30 + with self.assertRaises(SystemExit) as err: + check_gpio.disconnect_interface(gadget_name, + gpio_slot, + snap, + timeout) + + expected_calls = [call(gadget_name, + gpio_slot, + snap, + 'gpio')] + mock_snapd.return_value.disconnect.assert_has_calls(expected_calls) + mock_print.assert_called_with("Failed to disconnect gpio-499") + self.assertEqual(err.exception.code, 1) + + @patch('builtins.print') def test_check_node_exists(self, mock_print): # Mocking os.path.exists to return True with patch('os.path.exists', return_value=True): @@ -180,6 +206,27 @@ def test_check_node_not_exist(self): self.assertEqual(context.exception.args[0], "GPIO node of 499 not exist!") + @patch.dict(os.environ, {'SNAP_NAME': 'checkbox_snap'}) + @patch.dict(os.environ, {'SNAPD_TASK_TIMEOUT': '30'}) + @patch('check_gpio.connect_interface') + @patch('check_gpio.disconnect_interface') + def test_interface_test(self, + mock_disconnect, + mock_connect): + gadget_name = "gadget" + gpio_slot = "gpio-499" + mock_connect.side_effect = None + mock_disconnect.side_effect = None + with check_gpio.interface_test(gpio_slot, gadget_name): + mock_connect.assert_called_once_with(gadget_name, + gpio_slot, + 'checkbox_snap', + 30) + mock_disconnect.assert_called_once_with(gadget_name, + gpio_slot, + 'checkbox_snap', + 30) + if __name__ == '__main__': unittest.main() diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gpio/jobs.pxu b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gpio/jobs.pxu index 8205aa8d8..102cadfd5 100644 --- a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gpio/jobs.pxu +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gpio/jobs.pxu @@ -12,15 +12,6 @@ _description: Sprate by comma, and also colon to define a range of ports command: check_gpio.py check-gpio -c $EXPECTED_GADGET_GPIO -id: ce-oem-gpio/connect-slots -category_id: com.canonical.certification::gpio -plugin: shell -user: root -estimated_duration: 5 -_summary: Connect all GPIO slots with checkbox. -_purpose: Check if GPIO slots can connect with checkbox plug. -command: check_gpio.py connect - id: ce-oem-gpio-gadget-slots _summary: Generates a GPIO list that defined in the gadget snap _description: @@ -43,4 +34,4 @@ user: root category_id: com.canonical.certification::gpio estimated_duration: 5s flags: also-after-suspend -command: check_gpio.py check-node -n {gpio_number} +command: check_gpio.py check-node -n {gpio_number} -s {slot} diff --git a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gpio/test-plan.pxu b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gpio/test-plan.pxu index 097141420..df61bf292 100644 --- a/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gpio/test-plan.pxu +++ b/contrib/checkbox-ce-oem/checkbox-provider-ce-oem/units/gpio/test-plan.pxu @@ -21,5 +21,4 @@ bootstrap_include: ce-oem-gpio-gadget-slots include: ce-oem-gpio/check-slots - ce-oem-gpio/connect-slots ce-oem-gpio/node-export-test