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...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!