-
-
-class K8sJujuConnInitSuccessTestCase(asynctest.TestCase):
- def setUp(
- self,
- ):
- logging.disable(logging.CRITICAL)
-
- @asynctest.mock.patch("juju.controller.Controller.update_endpoints")
- @asynctest.mock.patch("juju.client.connector.Connector.connect")
- @asynctest.mock.patch("juju.controller.Controller.connection")
- @asynctest.mock.patch("n2vc.k8s_juju_conn.base64_to_cacert")
- @asynctest.mock.patch("n2vc.libjuju.Libjuju.__init__")
- def test_success(
- self,
- mock_libjuju=None,
- mock_base64_to_cacert=None,
- mock_connection=None,
- mock_connect=None,
- mock_update_endpoints=None,
- ):
- mock_libjuju.return_value = None
- loop = asyncio.get_event_loop()
- log = logging.getLogger()
- db = Mock()
- vca_config = {
- "secret": "secret",
- "cloud": "cloud",
- "k8s_cloud": "k8s_cloud",
- "user": "user",
- "host": "1.1.1.1",
- "port": 17070,
- "ca_cert": "cacert",
- }
- K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=log,
- loop=self.loop,
- vca_config=vca_config,
- on_update_db=None,
- )
-
- mock_libjuju.assert_called_once_with(
- endpoint="1.1.1.1:17070",
- api_proxy=None, # Not needed for k8s charms
- model_config={},
- username="user",
- password="secret",
- cacert=mock_base64_to_cacert.return_value,
- loop=loop,
- log=log,
- db=db,
- )
-
-
-class K8sJujuConnectorInitFailureTestCase(asynctest.TestCase):
- def setUp(
- self,
- ):
- self.loop = asyncio.get_event_loop()
- logging.disable(logging.CRITICAL)
- self.vca_config = {
- "secret": "secret",
- "api_proxy": "api_proxy",
- "cloud": "cloud",
- "k8s_cloud": "k8s_cloud",
- "user": "user",
- "host": "1.1.1.1",
- "port": 17070,
- "ca_cert": "cacert",
- }
-
- def test_missing_vca_config_host(self):
- db = Mock()
- self.vca_config.pop("host")
- with self.assertRaises(N2VCBadArgumentsException):
- self.k8s_juju_conn = K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=None,
- loop=self.loop,
- vca_config=self.vca_config,
- on_update_db=None,
- )
-
- def test_missing_vca_config_user(self):
- db = Mock()
- self.vca_config.pop("user")
- with self.assertRaises(N2VCBadArgumentsException):
- self.k8s_juju_conn = K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=None,
- loop=self.loop,
- vca_config=self.vca_config,
- on_update_db=None,
- )
-
- def test_missing_vca_config_secret(self):
- db = Mock()
- self.vca_config.pop("secret")
- with self.assertRaises(N2VCBadArgumentsException):
- self.k8s_juju_conn = K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=None,
- loop=self.loop,
- vca_config=self.vca_config,
- on_update_db=None,
- )
-
- def test_missing_vca_config_ca_cert(self):
- db = Mock()
- self.vca_config.pop("ca_cert")
- with self.assertRaises(N2VCBadArgumentsException):
- self.k8s_juju_conn = K8sJujuConnector(
- fs=fslocal.FsLocal(),
- db=db,
- log=None,
- loop=self.loop,
- vca_config=self.vca_config,
- on_update_db=None,
- )