How to use kubernetes - 10 common examples

def run_test(args, test_case):  # pylint: disable=too-many-branches,too-many-statements
  """Run a test."""
  util.load_kube_config()

  api_client = k8s_client.ApiClient()

  t = test_util.TestCase()
  t.class_name = "tfjob_test"
  namespace, name, env = ks_util.setup_ks_app(args)
  t.name = os.path.basename(name)

  try: # pylint: disable=too-many-nested-blocks
    util.run(["ks", "apply", env, "-c", args.component], cwd=args.app_dir)

    logging.info("Created job %s in namespaces %s", name, namespace)

    logging.info("Wait for conditions Failed")
    results = tf_job_client.wait_for_condition(
      api_client, namespace, name, ["Succeeded", "Failed"],
      status_callback=tf_job_client.log_status)

def create_namespace(self, ns_name):
        self.cluster.create_namespace(client.V1Namespace(metadata=client.V1ObjectMeta(name=ns_name)))

def setUpClass(cls):
        cls.config = base.get_e2e_configuration()
        cls.path_prefix = "kubernetes/e2e_test/test_yaml/"
        cls.test_namespace = "e2e-test-utils"
        k8s_client = client.api_client.ApiClient(configuration=cls.config)
        core_v1 = client.CoreV1Api(api_client=k8s_client)
        body = client.V1Namespace(metadata=client.V1ObjectMeta(name=cls.test_namespace))
        core_v1.create_namespace(body=body)

def refresh(self):
        """Refresh the underlying Kubernetes Api Pod object."""
        self.obj = client.CoreV1Api().read_namespaced_pod_status(
            name=self.name,
            namespace=self.namespace,
        )

account=None,
      namespace=namespace)

  # service component
  serviceComponent = "modelServer-service"
  generate_command = [ks, "generate", "tf-serving-service", serviceComponent]
  util.run(generate_command, cwd=app_dir)
  ks_deploy(
      app_dir,
      serviceComponent,
      params,
      env=None,
      account=None,
      namespace=namespace)

  core_api = k8s_client.CoreV1Api(api_client)
  deploy = core_api.read_namespaced_service(args.deploy_name, args.namespace)
  cluster_ip = deploy.spec.cluster_ip

  if not cluster_ip:
    raise ValueError("inception service wasn't assigned a cluster ip.")
  util.wait_for_deployment(
      api_client, namespace, args.deploy_name, timeout_minutes=10)
  logging.info("Verified TF serving started.")

def test_create_pod_from_yaml(self):
        """
        Should be able to create a pod.
        """
        k8s_client = client.api_client.ApiClient(configuration=self.config)
        utils.create_from_yaml(
            k8s_client, self.path_prefix + "core-pod.yaml")
        core_api = client.CoreV1Api(k8s_client)
        pod = core_api.read_namespaced_pod(name="myapp-pod",
                                           namespace="default")
        self.assertIsNotNone(pod)
        core_api.delete_namespaced_pod(
            name="myapp-pod", namespace="default",
            body={})

def __init__(self, get_cmd, env, image, labels={}):
        global ApiException
        global k8sclient
        from kubernetes import config
        from kubernetes import client as k8sclient
        from kubernetes.client.rest import ApiException

        config.load_kube_config()

        super(KubernetesServer, self).__init__(get_cmd, env)

        self._namespace = 'default'
        self._name = 'server-fixtures-%s' % uuid.uuid4()

        self._image = image
        self._run_cmd = get_cmd()
        self._labels = _merge_dicts(labels, {
            'server-fixtures': 'kubernetes-server-fixtures',
            'server-fixtures/session-id': CONFIG.session_id,
        })

        self._v1api = k8sclient.CoreV1Api()

def new(cls, name: str) -> 'Namespace':
        """Create a new Namespace with object backing.

        Args:
            name: The name of the new Namespace.

        Returns:
            A new Namespace instance.
        """
        return cls(client.V1Namespace(
            metadata=client.V1ObjectMeta(
                name=name
            )

def test_ns_create_new(self, mock_ns_read, mock_api, mock_log):
        # TODO: We should ideally replicate the correct API exception
        mock_ns_read.side_effect = ApiException()
        ns_create("a-namespace")
        mock_ns_read.assert_called_once_with("a-namespace")
        mock_api.create_namespace.assert_called_once()
        mock_log.info.assert_called_once_with('Created namespace "a-namespace"')

def wait_to_deployment_to_be_deleted(deployment_name, name_space, time_out=None):
    total_sleep_time = 0
    while True:
        try:
            resp = client.AppsV1Api().read_namespaced_deployment(name=deployment_name, namespace=name_space)
        except ApiException as e:
            if e.status == 404:
                print("Total time waiting for delete deployment {0}: {1} sec".format(deployment_name, total_sleep_time))
                break
        time.sleep(1)
        total_sleep_time += 1

        if time_out and total_sleep_time > time_out:
            raise Exception("Timeout waiting to delete deployment")