Refactoring of k8s manager and tests
Changes:
- Official kubernetes python lib
- Rewrite k8s api wrapper in OOP manner
- Use api where its possible instead of cli
- Remove virtlet code because its can be replaced with pod api
- Remove unused/oudated manager code
- Remove bug workaround in k8s upgrade template
- Remove netchecker obsolete code
- Remove unfinished test_rbd_flexvolume_driver
Change-Id: I446a240123282196a6ba54f588aea84791f175ba
Related-PROD: PROD-21700
diff --git a/tcp_tests/managers/k8s/nodes.py b/tcp_tests/managers/k8s/nodes.py
index c6d4dbe..4b0451e 100644
--- a/tcp_tests/managers/k8s/nodes.py
+++ b/tcp_tests/managers/k8s/nodes.py
@@ -12,70 +12,41 @@
# License for the specific language governing permissions and limitations
+from kubernetes import client
+
from tcp_tests.managers.k8s.base import K8sBaseResource
from tcp_tests.managers.k8s.base import K8sBaseManager
class K8sNode(K8sBaseResource):
- """docstring for ClassName"""
+ resource_type = 'node'
- def __repr__(self):
- return "<K8sNode: %s>" % self.name
+ def _read(self, **kwargs):
+ return self._manager.api.read_node(self.name, **kwargs)
- @property
- def name(self):
- return self.metadata.name
+ def _create(self, body, **kwargs):
+ return self._manager.api.create_node(body, **kwargs)
- @property
- def labels(self):
- return self.metadata.labels
+ def _patch(self, body, **kwargs):
+ return self._manager.api.patch_node(self.name, body, **kwargs)
- @labels.setter
- def labels(self, labels):
- current_labels = {
- label: None for label in self.labels
- }
- current_labels.update(labels)
- self.add_labels(labels=current_labels)
+ def _replace(self, body, **kwargs):
+ return self._manager.api.replace_node(self.name, body, **kwargs)
- def add_labels(self, labels):
- if not isinstance(labels, dict):
- raise TypeError("labels must be a dict!")
- body = {
- "metadata":
- {
- "labels": labels
- }
- }
- self._add_details(self._manager.update(body=body, name=self.name))
-
- def remove_labels(self, list_labels):
- labels = {label: None for label in list_labels}
- self.add_labels(labels=labels)
+ def _delete(self, **kwargs):
+ self._manager.api.delete_node(
+ self.name, client.V1DeleteOptions(), **kwargs)
class K8sNodeManager(K8sBaseManager):
- """docstring for ClassName"""
-
resource_class = K8sNode
- def _get(self, name, **kwargs):
- return self.api.read_namespaced_node(name=name, **kwargs)
+ @property
+ def api(self):
+ return self._cluster.api_core
- def _list(self, **kwargs):
- return self.api.list_namespaced_node(**kwargs)
+ def _list(self, namespace, **kwargs):
+ return self.api.list_node(**kwargs)
- def _create(self, body, **kwargs):
- return self.api.create_namespaced_node(body=body, **kwargs)
-
- def _replace(self, body, name, **kwargs):
- return self.api.replace_namespaced_node(body=body, name=name, **kwargs)
-
- def _delete(self, body, name, **kwargs):
- return self.api.delete_namespaced_node(body=body, name=name, **kwargs)
-
- def _deletecollection(self, **kwargs):
- return self.api.deletecollection_namespaced_node(**kwargs)
-
- def update(self, body, name, **kwargs):
- return self.api.patch_namespaced_node(body=body, name=name, **kwargs)
+ def _list_all(self, **kwargs):
+ return self._list(None, **kwargs)