Your advanced code editor is loading.

Please wait a moment.

#editor-loading { display: none; }

Licensed under the Apache License, Version 2.0 (the “License”);

you may not use this file except in compliance with the License.

You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software

distributed under the License is distributed on an “AS IS” BASIS,

WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

See the License for the specific language governing permissions and

limitations under the License.

import os import unittest

from kubernetes.e2e_test import base from kubernetes.client import api_client

from . import DynamicClient

class TestDiscoverer(unittest.TestCase):

@classmethod
          def setUpClass(cls):
          cls.config = base.get_e2e_configuration()
          
          def test_init_cache_from_file(self):
          client = DynamicClient(api_client.ApiClient(configuration=self.config))
          client.resources.get(api_version='v1', kind='Node')
          mtime1 = os.path.getmtime(client.resources._Discoverer__cache_file)
          client = DynamicClient(api_client.ApiClient(configuration=self.config))
          client.resources.get(api_version='v1', kind='Node')
          mtime2 = os.path.getmtime(client.resources._Discoverer__cache_file)
          test no Discoverer._write_cache called
          self.assertTrue(mtime1 == mtime2)
          def test_cache_decoder_resource_and_subresource(self):
          client = DynamicClient(api_client.ApiClient(configuration=self.config))
          first invalidate cache
          client.resources.invalidate_cache()
          do Discoverer.init
          client = DynamicClient(api_client.ApiClient(configuration=self.config))
          the resources of client will use _cache['resources'] in memory
          deploy1 = client.resources.get(kind='Deployment')
          do Discoverer.init
          client = DynamicClient(api_client.ApiClient(configuration=self.config))
          the resources of client will use _cache['resources'] decode from cache file
          deploy2 = client.resources.get(kind='Deployment')
          test Resource is the same
          self.assertTrue(deploy1 == deploy2)
          test Subresource is the same
          self.assertTrue(deploy1.status == deploy2.status)

Explanation

This test suite focuses on the DynamicClient class and its ability to dynamically discover and interact with Kubernetes API resources. Here’s a breakdown:

test_init_cache_from_file:

This test checks if the DynamicClient loads its resource cache from a file.
It creates two DynamicClient instances and retrieves the Node resource using client.resources.get().
The test ensures that the modification timestamp of the cache file remains unchanged between the two instances. This indicates that the cache is loaded from the file and not being rewritten.

test_cache_decoder_resource_and_subresource:

This test validates the cache decoding process for both resources and their subresources.
It first invalidates the cache using client.resources.invalidate_cache().
Then, it creates two DynamicClient instances.
The first instance retrieves the Deployment resource, causing the cache to be populated from the Kubernetes API.
The second instance retrieves the Deployment resource, now using the cached data.
The test verifies that the two retrieved Deployment instances (one from API, one from cache) are identical.
Finally, it asserts that the status subresource of the two Deployment instances are also identical.

Overall, this test suite confirms that DynamicClient correctly utilizes a cached file for resource discovery and that this cache accurately represents the actual resource data from the Kubernetes API. The caching behavior ensures efficiency by reducing the number of API calls for repeated requests.

Your advanced code editor is loading.

Please wait a moment.

Explanation

Graph

Symbols

We couldn't identify any entrypoints. If you believe this to be incorrect then please contact support.