KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use delete_namespace method in lisa

Best Python code snippet using lisa_python

test_k8s.py

Source:test_k8s.py

...32                    assert pod.get_phase() == PodPhase.RUNNING33                finally:34                    pod.delete()35                    assert pod.get_phase() == PodPhase.TERMINATING36                    k8s_backend.delete_namespace(namespace)37    def test_pod_from_template(self):38        template = {39          "apiVersion": "v1",40          "kind": "Pod",41          "metadata": {42            "name": "myapp-pod",43            "labels": {44              "app": "myapp"45            }46          },47          "spec": {48            "containers": [49              {50                "name": "myapp-container",51                "image": "busybox",52                "command": [53                  "sh",54                  "-c",55                  "echo Hello Kubernetes! && sleep 3600"56                ]57              }58            ]59          }60        }61        api_key = get_oc_api_token()62        with K8sBackend(api_key=api_key) as k8s_backend:63            namespace = k8s_backend.create_namespace()64            pod = Pod(namespace=namespace, from_template=template)65            try:66                pod.wait(200)67                assert pod.is_ready()68                assert pod.get_phase() == PodPhase.RUNNING69            finally:70                pod.delete()71                assert pod.get_phase() == PodPhase.TERMINATING72                k8s_backend.delete_namespace(namespace)73    def test_database_deployment(self):74        api_key = get_oc_api_token()75        with K8sBackend(api_key=api_key) as k8s_backend:76            namespace = k8s_backend.create_namespace()77            with DockerBackend() as backend:78                postgres_image = backend.ImageClass("centos/postgresql-10-centos7")79                postgres_image_metadata = postgres_image.get_metadata()80                # set up env variables81                db_env_variables = {"POSTGRESQL_USER": "user",82                                    "POSTGRESQL_PASSWORD": "pass",83                                    "POSTGRESQL_DATABASE": "db"}84                postgres_image_metadata.env_variables.update(db_env_variables)85                db_labels = {"app": "postgres"}86                db_service = Service(name="database", ports=["5432"], selector=db_labels,87                                     namespace=namespace,88                                     create_in_cluster=True)89                db_deployment = Deployment(name="database", selector=db_labels, labels=db_labels,90                                           image_metadata=postgres_image_metadata,91                                           namespace=namespace,92                                           create_in_cluster=True)93                try:94                    db_deployment.wait(200)95                    assert db_deployment.all_pods_ready()96                finally:97                    db_deployment.delete()98                    db_service.delete()99                    k8s_backend.delete_namespace(namespace)100    def test_list_pods(self):101        api_key = get_oc_api_token()102        with K8sBackend(api_key=api_key) as k8s_backend:103            namespace = k8s_backend.create_namespace()104            with DockerBackend() as backend:105                image = backend.ImageClass("openshift/hello-openshift")106                pod = image.run_in_pod(namespace=namespace)107                try:108                    pod.wait(200)109                    assert any(pod.name == p.name for p in k8s_backend.list_pods())110                finally:111                    pod.delete()112                    k8s_backend.delete_namespace(namespace)113    def test_list_services(self):114        api_key = get_oc_api_token()115        with K8sBackend(api_key=api_key) as k8s_backend:116            namespace = k8s_backend.create_namespace()117            labels = {"app": "postgres"}118            service = Service(name="database", ports=["5432"], selector=labels, namespace=namespace,119                              create_in_cluster=True)120            try:121                assert any(service.name == s.name for s in k8s_backend.list_services())122            finally:123                service.delete()124                k8s_backend.delete_namespace(namespace)125    def test_list_deployments(self):126        api_key = get_oc_api_token()127        with K8sBackend(api_key=api_key) as k8s_backend:128            namespace = k8s_backend.create_namespace()129            with DockerBackend() as backend:130                postgres_image = backend.ImageClass("centos/postgresql-10-centos7")131                postgres_image_metadata = postgres_image.get_metadata()132                # set up env variables133                db_env_variables = {"POSTGRESQL_USER": "user",134                                    "POSTGRESQL_PASSWORD": "pass",135                                    "POSTGRESQL_DATABASE": "db"}136                postgres_image_metadata.env_variables.update(db_env_variables)137                db_labels = {"app": "postgres"}138                db_deployment = Deployment(name="database", selector=db_labels, labels=db_labels,139                                           image_metadata=postgres_image_metadata,140                                           namespace=namespace,141                                           create_in_cluster=True)142                try:143                    db_deployment.wait(200)144                    assert db_deployment.all_pods_ready()145                    assert any(db_deployment.name == d.name for d in k8s_backend.list_deployments())146                finally:147                    db_deployment.delete()148                    k8s_backend.delete_namespace(namespace)149    def test_list_pod_for_namespace(self):150        api_key = get_oc_api_token()151        with K8sBackend(api_key=api_key) as k8s_backend:152            namespace1 = k8s_backend.create_namespace()153            namespace2 = k8s_backend.create_namespace()154            with DockerBackend() as backend:155                image = backend.ImageClass("openshift/hello-openshift")156                pod1 = image.run_in_pod(namespace=namespace1)157                try:158                    pod1.wait(200)159                    assert any(pod1.name == p.name for p in k8s_backend.list_pods(namespace1))160                    assert not any(pod1.name == p.name for p in k8s_backend.list_pods(namespace2))161                finally:162                    pod1.delete()163                    k8s_backend.delete_namespace(namespace1)164                    k8s_backend.delete_namespace(namespace2)165    def test_deployment_from_template(self):166        api_key = get_oc_api_token()167        with K8sBackend(api_key=api_key) as k8s_backend:168            namespace = k8s_backend.create_namespace()169            template = """170            apiVersion: apps/v1171            kind: Deployment172            metadata:173              name: hello-world174              labels:175                app: hello-world176            spec:177              replicas: 3178              selector:179                matchLabels:180                  app: hello-world181              template:182                metadata:183                  labels:184                    app: hello-world185                spec:186                  containers:187                  - name: hello-openshift188                    image: openshift/hello-openshift189            """190            test_deployment = Deployment(namespace=namespace, from_template=template,191                                         create_in_cluster=True)192            try:193                test_deployment.wait(200)194                assert test_deployment.all_pods_ready()195            finally:196                test_deployment.delete()197                k8s_backend.delete_namespace(namespace)198    def test_cleanup(self):199        api = get_core_api()200        # take just namespaces that are not in terminating state201        number_of_namespaces = len(202            [item for item in api.list_namespace().items if item.status.phase != "Terminating"])203        api_key = get_oc_api_token()204        with K8sBackend(api_key=api_key, cleanup=[K8sCleanupPolicy.NAMESPACES]) as k8s_backend:205            # create two namespaces206            k8s_backend.create_namespace()207            k8s_backend.create_namespace()208        # cleanup should delete two namespaces created with k8s backend209        assert len(210            [item for item in api.list_namespace().items211             if item.status.phase != "Terminating"]) == number_of_namespaces...

