def set_up(self):

self.vs = SoftLayer.VSManager(self.client,

SoftLayer.OrderingManager(self.client))

def test_list_instances(self):

results = self.vs.list_instances(hourly=True, monthly=True)

for result in results:

self.assertIn(result['id'], [100, 104])

self.assert_called_with('SoftLayer_Account', 'getVirtualGuests')

def test_list_instances_neither(self):

results = self.vs.list_instances(hourly=False, monthly=False)

for result in results:

self.assertIn(result['id'], [100, 104])

self.assert_called_with('SoftLayer_Account', 'getVirtualGuests')

def test_list_instances_monthly(self):

results = self.vs.list_instances(hourly=False, monthly=True)

for result in results:

self.assertIn(result['id'], [100])

self.assert_called_with('SoftLayer_Account', 'getMonthlyVirtualGuests')

def test_list_instances_hourly(self):

results = self.vs.list_instances(hourly=True, monthly=False)

for result in results:

self.assertIn(result['id'], [104])

self.assert_called_with('SoftLayer_Account', 'getHourlyVirtualGuests')

def test_list_instances_with_filters(self):

self.vs.list_instances(

hourly=True,

monthly=True,

tags=['tag1', 'tag2'],

cpus=2,

memory=1024,

hostname='hostname',

domain='example.com',

local_disk=True,

datacenter='dal05',

nic_speed=100,

public_ip='1.2.3.4',

private_ip='4.3.2.1',

)

_filter = {

'virtualGuests': {

'datacenter': {

'name': {'operation': '_= dal05'}},

'domain': {'operation': '_= example.com'},

'tagReferences': {

'tag': {'name': {

'operation': 'in',

'options': [{

'name': 'data', 'value': ['tag1', 'tag2']}]}}},

'maxCpu': {'operation': 2},

'localDiskFlag': {'operation': True},

'maxMemory': {'operation': 1024},

'hostname': {'operation': '_= hostname'},

'networkComponents': {'maxSpeed': {'operation': 100}},

'primaryIpAddress': {'operation': '_= 1.2.3.4'},

'primaryBackendIpAddress': {'operation': '_= 4.3.2.1'}

}

}

self.assert_called_with('SoftLayer_Account', 'getVirtualGuests',

filter=_filter)

def test_resolve_ids_ip(self):

_id = self.vs._get_ids_from_ip('172.16.240.2')

self.assertEqual(_id, [100, 104])

def test_resolve_ids_ip_private(self):

# Now simulate a private IP test

mock = self.set_mock('SoftLayer_Account', 'getVirtualGuests')

mock.side_effect = [[], [{'id': 99}]]

_id = self.vs._get_ids_from_ip('10.0.1.87')

self.assertEqual(_id, [99])

def test_resolve_ids_ip_invalid(self):

_id = self.vs._get_ids_from_ip('nope')

self.assertEqual(_id, [])

def test_resolve_ids_hostname(self):

_id = self.vs._get_ids_from_hostname('vs-test1')

self.assertEqual(_id, [100, 104])

def test_get_instance(self):

result = self.vs.get_instance(100)

self.assertEqual(fixtures.SoftLayer_Virtual_Guest.getObject, result)

self.assert_called_with('SoftLayer_Virtual_Guest', 'getObject',

identifier=100)

def test_get_create_options(self):

results = self.vs.get_create_options()

self.assertEqual(

fixtures.SoftLayer_Virtual_Guest.getCreateObjectOptions, results)

def test_cancel_instance(self):

result = self.vs.cancel_instance(1)

self.assertEqual(result, True)

self.assert_called_with('SoftLayer_Virtual_Guest', 'deleteObject',

identifier=1)

def test_reload_instance(self):

self.vs.reload_instance(1)

self.assert_called_with('SoftLayer_Virtual_Guest',

'reloadOperatingSystem',

args=('FORCE', {}),

identifier=1)

def test_reload_instance_posturi_sshkeys(self):

post_uri = 'http://test.sftlyr.ws/test.sh'

