Skip to content

Commit

Permalink
fixed multiple tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@XaverStiensmeier
XaverStiensmeier committed Oct 31, 2023
1 parent dcf2836 commit 400b350
Showing 1 changed file with 80 additions and 127 deletions.
207 changes: 80 additions & 127 deletions tests/test_create.py
View file
Original file line number Diff line number Diff line change
@@ -1,131 +1,87 @@
"""
Module to test create
"""
import os
from unittest import TestCase
from unittest.mock import patch, Mock, MagicMock, mock_open

from bibigrid.core import startup
from bibigrid.core.actions import create


class TestCreate(TestCase):
"""
Class to test create
"""

# pylint: disable=R0904
@patch("bibigrid.core.utility.handler.sshHandler.get_add_ssh_public_key_commands")
@patch("bibigrid.core.utility.handler.ssh_handler.get_add_ssh_public_key_commands")
@patch("bibigrid.core.utility.id_generation.generate_safe_cluster_id")
def test_init(self, mock_id, mock_ssh):
unique_id = 21
cluster_id = "21"
provider = MagicMock()
provider.cloud_specification["auth"]["project_name"] = "name"
key_name = create.KEY_PREFIX + provider.cloud_specification["auth"]["project_name"] \
+ create.SEPARATOR + str(unique_id)
mock_id.return_value = str(unique_id)
provider_dict = {'cloud_specification': {'auth': {'project_name': 'project_name'}}}
provider.__getitem__.side_effect = provider_dict.__getitem__
key_name = create.KEY_NAME.format(cluster_id=cluster_id)
mock_id.return_value = cluster_id
mock_ssh.return_value = [32]
c = create.Create([provider], [{}], "path", False)
self.assertEqual(str(unique_id), c.cluster_id)
self.assertEqual("ubuntu", c.ssh_user)
self.assertEqual([32], c.ssh_add_public_key_commands)
self.assertEqual(c.key_name, key_name)
creator = create.Create([provider], [{}], "path", startup.LOG, False)
self.assertEqual(cluster_id, creator.cluster_id)
self.assertEqual("ubuntu", creator.ssh_user)
self.assertEqual([32], creator.ssh_add_public_key_commands)
self.assertEqual(key_name, creator.key_name)
mock_id.assert_called_with([provider])

@patch("bibigrid.core.utility.handler.sshHandler.get_add_ssh_public_key_commands")
@patch("bibigrid.core.utility.handler.ssh_handler.get_add_ssh_public_key_commands")
@patch("bibigrid.core.utility.id_generation.generate_safe_cluster_id")
def test_init_with_cluster_id(self, mock_id, mock_ssh):
cluster_id = "21"
provider = MagicMock()
provider_dict = {'cloud_specification': {'auth': {'project_name': 'project_name'}}}
provider.__getitem__.side_effect = provider_dict.__getitem__
key_name = create.KEY_NAME.format(cluster_id=cluster_id)
mock_ssh.return_value = [32]
creator = create.Create([provider], [{}], "path", startup.LOG, False, cluster_id)
self.assertEqual(cluster_id, creator.cluster_id)
self.assertEqual("ubuntu", creator.ssh_user)
self.assertEqual([32], creator.ssh_add_public_key_commands)
self.assertEqual(key_name, creator.key_name)
mock_id.assert_not_called()

@patch("bibigrid.core.utility.handler.ssh_handler.get_add_ssh_public_key_commands")
@patch("bibigrid.core.utility.id_generation.generate_safe_cluster_id")
def test_init_username(self, mock_id, mock_ssh):
unique_id = 21
mock_id.return_value = str(unique_id)
cluster_id = "21"
mock_id.return_value = cluster_id
mock_ssh.return_value = [32]
c = create.Create([MagicMock()], [{"sshUser": "ssh"}], "path", False)
self.assertEqual("ssh", c.ssh_user)
creator = create.Create([MagicMock()], [{"sshUser": "ssh"}], "path", startup.LOG, False)
self.assertEqual("ssh", creator.ssh_user)

@patch("subprocess.check_output")
def test_generate_keypair(self, mock_subprocess):
provider = MagicMock()
provider.list_servers.return_value = []
c = create.Create([provider], [{}], "")
creator = create.Create([provider], [{}], "", startup.LOG)
public_key = "data"
with patch("builtins.open", mock_open(read_data=public_key)):
c.generate_keypair()
provider.create_keypair.assert_called_with(name=c.key_name, public_key=public_key)
mock_subprocess.assert_called_with(f'ssh-keygen -t ecdsa -f {create.KEY_FOLDER}{c.key_name} -P ""')

