56 lines
No EOL
2 KiB
Python
56 lines
No EOL
2 KiB
Python
import pytest
|
|
from mock import Mock
|
|
|
|
from util.config.provider import KubernetesConfigProvider
|
|
|
|
from test.fixtures import *
|
|
|
|
|
|
class TestKubernetesConfigProvider(KubernetesConfigProvider):
|
|
def __init__(self):
|
|
self.yaml_filename = 'yaml_filename'
|
|
self._service_token = 'service_token'
|
|
self._execute_k8s_api = Mock()
|
|
|
|
|
|
@pytest.mark.parametrize('directory,filename,expected', [
|
|
("directory", "file", "directory_file"),
|
|
("directory/dir", "file", "directory/dir_file"),
|
|
("directory/dir/", "file", "directory/dir_file"),
|
|
("directory", "file/test", "directory_file/test"),
|
|
])
|
|
def test_get_volume_path(directory, filename, expected):
|
|
provider = TestKubernetesConfigProvider()
|
|
|
|
assert expected == provider.get_volume_path(directory, filename)
|
|
|
|
|
|
@pytest.mark.parametrize('response,expected', [
|
|
(Mock(text="{\"data\": {\"license\":\"test\"}}", status_code=200), {"data": {"license":"test"}}),
|
|
(Mock(text="{\"data\": {\"license\":\"test\"}}", status_code=404), None),
|
|
])
|
|
def test_lookup_secret(response, expected):
|
|
provider = TestKubernetesConfigProvider()
|
|
provider._execute_k8s_api.return_value = response
|
|
assert expected == provider._lookup_secret()
|
|
|
|
|
|
@pytest.mark.parametrize('response,key,expected', [
|
|
(Mock(text="{\"data\": {\"license\":\"test\"}}", status_code=200), "license", True),
|
|
(Mock(text="{\"data\": {\"license\":\"test\"}}", status_code=200), "config.yaml", False),
|
|
(Mock(text="", status_code=404), "license", False),
|
|
])
|
|
def test_volume_file_exists(response, key, expected):
|
|
provider = TestKubernetesConfigProvider()
|
|
provider._execute_k8s_api.return_value = response
|
|
assert expected == provider.volume_file_exists(key)
|
|
|
|
|
|
@pytest.mark.parametrize('response,expected', [
|
|
(Mock(text="{\"data\": {\"extra_license\":\"test\"}}", status_code=200), ["license"]),
|
|
(Mock(text="", status_code=404), []),
|
|
])
|
|
def test_list_volume_directory(response, expected):
|
|
provider = TestKubernetesConfigProvider()
|
|
provider._execute_k8s_api.return_value = response
|
|
assert expected == provider.list_volume_directory("extra") |