self.vs.reload_instance(1, post_uri=post_uri, ssh_keys=[1701])

args = ('FORCE', {'customProvisionScriptUri': post_uri,

'sshKeyIds': [1701]})

self.assert_called_with('SoftLayer_Virtual_Guest',

'reloadOperatingSystem',

args=args,

identifier=1)

def test_reload_instance_with_new_os(self):

self.vs.reload_instance(1, image_id=1234)

args = ('FORCE', {'imageTemplateId': 1234})

self.assert_called_with('SoftLayer_Virtual_Guest',

'reloadOperatingSystem',

args=args,

identifier=1)

@mock.patch('SoftLayer.managers.vs.VSManager._generate_create_dict')

def test_create_verify(self, create_dict):

create_dict.return_value = {'test': 1, 'verify': 1}

self.vs.verify_create_instance(test=1, verify=1, tags=['test', 'tags'])

create_dict.assert_called_once_with(test=1, verify=1)

self.assert_called_with('SoftLayer_Virtual_Guest',

'generateOrderTemplate',

args=({'test': 1, 'verify': 1},))

@mock.patch('SoftLayer.managers.vs.VSManager._generate_create_dict')

def test_create_instance(self, create_dict):

create_dict.return_value = {'test': 1, 'verify': 1}

self.vs.create_instance(test=1, verify=1, tags='dev,green')

create_dict.assert_called_once_with(test=1, verify=1)

self.assert_called_with('SoftLayer_Virtual_Guest', 'createObject',

args=({'test': 1, 'verify': 1},))

self.assert_called_with('SoftLayer_Virtual_Guest', 'setTags',

args=('dev,green',),

identifier=100)

def test_create_instances(self):

self.vs.create_instances([{'cpus': 1,

'memory': 1024,

'hostname': 'server',

'domain': 'example.com',

'tags': 'dev,green'}])

args = ([{'domain': 'example.com',

'hourlyBillingFlag': True,

'localDiskFlag': True,

'maxMemory': 1024,

'hostname': 'server',

'startCpus': 1}],)

self.assert_called_with('SoftLayer_Virtual_Guest', 'createObjects',

args=args)

self.assert_called_with('SoftLayer_Virtual_Guest', 'setTags',

args=('dev,green',),

identifier=100)

def test_generate_os_and_image(self):

self.assertRaises(

ValueError,

self.vs._generate_create_dict,

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code=1,

image_id=1,

)

def test_generate_missing(self):

self.assertRaises(ValueError, self.vs._generate_create_dict)

self.assertRaises(ValueError, self.vs._generate_create_dict, cpus=1)

def test_generate_basic(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'hourlyBillingFlag': True,

}

self.assertEqual(data, assert_data)

def test_generate_monthly(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

hourly=False,

)

assert_data = {

'hourlyBillingFlag': False,

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

}

self.assertEqual(data, assert_data)

def test_generate_image_id(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

image_id="45",

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'blockDeviceTemplateGroup': {"globalIdentifier": "45"},

'hourlyBillingFlag': True,

}

self.assertEqual(data, assert_data)

def test_generate_dedicated(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

dedicated=True,

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'hourlyBillingFlag': True,

'dedicatedAccountHostOnlyFlag': True,

}

self.assertEqual(data, assert_data)

def test_generate_datacenter(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

datacenter="sng01",

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'hourlyBillingFlag': True,

'datacenter': {"name": 'sng01'},

}

self.assertEqual(data, assert_data)

def test_generate_public_vlan(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

public_vlan=1,

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'hourlyBillingFlag': True,

'primaryNetworkComponent': {"networkVlan": {"id": 1}},

}

self.assertEqual(data, assert_data)

def test_generate_private_vlan(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

private_vlan=1,

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'hourlyBillingFlag': True,

'primaryBackendNetworkComponent': {"networkVlan": {"id": 1}},

}

self.assertEqual(data, assert_data)

def test_generate_userdata(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

userdata="ICANHAZVSI",

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'hourlyBillingFlag': True,

'userData': [{'value': "ICANHAZVSI"}],

}

