Merge pull request #726 from nolar/fix-cluster-scoped-hierarchies

Fix absent namespaces for cluster-scoped resources in hierarchies

kopf/toolkits/hierarchies.py

@@ -2,6 +2,7 @@
 All the functions to properly build the object hierarchies.
 """
 import collections.abc
+import enum
 import warnings
 from typing import Any, Iterable, Iterator, Mapping, MutableMapping, Optional, Union, cast
 
@@ -13,6 +14,10 @@
 K8sObjects = Union[K8sObject, Iterable[K8sObject]]
 
 
+class _UNSET(enum.Enum):
+    token = enum.auto()
+
+
 def append_owner_reference(
         objs: K8sObjects,
         owner: Optional[bodies.Body] = None,
@@ -82,7 +87,7 @@ def remove_owner_reference(
 
 def label(
         objs: K8sObjects,
-        labels: Optional[Mapping[str, Union[None, str]]] = None,
+        labels: Union[Mapping[str, Union[None, str]], _UNSET] = _UNSET.token,
         *,
         forced: bool = False,
         nested: Optional[Union[str, Iterable[dicts.FieldSpec]]] = None,
@@ -97,9 +102,11 @@ def label(
         forced = force
 
     # Try to use the current object being handled if possible.
-    if labels is None:
+    if isinstance(labels, _UNSET):
         real_owner = _guess_owner(None)
         labels = real_owner.get('metadata', {}).get('labels', {})
+    if isinstance(labels, _UNSET):
+        raise RuntimeError("Impossible error: labels are not resolved.")  # for type-checking
 
     # Set labels based on the explicitly specified or guessed ones.
     for obj in cast(Iterator[K8sObject], dicts.walk(objs, nested=nested)):
@@ -124,7 +131,7 @@ def label(
 
 def harmonize_naming(
         objs: K8sObjects,
-        name: Optional[str] = None,
+        name: Union[None, str, _UNSET] = _UNSET.token,
         *,
         forced: bool = False,
         strict: bool = False,
@@ -145,9 +152,11 @@ def harmonize_naming(
     """
 
     # Try to use the current object being handled if possible.
-    if name is None:
+    if isinstance(name, _UNSET):
         real_owner = _guess_owner(None)
         name = real_owner.get('metadata', {}).get('name', None)
+    if isinstance(name, _UNSET):
+        raise RuntimeError("Impossible error: the name is not resolved.")  # for type-checking
     if name is None:
         raise LookupError("Name must be set explicitly: couldn't find it automatically.")
 
@@ -184,7 +193,7 @@ def harmonize_naming(
 
 def adjust_namespace(
         objs: K8sObjects,
-        namespace: Optional[str] = None,
+        namespace: Union[None, str, _UNSET] = _UNSET.token,
         *,
         forced: bool = False,
 ) -> None:
@@ -198,11 +207,11 @@ def adjust_namespace(
     """
 
     # Try to use the current object being handled if possible.
-    if namespace is None:
+    if isinstance(namespace, _UNSET):
         real_owner = _guess_owner(None)
         namespace = real_owner.get('metadata', {}).get('namespace', None)
-    if namespace is None:
-        raise LookupError("Namespace must be set explicitly: couldn't find it automatically.")
+    if isinstance(namespace, _UNSET):
+        raise RuntimeError("Impossible error: the namespace is not resolved.")  # for type-checking
 
     # Set namespace based on the explicitly specified or guessed namespace.
     for obj in cast(Iterator[K8sObject], dicts.walk(objs)):

tests/hierarchies/test_contextual_owner.py

@@ -108,10 +108,11 @@ def test_when_empty_for_name_harmonization(owner):
 
 
 def test_when_empty_for_namespace_adjustment(owner):
+    # An absent namespace means a cluster-scoped resource -- a valid case.
+    obj = {}
     owner._replace_with({})
-    with pytest.raises(LookupError) as e:
-        kopf.adjust_namespace([])
-    assert 'Namespace must be set explicitly' in str(e.value)
+    kopf.adjust_namespace(obj)
+    assert obj['metadata']['namespace'] is None
 
 
 def test_when_empty_for_adopting(owner):