def test_start_instance(self):
provider = MagicMock()
provider.list_servers.return_value = []
provider.create_server.return_value = 42
provider.add_auto_ip.return_value = {"floating_ip_address": 12}
c = create.Create([provider], [{}], "")
server_type = {"type": "testType", "image": "testImage"}
network = 21
external_network = "testExternal"
c.start_instance(provider, create.MASTER_IDENTIFIER, server_type, network, worker=False, volumes=2,
external_network=external_network)
provider.create_server.assert_called_with(name=create.MASTER_IDENTIFIER + create.SEPARATOR + c.cluster_id,
flavor=server_type["type"],
key_name=c.key_name,
image=server_type["image"],
network=network, volumes=2)
provider.add_auto_ip.assert_called_with(network=external_network, server=42)

def test_start_instance_worker(self):
provider = MagicMock()
provider.list_servers.return_value = []
provider.create_server.return_value = 42
provider.create_floating_ip.return_value = {"floating_ip_address": 12}
c = create.Create([provider], [{}], "")
server_type = {"type": "testType", "image": "testImage"}
network = 21
c.start_instance(provider, create.WORKER_IDENTIFIER, server_type, network, worker=True, volumes=None,
external_network=None)
provider.create_server.assert_called_with(
name=create.WORKER_IDENTIFIER.format(0) + create.SEPARATOR + c.cluster_id,
flavor=server_type["type"],
key_name=c.key_name,
image=server_type["image"],
network=network, volumes=None)
provider.create_floating_ip.assert_not_called()

@patch("bibigrid.models.returnThreading.ReturnThread")
def test_start_instances(self, return_mock):
provider = MagicMock()
provider.list_servers.return_value = []
external_network = "externalTest"
provider.get_external_netowrk.return_value = external_network
configuration = {"network": 42}
c = create.Create([provider], [configuration], "")
provider.get_external_network.return_value = 32
with patch.object(c, "prepare_vpn_or_master_args", return_value=(0, 1, 2)) as prepare_mock:
prepare_mock.return_value = (0, 1, 2)
c.start_instances({"network": 42}, provider)
prepare_mock.assert_called_with(configuration, provider)
provider.get_external_network.assert_called_with(configuration["network"])
return_mock.assert_called_with(target=c.start_instance,
args=[provider, 0, 1, configuration["network"], False, 2, 32])
creator.generate_keypair()
provider.create_keypair.assert_called_with(name=creator.key_name, public_key=public_key)
mock_subprocess.assert_called_with(f'ssh-keygen -t ecdsa -f {create.KEY_FOLDER}{creator.key_name} -P ""',
shell=True)

@patch("threading.Thread")
@patch("bibigrid.models.returnThreading.ReturnThread")
def test_start_instances_workers(self, return_mock, thread_mock):
provider = MagicMock()
provider.list_servers.return_value = []
external_network = "externalTest"
provider.get_external_netowrk.return_value = external_network
configuration = {"network": 42, "workerInstances": [{"count": 1}]}
c = create.Create([provider], [configuration], "")
provider.get_external_network.return_value = 32
with patch.object(c, "prepare_vpn_or_master_args", return_value=(0, 1, 2)) as prepare_mock:
prepare_mock.return_value = (0, 1, 2)
c.start_instances(configuration, provider)
thread_mock.assert_called_with(target=c.start_instance,
args=[provider, create.WORKER_IDENTIFIER, configuration["workerInstances"][0],
configuration["network"], True])
return_mock.assert_called()
# TODO: Rewrite start instance tests

def test_prepare_master_args(self):
provider = MagicMock()
provider.list_servers.return_value = []
external_network = "externalTest"
provider.get_external_netowrk.return_value = external_network
configuration = {"network": 42, "masterInstance": "Some"}
c = create.Create([provider], [configuration], "")
creator = create.Create([provider], [configuration], "", startup.LOG)
volume_return = [42]
with patch.object(c, "prepare_volumes", return_value=volume_return) as prepare_mock:
with patch.object(creator, "prepare_volumes", return_value=volume_return) as prepare_mock:
self.assertEqual((create.MASTER_IDENTIFIER, configuration["masterInstance"], volume_return),
c.prepare_vpn_or_master_args(configuration, provider))
creator.prepare_vpn_or_master_args(configuration, provider))
prepare_mock.assert_called_with(provider, [])

def test_prepare_vpn_args(self):
Expand All @@ -134,11 +90,11 @@ def test_prepare_vpn_args(self):
external_network = "externalTest"
provider.get_external_netowrk.return_value = external_network
configuration = {"network": 42, "vpnInstance": "Some"}
c = create.Create([provider], [configuration], "")
creator = create.Create([provider], [configuration], "", startup.LOG)
volume_return = [42]
with patch.object(c, "prepare_volumes", return_value=volume_return) as prepare_mock:
with patch.object(creator, "prepare_volumes", return_value=volume_return) as prepare_mock:
self.assertEqual((create.VPN_WORKER_IDENTIFIER, configuration["vpnInstance"], []),
c.prepare_vpn_or_master_args(configuration, provider))
creator.prepare_vpn_or_master_args(configuration, provider))
prepare_mock.assert_not_called()