self.assertEqual(data, assert_data)

def test_generate_network(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

nic_speed=9001,

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'hourlyBillingFlag': True,

'networkComponents': [{'maxSpeed': 9001}],

}

self.assertEqual(data, assert_data)

def test_generate_private_network_only(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

nic_speed=9001,

private=True

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'privateNetworkOnlyFlag': True,

'hourlyBillingFlag': True,

'networkComponents': [{'maxSpeed': 9001}],

}

self.assertEqual(data, assert_data)

def test_generate_post_uri(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

post_uri='https://example.com/boostrap.sh',

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'hourlyBillingFlag': True,

'postInstallScriptUri': 'https://example.com/boostrap.sh',

}

self.assertEqual(data, assert_data)

def test_generate_sshkey(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

ssh_keys=[543],

)

assert_data = {

'startCpus': 1,

'maxMemory': 1,

'hostname': 'test',

'domain': 'example.com',

'localDiskFlag': True,

'operatingSystemReferenceCode': "STRING",

'hourlyBillingFlag': True,

'sshKeys': [{'id': 543}],

}

self.assertEqual(data, assert_data)

def test_generate_no_disks(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING"

)

self.assertEqual(data.get('blockDevices'), None)

def test_generate_single_disk(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

disks=[50]

)

assert_data = {

'blockDevices': [

{"device": "0", "diskImage": {"capacity": 50}}]

}

self.assertTrue(data.get('blockDevices'))

self.assertEqual(data['blockDevices'], assert_data['blockDevices'])

def test_generate_multi_disk(self):

data = self.vs._generate_create_dict(

cpus=1,

memory=1,

hostname='test',

domain='example.com',

os_code="STRING",

disks=[50, 70, 100]

)

assert_data = {

'blockDevices': [

{"device": "0", "diskImage": {"capacity": 50}},

{"device": "2", "diskImage": {"capacity": 70}},

{"device": "3", "diskImage": {"capacity": 100}}]

}

self.assertTrue(data.get('blockDevices'))

self.assertEqual(data['blockDevices'], assert_data['blockDevices'])

def test_change_port_speed_public(self):

result = self.vs.change_port_speed(1, True, 100)

self.assertEqual(result, True)

self.assert_called_with('SoftLayer_Virtual_Guest',

'setPublicNetworkInterfaceSpeed',

identifier=1,

args=(100,))

def test_change_port_speed_private(self):

result = self.vs.change_port_speed(2, False, 10)

self.assertEqual(result, True)

self.assert_called_with('SoftLayer_Virtual_Guest',

'setPrivateNetworkInterfaceSpeed',

identifier=2,

args=(10,))

def test_rescue(self):

# Test rescue environment

result = self.vs.rescue(1234)

self.assertEqual(result, True)

self.assert_called_with('SoftLayer_Virtual_Guest',

'executeRescueLayer',

identifier=1234)

def test_edit_metadata(self):

# Test editing user data

result = self.vs.edit(100, userdata='my data')

self.assertEqual(result, True)

self.assert_called_with('SoftLayer_Virtual_Guest', 'setUserMetadata',

identifier=100,

args=(['my data'],))

def test_edit_blank(self):

# Now test a blank edit

self.assertTrue(self.vs.edit, 100)

def test_edit_full(self):

result = self.vs.edit(100,

hostname='new-host',

domain='new.sftlyr.ws',

notes='random notes')

self.assertEqual(result, True)

args = ({

'hostname': 'new-host',

'domain': 'new.sftlyr.ws',

'notes': 'random notes',

},)

self.assert_called_with('SoftLayer_Virtual_Guest', 'editObject',

identifier=100,

args=args)

def test_edit_tags(self):

# Test tag support

result = self.vs.edit(100, tags='dev,green')

self.assertEqual(result, True)

self.assert_called_with('SoftLayer_Virtual_Guest', 'setTags',

identifier=100,

args=('dev,green',))

def test_edit_tags_blank(self):

result = self.vs.edit(100, tags='')

self.assertEqual(result, True)

