Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use show_namespace method in tempest

Best Python code snippet using tempest_python

test_kubernetes.py

Source:test_kubernetes.py

1# -*- coding: utf-8 -*-2"""3    :codeauthor: :email:`Jeff Schroeder <jeffschroeder@computer.org>`4"""5# Import Python libs6from __future__ import absolute_import, print_function, unicode_literals7import base648from contextlib import contextmanager9# Import Salt Libs10import salt.utils.stringutils11from salt.ext import six12from salt.states import kubernetes13# Import Salt Testing Libs14from tests.support.mixins import LoaderModuleMockMixin15from tests.support.mock import MagicMock, patch16from tests.support.unit import TestCase, skipIf17@skipIf(18    kubernetes is False,19    "Probably Kubernetes client lib is not installed. \20                              Skipping test_kubernetes.py",21)22class KubernetesTestCase(TestCase, LoaderModuleMockMixin):23    """24    Test cases for salt.states.kubernetes25    """26    def setup_loader_modules(self):27        return {kubernetes: {"__env__": "base"}}28    @contextmanager29    def mock_func(self, func_name, return_value, test=False):30        """31        Mock any of the kubernetes state function return values and set32        the test options.33        """34        name = "kubernetes.{0}".format(func_name)35        mocked = {name: MagicMock(return_value=return_value)}36        with patch.dict(kubernetes.__salt__, mocked) as patched:37            with patch.dict(kubernetes.__opts__, {"test": test}):38                yield patched39    def make_configmap(self, name, namespace="default", data=None):40        return self.make_ret_dict(41            kind="ConfigMap", name=name, namespace=namespace, data=data,42        )43    def make_secret(self, name, namespace="default", data=None):44        secret_data = self.make_ret_dict(45            kind="Secret", name=name, namespace=namespace, data=data,46        )47        # Base64 all of the values just like kubectl does48        for key, value in six.iteritems(secret_data["data"]):49            secret_data["data"][key] = base64.b64encode(50                salt.utils.stringutils.to_bytes(value)51            )52        return secret_data53    def make_node_labels(self, name="minikube"):54        return {55            "kubernetes.io/hostname": name,56            "beta.kubernetes.io/os": "linux",57            "beta.kubernetes.io/arch": "amd64",58            "failure-domain.beta.kubernetes.io/region": "us-west-1",59        }60    def make_node(self, name="minikube"):61        node_data = self.make_ret_dict(kind="Node", name="minikube")62        node_data.update(63            {64                "api_version": "v1",65                "kind": "Node",66                "metadata": {67                    "annotations": {"node.alpha.kubernetes.io/ttl": "0"},68                    "labels": self.make_node_labels(name=name),69                    "name": name,70                    "namespace": None,71                    "self_link": "/api/v1/nodes/{name}".format(name=name),72                    "uid": "7811b8ae-c1a1-11e7-a55a-0800279fb61e",73                },74                "spec": {"external_id": name},75                "status": {},76            }77        )78        return node_data79    def make_namespace(self, name="default"):80        namespace_data = self.make_ret_dict(kind="Namespace", name=name)81        del namespace_data["data"]82        namespace_data.update(83            {84                "status": {"phase": "Active"},85                "spec": {"finalizers": ["kubernetes"]},86                "metadata": {87                    "name": name,88                    "namespace": None,89                    "labels": None,90                    "self_link": "/api/v1/namespaces/{namespace}".format(91                        namespace=name,92                    ),93                    "annotations": None,94                    "uid": "752fceeb-c1a1-11e7-a55a-0800279fb61e",95                },96            }97        )98        return namespace_data99    def make_ret_dict(self, kind, name, namespace=None, data=None):100        """101        Make a minimal example configmap or secret for using in mocks102        """103        assert kind in ("Secret", "ConfigMap", "Namespace", "Node")104        if data is None:105            data = {}106        self_link = "/api/v1/namespaces/{namespace}/{kind}s/{name}".format(107            namespace=namespace, kind=kind.lower(), name=name,108        )109        return_data = {110            "kind": kind,111            "data": data,112            "api_version": "v1",113            "metadata": {114                "name": name,115                "labels": None,116                "namespace": namespace,117                "self_link": self_link,118                "annotations": {"kubernetes.io/change-cause": "salt-call state.apply"},119            },120        }121        return return_data122    def test_configmap_present__fail(self):123        error = kubernetes.configmap_present(124            name="testme", data={1: 1}, source="salt://beyond/oblivion.jinja",125        )126        self.assertDictEqual(127            {128                "changes": {},129                "result": False,130                "name": "testme",131                "comment": "'source' cannot be used in combination with 'data'",132            },133            error,134        )135    def test_configmap_present__create_test_true(self):136        # Create a new configmap with test=True137        with self.mock_func("show_configmap", return_value=None, test=True):138            ret = kubernetes.configmap_present(139                name="example", data={"example.conf": "# empty config file"},140            )141            self.assertDictEqual(142                {143                    "comment": "The configmap is going to be created",144                    "changes": {},145                    "name": "example",146                    "result": None,147                },148                ret,149            )150    def test_configmap_present__create(self):151        # Create a new configmap152        with self.mock_func("show_configmap", return_value=None):153            cm = self.make_configmap(154                name="test", namespace="default", data={"foo": "bar"},155            )156            with self.mock_func("create_configmap", return_value=cm):157                actual = kubernetes.configmap_present(name="test", data={"foo": "bar"},)158                self.assertDictEqual(159                    {160                        "comment": "",161                        "changes": {"data": {"foo": "bar"}},162                        "name": "test",163                        "result": True,164                    },165                    actual,166                )167    def test_configmap_present__create_no_data(self):168        # Create a new configmap with no 'data' attribute169        with self.mock_func("show_configmap", return_value=None):170            cm = self.make_configmap(name="test", namespace="default",)171            with self.mock_func("create_configmap", return_value=cm):172                actual = kubernetes.configmap_present(name="test")173                self.assertDictEqual(174                    {175                        "comment": "",176                        "changes": {"data": {}},177                        "name": "test",178                        "result": True,179                    },180                    actual,181                )182    def test_configmap_present__replace_test_true(self):183        cm = self.make_configmap(184            name="settings",185            namespace="saltstack",186            data={"foobar.conf": "# Example configuration"},187        )188        with self.mock_func("show_configmap", return_value=cm, test=True):189            ret = kubernetes.configmap_present(190                name="settings",191                namespace="saltstack",192                data={"foobar.conf": "# Example configuration"},193            )194            self.assertDictEqual(195                {196                    "comment": "The configmap is going to be replaced",197                    "changes": {},198                    "name": "settings",199                    "result": None,200                },201                ret,202            )203    def test_configmap_present__replace(self):204        cm = self.make_configmap(name="settings", data={"action": "make=war"})205        # Replace an existing configmap206        with self.mock_func("show_configmap", return_value=cm):207            new_cm = cm.copy()208            new_cm.update({"data": {"action": "make=peace"}})209            with self.mock_func("replace_configmap", return_value=new_cm):210                actual = kubernetes.configmap_present(211                    name="settings", data={"action": "make=peace"},212                )213                self.assertDictEqual(214                    {215                        "comment": "The configmap is already present. Forcing recreation",216                        "changes": {"data": {"action": "make=peace"}},217                        "name": "settings",218                        "result": True,219                    },220                    actual,221                )222    def test_configmap_absent__noop_test_true(self):223        # Nothing to delete with test=True224        with self.mock_func("show_configmap", return_value=None, test=True):225            actual = kubernetes.configmap_absent(name="NOT_FOUND")226            self.assertDictEqual(227                {228                    "comment": "The configmap does not exist",229                    "changes": {},230                    "name": "NOT_FOUND",231                    "result": None,232                },233                actual,234            )235    def test_configmap_absent__test_true(self):236        # Configmap exists with test=True237        cm = self.make_configmap(name="deleteme", namespace="default")238        with self.mock_func("show_configmap", return_value=cm, test=True):239            actual = kubernetes.configmap_absent(name="deleteme")240            self.assertDictEqual(241                {242                    "comment": "The configmap is going to be deleted",243                    "changes": {},244                    "name": "deleteme",245                    "result": None,246                },247                actual,248            )249    def test_configmap_absent__noop(self):250        # Nothing to delete251        with self.mock_func("show_configmap", return_value=None):252            actual = kubernetes.configmap_absent(name="NOT_FOUND")253            self.assertDictEqual(254                {255                    "comment": "The configmap does not exist",256                    "changes": {},257                    "name": "NOT_FOUND",258                    "result": True,259                },260                actual,261            )262    def test_configmap_absent(self):263        # Configmap exists, delete it!264        cm = self.make_configmap(name="deleteme", namespace="default")265        with self.mock_func("show_configmap", return_value=cm):266            # The return from this module isn't used in the state267            with self.mock_func("delete_configmap", return_value={}):268                actual = kubernetes.configmap_absent(name="deleteme")269                self.assertDictEqual(270                    {271                        "comment": "ConfigMap deleted",272                        "changes": {273                            "kubernetes.configmap": {274                                "new": "absent",275                                "old": "present",276                            },277                        },278                        "name": "deleteme",279                        "result": True,280                    },281                    actual,282                )283    def test_secret_present__fail(self):284        actual = kubernetes.secret_present(285            name="sekret", data={"password": "monk3y"}, source="salt://nope.jinja",286        )287        self.assertDictEqual(288            {289                "changes": {},290                "result": False,291                "name": "sekret",292                "comment": "'source' cannot be used in combination with 'data'",293            },294            actual,295        )296    def test_secret_present__exists_test_true(self):297        secret = self.make_secret(name="sekret")298        new_secret = secret.copy()299        new_secret.update({"data": {"password": "uncle"}})300        # Secret exists already and needs replacing with test=True301        with self.mock_func("show_secret", return_value=secret):302            with self.mock_func("replace_secret", return_value=new_secret, test=True):303                actual = kubernetes.secret_present(304                    name="sekret", data={"password": "uncle"},305                )306                self.assertDictEqual(307                    {308                        "changes": {},309                        "result": None,310                        "name": "sekret",311                        "comment": "The secret is going to be replaced",312                    },313                    actual,314                )315    def test_secret_present__exists(self):316        # Secret exists and gets replaced317        secret = self.make_secret(name="sekret", data={"password": "booyah"})318        with self.mock_func("show_secret", return_value=secret):319            with self.mock_func("replace_secret", return_value=secret):320                actual = kubernetes.secret_present(321                    name="sekret", data={"password": "booyah"},322                )323                self.assertDictEqual(324                    {325                        "changes": {"data": ["password"]},326                        "result": True,327                        "name": "sekret",328                        "comment": "The secret is already present. Forcing recreation",329                    },330                    actual,331                )332    def test_secret_present__create(self):333        # Secret exists and gets replaced334        secret = self.make_secret(name="sekret", data={"password": "booyah"})335        with self.mock_func("show_secret", return_value=None):336            with self.mock_func("create_secret", return_value=secret):337                actual = kubernetes.secret_present(338                    name="sekret", data={"password": "booyah"},339                )340                self.assertDictEqual(341                    {342                        "changes": {"data": ["password"]},343                        "result": True,344                        "name": "sekret",345                        "comment": "",346                    },347                    actual,348                )349    def test_secret_present__create_no_data(self):350        # Secret exists and gets replaced351        secret = self.make_secret(name="sekret")352        with self.mock_func("show_secret", return_value=None):353            with self.mock_func("create_secret", return_value=secret):354                actual = kubernetes.secret_present(name="sekret")355                self.assertDictEqual(356                    {357                        "changes": {"data": []},358                        "result": True,359                        "name": "sekret",360                        "comment": "",361                    },362                    actual,363                )364    def test_secret_present__create_test_true(self):365        # Secret exists and gets replaced with test=True366        secret = self.make_secret(name="sekret")367        with self.mock_func("show_secret", return_value=None):368            with self.mock_func("create_secret", return_value=secret, test=True):369                actual = kubernetes.secret_present(name="sekret")370                self.assertDictEqual(371                    {372                        "changes": {},373                        "result": None,374                        "name": "sekret",375                        "comment": "The secret is going to be created",376                    },377                    actual,378                )379    def test_secret_absent__noop_test_true(self):380        with self.mock_func("show_secret", return_value=None, test=True):381            actual = kubernetes.secret_absent(name="sekret")382            self.assertDictEqual(383                {384                    "changes": {},385                    "result": None,386                    "name": "sekret",387                    "comment": "The secret does not exist",388                },389                actual,390            )391    def test_secret_absent__noop(self):392        with self.mock_func("show_secret", return_value=None):393            actual = kubernetes.secret_absent(name="passwords")394            self.assertDictEqual(395                {396                    "changes": {},397                    "result": True,398                    "name": "passwords",399                    "comment": "The secret does not exist",400                },401                actual,402            )403    def test_secret_absent__delete_test_true(self):404        secret = self.make_secret(name="credentials", data={"redis": "letmein"})405        with self.mock_func("show_secret", return_value=secret):406            with self.mock_func("delete_secret", return_value=secret, test=True):407                actual = kubernetes.secret_absent(name="credentials")408                self.assertDictEqual(409                    {410                        "changes": {},411                        "result": None,412                        "name": "credentials",413                        "comment": "The secret is going to be deleted",414                    },415                    actual,416                )417    def test_secret_absent__delete(self):418        secret = self.make_secret(name="foobar", data={"redis": "letmein"})419        deleted = {420            "status": None,421            "kind": "Secret",422            "code": None,423            "reason": None,424            "details": None,425            "message": None,426            "api_version": "v1",427            "metadata": {428                "self_link": "/api/v1/namespaces/default/secrets/foobar",429                "resource_version": "30292",430            },431        }432        with self.mock_func("show_secret", return_value=secret):433            with self.mock_func("delete_secret", return_value=deleted):434                actual = kubernetes.secret_absent(name="foobar")435                self.assertDictEqual(436                    {437                        "changes": {438                            "kubernetes.secret": {"new": "absent", "old": "present"},439                        },440                        "result": True,441                        "name": "foobar",442                        "comment": "Secret deleted",443                    },444                    actual,445                )446    def test_node_label_present__add_test_true(self):447        labels = self.make_node_labels()448        with self.mock_func("node_labels", return_value=labels, test=True):449            actual = kubernetes.node_label_present(450                name="com.zoo-animal", node="minikube", value="monkey",451            )452            self.assertDictEqual(453                {454                    "changes": {},455                    "result": None,456                    "name": "com.zoo-animal",457                    "comment": "The label is going to be set",458                },459                actual,460            )461    def test_node_label_present__add(self):462        node_data = self.make_node()463        # Remove some of the defaults to make it simpler464        node_data["metadata"]["labels"] = {465            "beta.kubernetes.io/os": "linux",466        }467        labels = node_data["metadata"]["labels"]468        with self.mock_func("node_labels", return_value=labels):469            with self.mock_func("node_add_label", return_value=node_data):470                actual = kubernetes.node_label_present(471                    name="failure-domain.beta.kubernetes.io/zone",472                    node="minikube",473                    value="us-central1-a",474                )475                self.assertDictEqual(476                    {477                        "comment": "",478                        "changes": {479                            "minikube.failure-domain.beta.kubernetes.io/zone": {480                                "new": {481                                    "failure-domain.beta.kubernetes.io/zone": "us-central1-a",482                                    "beta.kubernetes.io/os": "linux",483                                },484                                "old": {"beta.kubernetes.io/os": "linux"},485                            },486                        },487                        "name": "failure-domain.beta.kubernetes.io/zone",488                        "result": True,489                    },490                    actual,491                )492    def test_node_label_present__already_set(self):493        node_data = self.make_node()494        labels = node_data["metadata"]["labels"]495        with self.mock_func("node_labels", return_value=labels):496            with self.mock_func("node_add_label", return_value=node_data):497                actual = kubernetes.node_label_present(498                    name="failure-domain.beta.kubernetes.io/region",499                    node="minikube",500                    value="us-west-1",501                )502                self.assertDictEqual(503                    {504                        "changes": {},505                        "result": True,506                        "name": "failure-domain.beta.kubernetes.io/region",507                        "comment": "The label is already set and has the specified value",508                    },509                    actual,510                )511    def test_node_label_present__update_test_true(self):512        node_data = self.make_node()513        labels = node_data["metadata"]["labels"]514        with self.mock_func("node_labels", return_value=labels):515            with self.mock_func("node_add_label", return_value=node_data, test=True):516                actual = kubernetes.node_label_present(517                    name="failure-domain.beta.kubernetes.io/region",518                    node="minikube",519                    value="us-east-1",520                )521                self.assertDictEqual(522                    {523                        "changes": {},524                        "result": None,525                        "name": "failure-domain.beta.kubernetes.io/region",526                        "comment": "The label is going to be updated",527                    },528                    actual,529                )530    def test_node_label_present__update(self):531        node_data = self.make_node()532        # Remove some of the defaults to make it simpler533        node_data["metadata"]["labels"] = {534            "failure-domain.beta.kubernetes.io/region": "us-west-1",535        }536        labels = node_data["metadata"]["labels"]537        with self.mock_func("node_labels", return_value=labels):538            with self.mock_func("node_add_label", return_value=node_data):539                actual = kubernetes.node_label_present(540                    name="failure-domain.beta.kubernetes.io/region",541                    node="minikube",542                    value="us-east-1",543                )544                self.assertDictEqual(545                    {546                        "changes": {547                            "minikube.failure-domain.beta.kubernetes.io/region": {548                                "new": {549                                    "failure-domain.beta.kubernetes.io/region": "us-east-1"550                                },551                                "old": {552                                    "failure-domain.beta.kubernetes.io/region": "us-west-1"553                                },554                            }555                        },556                        "result": True,557                        "name": "failure-domain.beta.kubernetes.io/region",558                        "comment": "The label is already set, changing the value",559                    },560                    actual,561                )562    def test_node_label_absent__noop_test_true(self):563        labels = self.make_node_labels()564        with self.mock_func("node_labels", return_value=labels, test=True):565            actual = kubernetes.node_label_absent(566                name="non-existent-label", node="minikube",567            )568            self.assertDictEqual(569                {570                    "changes": {},571                    "result": None,572                    "name": "non-existent-label",573                    "comment": "The label does not exist",574                },575                actual,576            )577    def test_node_label_absent__noop(self):578        labels = self.make_node_labels()579        with self.mock_func("node_labels", return_value=labels):580            actual = kubernetes.node_label_absent(581                name="non-existent-label", node="minikube",582            )583            self.assertDictEqual(584                {585                    "changes": {},586                    "result": True,587                    "name": "non-existent-label",588                    "comment": "The label does not exist",589                },590                actual,591            )592    def test_node_label_absent__delete_test_true(self):593        labels = self.make_node_labels()594        with self.mock_func("node_labels", return_value=labels, test=True):595            actual = kubernetes.node_label_absent(596                name="failure-domain.beta.kubernetes.io/region", node="minikube",597            )598            self.assertDictEqual(599                {600                    "changes": {},601                    "result": None,602                    "name": "failure-domain.beta.kubernetes.io/region",603                    "comment": "The label is going to be deleted",604                },605                actual,606            )607    def test_node_label_absent__delete(self):608        node_data = self.make_node()609        labels = node_data["metadata"]["labels"].copy()610        node_data["metadata"]["labels"].pop("failure-domain.beta.kubernetes.io/region")611        with self.mock_func("node_labels", return_value=labels):612            with self.mock_func("node_remove_label", return_value=node_data):613                actual = kubernetes.node_label_absent(614                    name="failure-domain.beta.kubernetes.io/region", node="minikube",615                )616                self.assertDictEqual(617                    {618                        "result": True,619                        "changes": {620                            "kubernetes.node_label": {621                                "new": "absent",622                                "old": "present",623                            }624                        },625                        "comment": "Label removed from node",626                        "name": "failure-domain.beta.kubernetes.io/region",627                    },628                    actual,629                )630    def test_namespace_present__create_test_true(self):631        with self.mock_func("show_namespace", return_value=None, test=True):632            actual = kubernetes.namespace_present(name="saltstack")633            self.assertDictEqual(634                {635                    "changes": {},636                    "result": None,637                    "name": "saltstack",638                    "comment": "The namespace is going to be created",639                },640                actual,641            )642    def test_namespace_present__create(self):643        namespace_data = self.make_namespace(name="saltstack")644        with self.mock_func("show_namespace", return_value=None):645            with self.mock_func("create_namespace", return_value=namespace_data):646                actual = kubernetes.namespace_present(name="saltstack")647                self.assertDictEqual(648                    {649                        "changes": {"namespace": {"new": namespace_data, "old": {}}},650                        "result": True,651                        "name": "saltstack",652                        "comment": "",653                    },654                    actual,655                )656    def test_namespace_present__noop_test_true(self):657        namespace_data = self.make_namespace(name="saltstack")658        with self.mock_func("show_namespace", return_value=namespace_data, test=True):659            actual = kubernetes.namespace_present(name="saltstack")660            self.assertDictEqual(661                {662                    "changes": {},663                    "result": None,664                    "name": "saltstack",665                    "comment": "The namespace already exists",666                },667                actual,668            )669    def test_namespace_present__noop(self):670        namespace_data = self.make_namespace(name="saltstack")671        with self.mock_func("show_namespace", return_value=namespace_data):672            actual = kubernetes.namespace_present(name="saltstack")673            self.assertDictEqual(674                {675                    "changes": {},676                    "result": True,677                    "name": "saltstack",678                    "comment": "The namespace already exists",679                },680                actual,681            )682    def test_namespace_absent__noop_test_true(self):683        with self.mock_func("show_namespace", return_value=None, test=True):684            actual = kubernetes.namespace_absent(name="salt")685            self.assertDictEqual(686                {687                    "changes": {},688                    "result": None,689                    "name": "salt",690                    "comment": "The namespace does not exist",691                },692                actual,693            )694    def test_namespace_absent__noop(self):695        with self.mock_func("show_namespace", return_value=None):696            actual = kubernetes.namespace_absent(name="salt")697            self.assertDictEqual(698                {699                    "changes": {},700                    "result": True,701                    "name": "salt",702                    "comment": "The namespace does not exist",703                },704                actual,705            )706    def test_namespace_absent__delete_test_true(self):707        namespace_data = self.make_namespace(name="salt")708        with self.mock_func("show_namespace", return_value=namespace_data, test=True):709            actual = kubernetes.namespace_absent(name="salt")710            self.assertDictEqual(711                {712                    "changes": {},713                    "result": None,714                    "name": "salt",715                    "comment": "The namespace is going to be deleted",716                },717                actual,718            )719    def test_namespace_absent__delete_code_200(self):720        namespace_data = self.make_namespace(name="salt")721        deleted = namespace_data.copy()722        deleted["code"] = 200723        deleted.update({"code": 200, "message": None})724        with self.mock_func("show_namespace", return_value=namespace_data):725            with self.mock_func("delete_namespace", return_value=deleted):726                actual = kubernetes.namespace_absent(name="salt")727                self.assertDictEqual(728                    {729                        "changes": {730                            "kubernetes.namespace": {"new": "absent", "old": "present"}731                        },732                        "result": True,733                        "name": "salt",734                        "comment": "Terminating",735                    },736                    actual,737                )738    def test_namespace_absent__delete_status_terminating(self):739        namespace_data = self.make_namespace(name="salt")740        deleted = namespace_data.copy()741        deleted.update(742            {743                "code": None,744                "status": "Terminating namespace",745                "message": "Terminating this shizzzle yo",746            }747        )748        with self.mock_func("show_namespace", return_value=namespace_data):749            with self.mock_func("delete_namespace", return_value=deleted):750                actual = kubernetes.namespace_absent(name="salt")751                self.assertDictEqual(752                    {753                        "changes": {754                            "kubernetes.namespace": {"new": "absent", "old": "present"}755                        },756                        "result": True,757                        "name": "salt",758                        "comment": "Terminating this shizzzle yo",759                    },760                    actual,761                )762    def test_namespace_absent__delete_status_phase_terminating(self):763        # This is what kubernetes 1.8.0 looks like when deleting namespaces764        namespace_data = self.make_namespace(name="salt")765        deleted = namespace_data.copy()766        deleted.update(767            {"code": None, "message": None, "status": {"phase": "Terminating"}}768        )769        with self.mock_func("show_namespace", return_value=namespace_data):770            with self.mock_func("delete_namespace", return_value=deleted):771                actual = kubernetes.namespace_absent(name="salt")772                self.assertDictEqual(773                    {774                        "changes": {775                            "kubernetes.namespace": {"new": "absent", "old": "present"}776                        },777                        "result": True,778                        "name": "salt",779                        "comment": "Terminating",780                    },781                    actual,782                )783    def test_namespace_absent__delete_error(self):784        namespace_data = self.make_namespace(name="salt")785        deleted = namespace_data.copy()786        deleted.update({"code": 418, "message": "I' a teapot!", "status": None})787        with self.mock_func("show_namespace", return_value=namespace_data):788            with self.mock_func("delete_namespace", return_value=deleted):789                actual = kubernetes.namespace_absent(name="salt")790                self.assertDictEqual(791                    {792                        "changes": {},793                        "result": False,794                        "name": "salt",795                        "comment": "Something went wrong, response: {0}".format(796                            deleted,797                        ),798                    },799                    actual,...

