+ 'volumeMounts': container_obj._create_volume_mounts(),
+ 'tty': False,
+ 'stdin': True}
+ self.assertEqual(expected, container_obj.get_container_item())
+
+ def test__parse_commands_string(self):
+ container_obj = kubernetes.ContainerObject('cname', 'fake_sshkey')
+ self.assertEqual(['fake command'],
+ container_obj._parse_commands('fake command'))
+
+ def test__parse_commands_list(self):
+ container_obj = kubernetes.ContainerObject('cname', 'fake_sshkey')
+ self.assertEqual(['cmd1', 'cmd2'],
+ container_obj._parse_commands(['cmd1', 'cmd2']))
+
+ def test__parse_commands_exception(self):
+ container_obj = kubernetes.ContainerObject('cname', 'fake_sshkey')
+ with self.assertRaises(exceptions.KubernetesContainerCommandType):
+ container_obj._parse_commands({})
+
+
+class CustomResourceDefinitionObjectTestCase(base.BaseUnitTestCase):
+
+ def test__init(self):
+ template = {
+ 'metadata': {
+ 'name': 'newcrds.ctx_name.com'
+ },
+ 'spec': {
+ 'group': 'ctx_name.com',
+ 'version': 'v2',
+ 'scope': 'scope',
+ 'names': {'plural': 'newcrds',
+ 'singular': 'newcrd',
+ 'kind': 'Newcrd'}
+ }
+ }
+ crd_obj = kubernetes.CustomResourceDefinitionObject(
+ 'ctx_name', name='newcrd', version='v2', scope='scope')
+ self.assertEqual('newcrds.ctx_name.com', crd_obj._name)
+ self.assertEqual(template, crd_obj._template)
+
+ def test__init_missing_parameter(self):
+ with self.assertRaises(exceptions.KubernetesCRDObjectDefinitionError):
+ kubernetes.CustomResourceDefinitionObject('ctx_name',
+ noname='name')
+
+
+class NetworkObjectTestCase(base.BaseUnitTestCase):
+
+ def setUp(self):
+ self.net_obj = kubernetes.NetworkObject(name='fake_name',
+ plugin='fake_plugin',
+ args='fake_args')
+
+ def test__init_missing_parameter(self):
+ with self.assertRaises(
+ exceptions.KubernetesNetworkObjectDefinitionError):
+ kubernetes.NetworkObject('network_name', plugin='plugin')
+ with self.assertRaises(
+ exceptions.KubernetesNetworkObjectDefinitionError):
+ kubernetes.NetworkObject('network_name', args='args')
+
+ @mock.patch.object(kubernetes_utils, 'get_custom_resource_definition')
+ def test_crd(self, mock_get_crd):
+ mock_crd = mock.Mock()
+ mock_get_crd.return_value = mock_crd
+ net_obj = copy.deepcopy(self.net_obj)
+ self.assertEqual(mock_crd, net_obj.crd)
+
+ def test_template(self):
+ net_obj = copy.deepcopy(self.net_obj)
+ expected = {'apiVersion': 'group.com/v2',
+ 'kind': kubernetes.NetworkObject.KIND,
+ 'metadata': {
+ 'name': 'fake_name'},
+ 'plugin': 'fake_plugin',
+ 'args': 'fake_args'}
+ crd = mock.Mock()
+ crd.spec.group = 'group.com'
+ crd.spec.version = 'v2'
+ net_obj._crd = crd
+ self.assertEqual(expected, net_obj.template)
+
+ def test_group(self):
+ net_obj = copy.deepcopy(self.net_obj)
+ net_obj._crd = mock.Mock()
+ net_obj._crd.spec.group = 'fake_group'
+ self.assertEqual('fake_group', net_obj.group)
+
+ def test_version(self):
+ net_obj = copy.deepcopy(self.net_obj)
+ net_obj._crd = mock.Mock()
+ net_obj._crd.spec.version = 'version_4'
+ self.assertEqual('version_4', net_obj.version)
+
+ def test_plural(self):
+ net_obj = copy.deepcopy(self.net_obj)
+ net_obj._crd = mock.Mock()
+ net_obj._crd.spec.names.plural = 'name_ending_in_s'
+ self.assertEqual('name_ending_in_s', net_obj.plural)
+
+ def test_scope(self):
+ net_obj = copy.deepcopy(self.net_obj)
+ net_obj._crd = mock.Mock()
+ net_obj._crd.spec.scope = 'Cluster'
+ self.assertEqual('Cluster', net_obj.scope)
+
+ @mock.patch.object(kubernetes_utils, 'create_network')
+ def test_create(self, mock_create_network):
+ net_obj = copy.deepcopy(self.net_obj)
+ net_obj._scope = 'scope'
+ net_obj._group = 'group'
+ net_obj._version = 'version'
+ net_obj._plural = 'plural'
+ net_obj._template = 'template'
+ net_obj.create()
+ mock_create_network.assert_called_once_with(
+ 'scope', 'group', 'version', 'plural', 'template')
+
+ @mock.patch.object(kubernetes_utils, 'delete_network')
+ def test_delete(self, mock_delete_network):
+ net_obj = copy.deepcopy(self.net_obj)
+ net_obj._scope = 'scope'
+ net_obj._group = 'group'
+ net_obj._version = 'version'
+ net_obj._plural = 'plural'
+ net_obj._name = 'name'
+ net_obj.delete()
+ mock_delete_network.assert_called_once_with(
+ 'scope', 'group', 'version', 'plural', 'name')
+
+
+class ServiceNodePortObjectTestCase(base.BaseUnitTestCase):
+
+ def test__init(self):
+ with mock.patch.object(kubernetes.ServiceNodePortObject, '_add_port') \
+ as mock_add_port:
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80, 'name': 'web'}])
+
+ mock_add_port.assert_has_calls([mock.call(22, 'ssh', protocol='TCP'),
+ mock.call(80, 'web')])
+
+ @mock.patch.object(kubernetes.ServiceNodePortObject, '_add_port')
+ def test__init_missing_mandatory_parameters(self, *args):
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectDefinitionError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80}])
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectDefinitionError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'name': 'web'}])
+
+ @mock.patch.object(kubernetes.ServiceNodePortObject, '_add_port')
+ def test__init_missing_bad_name(self, *args):
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectNameError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80, 'name': '-web'}])
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectNameError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80, 'name': 'Web'}])
+ with self.assertRaises(
+ exceptions.KubernetesServiceObjectNameError):
+ kubernetes.ServiceNodePortObject(
+ 'fake_name', node_ports=[{'port': 80, 'name': 'web-'}])
+
+ def test__add_port(self):
+ nodeport_object = kubernetes.ServiceNodePortObject('fake_name')
+ port_ssh = {'name': 'ssh',
+ 'port': 22,
+ 'protocol': 'TCP'}
+ port_definition = {'port': 80,
+ 'protocol': 'TCP',
+ 'name': 'web',
+ 'targetPort': 10080,
+ 'nodePort': 30080}
+ port = copy.deepcopy(port_definition)
+ _port = port.pop('port')
+ name = port.pop('name')
+ nodeport_object._add_port(_port, name, **port)
+ self.assertEqual([port_ssh, port_definition],
+ nodeport_object.template['spec']['ports'])
+
+ @mock.patch.object(kubernetes_utils, 'create_service')
+ def test_create(self, mock_create_service):
+ nodeport_object = kubernetes.ServiceNodePortObject('fake_name')
+ nodeport_object.template = 'fake_template'
+ nodeport_object.create()
+ mock_create_service.assert_called_once_with('fake_template')
+
+ @mock.patch.object(kubernetes_utils, 'delete_service')
+ def test_delete(self, mock_delete_service):
+ nodeport_object = kubernetes.ServiceNodePortObject('fake_name')
+ nodeport_object.delete()
+ mock_delete_service.assert_called_once_with('fake_name-service')
+
+
+class KubernetesTemplate(base.BaseUnitTestCase):
+
+ def test_get_rc_by_name(self):
+ ctx_cfg = {
+ 'servers': {
+ 'host1': {'args': 'some data'}
+ }
+ }
+ k_template = kubernetes.KubernetesTemplate('k8s_name', ctx_cfg)
+ rc = k_template.get_rc_by_name('host1-k8s_name')
+ self.assertTrue(isinstance(rc, kubernetes.ReplicationControllerObject))
+
+ def test_get_rc_by_name_wrong_name(self):
+ ctx_cfg = {
+ 'servers': {
+ 'host1': {'args': 'some data'}
+ }
+ }
+ k_template = kubernetes.KubernetesTemplate('k8s_name', ctx_cfg)
+ self.assertIsNone(k_template.get_rc_by_name('wrong_host_name'))
+
+ def test_get_rc_by_name_no_rcs(self):
+ ctx_cfg = {'servers': {}}
+ k_template = kubernetes.KubernetesTemplate('k8s_name', ctx_cfg)
+ self.assertIsNone(k_template.get_rc_by_name('any_host_name'))