def test_prepare_args_keyerror(self):
Expand All @@ -147,35 +103,33 @@ def test_prepare_args_keyerror(self):
external_network = "externalTest"
provider.get_external_netowrk.return_value = external_network
configuration = {"network": 42}
c = create.Create([provider], [configuration], "")
creator = create.Create([provider], [configuration], "", startup.LOG)
volume_return = [42]
with patch.object(c, "prepare_volumes", return_value=volume_return) as prepare_mock:
with patch.object(creator, "prepare_volumes", return_value=volume_return) as prepare_mock:
with self.assertRaises(KeyError):
self.assertEqual((create.VPN_WORKER_IDENTIFIER, configuration["vpnInstance"], []),
c.prepare_vpn_or_master_args(configuration, provider))
creator.prepare_vpn_or_master_args(configuration, provider))
prepare_mock.assert_not_called()

@patch("bibigrid.core.utility.handler.sshHandler.ansible_preparation")
def test_setup_reachable_servers_master(self, mock_ansible):
@patch("bibigrid.core.utility.handler.ssh_handler.ansible_preparation")
def test_initialize_instances_master(self, mock_ansible):
provider = MagicMock()
provider.list_servers.return_value = []
configuration = {"masterInstance": 42}
c = create.Create([provider], [configuration], "")
floating_ip = 21
c.setup_reachable_servers(configuration, floating_ip)
mock_ansible.assert_called_with(floating_ip=floating_ip,
private_key=create.KEY_FOLDER + c.key_name,
username=c.ssh_user,
commands=[])
configuration = {"masterInstance": 42, "floating_ip": floating_ip}
creator = create.Create([provider], [configuration], "", startup.LOG)
creator.initialize_instances()
mock_ansible.assert_called_with(floating_ip=floating_ip, private_key=create.KEY_FOLDER + creator.key_name,
username=creator.ssh_user, commands=[], log=startup.LOG, gateway={})

def test_prepare_volumes_none(self):
provider = MagicMock()
provider.list_servers.return_value = []
provider.get_volume_by_id_or_name.return_value = 42
provider.create_volume_from_snapshot = 21
configuration = {"vpnInstance": 42}
c = create.Create([provider], [configuration], "")
self.assertEqual([], c.prepare_volumes(provider, []))
creator = create.Create([provider], [configuration], "", startup.LOG)
self.assertEqual(set(), creator.prepare_volumes(provider, []))

def test_prepare_volumes_volume(self):
provider = MagicMock()
Expand Down Expand Up @@ -225,36 +179,35 @@ def test_prepare_configurations_no_subnet(self):
subnet = ["subnet"]
provider.get_subnet_ids_by_network.return_value = subnet
configuration = {"network": 42}
c = create.Create([provider], [configuration], "")
c.prepare_configurations()
creator = create.Create([provider], [configuration], "", startup.LOG)
creator.prepare_configurations()
provider.get_subnet_ids_by_network.assert_called_with(42)
self.assertEqual(subnet, configuration["subnet"])
self.assertEqual(c.ssh_user, configuration["sshUser"])
self.assertEqual(creator.ssh_user, configuration["sshUser"])

def test_prepare_configurations_none(self):
provider = MagicMock()
provider.list_servers.return_value = []
configuration = {}
c = create.Create([provider], [configuration], "")
creator = create.Create([provider], [configuration], "", startup.LOG)
with self.assertRaises(KeyError):
c.prepare_configurations()
creator.prepare_configurations()

@patch("bibigrid.core.utility.ansibleConfigurator.configure_ansible_yaml")
@patch("bibigrid.core.utility.handler.sshHandler.execute_ssh")
@patch("bibigrid.core.utility.ansible_configurator.configure_ansible_yaml")
@patch("bibigrid.core.utility.handler.ssh_handler.execute_ssh")
def test_upload_playbooks(self, mock_ssh, mock_configure_ansible):
provider = MagicMock()
provider.list_servers.return_value = []
configuration = {}
c = create.Create([provider], [configuration], "")
c.master_ip = 42
c.upload_data()
mock_configure_ansible.assert_called_with(providers=c.providers,
configurations=c.configurations,
cluster_id=c.cluster_id)
mock_ssh.assert_called_with(floating_ip=c.master_ip, private_key=create.KEY_FOLDER + c.key_name,
username=c.ssh_user, filepaths=[(os.path.expanduser("/Documents/Repos/bibigrid/"
"resources/playbook/"),
"playbook")],
creator = create.Create([provider], [configuration], "", startup.LOG)
creator.master_ip = 42
creator.upload_data()
mock_configure_ansible.assert_called_with(providers=creator.providers, configurations=creator.configurations,
cluster_id=creator.cluster_id)
mock_ssh.assert_called_with(floating_ip=creator.master_ip, private_key=create.KEY_FOLDER + creator.key_name,
username=creator.ssh_user,
filepaths=[(os.path.expanduser("/Documents/Repos/bibigrid/"
"resources/playbook/"), "playbook")],
commands=['echo ansible_start'])

@patch("threading.Thread")
Expand Down

0 comments on commit 400b350

Please sign in to comment.