From 844a185f6b4b2e2347f085a8b34c5ee6e736786a Mon Sep 17 00:00:00 2001 From: anujsingh-cse Date: Fri, 9 Oct 2026 13:58:58 +0530 Subject: [PATCH 1/5] chore: add type annotations to DataSourceGCE Removing cloudinit.sources.DataSourceGCE from the mypy override list in pyproject.toml surfaces six errors. read_md seeds its result dict with None and False values, so mypy infers dict[str, bool | None] and rejects every later str, bytes and dict assignment. Annotate ret as Dict[str, Any], which is what the dict actually carries depending on which keys were populated. _parse_public_keys starts from an empty list that mypy cannot infer the element type of; annotate it as List[str]. self.metadata is declared as dict on the DataSource base class, and ret.get("meta-data") is Optional; read_md only reports success after populating "meta-data", so index it directly with a comment noting the guarantee. Refs GH-5445 --- cloudinit/sources/DataSourceGCE.py | 11 ++++++++--- pyproject.toml | 1 - 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/cloudinit/sources/DataSourceGCE.py b/cloudinit/sources/DataSourceGCE.py index 7c443b14183..70207616fd2 100644 --- a/cloudinit/sources/DataSourceGCE.py +++ b/cloudinit/sources/DataSourceGCE.py @@ -6,6 +6,7 @@ import json import logging from base64 import b64decode +from typing import Any, Dict, List from cloudinit import dmi, net, sources, url_helper, util from cloudinit.distros import ug_util @@ -159,7 +160,9 @@ def _get_data(self): else: LOG.debug(ret.get("reason")) return False - self.metadata = ret.get("meta-data") + # read_md only reports success after populating "meta-data", + # so the key is guaranteed present here. + self.metadata = ret["meta-data"] self.userdata_raw = ret.get("user-data") return True @@ -253,7 +256,7 @@ def _parse_public_keys(public_keys_data, default_user=None): # a list containing SSH public keys in the GCE specific key format # documented here: # https://cloud.google.com/compute/docs/instances/adding-removing-ssh-keys#sshkeyformat - public_keys = [] + public_keys: List[str] = [] if not public_keys_data: return public_keys for public_key in public_keys_data: @@ -273,7 +276,9 @@ def read_md(address=None, url_params=None, platform_check=True): if address is None: address = MD_V1_URL - ret = { + # The dict carries a mix of None, bool, str, bytes and dict values + # depending on which keys were populated, so its values are Any. + ret: Dict[str, Any] = { "meta-data": None, "user-data": None, "success": False, diff --git a/pyproject.toml b/pyproject.toml index 6812fe7b025..6eb65da1065 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,7 +53,6 @@ module = [ "cloudinit.sources.DataSourceConfigDrive", "cloudinit.sources.DataSourceEc2", "cloudinit.sources.DataSourceExoscale", - "cloudinit.sources.DataSourceGCE", "cloudinit.sources.DataSourceHetzner", "cloudinit.sources.DataSourceNoCloud", "cloudinit.sources.DataSourceOVF", From 04b11ea2f93594b7a855e3297f184cf8d3d7025d Mon Sep 17 00:00:00 2001 From: anujsingh-cse Date: Tue, 13 Oct 2026 22:48:39 +0530 Subject: [PATCH 2/5] fix(GCE): annotate every method and module function Follows the GH-5445 review guidance: full annotation coverage of the module, not only the errors check_untyped_defs surfaced. GoogleMetadataFetcher, every DataSourceGCE method and property, and all module functions are now typed. Concrete types: - get_value(path: str, is_text: bool, is_recursive: bool) -> Optional[str]; the constructor's retries are ints. - The datasource constructor takes (Dict[str, Any], distros.Distro, helpers.Paths) matching the base DataSource, and default_user is Optional[str] (None when no distro default exists). - launch_index is the only Optional property: GCE provides none. - get_instance_id -> str, get_public_ssh_keys -> List[str], get_hostname -> DataSourceHostname, availability_zone/region -> str. - _write_host_key_to_guest_attributes encodes into a separate local instead of reassigning the str parameter with bytes, which mypy surfaced once the signature was annotated. - _has_expired's decoded JSON is Dict[str, Any]; _parse_public_keys takes List[str] and Optional[str] and returns List[str]. - read_md now requires url_params: every fetch needs retry parameters, and the previous Optional-with-None default could only crash at the fetcher construction. The __main__ CLI block constructs the datasource defaults (no wait, 10s timeout, five retries, one second between) explicitly. Its callers in _get_data already pass the parameter. - platform_reports_gce -> bool and get_datasource_list(depends: List[str]) -> List[Any] (the mixed tuple list returned by list_from_depends). Refs GH-5445 --- cloudinit/sources/DataSourceGCE.py | 84 ++++++++++++++++++++---------- 1 file changed, 57 insertions(+), 27 deletions(-) diff --git a/cloudinit/sources/DataSourceGCE.py b/cloudinit/sources/DataSourceGCE.py index 70207616fd2..bcfa0531d72 100644 --- a/cloudinit/sources/DataSourceGCE.py +++ b/cloudinit/sources/DataSourceGCE.py @@ -6,9 +6,9 @@ import json import logging from base64 import b64decode -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional -from cloudinit import dmi, net, sources, url_helper, util +from cloudinit import distros, dmi, helpers, net, sources, url_helper, util from cloudinit.distros import ug_util from cloudinit.event import EventScope, EventType from cloudinit.net.dhcp import NoDHCPLeaseError @@ -38,13 +38,20 @@ class GoogleMetadataFetcher: - def __init__(self, metadata_address, num_retries, sec_between_retries): + def __init__( + self, + metadata_address: str, + num_retries: int, + sec_between_retries: int, + ) -> None: self.metadata_address = metadata_address self.num_retries = num_retries self.sec_between_retries = sec_between_retries - def get_value(self, path, is_text, is_recursive=False): - value = None + def get_value( + self, path: str, is_text: bool, is_recursive: bool = False + ) -> Optional[str]: + value: Optional[str] = None try: url = self.metadata_address + path if is_recursive: @@ -80,13 +87,18 @@ class DataSourceGCE(sources.DataSource): } } - def __init__(self, sys_cfg, distro, paths): + def __init__( + self, + sys_cfg: Dict[str, Any], + distro: distros.Distro, + paths: helpers.Paths, + ) -> None: sources.DataSource.__init__(self, sys_cfg, distro, paths) - self.default_user = None + self.default_user: Optional[str] = None if distro: users, _groups = ug_util.normalize_users_groups(sys_cfg, distro) self.default_user, _user_config = ug_util.extract_default(users) - self.metadata = dict() + self.metadata: Dict[str, Any] = dict() self.ds_cfg = util.mergemanydict( [ util.get_cfg_by_path(sys_cfg, ["datasource", "GCE"], {}), @@ -95,7 +107,7 @@ def __init__(self, sys_cfg, distro, paths): ) self.metadata_address = self.ds_cfg["metadata_url"] - def _get_data(self): + def _get_data(self) -> bool: url_params = self.get_url_params() ret = {} if self.perform_dhcp_setup: @@ -167,33 +179,38 @@ def _get_data(self): return True @property - def launch_index(self): + def launch_index(self) -> None: # GCE does not provide launch_index property. return None - def get_instance_id(self): + def get_instance_id(self) -> str: return self.metadata["instance-id"] - def get_public_ssh_keys(self): + def get_public_ssh_keys(self) -> List[str]: public_keys_data = self.metadata["public-keys-data"] return _parse_public_keys(public_keys_data, self.default_user) - def publish_host_keys(self, hostkeys): + def publish_host_keys(self, hostkeys: List[Any]) -> None: for key in hostkeys: _write_host_key_to_guest_attributes(*key) - def get_hostname(self, fqdn=False, resolve_ip=False, metadata_only=False): + def get_hostname( + self, + fqdn: bool = False, + resolve_ip: bool = False, + metadata_only: bool = False, + ) -> DataSourceHostname: # GCE has long FDQN's and has asked for short hostnames. return DataSourceHostname( self.metadata["local-hostname"].split(".")[0], False ) @property - def availability_zone(self): + def availability_zone(self) -> str: return self.metadata["availability-zone"] @property - def region(self): + def region(self) -> str: return self.availability_zone.rsplit("-", 1)[0] @@ -201,12 +218,12 @@ class DataSourceGCELocal(DataSourceGCE): perform_dhcp_setup = True -def _write_host_key_to_guest_attributes(key_type, key_value): +def _write_host_key_to_guest_attributes(key_type: str, key_value: str) -> None: url = "%s/%s/%s" % (GUEST_ATTRIBUTES_URL, HOSTKEY_NAMESPACE, key_type) - key_value = key_value.encode("utf-8") + encoded_value = key_value.encode("utf-8") resp = url_helper.readurl( url=url, - data=key_value, + data=encoded_value, headers=HEADERS, request_method="PUT", check_status=False, @@ -217,7 +234,7 @@ def _write_host_key_to_guest_attributes(key_type, key_value): LOG.debug("Unable to write %s host key to guest attributes.", key_type) -def _has_expired(public_key): +def _has_expired(public_key: str) -> bool: # Check whether an SSH key is expired. Public key input is a single SSH # public key in the GCE specific key format documented here: # https://cloud.google.com/compute/docs/instances/adding-removing-ssh-keys#sshkeyformat @@ -232,7 +249,7 @@ def _has_expired(public_key): return False try: - json_obj = json.loads(json_str) + json_obj: Dict[str, Any] = json.loads(json_str) except ValueError: return False @@ -251,7 +268,9 @@ def _has_expired(public_key): return datetime.datetime.now(datetime.timezone.utc) > expire_time -def _parse_public_keys(public_keys_data, default_user=None): +def _parse_public_keys( + public_keys_data: List[str], default_user: Optional[str] = None +) -> List[str]: # Parse the SSH key data for the default user account. Public keys input is # a list containing SSH public keys in the GCE specific key format # documented here: @@ -271,7 +290,11 @@ def _parse_public_keys(public_keys_data, default_user=None): return public_keys -def read_md(address=None, url_params=None, platform_check=True): +def read_md( + url_params: sources.URLParams, + address: Optional[str] = None, + platform_check: bool = True, +) -> Dict[str, Any]: if address is None: address = MD_V1_URL @@ -307,7 +330,7 @@ def read_md(address=None, url_params=None, platform_check=True): metadata_fetcher = GoogleMetadataFetcher( address, url_params.num_retries, url_params.sec_between_retries ) - md = {} + md: Dict[str, Any] = {} # Iterate over url_map keys to get metadata items. for mkey, paths, required, is_text, is_recursive in url_map: value = None @@ -356,7 +379,7 @@ def read_md(address=None, url_params=None, platform_check=True): return ret -def platform_reports_gce(): +def platform_reports_gce() -> bool: pname = dmi.read_dmi_data("system-product-name") or "N/A" if pname == "Google Compute Engine" or pname == "Google": return True @@ -380,7 +403,7 @@ def platform_reports_gce(): # Return a list of data sources that match this set of dependencies. -def get_datasource_list(depends): +def get_datasource_list(depends: List[str]) -> List[Any]: return sources.list_from_depends(depends, datasources) @@ -404,7 +427,14 @@ def get_datasource_list(depends): default=True, ) args = parser.parse_args() - data = read_md(address=args.endpoint, platform_check=args.platform_check) + # The datasource defaults from cloudinit.sources: no extra wait, a 10s + # timeout, five 404 retries, one second between retries. + cli_url_params = sources.URLParams(-1, 10, 5, 1) + data = read_md( + url_params=cli_url_params, + address=args.endpoint, + platform_check=args.platform_check, + ) if "user-data" in data: # user-data is bytes not string like other things. Handle it specially. # If it can be represented as utf-8 then do so. Otherwise print base64 From 161c412f8ad5e893ff1146115d53bc448fc64904 Mon Sep 17 00:00:00 2001 From: anujsingh-cse Date: Tue, 13 Oct 2026 23:31:56 +0530 Subject: [PATCH 3/5] fix(GCE): keep get_hostname signature identical to the base test_init.py::test_get_hostname_subclass_support compares inspect.getfullargspec across every DataSource override, annotations included, so the GCE override must match the untyped base exactly; a typed mirror fails the conformance test on the override contract. --- cloudinit/sources/DataSourceGCE.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/cloudinit/sources/DataSourceGCE.py b/cloudinit/sources/DataSourceGCE.py index bcfa0531d72..f651514ba7e 100644 --- a/cloudinit/sources/DataSourceGCE.py +++ b/cloudinit/sources/DataSourceGCE.py @@ -194,12 +194,10 @@ def publish_host_keys(self, hostkeys: List[Any]) -> None: for key in hostkeys: _write_host_key_to_guest_attributes(*key) - def get_hostname( - self, - fqdn: bool = False, - resolve_ip: bool = False, - metadata_only: bool = False, - ) -> DataSourceHostname: + # Signature matches the untyped DataSource.get_hostname exactly: + # test_init.py::test_get_hostname_subclass_support compares + # inspect.getfullargspec across the override, annotations included. + def get_hostname(self, fqdn=False, resolve_ip=False, metadata_only=False): # GCE has long FDQN's and has asked for short hostnames. return DataSourceHostname( self.metadata["local-hostname"].split(".")[0], False From 441d35196cc12665870c2fe2cad5966e10e01bb7 Mon Sep 17 00:00:00 2001 From: anujsingh-cse Date: Sat, 10 Oct 2026 09:12:25 +0530 Subject: [PATCH 4/5] test(azure): parenthesize implicit concatenations for ruff ISC004 #6958 added base64 error-message parameters to test_azure_helper.py with implicitly concatenated strings inside collection literals, which the pinned ruff flags as ISC004. CI for every open PR runs against the merge ref, so this fails check_format across the board. Wrap the two concatenations in parentheses; the string values are unchanged. --- tests/unittests/sources/test_azure_helper.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/tests/unittests/sources/test_azure_helper.py b/tests/unittests/sources/test_azure_helper.py index b72ce830e46..d222bc7f1f3 100644 --- a/tests/unittests/sources/test_azure_helper.py +++ b/tests/unittests/sources/test_azure_helper.py @@ -1562,8 +1562,10 @@ def test_empty_custom_data_passes(self): [ ( "ABCDE", # invalid length - "Invalid base64-encoded string: number of data characters " - "(5) cannot be 1 more than a multiple of 4", + ( + "Invalid base64-encoded string: number of data characters " + "(5) cannot be 1 more than a multiple of 4" + ), ), ( "ab", # incorrect padding @@ -1571,8 +1573,10 @@ def test_empty_custom_data_passes(self): ), ( "not_base64", # bad character stripped away - "Invalid base64-encoded string: number of data characters " - "(9) cannot be 1 more than a multiple of 4", + ( + "Invalid base64-encoded string: number of data characters " + "(9) cannot be 1 more than a multiple of 4" + ), ), ], ) From 005ec4b07cdb0a98433fc52a5f364456a5a1fca6 Mon Sep 17 00:00:00 2001 From: anujsingh-cse Date: Mon, 26 Oct 2026 14:08:02 +0530 Subject: [PATCH 5/5] chore(gce): tighten host key and datasource list types publish_host_keys always receives (type, value) pairs and get_datasource_list returns DataSource classes, so neither needs Any. get_hostname remains unannotated: test_init compares inspect.getfullargspec against the untyped DataSource base, annotations included. --- cloudinit/sources/DataSourceGCE.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cloudinit/sources/DataSourceGCE.py b/cloudinit/sources/DataSourceGCE.py index f651514ba7e..14c5848b02a 100644 --- a/cloudinit/sources/DataSourceGCE.py +++ b/cloudinit/sources/DataSourceGCE.py @@ -6,7 +6,7 @@ import json import logging from base64 import b64decode -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple, Type from cloudinit import distros, dmi, helpers, net, sources, url_helper, util from cloudinit.distros import ug_util @@ -190,7 +190,7 @@ def get_public_ssh_keys(self) -> List[str]: public_keys_data = self.metadata["public-keys-data"] return _parse_public_keys(public_keys_data, self.default_user) - def publish_host_keys(self, hostkeys: List[Any]) -> None: + def publish_host_keys(self, hostkeys: List[Tuple[str, str]]) -> None: for key in hostkeys: _write_host_key_to_guest_attributes(*key) @@ -401,7 +401,7 @@ def platform_reports_gce() -> bool: # Return a list of data sources that match this set of dependencies. -def get_datasource_list(depends: List[str]) -> List[Any]: +def get_datasource_list(depends: List[str]) -> List[Type[sources.DataSource]]: return sources.list_from_depends(depends, datasources)