fix: get-kubeconfig merge into empty lists (#648)

Author: merll

linodecli/plugins/get-kubeconfig.py

@@ -90,7 +90,7 @@ def call(args, context):
         else cluster_config
     )
     if parsed.dry_run:
-        print(cluster_config)
+        print(yaml.dump(cluster_config))
     else:
         _dump_config(kubeconfig_path, cluster_config)
 
@@ -146,27 +146,35 @@ def _dump_config(filepath, data):
         yaml.dump(data, file_descriptor)
 
 
-# Merges the lists in the provided dicts. If non-list properties of the two
-# dicts differ, uses the value from the first dict.
 def _merge_dict(dict_1, dict_2):
+    """
+    Merges two dicts:
+    * Lists that are present in both dicts are merged together by their "name" key
+      (preferring duplicate values in the first dict)
+    * `None` or missing keys in the first dict are set to the second dict's value
+    * Other values are preferred from the first dict
+    """
     # Return a new dict to prevent any accidental mutations
     result = {}
 
-    for key in dict_1:
-        if not isinstance(dict_1[key], list):
-            result[key] = dict_1[key]
-            continue
-
-        merge_map = {sub["name"]: sub for sub in dict_1[key]}
-
-        for sub in dict_2[key]:
-            # If the name is already in the merge map, skip
-            if sub["name"] in merge_map:
-                continue
-
-            merge_map[sub["name"]] = sub
-
-        # Convert back to a list
-        result[key] = list(merge_map.values())
+    for key, dict_1_value in dict_1.items():
+        if dict_1_value is None and (dict_2_value := dict_2.get(key)):
+            # Replace null value in previous config
+            result[key] = dict_2_value
+        elif isinstance(dict_1_value, list) and (
+            dict_2_value := dict_2.get(key)
+        ):
+            merge_map = {sub["name"]: sub for sub in dict_1_value}
+            for list_2_item in dict_2_value:
+                if (list_2_name := list_2_item["name"]) not in merge_map:
+                    merge_map[list_2_name] = list_2_item
+            # Convert back to a list
+            result[key] = list(merge_map.values())
+        else:
+            result[key] = dict_1_value
+
+    # Process keys missing in dict_1
+    for key in set(dict_2.keys()).difference(dict_1.keys()):
+        result[key] = dict_2[key]
 
     return result

tests/unit/test_plugin_kubeconfig.py

@@ -42,7 +42,6 @@
 - item:
     property-1: a
     property-2: b
-    property-3: c
   name: item-1
 - item:
     property-1: d
@@ -51,6 +50,12 @@
 dictionary: {"foo": "bar"}
 """
 
+TEST_YAML_EMPTY_CONFIG = """
+name: testing-kubeconfig
+things: null
+items: null
+"""
+
 
 # Test the output of --help
 def test_print_help():
@@ -199,6 +204,37 @@ def test_merge(mock_cli, fake_kubeconfig_file):
     assert result["dictionary"] == yaml_a["dictionary"]
 
 
+def test_merge_to_empty_config(mock_cli, fake_kubeconfig_file_without_entries):
+    stdout_buf = io.StringIO()
+    mock_cli.call_operation = mock_call_operation
+
+    file_path = fake_kubeconfig_file_without_entries
+
+    try:
+        with contextlib.redirect_stdout(stdout_buf):
+            plugin.call(
+                [
+                    "--label",
+                    "nonempty_data",
+                    "--kubeconfig",
+                    file_path,
+                    "--dry-run",
+                ],
+                PluginContext("REALTOKEN", mock_cli),
+            )
+    except SystemExit as err:
+        assert err.code == 0
+
+    result = yaml.safe_load(stdout_buf.getvalue())
+    yaml_a = yaml.safe_load(TEST_YAML_EMPTY_CONFIG)
+    yaml_b = yaml.safe_load(TEST_YAML_CONTENT_B)
+
+    assert result["name"] == yaml_a["name"]
+    assert result["things"] is None
+    assert result["items"] == yaml_b["items"]
+    assert result["dictionary"] == yaml_b["dictionary"]
+
+
 @pytest.fixture(scope="session", autouse=True)
 def fake_kubeconfig_file():
     with tempfile.NamedTemporaryFile(delete=False) as fp:
@@ -220,6 +256,17 @@ def fake_empty_file():
     os.remove(file_path)
 
 
+@pytest.fixture(scope="session", autouse=True)
+def fake_kubeconfig_file_without_entries():
+    with tempfile.NamedTemporaryFile("wt", delete=False) as fp:
+        fp.write(TEST_YAML_EMPTY_CONFIG)
+        file_path = fp.name
+
+    yield file_path
+
+    os.remove(file_path)
+
+
 def mock_call_operation(command, action, **kwargs):
     if (
         command == "lke"