Merge pull request #120 from ClarkSource/fix-context-values

fix: make sure context values are not overwritten

k8t/project.py

@@ -8,33 +8,33 @@
 # THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 
 import os
-from typing import List
+from typing import List, Optional
 
 
 def check_directory(path: str) -> bool:
     return os.path.exists(os.path.join(path, ".k8t"))
 
 
-def get_base_dir(root: str, cluster: str, environment: str) -> str:
+def get_base_dir(root: str, cluster: Optional[str], environment: Optional[str]) -> str:
     base_path = root
 
     if cluster is not None:
         base_path = os.path.join(base_path, "clusters", cluster)
 
         if not os.path.isdir(base_path):
-            raise RuntimeError("No such cluster: {}".format(cluster))
+            raise RuntimeError(f"No such cluster: {cluster}")
 
     if environment is not None:
         base_path = os.path.join(base_path, "environments", environment)
 
         if not os.path.isdir(base_path):
-            raise RuntimeError("No such environment: {}".format(environment))
+            raise RuntimeError(f"No such environment: {environment}")
 
     return base_path
 
 
 # pylint: disable=too-many-arguments
-def find_files(root: str, cluster: str, environment: str, name: str, file_ok=True, dir_ok=True) -> List[str]:
+def find_files(root: str, cluster: Optional[str], environment: Optional[str], name: str, file_ok=True, dir_ok=True) -> List[str]:
     def check(path):
         return (file_ok and os.path.isfile(path)) or (dir_ok and os.path.isdir(root_path))
 
@@ -58,7 +58,7 @@ def check(path):
         cluster_path = os.path.join(root, "clusters", cluster)
 
         if not os.path.isdir(cluster_path):
-            raise RuntimeError("no such cluster: {}".format(cluster))
+            raise RuntimeError(f"no such cluster: {cluster}")
 
         file_path = os.path.join(cluster_path, name)
 
@@ -74,6 +74,6 @@ def check(path):
                 env_found = True
 
     if not env_found:
-        raise RuntimeError("no such environment: {}".format(environment))
+        raise RuntimeError(f"no such environment: {environment}")
 
     return files

k8t/util.py

@@ -10,13 +10,14 @@
 import copy
 import json
 import logging
-import math
-import bitmath
 import os
 import shutil
+
 from functools import reduce
 from typing import Any, Dict, List, Tuple
 
+import bitmath
+
 from click import secho  # pylint: disable=E0401
 from ruamel.yaml import YAML  # pylint: disable=E0401
 from simple_tools.interaction import confirm  # pylint: disable=E0401
@@ -41,7 +42,7 @@ def touch(fname: str, mode=0o666, dir_fd=None, **kwargs) -> None:
 def makedirs(path, warn_exists=True):
     if os.path.exists(path):
         if warn_exists:
-            if confirm("directory {} already exists, go ahead?".format(path)):
+            if confirm(f"directory {path} already exists, go ahead?"):
                 secho(f"Directory exists: {path}", fg="yellow")
 
                 return
@@ -84,9 +85,11 @@ def merge(d_1: dict, d_2: dict, path=None, method="ltr"):
                 elif method == "ask":
                     raise NotImplementedError('Merge method "ask"')
                 elif method == "crash":
-                    raise Exception("Conflict at %s" % ".".join(path + [str(key)]))
+                    value_path = ".".join(path + [str(key)])
+
+                    raise RuntimeError(f"Conflict at {value_path}")
                 else:
-                    raise Exception("Invalid merge method: %s" % method)
+                    raise RuntimeError(f"Invalid merge method: {method}")
         else:
             d_1[key] = d_2[key]
 
@@ -107,7 +110,8 @@ def load_yaml(path: str) -> dict:
 
     with open(path, "r") as stream:
         yaml = YAML(typ="safe", pure=True)
-        return yaml.load(stream) or dict()
+
+        return yaml.load(stream) or {}
 
 
 def load_cli_value(key: str, value: str) -> Tuple[str, Any]:

k8t/values.py

@@ -22,12 +22,12 @@ def load_all(root: str, cluster: str, environment: str, method: str) -> Dict[str
 
     LOGGER.debug("using value files: %s", values)
 
-    base = dict()
+    context = dict()
 
     if cluster is not None:
-        base['cluster'] = cluster
+        context['cluster'] = cluster
 
     if environment is not None:
-        base['environment'] = environment
+        context['environment'] = environment
 
-    return deep_merge(base, *[load_yaml(f) for f in values], method=method)
+    return deep_merge(*[load_yaml(f) for f in values], context, method=method)

tests/resources/bad/values.yaml

@@ -1,2 +1,2 @@
 test:
-  foo: bar
+  foo: bar

tests/resources/good/values.yaml

@@ -2,3 +2,7 @@ cls: none
 env: none
 foo: 1
 bar: test
+
+# these should be overwritten
+cluster: foobar
+environment: foobaz