Commit a0be9787 by Chif Gergő

Contract test dictionaries

parent 0f9c85a8
Pipeline #755 failed with stage
in 37 seconds
......@@ -13,7 +13,8 @@ class InstanceInterface:
def create_vm_from_template(self, name, image, flavor, networks):
raise NotImplementedError
def create_multiple_vm_from_template(self, image, flavor, networks, number):
def create_multiple_vm_from_template(self, image, flavor,
networks, number):
raise NotImplementedError
def get_vm(self, name_or_id):
......
......@@ -42,12 +42,12 @@ class BlockDeviceMapping:
def __init__(self, boot_index, uuid, source_type, volume_size,
destination_type, delete_on_termination, disk_bus):
self.boot_index = boot_index
self.uuid=uuid
self.source_type=source_type
self.volume_size=volume_size
self.destination_type=destination_type
self.delete_on_termination=delete_on_termination
self.disk_bus=disk_bus
self.uuid = uuid
self.source_type = source_type
self.volume_size = volume_size
self.destination_type = destination_type
self.delete_on_termination = delete_on_termination
self.disk_bus = disk_bus
def JSON(self):
return json.dumps(self.__dict__)
# This file is for testing the openstack api access
if __name__ == "__main__":
from sys import path
from os.path import dirname as dir
path.append(dir(path[0]))
from implementation.vm.instance import OSVirtualMachineManager
from interface.vm.resources import BlockDeviceMapping, Flavor
import openstack
from implementation.network.OpenstackFloatingIPManager import OpenstackFloatingIPManager
from implementation.network.OpenstackNetworkManager import OpenstackNetworkManager
from implementation.network.OpenstackPortManager import OpenstackPortManager
from implementation.network.OpenstackRouterManager import OpenstackRouterManager
from implementation.network.OpenstackSubnetManager import OpenstackSubnetManager
# from implementation.network.OpenstackFloatingIPManager import OpenstackFloatingIPManager
# from implementation.network.OpenstackNetworkManager import OpenstackNetworkManager
# from implementation.network.OpenstackPortManager import OpenstackPortManager
# from implementation.network.OpenstackRouterManager import OpenstackRouterManager
# from implementation.network.OpenstackSubnetManager import OpenstackSubnetManager
# openstack.enable_logging(debug=True)
conn = openstack.connect(cloud='openstack')
auth = {
"auth_url": "http://10.34.0.113/identity/v3",
"username": "admin",
"project_id": "2db5309f541c4466bc80bc534cf579d7",
"project_name": "admin",
"password": "64c7ee341d03844c548c",
"user_domain_name": "Default",
"user_domain_id": "Default",
"region_name": "RegionOne",
}
interface = OSVirtualMachineManager(auth=auth)
block_dev_map = BlockDeviceMapping(
boot_index=0,
......@@ -23,7 +38,7 @@ block_dev_map = BlockDeviceMapping(
)
networks = [{"uuid": "c03d0d4b-413e-4cc6-9ebe-c0b5ca0dac3a"}]
interface = OSVirtualMachineManager(conn)
print('#'*40)
# interface.create_vm_from_template('new_server', resource, "cirros-0.4.0-x86_64-disk")
......
import unittest
from interface.image.Image import Image
class TestImage(unittest.TestCase):
def test_fields(self):
image = Image('id', 'name', 'format')
self.assertEqual(image.id, 'id')
self.assertEqual(image.name, 'name')
self.assertEqual(image.format, 'format')
if __name__ == '__main__':
unittest.main()
import unittest
from interface.storage.Snapshot import Snapshot
class TestSnapshot(unittest.TestCase):
def test_fields(self):
snapshot = Snapshot('id', 'volume_id', 1, 'status', 'created_at')
self.assertEqual(snapshot.id, 'id')
self.assertEqual(snapshot.volume_id, 'volume_id')
self.assertEqual(snapshot.size, 1)
self.assertEqual(snapshot.status, 'status')
self.assertEqual(snapshot.created_at, 'created_at')
if __name__ == '__main__':
unittest.main()
import unittest
from unittest.mock import MagicMock
from unittest.mock import MagicMock, create_autospec
from implementation.vm.instance import OSVirtualMachineManager
from interface.vm.resources import Instance
import openstack
servers = [
MagicMock(
......@@ -9,7 +9,7 @@ servers = [
name="test1",
flavorRef="flav1",
imageRef="image1",
networks= [{"uuid": "network1"}],
networks=[{"uuid": "network1"}],
terminated_at="20200320-15-31",
launched_at="20190510-13-22",
disks=["disk1"],
......@@ -28,6 +28,8 @@ class MockOpenStackCompute(MagicMock):
self.compute.wait_for_server = MagicMock(return_value=True)
self.compute.list_servers = MagicMock(return_value=servers)
self.compute.get_server = create_autospec()
class InstanceCreateTestCase(unittest.TestCase):
def setUp(self):
......@@ -40,9 +42,12 @@ class InstanceCreateTestCase(unittest.TestCase):
pass
def test_create_from_template(self):
instance = self.manager.create_vm_from_template('test', 'imageid', 'flavorid',
instance = self.manager.create_vm_from_template('test1',
'imageid',
'flavorid',
['networkid1'])
self.conn.compute.create_server.assert_called()
self.assertEqual(instance.name, servers[0].name)
def test_create_from_template_params(self):
self.manager.create_vm_from_template('test', 'imageid', 'flavorid',
......@@ -53,3 +58,6 @@ class InstanceCreateTestCase(unittest.TestCase):
imageRef="uuid2",
networks=['networkid1'],
min_count=1)
def test_suspend(self):
self.compute.suspend_server()
import unittest
from interface.storage.Volume import Volume
class TestVolume(unittest.TestCase):
def test_fields(self):
volume = Volume('id', 'image_id', 1, False, 'status', 'created_at')
self.assertEqual(volume.id, 'id')
self.assertEqual(volume.image_id, 'image_id')
self.assertEqual(volume.size, 1)
self.assertEqual(volume.bootable, False)
self.assertEqual(volume.status, 'status')
self.assertEqual(volume.created_at, 'created_at')
if __name__ == '__main__':
unittest.main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment