@patch("n2vc.k8s_helm_base_conn.EnvironConfig")
async def setUp(self, mock_env):
- mock_env.return_value = {
- "stablerepourl": "https://charts.helm.sh/stable"
- }
+ mock_env.return_value = {"stablerepourl": "https://charts.helm.sh/stable"}
self.db = Mock(DbMemory())
self.fs = asynctest.Mock(FsLocal())
self.fs.path = "./tmp/"
self.kdu_instance = "stable-openldap-0005399828"
self.helm_conn.generate_kdu_instance_name = Mock(return_value=self.kdu_instance)
self.helm_conn._get_namespaces = asynctest.CoroutineMock(return_value=[])
- self.helm_conn._namespace_exists = asynctest.CoroutineMock(side_effect=self.helm_conn._namespace_exists)
+ self.helm_conn._namespace_exists = asynctest.CoroutineMock(
+ side_effect=self.helm_conn._namespace_exists
+ )
self.helm_conn._create_namespace = asynctest.CoroutineMock()
await self.helm_conn.install(
async def test_namespace_exists(self):
self.helm_conn._get_namespaces = asynctest.CoroutineMock()
- self.helm_conn._get_namespaces.return_value = ['testk8s', 'kube-system']
+ self.helm_conn._get_namespaces.return_value = ["testk8s", "kube-system"]
result = await self.helm_conn._namespace_exists(self.cluster_id, self.namespace)
self.helm_conn._get_namespaces.assert_called_once()
self.assertEqual(result, True)
self.helm_conn._get_namespaces.reset_mock()
- result = await self.helm_conn._namespace_exists(self.cluster_id, 'none-exists-namespace')
+ result = await self.helm_conn._namespace_exists(
+ self.cluster_id, "none-exists-namespace"
+ )
self.helm_conn._get_namespaces.assert_called_once()
self.assertEqual(result, False)