from unittest import TestCase, mock
from osm_mon.collector.utils.openstack import OpenstackUtils
+from osm_mon.core.exceptions import CertificateNotCreated
-@mock.patch('osm_mon.collector.utils.openstack.session')
+@mock.patch("osm_mon.collector.utils.openstack.session")
class OpenstackUtilsTest(TestCase):
-
def setUp(self):
super().setUp()
def test_session_without_insecure(self, mock_session):
creds = {
- 'config': {
- },
- 'vim_url': 'url',
- 'vim_user': 'user',
- 'vim_password': 'password',
- 'vim_tenant_name': 'tenant_name'
+ "config": {},
+ "vim_url": "url",
+ "vim_user": "user",
+ "vim_password": "password",
+ "vim_tenant_name": "tenant_name",
}
OpenstackUtils.get_session(creds)
mock_session.Session.assert_called_once_with(
- auth=mock.ANY, verify=True, timeout=10)
+ auth=mock.ANY, verify=True, timeout=10
+ )
def test_session_with_insecure(self, mock_session):
creds = {
- 'config': {
- 'insecure': True
- },
- 'vim_url': 'url',
- 'vim_user': 'user',
- 'vim_password': 'password',
- 'vim_tenant_name': 'tenant_name'
+ "config": {"insecure": True},
+ "vim_url": "url",
+ "vim_user": "user",
+ "vim_password": "password",
+ "vim_tenant_name": "tenant_name",
}
OpenstackUtils.get_session(creds)
mock_session.Session.assert_called_once_with(
- auth=mock.ANY, verify=False, timeout=10)
+ auth=mock.ANY, verify=False, timeout=10
+ )
def test_session_with_insecure_false(self, mock_session):
creds = {
- 'config': {
- 'insecure': False
- },
- 'vim_url': 'url',
- 'vim_user': 'user',
- 'vim_password': 'password',
- 'vim_tenant_name': 'tenant_name'
+ "config": {"insecure": False},
+ "vim_url": "url",
+ "vim_user": "user",
+ "vim_password": "password",
+ "vim_tenant_name": "tenant_name",
}
OpenstackUtils.get_session(creds)
mock_session.Session.assert_called_once_with(
- auth=mock.ANY, verify=True, timeout=10)
+ auth=mock.ANY, verify=True, timeout=10
+ )
+
+ @mock.patch("osm_mon.collector.utils.openstack.OpenstackUtils._create_file_cert")
+ def test_session_with_ca_cert_content(self, mock_create_file_cert, mock_session):
+ creds = {
+ "_id": "1234",
+ "config": {"ca_cert_content": "test"},
+ "vim_url": "url",
+ "vim_user": "user",
+ "vim_password": "password",
+ "vim_tenant_name": "tenant_name",
+ }
+ mock_create_file_cert.return_value = {"ca_cert": "testfile"}
+ OpenstackUtils.get_session(creds)
+ mock_session.Session.assert_called_once_with(
+ auth=mock.ANY, verify="testfile", timeout=10
+ )
+
+ @mock.patch("osm_mon.collector.utils.openstack.makedirs", return_value="")
+ @mock.patch("osm_mon.collector.utils.openstack.path")
+ def test_create_file_cert(self, mock_path, mock_makedirs, mock_session):
+ vim_config = {"ca_cert_content": "test"}
+ target_id = "1234"
+ mock_path.isdir.return_value = False
+
+ with mock.patch("builtins.open", mock.mock_open()) as mocked_file:
+ OpenstackUtils._create_file_cert(vim_config, target_id)
+ mock_makedirs.assert_called_once_with("/app/osm_mon/certs/1234")
+ mocked_file.assert_called_once_with(
+ f"/app/osm_mon/certs/{target_id}/ca_cert", "w"
+ )
+ assert vim_config["ca_cert"] == f"/app/osm_mon/certs/{target_id}/ca_cert"
+
+ @mock.patch("osm_mon.collector.utils.openstack.makedirs")
+ @mock.patch("osm_mon.collector.utils.openstack.path")
+ def test_create_file_cert_exists(self, mock_path, mock_makedirs, mock_session):
+ vim_config = {"ca_cert_content": "test"}
+ target_id = "1234"
+ mock_path.isdir.return_value = True
+
+ with mock.patch("builtins.open", mock.mock_open()) as mocked_file:
+ OpenstackUtils._create_file_cert(vim_config, target_id)
+ mock_makedirs.assert_not_called()
+ mocked_file.assert_called_once_with(
+ f"/app/osm_mon/certs/{target_id}/ca_cert", "w"
+ )
+ assert vim_config["ca_cert"] == f"/app/osm_mon/certs/{target_id}/ca_cert"
+
+ @mock.patch("osm_mon.collector.utils.openstack.makedirs", side_effect=Exception)
+ @mock.patch("osm_mon.collector.utils.openstack.path")
+ def test_create_file_cert_makedirs_except(
+ self, mock_path, mock_makedirs, mock_session
+ ):
+ vim_config = {"ca_cert_content": "test"}
+ target_id = "1234"
+ mock_path.isdir.return_value = False
+
+ with mock.patch("builtins.open", mock.mock_open()) as mocked_file:
+ with self.assertRaises(CertificateNotCreated):
+ OpenstackUtils._create_file_cert(vim_config, target_id)
+ mock_makedirs.assert_called_once_with("/app/osm_mon/certs/1234")
+ mocked_file.assert_not_called()
+ assert vim_config["ca_cert_content"] == "test"
+
+ @mock.patch("osm_mon.collector.utils.openstack.makedirs", return_value="")
+ @mock.patch("osm_mon.collector.utils.openstack.path")
+ def test_create_file_cert_open_excepts(
+ self, mock_path, mock_makedirs, mock_session
+ ):
+ vim_config = {"ca_cert_content": "test"}
+ target_id = "1234"
+ mock_path.isdir.return_value = False
+
+ with mock.patch("builtins.open", mock.mock_open()) as mocked_file:
+ mocked_file.side_effect = Exception
+ with self.assertRaises(CertificateNotCreated):
+ OpenstackUtils._create_file_cert(vim_config, target_id)
+ mock_makedirs.assert_called_once_with("/app/osm_mon/certs/1234")
+ mocked_file.assert_called_once_with(
+ f"/app/osm_mon/certs/{target_id}/ca_cert", "w"
+ )
+ assert vim_config["ca_cert_content"] == "test"