How to use list_namespaces method in tempest

Best Python code snippet using tempest_python

cli.py

Source:cli.py

1import click, yaml, errno, os2from ksns.client import K8s3from ksns.const import homedir, env_kubeconfig, path_kubeconfig, NAMESPACE_STATUS4@click.group()5def main():6    """ 7NOTE: Set the KUBECONFIG environment variable or ~/.kube/config will be considered8\b9USAGE: 10    1. list namespaces : ksns list11    2. switch namespaces: ksns ns <namespace_name>12    """13    pass14@main.command()15def list() -> None:16    """ List of namespaces in context """17    if not env_kubeconfig:18        kubeconfig = homedir + path_kubeconfig19    else:20        kubeconfig = env_kubeconfig21    if not os.path.exists(kubeconfig) and not env_kubeconfig:22        click.secho(f"You need to store kubeconfig file in path '{kubeconfig}'", fg='red', nl=True)23        click.secho(f"OR set environment variable via, export KUBECONFIG=<kubeconfig path>", fg='red', nl=True)24        raise click.Abort()25    try:26        kctx = K8s(configuration_yaml=kubeconfig)27        list_namespaces = kctx.client.list_namespace(watch=False)28        current_namespace = kctx.config['contexts'][0]['context']['namespace']29        for i in list_namespaces.items:30            namespace = i.metadata.name31            if namespace == current_namespace:32                fg = 'yellow'33            else: 34                fg = 'green'35            click.secho(namespace, fg=fg, nl=True)36        click.secho(f"note: using config: {kubeconfig}", fg='blue', nl=True)37    except Exception as error:38        click.secho(f"Caught this error: '{repr(error)}'", fg='red', nl=True)39@main.command()40@click.argument('namespace', type=str, default='default')41def ns(namespace) -> None:42    """ Switch to another namespace: <namespace_name>"""43    namespace_arr = []44    if not env_kubeconfig:45        kubeconfig = homedir + path_kubeconfig46    else:47        kubeconfig = env_kubeconfig48    if not os.path.exists(kubeconfig) and not env_kubeconfig:49        click.secho(f"You need to store kubeconfig file in path '{kubeconfig}'", fg='red', nl=True)50        click.secho(f"OR set environment variable via, export KUBECONFIG=<kubeconfig path>", fg='red', nl=True)51        raise click.Abort()52    try:53        kctx = K8s(configuration_yaml=kubeconfig)54        list_namespaces = kctx.client.list_namespace(watch=False)55        for i in list_namespaces.items:56            namespace_arr.append(i.metadata.name)57        if namespace not in namespace_arr:58            click.secho(f"Specified namespace '{namespace}' is not valid.", fg='red', nl=True)59            raise click.Abort()60        current_namespace = kctx.config['contexts'][0]['context']['namespace']61        with open(kubeconfig) as file:62            getconfig = yaml.load(file, Loader=yaml.FullLoader)63        with open(kubeconfig, 'w') as config_file:64            getconfig['contexts'][0]['context']['namespace'] = namespace65            sort_file = yaml.dump(getconfig, config_file, sort_keys=True)66        67        click.secho(f"switched to {namespace} namespace", fg='green', nl=True)68        click.secho(f"using config: {kubeconfig}", fg='blue', nl=True)69    except Exception as error:70        click.secho(f"Caught this error: '{repr(error)}'", fg='red', nl=True)71        72    73if __name__ == "__main__":...

delete_namespace.py

Source:delete_namespace.py

...17args = parser.parse_args()18namespace_name = args.namespace_name19verbose = args.verbose20cloudmap = boto3.client('servicediscovery')21list_namespaces = cloudmap.list_namespaces()22if list_namespaces['ResponseMetadata']['HTTPStatusCode'] < 400:23    # Find namespace24    for namespace in list_namespaces['Namespaces']:25        if namespace['Name'] == namespace_name:26            # Services27            list_services = cloudmap.list_services(Filters=[{'Name': 'NAMESPACE_ID','Values':[namespace['Id']],'Condition': 'EQ'}])28            for service in list_services['Services']:29                # Service Instances30                list_instances = cloudmap.list_instances(ServiceId=service['Id'])31                for instance in list_instances['Instances']:32                    cloudmap.deregister_instance(ServiceId=service['Id'],InstanceId=instance['Id'])33                    if verbose:34                        print("Deregistered: " + instance['Id'] + ' from service '+ service['Name'])35                    ...

quickstart.py

Source:quickstart.py

...13# See the License for the specific language governing permissions and14# limitations under the License.15# [START servicedirectory_quickstart]16from google.cloud import servicedirectory_v117def list_namespaces(project_id, location_id):18    """Lists all namespaces in the given location."""19    client = servicedirectory_v1.RegistrationServiceClient()20    response = client.list_namespaces(21        parent=f'projects/{project_id}/locations/{location_id}')22    print(f'Listed namespaces in {location_id}.')23    for namespace in response:24        print(f'Namespace: {namespace.name}')25    return response...

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.