diff --git a/cloudinit/config/cc_rpi_connect.py b/cloudinit/config/cc_rpi_connect.py index b1b234f4c06..d084bbe3eb2 100644 --- a/cloudinit/config/cc_rpi_connect.py +++ b/cloudinit/config/cc_rpi_connect.py @@ -19,22 +19,17 @@ "id": "cc_rpi_connect", "distros": ["raspberry-pi-os"], "frequency": PER_INSTANCE, - "activate_by_schema_keys": [ - ENABLE_RPI_CONNECT_KEY - ] + "activate_by_schema_keys": [ENABLE_RPI_CONNECT_KEY], } + def configure_rpi_connect(enable: bool) -> None: LOG.debug(f"Configuring rpi-connect: {enable}") num = 0 if enable else 1 try: - subp.subp([ - "/usr/bin/raspi-config", - "do_rpi_connect", - str(num) - ]) + subp.subp(["/usr/bin/raspi-config", "do_rpi_connect", str(num)]) except subp.ProcessExecutionError as e: LOG.error(f"Failed to configure rpi-connect: {e}") @@ -47,4 +42,6 @@ def handle(name: str, cfg: Config, cloud: Cloud, args: list) -> None: if isinstance(enable, bool): configure_rpi_connect(enable) else: - LOG.warning(f"Invalid value for {ENABLE_RPI_CONNECT_KEY}: {enable}") + LOG.warning( + f"Invalid value for {ENABLE_RPI_CONNECT_KEY}: " + str(enable) + ) diff --git a/cloudinit/config/cc_rpi_interfaces.py b/cloudinit/config/cc_rpi_interfaces.py index 6fc3509cf65..9f5e4929479 100644 --- a/cloudinit/config/cc_rpi_interfaces.py +++ b/cloudinit/config/cc_rpi_interfaces.py @@ -19,7 +19,7 @@ "i2c": "do_i2c", "serial": "do_serial", "onewire": "do_onewire", - "remote_gpio": "do_rgpio" + "remote_gpio": "do_rgpio", } RASPI_CONFIG_SERIAL_CONS_FN = "do_serial_cons" RASPI_CONFIG_SERIAL_HW_FN = "do_serial_hw" @@ -28,28 +28,25 @@ "id": "cc_rpi_interfaces", "distros": ["raspberry-pi-os"], "frequency": PER_INSTANCE, - "activate_by_schema_keys": [ - RPI_INTERFACES_KEY - ] + "activate_by_schema_keys": [RPI_INTERFACES_KEY], } + # TODO: test def require_reboot(cfg: Config) -> None: cfg["power_state"] = cfg.get("power_state", {}) cfg["power_state"]["mode"] = cfg["power_state"].get("mode", "reboot") cfg["power_state"]["condition"] = True + def is_pifive() -> bool: try: - subp.subp([ - "/usr/bin/raspi-config", - "nonint", - "is_pifive" - ]) + subp.subp(["/usr/bin/raspi-config", "nonint", "is_pifive"]) return True except subp.ProcessExecutionError: return False + def configure_serial_interface(cfg: dict | bool, instCfg: Config) -> None: enable_console = False enable_hw = False @@ -58,7 +55,7 @@ def configure_serial_interface(cfg: dict | bool, instCfg: Config) -> None: enable_console = cfg.get("console", False) enable_hw = cfg.get("hardware", False) elif isinstance(cfg, bool): - # default to enabling console as if < pi5 + # default to enabling console as if < pi5 # this will also enable the hardware enable_console = cfg @@ -69,48 +66,58 @@ def configure_serial_interface(cfg: dict | bool, instCfg: Config) -> None: enable_hw = True try: - subp.subp([ - "/usr/bin/raspi-config", - "nonint", - RASPI_CONFIG_SERIAL_CONS_FN, - str(0 if enable_console else 1) - ]) - - try: - subp.subp([ + subp.subp( + [ "/usr/bin/raspi-config", "nonint", - RASPI_CONFIG_SERIAL_HW_FN, - str(0 if enable_hw else 1) - ]) + RASPI_CONFIG_SERIAL_CONS_FN, + str(0 if enable_console else 1), + ] + ) + + try: + subp.subp( + [ + "/usr/bin/raspi-config", + "nonint", + RASPI_CONFIG_SERIAL_HW_FN, + str(0 if enable_hw else 1), + ] + ) except subp.ProcessExecutionError as e: LOG.error(f"Failed to configure serial hardware: {e}") - + require_reboot(instCfg) except subp.ProcessExecutionError as e: LOG.error(f"Failed to configure serial console: {e}") - + def configure_interface(iface: str, enable: bool) -> None: - assert iface in SUPPORTED_INTERFACES.keys() \ - and iface != "serial", \ - f"Unsupported interface: {iface}" + assert ( + iface in SUPPORTED_INTERFACES.keys() and iface != "serial" + ), f"Unsupported interface: {iface}" try: - subp.subp([ - "/usr/bin/raspi-config", - "nonint", - SUPPORTED_INTERFACES[iface], - str(0 if enable else 1) - ]) + subp.subp( + [ + "/usr/bin/raspi-config", + "nonint", + SUPPORTED_INTERFACES[iface], + str(0 if enable else 1), + ] + ) except subp.ProcessExecutionError as e: LOG.error(f"Failed to configure {iface}: {e}") + def handle(name: str, cfg: Config, cloud: Cloud, args: list) -> None: if RPI_INTERFACES_KEY not in cfg: return elif not isinstance(cfg[RPI_INTERFACES_KEY], dict): - LOG.warning(f"Invalid value for {RPI_INTERFACES_KEY}: {cfg[RPI_INTERFACES_KEY]}") + LOG.warning( + f"Invalid value for {RPI_INTERFACES_KEY}: " + + cfg[RPI_INTERFACES_KEY] + ) return elif not cfg[RPI_INTERFACES_KEY]: LOG.debug(f"Empty value for {RPI_INTERFACES_KEY}. Skipping...") @@ -126,11 +133,15 @@ def handle(name: str, cfg: Config, cloud: Cloud, args: list) -> None: if key == "serial": if not isinstance(enable, dict) and not isinstance(enable, bool): - LOG.warning(f"Invalid value for {RPI_INTERFACES_KEY}.{key}: {enable}") + LOG.warning( + f"Invalid value for {RPI_INTERFACES_KEY}.{key}: {enable}" + ) continue configure_serial_interface(enable, cfg) if isinstance(enable, bool): configure_interface(key, enable) else: - LOG.warning(f"Invalid value for {RPI_INTERFACES_KEY}.{key}: {enable}") + LOG.warning( + f"Invalid value for {RPI_INTERFACES_KEY}.{key}: {enable}" + ) diff --git a/cloudinit/config/cc_rpi_userdata.py b/cloudinit/config/cc_rpi_userdata.py index 74969b187c3..ca10d1dfa04 100644 --- a/cloudinit/config/cc_rpi_userdata.py +++ b/cloudinit/config/cc_rpi_userdata.py @@ -5,7 +5,8 @@ # This file is part of cloud-init. See LICENSE file for license information. import logging -import os, time +import os +import time import subprocess from cloudinit import subp from cloudinit.cloud import Cloud @@ -23,46 +24,57 @@ meta: MetaSchema = { "id": "cc_rpi_userdata", "distros": ["raspberry-pi-os"], - "frequency": PER_ALWAYS, # Run every boot to trigger setup wizard even when no settings - "activate_by_schema_keys": [ - DISABLE_PIWIZ_KEY, - RPI_USERCONF_KEY - ] + # Run every boot to trigger setup wizard even when no settings + "frequency": PER_ALWAYS, + "activate_by_schema_keys": [DISABLE_PIWIZ_KEY, RPI_USERCONF_KEY], } + def bool_to_str(value: bool | None) -> str: return "Yes" if value else "No" + def get_fwloc_or_default() -> str: fwloc = None try: # Run the command and capture the output fwloc = subp.subp( - ['/usr/lib/raspberrypi-sys-mods/get_fw_loc'], decode='strict') \ - .stdout.strip() + ["/usr/lib/raspberrypi-sys-mods/get_fw_loc"], decode="strict" + ).stdout.strip() # If the output is empty, set the default value if not fwloc: - fwloc = '/boot' + fwloc = "/boot" except subp.ProcessExecutionError: # If the command fails, set the default value - fwloc = '/boot' + fwloc = "/boot" return fwloc -def run_userconf_service(base: str | None, passwd_override: str | None = None) -> bool: + +def run_userconf_service( + base: str | None, passwd_override: str | None = None +) -> bool: try: # reset the TTY device os.system(f"echo 'reset\\r\\n' > {USERCONF_SERVICE_TTY}") time.sleep(1) # Execute the command on different tty - result = subp.subp([ - "openvt", - "-s", "-f", "-w", "-c", USERCONF_SERVICE_TTY[-1], - "--", "/usr/lib/userconf-pi/userconf-service"], + result = subp.subp( + [ + "openvt", + "-s", + "-f", + "-w", + "-c", + USERCONF_SERVICE_TTY[-1], + "--", + "/usr/lib/userconf-pi/userconf-service", + ], timeout=(None if not passwd_override else 10), - decode='strict') - + decode="strict", + ) + if base: try: os.remove(f"{base}/userconf.txt") @@ -94,7 +106,10 @@ def run_userconf_service(base: str | None, passwd_override: str | None = None) - pass return False -def run_service(passwd_override: str | None = None, user_override: str | None = None) -> bool: + +def run_service( + passwd_override: str | None = None, user_override: str | None = None +) -> bool: # Ensure the TTY exists before trying to open it if not os.path.exists(USERCONF_SERVICE_TTY): if not passwd_override: @@ -103,10 +118,13 @@ def run_service(passwd_override: str | None = None, user_override: str | None = else: LOG.debug(f"TTY device {USERCONF_SERVICE_TTY} does not exist.") - # should never happen and not solvable by the user - assert (passwd_override is None and user_override is None) or \ - (passwd_override is not None and user_override is not None), \ - "Internal error: User override is required when password override is provided." + # should never happen and not solvable by the user + assert (passwd_override is None and user_override is None) or ( + passwd_override is not None and user_override is not None + ), ( + "Internal error: User override is required when password " + + "override is provided." + ) base: str | None = None if passwd_override: @@ -129,11 +147,23 @@ def run_service(passwd_override: str | None = None, user_override: str | None = LOG.debug("Userconf-pi service loop finished.") return True -def configure_pizwiz(cfg: Config, disable: bool, passwd_override: str | None, user_override: str | None = None) -> None: - LOG.debug("Configuring piwiz with disable_piwiz=%s, passwd_override=%s, user_override=%s", disable, bool_to_str(passwd_override), bool_to_str(user_override)) + +def configure_pizwiz( + cfg: Config, + disable: bool, + passwd_override: str | None, + user_override: str | None = None, +) -> None: + LOG.debug( + "Configuring piwiz with disable_piwiz=%s, passwd_override=%s, " + + "user_override=%s", + bool_to_str(disable), + bool_to_str(passwd_override != None), + bool_to_str(user_override != None), + ) if disable: - # execute cancel rename script to ensure + # execute cancel rename script to ensure # piwiz isn't started (on desktop) os.system("/usr/bin/cancel-rename pi") else: @@ -149,25 +179,34 @@ def configure_pizwiz(cfg: Config, disable: bool, passwd_override: str | None, us elif "pi" not in cfg["users"]: cfg["users"].append("pi") + def handle(name: str, cfg: Config, cloud: Cloud, args: list) -> None: disable_piwiz: bool = False password_override: str | None = None user_override: str | None = None - if os.path.exists(MODULE_DEACTIVATION_FILE) or not os.path.exists("/usr/lib/userconf-pi"): - LOG.debug("Userconf-Pi: deactivation file detected or userconf-pi not installed. Skipping...") + if os.path.exists(MODULE_DEACTIVATION_FILE) or not os.path.exists( + "/usr/lib/userconf-pi" + ): + LOG.debug( + "Userconf-Pi: deactivation file detected or userconf-pi " + + "not installed. Skipping..." + ) return if RPI_USERCONF_KEY in cfg: # expect it to be a dictionary userconf = cfg[RPI_USERCONF_KEY] - # look over miss configuration to + # look over miss configuration to if isinstance(userconf, dict) and "password" in userconf: password_override = userconf["password"] # user key is optional with default to pi user_override = userconf.get("user", "pi") - LOG.debug(f"Userconf override: user={user_override}, password=") + LOG.debug( + f"Userconf override: user={user_override}, " + + "password=" + ) else: LOG.error(f"Invalid userconf-pi configuration: {userconf}") @@ -175,6 +214,9 @@ def handle(name: str, cfg: Config, cloud: Cloud, args: list) -> None: if isinstance(cfg[DISABLE_PIWIZ_KEY], bool): disable_piwiz = cfg[DISABLE_PIWIZ_KEY] else: - LOG.error(f"Invalid {DISABLE_PIWIZ_KEY} configuration: {cfg[DISABLE_PIWIZ_KEY]}") + LOG.error( + f"Invalid {DISABLE_PIWIZ_KEY} configuration: " + + cfg[DISABLE_PIWIZ_KEY] + ) configure_pizwiz(cfg, disable_piwiz, password_override, user_override) diff --git a/cloudinit/distros/raspberry-pi-os.py b/cloudinit/distros/raspberry-pi-os.py index 7967f8dd114..c0bfa097e1a 100644 --- a/cloudinit/distros/raspberry-pi-os.py +++ b/cloudinit/distros/raspberry-pi-os.py @@ -13,26 +13,33 @@ def __init__(self, name, cfg, paths): super().__init__(name, cfg, paths) def set_keymap(self, layout: str, model: str, variant: str, options: str): - """Currently Raspberry Pi OS sys-mods only supports setting the layout""" + """Currently Raspberry Pi OS sys-mods only supports + setting the layout""" - subp.subp([ - "/usr/lib/raspberrypi-sys-mods/imager_custom", - "set_keymap", - layout - ]) + subp.subp( + [ + "/usr/lib/raspberrypi-sys-mods/imager_custom", + "set_keymap", + layout, + ] + ) def apply_locale(self, locale, out_fn=None, keyname="LANG"): try: - subp.subp([ - "/usr/bin/raspi-config", - "nonint", - "do_change_locale", - f'{locale}' - ]) + subp.subp( + [ + "/usr/bin/raspi-config", + "nonint", + "do_change_locale", + f"{locale}", + ] + ) except subp.ProcessExecutionError: - subp.subp([ - "/usr/bin/raspi-config", - "nonint", - "do_change_locale", - f'{locale}.UTF-8' - ]) + subp.subp( + [ + "/usr/bin/raspi-config", + "nonint", + "do_change_locale", + f"{locale}.UTF-8", + ] + ) diff --git a/tests/unittests/config/test_cc_rpi_connect.py b/tests/unittests/config/test_cc_rpi_connect.py index bb63c17d89a..3368c060ce7 100644 --- a/tests/unittests/config/test_cc_rpi_connect.py +++ b/tests/unittests/config/test_cc_rpi_connect.py @@ -14,7 +14,7 @@ def is_notPi() -> bool: @mock.patch('cloudinit.subp.subp') class TestCCRPiConnect(CiTestCase): - \"""Tests work in progress. Just partially implemented to + \"""Tests work in progress. Just partially implemented to show the idea.\""" @mock.patch('cloudinit.subp.subp') @@ -59,9 +59,10 @@ class TestCCRPiConnectSchema: "config, error_msg", [ ({ENABLE_RPI_CONNECT_KEY: True}, None), - ({ - ENABLE_RPI_CONNECT_KEY: "true" - }, "'true' is not of type 'boolean'") + ( + {ENABLE_RPI_CONNECT_KEY: "true"}, + "'true' is not of type 'boolean'", + ), ], ) def test_schema_validation(self, config, error_msg): diff --git a/tests/unittests/config/test_cc_rpi_interfaces.py b/tests/unittests/config/test_cc_rpi_interfaces.py index e66125b6216..ffd9a8ab9ef 100644 --- a/tests/unittests/config/test_cc_rpi_interfaces.py +++ b/tests/unittests/config/test_cc_rpi_interfaces.py @@ -28,9 +28,9 @@ def test_configure_spi_interface(self, mock_subp): } handle("cc_rpi_interfaces", config, mock.Mock(), []) mock_subp.assert_called_with([ - '/usr/bin/raspi-config', - 'nonint', - SUPPORTED_INTERFACES["spi"], + '/usr/bin/raspi-config', + 'nonint', + SUPPORTED_INTERFACES["spi"], '0']) @mock.patch('cloudinit.subp.subp') @@ -78,13 +78,13 @@ def test_get_enabled_interfaces(self, mock_subp, mock_path_exists): } handle("cc_rpi_interfaces", config, mock.Mock(), []) # Assert all interface enabling commands were called - mock_subp.assert_any_call(['/usr/bin/raspi-config', + mock_subp.assert_any_call(['/usr/bin/raspi-config', 'nonint', 'do_spi', '0']) - mock_subp.assert_any_call(['/usr/bin/raspi-config', + mock_subp.assert_any_call(['/usr/bin/raspi-config', 'nonint', 'do_i2c', '0']) - mock_subp.assert_any_call(['/usr/bin/raspi-config', + mock_subp.assert_any_call(['/usr/bin/raspi-config', 'nonint', 'do_onewire', '0']) - mock_subp.assert_any_call(['/usr/bin/raspi-config', + mock_subp.assert_any_call(['/usr/bin/raspi-config', 'nonint', 'do_rgpio', '0']) """ @@ -95,24 +95,22 @@ class TestCCRPiInterfacesSchema: "config, error_msg", [ ({RPI_INTERFACES_KEY: {"spi": True, "i2c": False}}, None), - ({ - RPI_INTERFACES_KEY: {"spi": "true"} - }, "'true' is not of type 'boolean'"), - ({ - RPI_INTERFACES_KEY: { - "serial": { - "console": True, - "hardware": False + ( + {RPI_INTERFACES_KEY: {"spi": "true"}}, + "'true' is not of type 'boolean'", + ), + ( + { + RPI_INTERFACES_KEY: { + "serial": {"console": True, "hardware": False} } - } - }, None), - ({ - RPI_INTERFACES_KEY: { - "serial": { - "console": 123 - } - } - }, "123 is not of type 'boolean'") + }, + None, + ), + ( + {RPI_INTERFACES_KEY: {"serial": {"console": 123}}}, + "123 is not of type 'boolean'", + ), ], ) def test_schema_validation(self, config, error_msg): diff --git a/tests/unittests/config/test_cc_rpi_userdata.py b/tests/unittests/config/test_cc_rpi_userdata.py index b761c0a8af9..f26fbcba7c0 100644 --- a/tests/unittests/config/test_cc_rpi_userdata.py +++ b/tests/unittests/config/test_cc_rpi_userdata.py @@ -1,7 +1,10 @@ # This file is part of cloud-init. See LICENSE file for license information. import pytest -from cloudinit.config.cc_rpi_userdata import DISABLE_PIWIZ_KEY, RPI_USERCONF_KEY +from cloudinit.config.cc_rpi_userdata import ( + DISABLE_PIWIZ_KEY, + RPI_USERCONF_KEY, +) from cloudinit.config.schema import ( SchemaValidationError, get_schema, @@ -16,7 +19,7 @@ def is_notPi() -> bool: @mock.patch('cloudinit.subp.subp') class TestCCRPiUserdata(CiTestCase): - \"""Tests work in progress. Just partially implemented + \"""Tests work in progress. Just partially implemented to show the idea.\""" @mock.patch('subprocess.run') @@ -59,21 +62,27 @@ def test_default_user_still_exists(self, mock_listdir): assert 'pi' in os.listdir('/home') """ + @skipUnlessJsonSchema() class TestCCRPiUserdataSchema: @pytest.mark.parametrize( "config, error_msg", [ ({DISABLE_PIWIZ_KEY: True}, None), - ({ - RPI_USERCONF_KEY: { - "password": "hashedpassword", "user": "newuser" - } - }, None), + ( + { + RPI_USERCONF_KEY: { + "password": "hashedpassword", + "user": "newuser", + } + }, + None, + ), ({DISABLE_PIWIZ_KEY: "true"}, "'true' is not of type 'boolean'"), - ({ - RPI_USERCONF_KEY: {"password": 12345} - }, "12345 is not of type 'string'") + ( + {RPI_USERCONF_KEY: {"password": 12345}}, + "12345 is not of type 'string'", + ), ], ) def test_schema_validation(self, config, error_msg):