self.assert_called_with('SoftLayer_Virtual_Guest', 'setTags',

identifier=100,

args=('',))

def test_captures(self):

# capture only the OS disk

result = self.vs.capture(1, 'a')

expected = fixtures.SoftLayer_Virtual_Guest.createArchiveTransaction

self.assertEqual(result, expected)

args = ('a', [], None)

self.assert_called_with('SoftLayer_Virtual_Guest',

'createArchiveTransaction',

args=args,

identifier=1)

def test_capture_additional_disks(self):

# capture all the disks, minus the swap

# make sure the data is carried along with it

result = self.vs.capture(1, 'a', additional_disks=True)

expected = fixtures.SoftLayer_Virtual_Guest.createArchiveTransaction

self.assertEqual(result, expected)

args = ('a', [{'device': 0, 'mountType': 'Disk', 'uuid': 1},

{'device': 3, 'mountType': 'Disk', 'uuid': 3}], None)

self.assert_called_with('SoftLayer_Virtual_Guest',

'createArchiveTransaction',

args=args,

identifier=1)

def test_upgrade(self):

mock = self.set_mock('SoftLayer_Product_Package', 'getAllObjects')

mock.return_value = [

{'id': 46, 'name': 'Virtual Servers',

'description': 'Virtual Server Instances',

'type': {'keyName': 'VIRTUAL_SERVER_INSTANCE'}, 'isActive': 1},

]

# test single upgrade

result = self.vs.upgrade(1, cpus=4, public=False)

self.assertEqual(result, True)

self.assert_called_with('SoftLayer_Product_Order', 'placeOrder')

call = self.calls('SoftLayer_Product_Order', 'placeOrder')[0]

order_container = call.args[0]

self.assertEqual(order_container['prices'], [{'id': 1007}])

self.assertEqual(order_container['virtualGuests'], [{'id': 1}])

def test_upgrade_blank(self):

# Now test a blank upgrade

result = self.vs.upgrade(1)

self.assertEqual(result, False)

self.assertEqual(self.calls('SoftLayer_Product_Order', 'placeOrder'),

[])

def test_upgrade_full(self):

# Testing all parameters Upgrade

result = self.vs.upgrade(1,

cpus=4,

memory=2,

nic_speed=1000,

public=True)

self.assertEqual(result, True)

self.assert_called_with('SoftLayer_Product_Order', 'placeOrder')

call = self.calls('SoftLayer_Product_Order', 'placeOrder')[0]

order_container = call.args[0]

self.assertIn({'id': 1144}, order_container['prices'])

self.assertIn({'id': 1133}, order_container['prices'])

self.assertIn({'id': 1122}, order_container['prices'])

self.assertEqual(order_container['virtualGuests'], [{'id': 1}])

def test_upgrade_skips_location_based_prices(self):

# Test that no prices that have locationGroupId set are used

self.assertRaises(exceptions.SoftLayerError,

self.vs.upgrade, 1, cpus=55, memory=2, public=True)

def test_get_item_id_for_upgrade(self):

item_id = 0

package_items = self.client['Product_Package'].getItems(id=46)

for item in package_items:

if ((item['prices'][0]['categories'][0]['id'] == 3)

and (item.get('capacity') == '2')):

item_id = item['prices'][0]['id']

break

def set_up(self):

self.client = mock.MagicMock()

self.vs = SoftLayer.VSManager(self.client)

self.guestObject = self.client['Virtual_Guest'].getObject

@mock.patch('SoftLayer.managers.vs.VSManager.wait_for_ready')

def test_wait_interface(self, ready):

# verify interface to wait_for_ready is intact

self.vs.wait_for_transaction(1, 1)

ready.assert_called_once_with(1, 1, delay=1, pending=True)

def test_active_not_provisioned(self):

# active transaction and no provision date should be false

self.guestObject.side_effect = [

{'activeTransaction': {'id': 1}},

]

value = self.vs.wait_for_ready(1, 0)

self.assertFalse(value)

def test_active_and_provisiondate(self):

# active transaction and provision date should be True