list.py

Source:list.py

1# -*- coding: utf-8 -*- #2# Copyright 2017 Google LLC. All Rights Reserved.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8#    http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15"""Command for listing available services."""16from __future__ import absolute_import17from __future__ import division18from __future__ import unicode_literals19from googlecloudsdk.api_lib.run import global_methods20from googlecloudsdk.calliope import base21from googlecloudsdk.command_lib.run import commands22from googlecloudsdk.command_lib.run import connection_context23from googlecloudsdk.command_lib.run import flags24from googlecloudsdk.command_lib.run import pretty_print25from googlecloudsdk.command_lib.run import resource_args26from googlecloudsdk.command_lib.run import serverless_operations27from googlecloudsdk.command_lib.util.concepts import concept_parsers28from googlecloudsdk.command_lib.util.concepts import presentation_specs29from googlecloudsdk.core import log30@base.ReleaseTracks(base.ReleaseTrack.BETA, base.ReleaseTrack.GA)31class List(commands.List):32  """List available services."""33  detailed_help = {34      'DESCRIPTION': """\35          {description}36          """,37      'EXAMPLES': """\38          To list available services:39              $ {command}40          """,41  }42  @classmethod43  def CommonArgs(cls, parser):44    # Flags specific to connecting to a cluster45    cluster_group = flags.GetClusterArgGroup(parser)46    namespace_presentation = presentation_specs.ResourcePresentationSpec(47        '--namespace',48        resource_args.GetNamespaceResourceSpec(),49        'Namespace to list services in.',50        required=True,51        prefixes=False)52    concept_parsers.ConceptParser(53        [namespace_presentation]).AddToParser(cluster_group)54    parser.display_info.AddUriFunc(cls._GetResourceUri)55  @classmethod56  def Args(cls, parser):57    cls.CommonArgs(parser)58  def _SetFormat(self, args, show_region=False, show_namespace=False):59    """Set display format for output.60    Args:61      args: Namespace, the args namespace62      show_region: bool, True to show region of listed services63      show_namespace: bool, True to show namespace of listed services64    """65    columns = [66        pretty_print.READY_COLUMN,67        'firstof(id,metadata.name):label=SERVICE',68    ]69    if show_region:70      columns.append('region:label=REGION')71    if show_namespace:72      columns.append('namespace:label=NAMESPACE')73    columns.extend([74        'domain:label=URL',75        'last_modifier:label="LAST DEPLOYED BY"',76        'last_transition_time:label="LAST DEPLOYED AT"',77    ])78    args.GetDisplayInfo().AddFormat(79        'table({})'.format(','.join(columns)))80  def Run(self, args):81    """List available services."""82    is_managed = flags.GetPlatform() == flags.PLATFORM_MANAGED83    if is_managed and not args.IsSpecified('region'):84      self._SetFormat(args, show_region=True)85      client = global_methods.GetServerlessClientInstance()86      self.SetPartialApiEndpoint(client.url)87      args.CONCEPTS.namespace.Parse()  # Error if no proj.88      # Don't consider region property here, we'll default to all regions89      return commands.SortByName(global_methods.ListServices(client))90    else:91      conn_context = connection_context.GetConnectionContext(92          args, product=connection_context.Product.RUN)93      self._SetFormat(94          args, show_region=is_managed, show_namespace=(not is_managed))95      namespace_ref = args.CONCEPTS.namespace.Parse()96      with serverless_operations.Connect(conn_context) as client:97        self.SetCompleteApiEndpoint(conn_context.endpoint)98        if not is_managed:99          location_msg = ' in [{}]'.format(conn_context.cluster_location)100          log.status.Print('For cluster [{cluster}]{zone}:'.format(101              cluster=conn_context.cluster_name,102              zone=location_msg if conn_context.cluster_location else ''))103        return commands.SortByName(client.ListServices(namespace_ref))104@base.ReleaseTracks(base.ReleaseTrack.ALPHA)105class AlphaList(List):106  """List available services."""107  @classmethod108  def Args(cls, parser):109    cls.CommonArgs(parser)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.