From 729df58739ae3225760d5a3e23840674c1ed94ed Mon Sep 17 00:00:00 2001 From: Eric Krichbaum Date: Sat, 31 Oct 2026 15:09:55 -0500 Subject: [PATCH 1/8] Clarify S3 tagging permissions in AWS documentation Added note about S3 bucket tagging permissions and implications for discovery. --- docs/source_aws.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/docs/source_aws.md b/docs/source_aws.md index 91b9071a..f4b4cf06 100644 --- a/docs/source_aws.md +++ b/docs/source_aws.md @@ -104,6 +104,16 @@ For example, the following supplemental policy enables all three: } ``` +The policy includes `s3:GetBucketTagging` for optional S3 tag synchronization. +If bucket tagging is desired, grant this permission to the inventory role used +by your AWS profile (or every selected member-account role in organization mode) +and configure `tag_keys` to select the tags to copy to NetBox. + +**Current S3 discovery always requests bucket tags**, even when `tag_keys` is +empty. Omitting `s3:GetBucketTagging` can therefore cause an authorization denial +that skips the affected bucket and postpones orphaning/pruning for this source, +as described below; it does not simply disable tag synchronization. + Where your IAM policy supports resource-level permissions, restrict the resource-specific actions to the intended buckets, functions, and EKS resources. Listing actions still need their documented wildcard scope. No S3 object reads, From b740ec4dbb4610513fe1c8dc8c2b72b1dde18cf1 Mon Sep 17 00:00:00 2001 From: Eric Krichbaum Date: Sat, 31 Oct 2026 15:10:34 -0500 Subject: [PATCH 2/8] Refactor custom fields processing in object_classes.py Refactor custom field handling and remove unused code. --- module/netbox/object_classes.py | 92 +++++++-------------------------- 1 file changed, 20 insertions(+), 72 deletions(-) diff --git a/module/netbox/object_classes.py b/module/netbox/object_classes.py index c382d326..b405fa5a 100644 --- a/module/netbox/object_classes.py +++ b/module/netbox/object_classes.py @@ -57,15 +57,7 @@ class to handle and determine NetBox interface types "200gbase-x-qsfp56": 200_000, "200gbase-x-cfp2": 200_000, "400gbase-x-qsfpdd": 400_000, - "400gbase-x-osfp": 400_000, - "1gfc-sfp": 1, - "2gfc-sfp": 2, - "4gfc-sfp": 4, - "8gfc-sfpp": 8, - "16gfc-sfpp": 16, - "32gfc-sfp28": 32, - "64gfc-qsfpp": 64, - "128gfc-qsfp28": 128 + "400gbase-x-osfp": 400_000 } # assign common types for an interface speed value @@ -698,21 +690,21 @@ def update(self, data=None, read_from_netbox=False, source=None): # object custom fields which get merged into the update from the current # data, and comparing reduced to unreduced values would report a change # on every run. - def reduce_custom_fields_for_write(custom_field_data: dict) -> dict: + def reduce_object_custom_fields_to_ids(custom_field_data: dict) -> dict: reduced_data = dict(custom_field_data) for f_name, f_value in custom_field_data.items(): # Check for custom field type custom_field = self.inventory.get_by_data(NBCustomField, data={"name": f_name}) if custom_field is None: - field_type = None - else: - field_type = grab(custom_field, "data.type") + continue - # custom fields read from the NetBox API report the type as - # a dict like {"value": "multiobject", "label": "Multiple objects"} - if isinstance(field_type, dict): - field_type = field_type.get("value") + field_type = grab(custom_field, "data.type") + + # custom fields read from the NetBox API report the type as + # a dict like {"value": "multiobject", "label": "Multiple objects"} + if isinstance(field_type, dict): + field_type = field_type.get("value") # Handle object type custom fields - need only ID if field_type == "object" and isinstance(f_value, dict) and \ @@ -729,41 +721,11 @@ def reduce_custom_fields_for_write(custom_field_data: dict) -> dict: if ids: reduced_data[f_name] = ids - # Handle selection custom fields - need only the raw choice value. - # The NetBox API returns a selection on read as - # {"value": "business-hours", "label": "Business Hours"} but on - # write (POST/PATCH) it expects the bare value; sending the dict - # back is rejected as an invalid choice (issue #598). Reducing it - # here also keeps change detection stable, so an unmanaged selection - # field that is merged in from the current NetBox data is not - # reported as changed on every run. - elif field_type == "select" and isinstance(f_value, dict) and \ - "value" in f_value: - reduced_data[f_name] = f_value.get("value") - - # Handle multi-selection custom fields - need a list of raw values. - # On read NetBox returns a list of {"value", "label"} dicts. - elif field_type in ("multiselect", "multi-select") and isinstance(f_value, list): - values = [item.get("value") if isinstance(item, dict) and "value" in item else item - for item in f_value] - reduced_data[f_name] = values - - # Fallback for a custom field that is not in the inventory, so - # its type is unknown: a dict consisting of exactly the "value" - # and "label" keys is how NetBox represents a selection choice on - # read, so reduce it to its bare value too. The reporter of #598 - # noted the offending field was one netbox-sync does not manage; - # object custom fields read as {"id","url","display",...} have - # more keys and are left untouched by this exact-key check. - elif field_type is None and isinstance(f_value, dict) and \ - set(f_value.keys()) == {"value", "label"}: - reduced_data[f_name] = f_value.get("value") - return reduced_data - current_value = reduce_custom_fields_for_write(current_value) + current_value = reduce_object_custom_fields_to_ids(current_value) current_value_str = str(current_value) - new_value = {**current_value, **reduce_custom_fields_for_write(new_value)} + new_value = {**current_value, **reduce_object_custom_fields_to_ids(new_value)} new_value_str = str(new_value) elif isinstance(new_value, (NetBoxObject, NBObjectList)): new_value_str = str(new_value.get_display_name()) @@ -873,6 +835,9 @@ def get_display_name(self, data=None, including_second_key=False): read_from_netbox = True if secondary_key_value.get("id", 0) != 0 else False secondary_key_value = self.get_display_name(data=secondary_key_value) + if secondary_key_value is None and include_secondary_key_if_present: + return my_name + if secondary_key_value is None and read_from_netbox is False and include_secondary_key_if_present is False: log.warning(f"Unable to determine second key '{secondary_key}' for {self.name} '{my_name}', " f"got: {org_secondary_key_value}") @@ -1562,6 +1527,7 @@ class NBVRF(NetBoxObject): object_type = "ipam.vrf" primary_key = "name" prune = False + read_only = True def __init__(self, *args, **kwargs): self.data_model = { @@ -1935,6 +1901,7 @@ class NBCluster(NetBoxObject): object_type = "virtualization.cluster" primary_key = "name" secondary_key = "scope_id" + include_secondary_key_if_present = True prune = False def __init__(self, *args, **kwargs): @@ -2032,7 +1999,6 @@ def __init__(self, *args, **kwargs): "memory": int, "disk": int, "comments": str, - "description": 200, "primary_ip4": NBIPAddress, "primary_ip6": NBIPAddress, "site": NBSite, @@ -2073,7 +2039,6 @@ def __init__(self, *args, **kwargs): "untagged_vlan": NBVLAN, "tagged_vlans": NBVLANList, "description": 200, - "custom_fields": NBCustomField, "tags": NBTagList } super().__init__(*args, **kwargs) @@ -2412,11 +2377,10 @@ def get_device_vm(self): def remove_interface_association(self): o_id = self.data.get("assigned_object_id") o_type = self.data.get("assigned_object_type") - o_interface = self.get_interface() + o_device = self.get_device_vm() - # NetBox refuses to unassign a MAC address which is still the primary MAC of its interface - if grab(o_interface, "data.primary_mac_address") is self: - o_interface.unset_attribute("primary_mac_address") + if grab(o_device, "data.primary_mac_address") is self: + o_device.unset_attribute("primary_mac_address") if o_id is not None: self.unset_attribute("assigned_object_id") @@ -2424,22 +2388,6 @@ def remove_interface_association(self): self.unset_attribute("assigned_object_type") -class NBIPAddressList(NBObjectList): - member_type = NBIPAddress - - def get_nb_reference(self): - """ - return None if one IP address is unresolvable - """ - return_list = list() - for ip_address in self: - if ip_address.nb_id == 0: - return None - - return_list.append(ip_address.nb_id) - - return return_list - class NBFHRPGroupItem(NetBoxObject): name = "FHRP group" api_path = "ipam/fhrp-groups" @@ -2451,7 +2399,7 @@ class NBFHRPGroupItem(NetBoxObject): def __init__(self, *args, **kwargs): self.data_model = { "group_id": int, - "ip_addresses": NBIPAddressList, + "ip_addresses": NBIPAddress, "description": 200, "tags": NBTagList, "custom_fields": NBCustomField From 4cf4803fbe87e3cd8e70cf0b5db1003c8a5fe291 Mon Sep 17 00:00:00 2001 From: Eric Krichbaum Date: Sat, 31 Oct 2026 15:20:07 -0500 Subject: [PATCH 3/8] Refactor custom field handling and update data models --- module/netbox/object_classes.py | 88 +++++++++++++++++++++++++++------ 1 file changed, 72 insertions(+), 16 deletions(-) diff --git a/module/netbox/object_classes.py b/module/netbox/object_classes.py index b405fa5a..b718b749 100644 --- a/module/netbox/object_classes.py +++ b/module/netbox/object_classes.py @@ -57,7 +57,15 @@ class to handle and determine NetBox interface types "200gbase-x-qsfp56": 200_000, "200gbase-x-cfp2": 200_000, "400gbase-x-qsfpdd": 400_000, - "400gbase-x-osfp": 400_000 + "400gbase-x-osfp": 400_000, + "1gfc-sfp": 1, + "2gfc-sfp": 2, + "4gfc-sfp": 4, + "8gfc-sfpp": 8, + "16gfc-sfpp": 16, + "32gfc-sfp28": 32, + "64gfc-qsfpp": 64, + "128gfc-qsfp28": 128 } # assign common types for an interface speed value @@ -690,21 +698,21 @@ def update(self, data=None, read_from_netbox=False, source=None): # object custom fields which get merged into the update from the current # data, and comparing reduced to unreduced values would report a change # on every run. - def reduce_object_custom_fields_to_ids(custom_field_data: dict) -> dict: + def reduce_custom_fields_for_write(custom_field_data: dict) -> dict: reduced_data = dict(custom_field_data) for f_name, f_value in custom_field_data.items(): # Check for custom field type custom_field = self.inventory.get_by_data(NBCustomField, data={"name": f_name}) if custom_field is None: - continue + field_type = None + else: + field_type = grab(custom_field, "data.type") - field_type = grab(custom_field, "data.type") - - # custom fields read from the NetBox API report the type as - # a dict like {"value": "multiobject", "label": "Multiple objects"} - if isinstance(field_type, dict): - field_type = field_type.get("value") + # custom fields read from the NetBox API report the type as + # a dict like {"value": "multiobject", "label": "Multiple objects"} + if isinstance(field_type, dict): + field_type = field_type.get("value") # Handle object type custom fields - need only ID if field_type == "object" and isinstance(f_value, dict) and \ @@ -721,11 +729,41 @@ def reduce_object_custom_fields_to_ids(custom_field_data: dict) -> dict: if ids: reduced_data[f_name] = ids + # Handle selection custom fields - need only the raw choice value. + # The NetBox API returns a selection on read as + # {"value": "business-hours", "label": "Business Hours"} but on + # write (POST/PATCH) it expects the bare value; sending the dict + # back is rejected as an invalid choice (issue #598). Reducing it + # here also keeps change detection stable, so an unmanaged selection + # field that is merged in from the current NetBox data is not + # reported as changed on every run. + elif field_type == "select" and isinstance(f_value, dict) and \ + "value" in f_value: + reduced_data[f_name] = f_value.get("value") + + # Handle multi-selection custom fields - need a list of raw values. + # On read NetBox returns a list of {"value", "label"} dicts. + elif field_type in ("multiselect", "multi-select") and isinstance(f_value, list): + values = [item.get("value") if isinstance(item, dict) and "value" in item else item + for item in f_value] + reduced_data[f_name] = values + + # Fallback for a custom field that is not in the inventory, so + # its type is unknown: a dict consisting of exactly the "value" + # and "label" keys is how NetBox represents a selection choice on + # read, so reduce it to its bare value too. The reporter of #598 + # noted the offending field was one netbox-sync does not manage; + # object custom fields read as {"id","url","display",...} have + # more keys and are left untouched by this exact-key check. + elif field_type is None and isinstance(f_value, dict) and \ + set(f_value.keys()) == {"value", "label"}: + reduced_data[f_name] = f_value.get("value") + return reduced_data - current_value = reduce_object_custom_fields_to_ids(current_value) + current_value = reduce_custom_fields_for_write(current_value) current_value_str = str(current_value) - new_value = {**current_value, **reduce_object_custom_fields_to_ids(new_value)} + new_value = {**current_value, **reduce_custom_fields_for_write(new_value)} new_value_str = str(new_value) elif isinstance(new_value, (NetBoxObject, NBObjectList)): new_value_str = str(new_value.get_display_name()) @@ -1527,7 +1565,6 @@ class NBVRF(NetBoxObject): object_type = "ipam.vrf" primary_key = "name" prune = False - read_only = True def __init__(self, *args, **kwargs): self.data_model = { @@ -1999,6 +2036,7 @@ def __init__(self, *args, **kwargs): "memory": int, "disk": int, "comments": str, + "description": 200, "primary_ip4": NBIPAddress, "primary_ip6": NBIPAddress, "site": NBSite, @@ -2039,6 +2077,7 @@ def __init__(self, *args, **kwargs): "untagged_vlan": NBVLAN, "tagged_vlans": NBVLANList, "description": 200, + "custom_fields": NBCustomField, "tags": NBTagList } super().__init__(*args, **kwargs) @@ -2377,10 +2416,11 @@ def get_device_vm(self): def remove_interface_association(self): o_id = self.data.get("assigned_object_id") o_type = self.data.get("assigned_object_type") - o_device = self.get_device_vm() + o_interface = self.get_interface() - if grab(o_device, "data.primary_mac_address") is self: - o_device.unset_attribute("primary_mac_address") + # NetBox refuses to unassign a MAC address which is still the primary MAC of its interface + if grab(o_interface, "data.primary_mac_address") is self: + o_interface.unset_attribute("primary_mac_address") if o_id is not None: self.unset_attribute("assigned_object_id") @@ -2388,6 +2428,22 @@ def remove_interface_association(self): self.unset_attribute("assigned_object_type") +class NBIPAddressList(NBObjectList): + member_type = NBIPAddress + + def get_nb_reference(self): + """ + return None if one IP address is unresolvable + """ + return_list = list() + for ip_address in self: + if ip_address.nb_id == 0: + return None + + return_list.append(ip_address.nb_id) + + return return_list + class NBFHRPGroupItem(NetBoxObject): name = "FHRP group" api_path = "ipam/fhrp-groups" @@ -2399,7 +2455,7 @@ class NBFHRPGroupItem(NetBoxObject): def __init__(self, *args, **kwargs): self.data_model = { "group_id": int, - "ip_addresses": NBIPAddress, + "ip_addresses": NBIPAddressList, "description": 200, "tags": NBTagList, "custom_fields": NBCustomField From 6e6ba7c395ead2fb13fcba2627c4b3f4ea515167 Mon Sep 17 00:00:00 2001 From: Eric Krichbaum Date: Sat, 31 Oct 2026 15:20:26 -0500 Subject: [PATCH 4/8] Implement test for unscoped cluster scope warning Add test for unscoped cluster warning in logging. --- tests/test_cluster_scope.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/test_cluster_scope.py b/tests/test_cluster_scope.py index a43eb4a6..17b3d523 100644 --- a/tests/test_cluster_scope.py +++ b/tests/test_cluster_scope.py @@ -46,6 +46,13 @@ def test_a_site_object_survives_resolving(inventory): assert cluster.data.get("scope_type") == "dcim.site", "scope type was dropped" +def test_unscoped_cluster_has_no_missing_scope_warning(inventory, caplog): + cluster = inventory.add_object(NBCluster, data={"name": "global"}) + + assert cluster.get_display_name(including_second_key=True) == "global" + assert "Unable to determine second key 'scope_id'" not in caplog.text + + def test_source_scopes_a_cluster_with_the_site_object(vcsim, inventory, load_config, vmware_settings): """A dict here is sent to NetBox as is and rejected with 'scope_id: A valid integer is required.'""" from module.sources import instantiate_sources From f44a108ab7263e2ced3083ddaff6b25fe0d6fe92 Mon Sep 17 00:00:00 2001 From: Eric Krichbaum Date: Sat, 31 Oct 2026 19:05:35 -0500 Subject: [PATCH 5/8] Refactor service_object method for clarity and efficiency Previous adoption of original code dropped the required custom fields to match. This patch adds a guide for empty/missing and still allowing for true name collision. --- module/sources/aws/connection.py | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/module/sources/aws/connection.py b/module/sources/aws/connection.py index 8d139652..7146cc28 100644 --- a/module/sources/aws/connection.py +++ b/module/sources/aws/connection.py @@ -324,8 +324,26 @@ def resource_name(self, name, identity, limit): return f"{name[:limit - 13]}-{suffix}" def service_object(self, object_type, identity, data): - matches = [item for item in self.inventory.get_all_items(object_type) + items = self.inventory.get_all_items(object_type) + matches = [item for item in items if (item.data.get("custom_fields") or {}).get("aws_resource_id") == identity] + parent = "virtual_machine" if object_type is NBVirtualDisk else "cluster" if object_type is NBVM else "group" + if not matches: + name_matches = [item for item in items + if item.data.get("name") == data["name"] and + item.data.get(parent) is data.get(parent)] + if len(name_matches) == 1: + candidate = name_matches[0] + fields = candidate.data.get("custom_fields") or {} + candidate_identity = fields.get("aws_resource_id") + candidate_service = fields.get("aws_service") + identity_missing = candidate_identity is None or ( + isinstance(candidate_identity, str) and not candidate_identity.strip()) + service = (data.get("custom_fields") or {}).get("aws_service") + service_matches = candidate_service in (None, "", service) + source_managed = candidate.source is self or self.source_tag in candidate.get_tags() + if identity_missing and service_matches and source_managed: + matches = [candidate] if len(matches) > 1: raise ValueError(f"Multiple NetBox {object_type.name} objects match AWS resource '{identity}'") if matches: @@ -334,9 +352,8 @@ def service_object(self, object_type, identity, data): raise ValueError(f"AWS resource '{identity}' belongs to another source") item.update(data=data, source=self) else: - parent = "virtual_machine" if object_type is NBVirtualDisk else "cluster" if object_type is NBVM else "group" if any(item.data.get("name") == data["name"] and item.data.get(parent) is data.get(parent) - for item in self.inventory.get_all_items(object_type)): + for item in items): raise ValueError(f"NetBox {object_type.name} name collision for AWS resource '{identity}'") item = self.inventory.add_object(object_type, data=data, source=self) self.reconcile_tags(item, data["tags"]) From 714a19d7fcca303d831a8212cd93a39c378b07c4 Mon Sep 17 00:00:00 2001 From: Eric Krichbaum Date: Sat, 31 Oct 2026 19:06:00 -0500 Subject: [PATCH 6/8] Refactor AWS service tests for clarity and coverage Updated assertions for name collision and bucket checks in AWS service tests. Added new tests for EKS cluster handling with missing identity fields and conflicting identity fields. --- tests/test_aws_services.py | 57 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 55 insertions(+), 2 deletions(-) diff --git a/tests/test_aws_services.py b/tests/test_aws_services.py index 2c7160d3..0fb693af 100644 --- a/tests/test_aws_services.py +++ b/tests/test_aws_services.py @@ -265,10 +265,13 @@ def test_s3_conflict_skips_dependencies_but_continues(make_source, inventory, ca "name": "inventory-bucket", "virtual_machine": holder, "size": 0, }) source.apply() - assert "name collision" in caplog.text + assert ("name collision" in caplog.text) == (stage == "bucket") buckets = [disk.data["name"] for disk in inventory.get_all_items(NBVirtualDisk) if disk.data.get("custom_fields", {}).get("aws_service") == "s3"] - assert buckets == (["second"] if stage == "bucket" else []) + assert buckets == (["second"] if stage == "bucket" else ["inventory-bucket", "second"]) + assert cluster.data["custom_fields"]["aws_resource_id"] == identity + if stage == "holder": + assert holder.data["custom_fields"]["aws_resource_id"] == identity assert any(vm.data.get("custom_fields", {}).get("aws_instance_id") == INSTANCE for vm in inventory.get_all_items(NBVM)) @@ -295,6 +298,56 @@ def test_eks_conflict_skips_workers_and_continues_next_region(make_source, inven assert "i-next-region" in instances +@pytest.mark.parametrize("fields", [ + {}, + {"aws_resource_id": "", "aws_service": ""}, +]) +def test_eks_cluster_with_missing_identity_fields_is_reconciled(make_source, inventory, fields, caplog): + source = make_source("sync_eks = True\nsync_private_ips = False", data=service_data()) + group = inventory.add_object(NBClusterGroup, data={"name": f"AWS {ACCOUNT}"}) + cluster = inventory.add_object(NBCluster, data={ + "name": f"EKS {ACCOUNT}/{REGION}/inventory-cluster", + "group": group, + "custom_fields": fields, + }) + cluster.add_tags(source.source_tag) + + source.apply() + + assert not any("Skipping EKS cluster 'inventory-cluster'" in message + for message in caplog.messages) + assert len([item for item in inventory.get_all_items(NBCluster) + if item.data.get("name") == cluster.data["name"]]) == 1 + assert cluster.data["custom_fields"]["aws_resource_id"] == CLUSTER_ARN + assert cluster.data["custom_fields"]["aws_service"] == "eks" + worker = next(vm for vm in inventory.get_all_items(NBVM) + if vm.data.get("custom_fields", {}).get("aws_instance_id") == INSTANCE) + assert worker.data["cluster"] is cluster + + +@pytest.mark.parametrize("fields", [ + {"aws_resource_id": f"arn:aws:eks:{REGION}:{ACCOUNT}:cluster/another-cluster", "aws_service": "eks"}, + {"aws_resource_id": "", "aws_service": "lambda"}, +]) +def test_eks_cluster_does_not_adopt_conflicting_identity_fields(make_source, inventory, fields, caplog): + source = make_source("sync_eks = True\nsync_private_ips = False", data=service_data()) + source.custom_fields() + group = inventory.add_object(NBClusterGroup, data={"name": f"AWS {ACCOUNT}"}) + cluster = inventory.add_object(NBCluster, data={ + "name": f"EKS {ACCOUNT}/{REGION}/inventory-cluster", + "group": group, + "custom_fields": fields, + }) + cluster.add_tags(source.source_tag) + + source.apply() + + assert "Skipping EKS cluster 'inventory-cluster'" in caplog.text + assert cluster.data["custom_fields"] == fields + assert not any(vm.data.get("custom_fields", {}).get("aws_instance_id") == INSTANCE + for vm in inventory.get_all_items(NBVM)) + + def test_lambda_function_conflict_continues_next_function(make_source, inventory, caplog): data = service_data() source = make_source("sync_lambda = True\nsync_private_ips = False", data=data) From 8bd996b15a4748164d2bd8c80158e4023e006679 Mon Sep 17 00:00:00 2001 From: Eric Krichbaum Date: Sat, 31 Oct 2026 21:51:16 -0500 Subject: [PATCH 7/8] Fix secondary key value assignment logic --- module/netbox/object_classes.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/module/netbox/object_classes.py b/module/netbox/object_classes.py index b718b749..073acea6 100644 --- a/module/netbox/object_classes.py +++ b/module/netbox/object_classes.py @@ -862,7 +862,6 @@ def get_display_name(self, data=None, including_second_key=False): [enforce_secondary_key, including_second_key, include_secondary_key_if_present]: secondary_key_value = this_data_set.get(secondary_key) - org_secondary_key_value = str(secondary_key_value) read_from_netbox = False if isinstance(secondary_key_value, NetBoxObject): @@ -873,6 +872,8 @@ def get_display_name(self, data=None, including_second_key=False): read_from_netbox = True if secondary_key_value.get("id", 0) != 0 else False secondary_key_value = self.get_display_name(data=secondary_key_value) + org_secondary_key_value = str(secondary_key_value) + if secondary_key_value is None and include_secondary_key_if_present: return my_name From 4c3a6a922f0d1e5407fb512c71a69a4bdf435635 Mon Sep 17 00:00:00 2001 From: Eric Krichbaum Date: Sat, 31 Oct 2026 21:53:38 -0500 Subject: [PATCH 8/8] Add test for VM lookup without cluster metadata Added a test to ensure VM lookup does not serialize cluster metadata. --- tests/test_cluster_scope.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/tests/test_cluster_scope.py b/tests/test_cluster_scope.py index 17b3d523..b609242f 100644 --- a/tests/test_cluster_scope.py +++ b/tests/test_cluster_scope.py @@ -13,10 +13,11 @@ that object again, otherwise every run reports a change. """ import logging +from datetime import datetime import pytest -from module.netbox.object_classes import NBCluster, NBClusterType, NBSite, NBSiteGroup +from module.netbox.object_classes import NBCluster, NBClusterType, NBVM, NBSite, NBSiteGroup def test_cluster_scope_is_resolved_from_an_id(inventory): @@ -53,6 +54,16 @@ def test_unscoped_cluster_has_no_missing_scope_warning(inventory, caplog): assert "Unable to determine second key 'scope_id'" not in caplog.text +def test_vm_lookup_does_not_serialize_cluster_metadata(inventory): + cluster = inventory.add_object(NBCluster, data={ + "name": "eks-cluster", + "custom_fields": {"aws_metadata": {"createdAt": datetime(2026, 10, 23)}}, + }) + vm = inventory.add_object(NBVM, data={"name": "worker", "cluster": cluster}) + + assert inventory.get_by_data(NBVM, data={"name": "worker", "cluster": cluster}) is vm + + def test_source_scopes_a_cluster_with_the_site_object(vcsim, inventory, load_config, vmware_settings): """A dict here is sent to NetBox as is and rejected with 'scope_id: A valid integer is required.'""" from module.sources import instantiate_sources