self.guestObject.side_effect = [

{'activeTransaction': {'id': 1},

'provisionDate': 'aaa'},

]

value = self.vs.wait_for_ready(1, 1)

self.assertTrue(value)

def test_active_provision_pending(self):

# active transaction and provision date

# and pending should be false

self.guestObject.side_effect = [

{'activeTransaction': {'id': 1}, 'provisionDate': 'aaa'},

]

value = self.vs.wait_for_ready(1, 0, pending=True)

self.assertFalse(value)

def test_active_reload(self):

# actively running reload

self.guestObject.side_effect = [

{

'activeTransaction': {'id': 1},

'provisionDate': 'aaa',

'lastOperatingSystemReload': {'id': 1},

},

]

value = self.vs.wait_for_ready(1, 0)

self.assertFalse(value)

def test_reload_no_pending(self):

# reload complete, maintance transactions

self.guestObject.side_effect = [

{

'activeTransaction': {'id': 2},

'provisionDate': 'aaa',

'lastOperatingSystemReload': {'id': 1},

},

]

value = self.vs.wait_for_ready(1, 1)

self.assertTrue(value)

def test_reload_pending(self):

# reload complete, pending maintance transactions

self.guestObject.side_effect = [

{

'activeTransaction': {'id': 2},

'provisionDate': 'aaa',

'lastOperatingSystemReload': {'id': 1},

},

]

value = self.vs.wait_for_ready(1, 0, pending=True)

self.assertFalse(value)

@mock.patch('time.sleep')

def test_ready_iter_once_incomplete(self, _sleep):

self.guestObject = self.client['Virtual_Guest'].getObject

# no iteration, false

self.guestObject.side_effect = [

{'activeTransaction': {'id': 1}},

]

value = self.vs.wait_for_ready(1, 0)

self.assertFalse(value)

self.assertFalse(_sleep.called)

@mock.patch('time.sleep')

def test_iter_once_complete(self, _sleep):

# no iteration, true

self.guestObject.side_effect = [

{'provisionDate': 'aaa'},

]

value = self.vs.wait_for_ready(1, 1)

self.assertTrue(value)

self.assertFalse(_sleep.called)

@mock.patch('time.sleep')

def test_iter_four_complete(self, _sleep):

# test 4 iterations with positive match

self.guestObject.side_effect = [

{'activeTransaction': {'id': 1}},

{'activeTransaction': {'id': 1}},

{'activeTransaction': {'id': 1}},

{'provisionDate': 'aaa'},

]

value = self.vs.wait_for_ready(1, 4)

self.assertTrue(value)

_sleep.assert_has_calls([mock.call(1), mock.call(1), mock.call(1)])

self.guestObject.assert_has_calls([

mock.call(id=1, mask=mock.ANY), mock.call(id=1, mask=mock.ANY),

mock.call(id=1, mask=mock.ANY), mock.call(id=1, mask=mock.ANY),

])

@mock.patch('time.time')

@mock.patch('time.sleep')

def test_iter_two_incomplete(self, _sleep, _time):

# test 2 iterations, with no matches

self.guestObject.side_effect = [

{'activeTransaction': {'id': 1}},

{'activeTransaction': {'id': 1}},

{'provisionDate': 'aaa'}

]

_time.side_effect = [0, 1, 2]

value = self.vs.wait_for_ready(1, 2)

self.assertFalse(value)

_sleep.assert_called_once_with(1)

self.guestObject.assert_has_calls([

mock.call(id=1, mask=mock.ANY),

mock.call(id=1, mask=mock.ANY),

])

@mock.patch('time.time')

@mock.patch('time.sleep')

def test_iter_20_incomplete(self, _sleep, _time):

"""Wait for up to 20 seconds (sleeping for 10 seconds) for a server."""

self.guestObject.return_value = {'activeTransaction': {'id': 1}}

_time.side_effect = [0, 10, 20]

value = self.vs.wait_for_ready(1, 20, delay=10)

self.assertFalse(value)

self.guestObject.assert_has_calls([mock.call(id=1, mask=mock.ANY)])