environment.py

Source:environment.py

1"""2before_step(context, step), after_step(context, step)3    These run before and after every step.4    The step passed in is an instance of Step.5before_scenario(context, scenario), after_scenario(context, scenario)6    These run before and after each scenario is run.7    The scenario passed in is an instance of Scenario.8before_feature(context, feature), after_feature(context, feature)9    These run before and after each feature file is exercised.10    The feature passed in is an instance of Feature.11before_all(context), after_all(context)12    These run before and after the whole shooting match.13"""14from steps.command import Command15from steps.environment import ctx16from behave.model_core import Status17import os18import semver19cmd = Command()20if os.getenv("DELETE_NAMESPACE") in ["always", "never", "keepwhenfailed"]:21    delete_namespace = os.getenv("DELETE_NAMESPACE")22else:23    delete_namespace = "keepwhenfailed"24def before_all(_context):25    if ctx.cli == "oc":26        output, code = cmd.run("oc version --client | grep Client")27        assert code == 0, f"Checking oc version failed: {output}"28        oc_ver = output.split()[2]29        assert semver.compare(oc_ver, "4.5.0") > 0, f"oc version is required 4.5+, but is {oc_ver}."30        namespace = os.getenv("TEST_NAMESPACE")31        output, code = cmd.run(f"oc project {namespace}")32        assert code == 0, f"Cannot set default namespace to {namespace}, reason: {output}"33    start_sbo = os.getenv("TEST_ACCEPTANCE_START_SBO")34    assert start_sbo is not None, "TEST_ACCEPTANCE_START_SBO is not set. It should be one of local, remote or operator-hub"35    assert start_sbo in {"local", "remote", "operator-hub"}, "TEST_ACCEPTANCE_START_SBO should be one of local, remote or operator-hub"36    if start_sbo == "local":37        assert not str(os.getenv("TEST_ACCEPTANCE_SBO_STARTED")).startswith("FAILED"), "TEST_ACCEPTANCE_SBO_STARTED shoud not be FAILED."38    elif start_sbo == "remote":39        output, code = cmd.run(40            f"{ctx.cli} get deployment --all-namespaces -o json"41            + " | jq -rc '.items[] | select(.metadata.name == \"service-binding-operator\").metadata.namespace'")42        assert code == 0, f"Non-zero return code while trying to detect namespace for SBO: {output}"43        output = str(output).strip()44        assert output != "", "Unable to find SBO's deployment in any namespace."45        _context.sbo_namespace = output46    else:47        assert False, f"TEST_ACCEPTANCE_START_SBO={start_sbo} is currently unsupported."48    ctx.no_scenarios_failed = True49def before_scenario(_context, _scenario):50    _context.bindings = dict()51    output, code = cmd.run(f'{ctx.cli} get ns default -o jsonpath="{{.metadata.name}}"')52    assert code == 0, f"Checking connection to OS cluster by getting the 'default' project failed: {output}"53def after_scenario(_context, scenario):54    if "namespace" in _context:55        namespace = _context.namespace.name56    elif os.getenv("TEST_NAMESPACE"):57        namespace = os.getenv("TEST_NAMESPACE")58    else:59        print("No namespace set in context nor TEST_NAMESPACE env variable is set, skipping deletion")60        return61    if delete_namespace == "always":62        delete_current_namespace(namespace)63    elif (delete_namespace == "keepwhenfailed"):64        if scenario.status == Status.passed and (ctx.no_scenarios_failed):65            delete_current_namespace(namespace)66        else:67            ctx.no_scenarios_failed = False68            print(f"Deleting namespace {namespace} skipped, since one of the scenarios failed.")69    elif delete_namespace == "never":70        print(f"Namespace {namespace} deletion skipped.")71def delete_current_namespace(name):72    output, code = cmd.run(f"{ctx.cli} delete namespace {name} --ignore-not-found --timeout=1800s")73    assert code == 0, f"Deletion of namespace failed: {output}"...

fixtures.py

Source:fixtures.py

...16@pytest.fixture(scope='function')17async def kubernetes(k8s_config):18    async with K8SContextManager(k8s_config) as context:19        cm = ClusterManager(context)20        await cm.delete_namespace('aiocluster-test')21        await cm.create_namespace('aiocluster-test')22        yield cm23        await cm.delete_namespace('aiocluster-test')24@pytest.fixture(scope='function')25async def nomad(nomad_config):26    async with NomadContextManager(nomad_config) as context:27        cm = ClusterManager(context)28        await cm.delete_namespace('aiocluster-test')29        await cm.create_namespace('aiocluster-test')30        yield cm...

mds_parse.py

Source:mds_parse.py

2import sys3import re4import xml.etree.ElementTree as ET5ns_regex = re.compile( '^(\{[^}]*\})(.*)$' )6def delete_namespace( name ):7    m = ns_regex.match( name )8    return m.group( 2 )9def resolve_URL( URL ):10    URL = URL.replace( 'http://id.loc.gov/authorities/subjects/', '' )11    URL = URL.replace( 'http://id.loc.gov/authorities/names/', '' )12    return URL13parser = ET.iterparse( sys.argv[ 1 ], events = ( 'start', 'end' ) );14parser = iter( parser );15ev, root = parser.next();16for ev, elt in parser:17    tag = delete_namespace( elt.tag )18    19    if ev == 'start':20        if tag == 'Description':21            for k in elt.attrib:22                val = elt.attrib[ k ]23                if delete_namespace( k ) == 'about':24                    this = resolve_URL( val )25    if ev == 'end':26        if tag == 'Description':27            this = ''28        if tag == 'authoritativeLabel':29            print elt.text.encode( 'utf-8', 'ignore' )30    31    root.clear();32    ...

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.