diff options
Diffstat (limited to 'openstackclient/tests')
| -rw-r--r-- | openstackclient/tests/functional/compute/v2/test_server.py | 495 | ||||
| -rw-r--r-- | openstackclient/tests/functional/network/v2/test_router.py | 235 | ||||
| -rw-r--r-- | openstackclient/tests/functional/volume/v1/test_volume_type.py | 71 | ||||
| -rw-r--r-- | openstackclient/tests/functional/volume/v2/test_volume_type.py | 87 | ||||
| -rw-r--r-- | openstackclient/tests/unit/compute/v2/test_server.py | 146 | ||||
| -rw-r--r-- | openstackclient/tests/unit/volume/v1/fakes.py | 31 | ||||
| -rw-r--r-- | openstackclient/tests/unit/volume/v1/test_type.py | 237 | ||||
| -rw-r--r-- | openstackclient/tests/unit/volume/v2/fakes.py | 31 | ||||
| -rw-r--r-- | openstackclient/tests/unit/volume/v2/test_type.py | 276 | ||||
| -rw-r--r-- | openstackclient/tests/unit/volume/v2/test_volume_host.py | 31 |
10 files changed, 1385 insertions, 255 deletions
diff --git a/openstackclient/tests/functional/compute/v2/test_server.py b/openstackclient/tests/functional/compute/v2/test_server.py index 119ef05c..4d8019b2 100644 --- a/openstackclient/tests/functional/compute/v2/test_server.py +++ b/openstackclient/tests/functional/compute/v2/test_server.py @@ -12,12 +12,12 @@ import json import time +import uuid -from tempest.lib.common.utils import data_utils +from tempest.lib import exceptions from openstackclient.tests.functional import base from openstackclient.tests.functional.volume.v2 import test_volume -from tempest.lib import exceptions class ServerTests(base.TestCase): @@ -27,23 +27,30 @@ class ServerTests(base.TestCase): def get_flavor(cls): # NOTE(rtheis): Get cirros256 or m1.tiny flavors since functional # tests may create other flavors. - flavors = cls.openstack('flavor list -c Name -f value').split('\n') + flavors = json.loads(cls.openstack( + "flavor list -f json " + )) server_flavor = None for flavor in flavors: - if flavor in ['m1.tiny', 'cirros256']: - server_flavor = flavor + if flavor['Name'] in ['m1.tiny', 'cirros256']: + server_flavor = flavor['Name'] break return server_flavor @classmethod def get_image(cls): - # NOTE(rtheis): Get cirros image since functional tests may - # create other images. - images = cls.openstack('image list -c Name -f value').split('\n') + # NOTE(rtheis): Get first Cirros image since functional tests may + # create other images. Image may be named '-uec' or + # '-disk'. + images = json.loads(cls.openstack( + "image list -f json " + )) server_image = None for image in images: - if image.startswith('cirros-') and image.endswith('-uec'): - server_image = image + if (image['Name'].startswith('cirros-') and + (image['Name'].endswith('-uec') or + image['Name'].endswith('-disk'))): + server_image = image['Name'] break return server_image @@ -52,92 +59,84 @@ class ServerTests(base.TestCase): try: # NOTE(rtheis): Get private network since functional tests may # create other networks. - raw_output = cls.openstack('network show private -c id -f value') + cmd_output = json.loads(cls.openstack( + 'network show private -f json' + )) except exceptions.CommandFailed: return '' - return ' --nic net-id=' + raw_output.strip('\n') + return '--nic net-id=' + cmd_output['id'] def server_create(self, name=None): - """Create server. Add cleanup.""" - name = name or data_utils.rand_uuid() - opts = self.get_opts(self.FIELDS) - raw_output = self.openstack('--debug server create --flavor ' + - self.flavor_name + - ' --image ' + self.image_name + - self.network_arg + ' ' + - name + opts) - if not raw_output: + """Create server, with cleanup""" + name = name or uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'server create -f json ' + + '--flavor ' + self.flavor_name + ' ' + + '--image ' + self.image_name + ' ' + + self.network_arg + ' ' + + '--wait ' + + name + )) + if not cmd_output: self.fail('Server has not been created!') self.addCleanup(self.server_delete, name) - - def server_list(self, params=[]): - """List servers.""" - opts = self.get_opts(params) - return self.openstack('server list' + opts) + self.assertEqual( + name, + cmd_output["name"], + ) + return cmd_output def server_delete(self, name): - """Delete server by name.""" + """Delete server by name""" self.openstack('server delete ' + name) def setUp(self): - """Set necessary variables and create server.""" + """Select common resources""" super(ServerTests, self).setUp() self.flavor_name = self.get_flavor() self.image_name = self.get_image() self.network_arg = self.get_network() - self.NAME = data_utils.rand_name('TestServer') - self.OTHER_NAME = data_utils.rand_name('TestServer') - self.HEADERS = ['"Name"'] - self.FIELDS = ['name'] - self.IP_POOL = 'public' - self.server_create(self.NAME) - - def test_server_rename(self): - """Test server rename command. - - Test steps: - 1) Boot server in setUp - 2) Rename server - 3) Check output - 4) Rename server back to original name - """ - raw_output = self.openstack('server set --name ' + self.OTHER_NAME + - ' ' + self.NAME) - self.assertOutput("", raw_output) - self.assertNotIn(self.NAME, self.server_list(['Name'])) - self.assertIn(self.OTHER_NAME, self.server_list(['Name'])) - self.openstack('server set --name ' + self.NAME + ' ' + - self.OTHER_NAME) - def test_server_list(self): - """Test server list command. + """Test server list, set""" + cmd_output = self.server_create() + name1 = cmd_output['name'] + cmd_output = self.server_create() + name2 = cmd_output['name'] + self.wait_for_status(name1, "ACTIVE") + self.wait_for_status(name2, "ACTIVE") - Test steps: - 1) Boot server in setUp - 2) List servers - 3) Check output - """ - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('server list' + opts) - self.assertIn(self.NAME, raw_output) + cmd_output = json.loads(self.openstack( + 'server list -f json' + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + self.assertIn(name2, col_name) - def test_server_show(self): - """Test server create, server delete commands""" - name1 = data_utils.rand_name('TestServer') + # Test list --status PAUSED + raw_output = self.openstack('server pause ' + name2) + self.assertEqual("", raw_output) + self.wait_for_status(name2, "PAUSED") cmd_output = json.loads(self.openstack( - 'server create -f json ' + - '--flavor ' + self.flavor_name + ' ' + - '--image ' + self.image_name + ' ' + - self.network_arg + ' ' + - name1 + 'server list -f json ' + + '--status ACTIVE' )) - self.assertIsNotNone(cmd_output["id"]) - self.addCleanup(self.openstack, 'server delete ' + name1) - self.assertEqual( - name1, - cmd_output["name"], - ) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + self.assertNotIn(name2, col_name) + cmd_output = json.loads(self.openstack( + 'server list -f json ' + + '--status PAUSED' + )) + col_name = [x["Name"] for x in cmd_output] + self.assertNotIn(name1, col_name) + self.assertIn(name2, col_name) + + def test_server_set(self): + """Test server create, delete, set, show""" + cmd_output = self.server_create() + name = cmd_output['name'] + # self.wait_for_status(name, "ACTIVE") # Have a look at some other fields flavor = json.loads(self.openstack( @@ -165,171 +164,176 @@ class ServerTests(base.TestCase): cmd_output["image"], ) - def test_server_metadata(self): - """Test command to set server metadata. + # Test properties set + raw_output = self.openstack( + 'server set ' + + '--property a=b --property c=d ' + + name + ) + self.assertOutput('', raw_output) + + cmd_output = json.loads(self.openstack( + 'server show -f json ' + + name + )) + # Really, shouldn't this be a list? + self.assertEqual( + "a='b', c='d'", + cmd_output['properties'], + ) - Test steps: - 1) Boot server in setUp - 2) Set properties for server - 3) Check server properties in server show output - 4) Unset properties for server - 5) Check server properties in server show output - """ - self.wait_for_status("ACTIVE") - # metadata raw_output = self.openstack( - 'server set --property a=b --property c=d ' + self.NAME) - opts = self.get_opts(["name", "properties"]) - raw_output = self.openstack('server show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\na='b', c='d'\n", raw_output) + 'server unset ' + + '--property a ' + + name + ) + cmd_output = json.loads(self.openstack( + 'server show -f json ' + + name + )) + self.assertEqual( + "c='d'", + cmd_output['properties'], + ) + # Test set --name + new_name = uuid.uuid4().hex raw_output = self.openstack( - 'server unset --property a ' + self.NAME) - opts = self.get_opts(["name", "properties"]) - raw_output = self.openstack('server show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\nc='d'\n", raw_output) - - def test_server_suspend_resume(self): - """Test server suspend and resume commands. - - Test steps: - 1) Boot server in setUp - 2) Suspend server - 3) Check for SUSPENDED server status - 4) Resume server - 5) Check for ACTIVE server status - """ - self.wait_for_status("ACTIVE") - # suspend - raw_output = self.openstack('server suspend ' + self.NAME) - self.assertEqual("", raw_output) - self.wait_for_status("SUSPENDED") - # resume - raw_output = self.openstack('server resume ' + self.NAME) - self.assertEqual("", raw_output) - self.wait_for_status("ACTIVE") + 'server set ' + + '--name ' + new_name + ' ' + + name + ) + self.assertOutput("", raw_output) + cmd_output = json.loads(self.openstack( + 'server show -f json ' + + new_name + )) + self.assertEqual( + new_name, + cmd_output["name"], + ) + # Put it back so we clean up properly + raw_output = self.openstack( + 'server set ' + + '--name ' + name + ' ' + + new_name + ) + self.assertOutput("", raw_output) - def test_server_lock_unlock(self): - """Test server lock and unlock commands. + def test_server_actions(self): + """Test server action pairs - Test steps: - 1) Boot server in setUp - 2) Lock server - 3) Check output - 4) Unlock server - 5) Check output + suspend/resume + pause/unpause + rescue/unrescue + lock/unlock """ - self.wait_for_status("ACTIVE") - # lock - raw_output = self.openstack('server lock ' + self.NAME) - self.assertEqual("", raw_output) - # unlock - raw_output = self.openstack('server unlock ' + self.NAME) + cmd_output = self.server_create() + name = cmd_output['name'] + + # suspend + raw_output = self.openstack('server suspend ' + name) self.assertEqual("", raw_output) + self.wait_for_status(name, "SUSPENDED") - def test_server_pause_unpause(self): - """Test server pause and unpause commands. + # resume + raw_output = self.openstack('server resume ' + name) + self.assertEqual("", raw_output) + self.wait_for_status(name, "ACTIVE") - Test steps: - 1) Boot server in setUp - 2) Pause server - 3) Check for PAUSED server status - 4) Unpause server - 5) Check for ACTIVE server status - """ - self.wait_for_status("ACTIVE") # pause - raw_output = self.openstack('server pause ' + self.NAME) + raw_output = self.openstack('server pause ' + name) self.assertEqual("", raw_output) - self.wait_for_status("PAUSED") + self.wait_for_status(name, "PAUSED") + # unpause - raw_output = self.openstack('server unpause ' + self.NAME) + raw_output = self.openstack('server unpause ' + name) self.assertEqual("", raw_output) - self.wait_for_status("ACTIVE") - - def test_server_rescue_unrescue(self): - """Test server rescue and unrescue commands. + self.wait_for_status(name, "ACTIVE") - Test steps: - 1) Boot server in setUp - 2) Rescue server - 3) Check for RESCUE server status - 4) Unrescue server - 5) Check for ACTIVE server status - """ - self.wait_for_status("ACTIVE") # rescue - opts = self.get_opts(["adminPass"]) - raw_output = self.openstack('server rescue ' + self.NAME + opts) + raw_output = self.openstack('server rescue ' + name) self.assertNotEqual("", raw_output) - self.wait_for_status("RESCUE") + self.wait_for_status(name, "RESCUE") + # unrescue - raw_output = self.openstack('server unrescue ' + self.NAME) + raw_output = self.openstack('server unrescue ' + name) + self.assertEqual("", raw_output) + self.wait_for_status(name, "ACTIVE") + + # lock + raw_output = self.openstack('server lock ' + name) + self.assertEqual("", raw_output) + # NOTE(dtroyer): No way to verify this status??? + + # unlock + raw_output = self.openstack('server unlock ' + name) self.assertEqual("", raw_output) - self.wait_for_status("ACTIVE") + # NOTE(dtroyer): No way to verify this status??? def test_server_attach_detach_floating_ip(self): - """Test commands to attach and detach floating IP for server. - - Test steps: - 1) Boot server in setUp - 2) Create floating IP - 3) Add floating IP to server - 4) Check for floating IP in server show output - 5) Remove floating IP from server - 6) Check that floating IP is not in server show output - 7) Delete floating IP - 8) Check output - """ - self.wait_for_status("ACTIVE") + """Test floating ip create/delete; server add/remove floating ip""" + cmd_output = self.server_create() + name = cmd_output['name'] + self.wait_for_status(name, "ACTIVE") + # attach ip - opts = self.get_opts(["id", "floating_ip_address"]) - raw_output = self.openstack('floating ip create ' + - self.IP_POOL + - opts) - ip, ipid, rol = tuple(raw_output.split('\n')) - self.assertNotEqual("", ipid) - self.assertNotEqual("", ip) - raw_output = self.openstack('server add floating ip ' + self.NAME + - ' ' + ip) + cmd_output = json.loads(self.openstack( + 'floating ip create -f json ' + + 'public' + )) + floating_ip = cmd_output['floating_ip_address'] + self.assertNotEqual('', cmd_output['id']) + self.assertNotEqual('', floating_ip) + self.addCleanup( + self.openstack, + 'floating ip delete ' + cmd_output['id'] + ) + + raw_output = self.openstack( + 'server add floating ip ' + + name + ' ' + + floating_ip + ) self.assertEqual("", raw_output) - raw_output = self.openstack('server show ' + self.NAME) - self.assertIn(ip, raw_output) + cmd_output = json.loads(self.openstack( + 'server show -f json ' + + name + )) + self.assertIn( + floating_ip, + cmd_output['addresses'], + ) # detach ip - raw_output = self.openstack('server remove floating ip ' + self.NAME + - ' ' + ip) - self.assertEqual("", raw_output) - raw_output = self.openstack('server show ' + self.NAME) - self.assertNotIn(ip, raw_output) - raw_output = self.openstack('floating ip delete ' + ipid) + raw_output = self.openstack( + 'server remove floating ip ' + + name + ' ' + + floating_ip + ) self.assertEqual("", raw_output) + cmd_output = json.loads(self.openstack( + 'server show -f json ' + + name + )) + self.assertNotIn( + floating_ip, + cmd_output['addresses'], + ) + def test_server_reboot(self): - """Test server reboot command. + """Test server reboot""" + cmd_output = self.server_create() + name = cmd_output['name'] - Test steps: - 1) Boot server in setUp - 2) Reboot server - 3) Check for ACTIVE server status - """ - self.wait_for_status("ACTIVE") # reboot - raw_output = self.openstack('server reboot ' + self.NAME) + raw_output = self.openstack('server reboot ' + name) self.assertEqual("", raw_output) - self.wait_for_status("ACTIVE") + self.wait_for_status(name, "ACTIVE") def test_server_boot_from_volume(self): - """Test server create from volume, server delete - - Test steps: - 1) Create volume from image - 2) Create empty volume - 3) Create server from new volumes - 4) Check for ACTIVE new server status - 5) Check volumes attached to server - """ - # server_image = self.get_image() + """Test server create from volume, server delete""" # get volume status wait function volume_wait_for = test_volume.VolumeTests( methodName='wait_for', @@ -348,7 +352,7 @@ class ServerTests(base.TestCase): image_size = 1 # create volume from image - volume_name = data_utils.rand_name('volume', self.image_name) + volume_name = uuid.uuid4().hex cmd_output = json.loads(self.openstack( 'volume create -f json ' + '--image ' + self.image_name + ' ' + @@ -364,7 +368,7 @@ class ServerTests(base.TestCase): volume_wait_for("volume", volume_name, "available") # create empty volume - empty_volume_name = data_utils.rand_name('TestVolume') + empty_volume_name = uuid.uuid4().hex cmd_output = json.loads(self.openstack( 'volume create -f json ' + '--size ' + str(image_size) + ' ' + @@ -379,13 +383,14 @@ class ServerTests(base.TestCase): volume_wait_for("volume", empty_volume_name, "available") # create server - server_name = data_utils.rand_name('TestServer') + server_name = uuid.uuid4().hex server = json.loads(self.openstack( 'server create -f json ' + '--flavor ' + self.flavor_name + ' ' + '--volume ' + volume_name + ' ' + '--block-device-mapping vdb=' + empty_volume_name + ' ' + self.network_arg + ' ' + + '--wait ' + server_name )) self.assertIsNotNone(server["id"]) @@ -394,7 +399,6 @@ class ServerTests(base.TestCase): server_name, server['name'], ) - volume_wait_for("server", server_name, "ACTIVE") # check volumes cmd_output = json.loads(self.openstack( @@ -437,25 +441,78 @@ class ServerTests(base.TestCase): cmd_output['status'], ) - def wait_for_status(self, expected_status='ACTIVE', wait=900, interval=30): + def test_server_create_with_none_network(self): + """Test server create with none network option.""" + server_name = uuid.uuid4().hex + server = json.loads(self.openstack( + # auto/none enable in nova micro version (v2.37+) + '--os-compute-api-version 2.latest ' + + 'server create -f json ' + + '--flavor ' + self.flavor_name + ' ' + + '--image ' + self.image_name + ' ' + + '--nic none ' + + server_name + )) + self.assertIsNotNone(server["id"]) + self.addCleanup(self.openstack, 'server delete --wait ' + server_name) + self.assertEqual(server_name, server['name']) + self.wait_for_status(server_name, "ACTIVE") + server = json.loads(self.openstack( + 'server show -f json ' + server_name + )) + self.assertIsNotNone(server['addresses']) + self.assertEqual('', server['addresses']) + + def test_server_create_with_empty_network_option_latest(self): + """Test server create with empty network option in nova 2.latest.""" + server_name = uuid.uuid4().hex + try: + self.openstack( + # auto/none enable in nova micro version (v2.37+) + '--os-compute-api-version 2.latest ' + + 'server create -f json ' + + '--flavor ' + self.flavor_name + ' ' + + '--image ' + self.image_name + ' ' + + server_name + ) + except exceptions.CommandFailed as e: + self.assertIn('nics are required after microversion 2.36', + e.stderr) + else: + self.fail('CommandFailed should be raised.') + + def wait_for_status( + self, + name, + expected_status='ACTIVE', + wait=900, + interval=10, + ): """Wait until server reaches expected status.""" # TODO(thowe): Add a server wait command to osc failures = ['ERROR'] total_sleep = 0 - opts = self.get_opts(['status']) while total_sleep < wait: - status = self.openstack('server show ' + self.NAME + opts) - status = status.rstrip() - print('Waiting for {} current status: {}'.format(expected_status, - status)) + cmd_output = json.loads(self.openstack( + 'server show -f json ' + + name + )) + status = cmd_output['status'] + print('Waiting for {}, current status: {}'.format( + expected_status, + status, + )) if status == expected_status: break self.assertNotIn(status, failures) time.sleep(interval) total_sleep += interval - status = self.openstack('server show ' + self.NAME + opts) - status = status.rstrip() + cmd_output = json.loads(self.openstack( + 'server show -f json ' + + name + )) + status = cmd_output['status'] self.assertEqual(status, expected_status) # give it a little bit more time time.sleep(5) diff --git a/openstackclient/tests/functional/network/v2/test_router.py b/openstackclient/tests/functional/network/v2/test_router.py index 789c3825..443f68b2 100644 --- a/openstackclient/tests/functional/network/v2/test_router.py +++ b/openstackclient/tests/functional/network/v2/test_router.py @@ -10,6 +10,7 @@ # License for the specific language governing permissions and limitations # under the License. +import json import uuid from openstackclient.tests.functional import base @@ -17,34 +18,210 @@ from openstackclient.tests.functional import base class RouterTests(base.TestCase): """Functional tests for router. """ - NAME = uuid.uuid4().hex - HEADERS = ['Name'] - FIELDS = ['name'] - - @classmethod - def setUpClass(cls): - opts = cls.get_opts(cls.FIELDS) - raw_output = cls.openstack('router create ' + cls.NAME + opts) - expected = cls.NAME + '\n' - cls.assertOutput(expected, raw_output) - - @classmethod - def tearDownClass(cls): - raw_output = cls.openstack('router delete ' + cls.NAME) - cls.assertOutput('', raw_output) + + def test_router_create_and_delete(self): + """Test create options, delete""" + name1 = uuid.uuid4().hex + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'router create -f json ' + + name1 + )) + self.assertEqual( + name1, + cmd_output["name"], + ) + cmd_output = json.loads(self.openstack( + 'router create -f json ' + + name2 + )) + self.assertEqual( + name2, + cmd_output["name"], + ) + + del_output = self.openstack( + 'router delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) def test_router_list(self): - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('router list' + opts) - self.assertIn(self.NAME, raw_output) - - def test_router_set(self): - self.openstack('router set --disable ' + self.NAME) - opts = self.get_opts(['name', 'admin_state_up']) - raw_output = self.openstack('router show ' + self.NAME + opts) - self.assertEqual("DOWN\n" + self.NAME + "\n", raw_output) - - def test_router_show(self): - opts = self.get_opts(self.FIELDS) - raw_output = self.openstack('router show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\n", raw_output) + """Test create, list filter""" + # Get project IDs + cmd_output = json.loads(self.openstack('token issue -f json ')) + auth_project_id = cmd_output['project_id'] + + cmd_output = json.loads(self.openstack('project list -f json ')) + admin_project_id = None + demo_project_id = None + for p in cmd_output: + if p['Name'] == 'admin': + admin_project_id = p['ID'] + if p['Name'] == 'demo': + demo_project_id = p['ID'] + + # Verify assumptions: + # * admin and demo projects are present + # * demo and admin are distinct projects + # * tests run as admin + self.assertIsNotNone(admin_project_id) + self.assertIsNotNone(demo_project_id) + self.assertNotEqual(admin_project_id, demo_project_id) + self.assertEqual(admin_project_id, auth_project_id) + + name1 = uuid.uuid4().hex + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'router create -f json ' + + '--disable ' + + name1 + )) + self.assertEqual( + name1, + cmd_output["name"], + ) + self.assertEqual( + "DOWN", + cmd_output["admin_state_up"], + ) + self.assertEqual( + admin_project_id, + cmd_output["project_id"], + ) + cmd_output = json.loads(self.openstack( + 'router create -f json ' + + '--project ' + demo_project_id + + ' ' + name2 + )) + self.assertEqual( + name2, + cmd_output["name"], + ) + self.assertEqual( + "UP", + cmd_output["admin_state_up"], + ) + self.assertEqual( + demo_project_id, + cmd_output["project_id"], + ) + + # Test list --project + cmd_output = json.loads(self.openstack( + 'router list -f json ' + + '--project ' + demo_project_id + )) + names = [x["Name"] for x in cmd_output] + self.assertNotIn(name1, names) + self.assertIn(name2, names) + + # Test list --disable + cmd_output = json.loads(self.openstack( + 'router list -f json ' + + '--disable ' + )) + names = [x["Name"] for x in cmd_output] + self.assertIn(name1, names) + self.assertNotIn(name2, names) + + # Test list --name + cmd_output = json.loads(self.openstack( + 'router list -f json ' + + '--name ' + name1 + )) + names = [x["Name"] for x in cmd_output] + self.assertIn(name1, names) + self.assertNotIn(name2, names) + + # Test list --long + cmd_output = json.loads(self.openstack( + 'router list -f json ' + + '--long ' + )) + names = [x["Name"] for x in cmd_output] + self.assertIn(name1, names) + self.assertIn(name2, names) + + del_output = self.openstack( + 'router delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) + + def test_router_set_show_unset(self): + """Tests create router, set, unset, show, delete""" + + name = uuid.uuid4().hex + new_name = name + "_" + cmd_output = json.loads(self.openstack( + 'router create -f json ' + + '--description aaaa ' + + name + )) + self.assertEqual( + name, + cmd_output["name"], + ) + self.assertEqual( + 'aaaa', + cmd_output["description"], + ) + + # Test set --disable + cmd_output = self.openstack( + 'router set ' + + '--name ' + new_name + + ' --description bbbb ' + + '--disable ' + + name + ) + self.assertOutput('', cmd_output) + + cmd_output = json.loads(self.openstack( + 'router show -f json ' + + new_name + )) + self.assertEqual( + new_name, + cmd_output["name"], + ) + self.assertEqual( + 'bbbb', + cmd_output["description"], + ) + self.assertEqual( + 'DOWN', + cmd_output["admin_state_up"], + ) + + # Test set --ha --distributed + cmd_output = self.openstack( + 'router set ' + + '--distributed ' + + '--external-gateway public ' + + new_name + ) + self.assertOutput('', cmd_output) + + cmd_output = json.loads(self.openstack( + 'router show -f json ' + + new_name + )) + self.assertEqual( + True, + cmd_output["distributed"], + ) + self.assertIsNotNone(cmd_output["external_gateway_info"]) + + # Test unset + cmd_output = self.openstack( + 'router unset ' + + '--external-gateway ' + + new_name + ) + cmd_output = json.loads(self.openstack( + 'router show -f json ' + + new_name + )) + self.assertIsNone(cmd_output["external_gateway_info"]) + + del_output = self.openstack( + 'router delete ' + new_name) + self.assertOutput('', del_output) diff --git a/openstackclient/tests/functional/volume/v1/test_volume_type.py b/openstackclient/tests/functional/volume/v1/test_volume_type.py index 955759b6..d1842795 100644 --- a/openstackclient/tests/functional/volume/v1/test_volume_type.py +++ b/openstackclient/tests/functional/volume/v1/test_volume_type.py @@ -87,3 +87,74 @@ class VolumeTypeTests(common.BaseVolumeTests): time.sleep(5) raw_output = self.openstack(cmd) self.assertOutput('', raw_output) + + # NOTE: Add some basic funtional tests with the old format to + # make sure the command works properly, need to change + # these to new test format when beef up all tests for + # volume tye commands. + def test_encryption_type(self): + encryption_type = uuid.uuid4().hex + # test create new encryption type + opts = self.get_opts(['encryption']) + raw_output = self.openstack( + 'volume type create ' + '--encryption-provider LuksEncryptor ' + '--encryption-cipher aes-xts-plain64 ' + '--encryption-key-size 128 ' + '--encryption-control-location front-end ' + + encryption_type + opts) + expected = ["provider='LuksEncryptor'", + "cipher='aes-xts-plain64'", + "key_size='128'", + "control_location='front-end'"] + for attr in expected: + self.assertIn(attr, raw_output) + # test show encryption type + opts = self.get_opts(['encryption']) + raw_output = self.openstack( + 'volume type show --encryption-type ' + encryption_type + opts) + expected = ["provider='LuksEncryptor'", + "cipher='aes-xts-plain64'", + "key_size='128'", + "control_location='front-end'"] + for attr in expected: + self.assertIn(attr, raw_output) + # test list encryption type + opts = self.get_opts(['Encryption']) + raw_output = self.openstack( + 'volume type list --encryption-type ' + opts) + expected = ["provider='LuksEncryptor'", + "cipher='aes-xts-plain64'", + "key_size='128'", + "control_location='front-end'"] + for attr in expected: + self.assertIn(attr, raw_output) + # test set new encryption type + raw_output = self.openstack( + 'volume type set ' + '--encryption-provider LuksEncryptor ' + '--encryption-cipher aes-xts-plain64 ' + '--encryption-key-size 128 ' + '--encryption-control-location front-end ' + + self.NAME) + self.assertEqual('', raw_output) + opts = self.get_opts(['encryption']) + raw_output = self.openstack( + 'volume type show --encryption-type ' + self.NAME + opts) + expected = ["provider='LuksEncryptor'", + "cipher='aes-xts-plain64'", + "key_size='128'", + "control_location='front-end'"] + for attr in expected: + self.assertIn(attr, raw_output) + # test unset encryption type + raw_output = self.openstack( + 'volume type unset --encryption-type ' + self.NAME) + self.assertEqual('', raw_output) + opts = self.get_opts(['encryption']) + raw_output = self.openstack( + 'volume type show --encryption-type ' + self.NAME + opts) + self.assertEqual('\n', raw_output) + # test delete encryption type + raw_output = self.openstack('volume type delete ' + encryption_type) + self.assertEqual('', raw_output) diff --git a/openstackclient/tests/functional/volume/v2/test_volume_type.py b/openstackclient/tests/functional/volume/v2/test_volume_type.py index b4df5b2d..a5d0a767 100644 --- a/openstackclient/tests/functional/volume/v2/test_volume_type.py +++ b/openstackclient/tests/functional/volume/v2/test_volume_type.py @@ -102,3 +102,90 @@ class VolumeTypeTests(common.BaseVolumeTests): time.sleep(5) raw_output = self.openstack(cmd) self.assertOutput('', raw_output) + + # NOTE: Add some basic funtional tests with the old format to + # make sure the command works properly, need to change + # these to new test format when beef up all tests for + # volume tye commands. + def test_encryption_type(self): + encryption_type = uuid.uuid4().hex + # test create new encryption type + opts = self.get_opts(['encryption']) + raw_output = self.openstack( + 'volume type create ' + '--encryption-provider LuksEncryptor ' + '--encryption-cipher aes-xts-plain64 ' + '--encryption-key-size 128 ' + '--encryption-control-location front-end ' + + encryption_type + opts) + expected = ["provider='LuksEncryptor'", + "cipher='aes-xts-plain64'", + "key_size='128'", + "control_location='front-end'"] + for attr in expected: + self.assertIn(attr, raw_output) + # test show encryption type + opts = self.get_opts(['encryption']) + raw_output = self.openstack( + 'volume type show --encryption-type ' + encryption_type + opts) + expected = ["provider='LuksEncryptor'", + "cipher='aes-xts-plain64'", + "key_size='128'", + "control_location='front-end'"] + for attr in expected: + self.assertIn(attr, raw_output) + # test list encryption type + opts = self.get_opts(['Encryption']) + raw_output = self.openstack( + 'volume type list --encryption-type ' + opts) + expected = ["provider='LuksEncryptor'", + "cipher='aes-xts-plain64'", + "key_size='128'", + "control_location='front-end'"] + for attr in expected: + self.assertIn(attr, raw_output) + # test set existing encryption type + raw_output = self.openstack( + 'volume type set ' + '--encryption-key-size 256 ' + '--encryption-control-location back-end ' + + encryption_type) + self.assertEqual('', raw_output) + opts = self.get_opts(['encryption']) + raw_output = self.openstack( + 'volume type show --encryption-type ' + encryption_type + opts) + expected = ["provider='LuksEncryptor'", + "cipher='aes-xts-plain64'", + "key_size='256'", + "control_location='back-end'"] + for attr in expected: + self.assertIn(attr, raw_output) + # test set new encryption type + raw_output = self.openstack( + 'volume type set ' + '--encryption-provider LuksEncryptor ' + '--encryption-cipher aes-xts-plain64 ' + '--encryption-key-size 128 ' + '--encryption-control-location front-end ' + + self.NAME) + self.assertEqual('', raw_output) + opts = self.get_opts(['encryption']) + raw_output = self.openstack( + 'volume type show --encryption-type ' + self.NAME + opts) + expected = ["provider='LuksEncryptor'", + "cipher='aes-xts-plain64'", + "key_size='128'", + "control_location='front-end'"] + for attr in expected: + self.assertIn(attr, raw_output) + # test unset encryption type + raw_output = self.openstack( + 'volume type unset --encryption-type ' + self.NAME) + self.assertEqual('', raw_output) + opts = self.get_opts(['encryption']) + raw_output = self.openstack( + 'volume type show --encryption-type ' + self.NAME + opts) + self.assertEqual('\n', raw_output) + # test delete encryption type + raw_output = self.openstack('volume type delete ' + encryption_type) + self.assertEqual('', raw_output) diff --git a/openstackclient/tests/unit/compute/v2/test_server.py b/openstackclient/tests/unit/compute/v2/test_server.py index 4cac990e..249902bc 100644 --- a/openstackclient/tests/unit/compute/v2/test_server.py +++ b/openstackclient/tests/unit/compute/v2/test_server.py @@ -467,6 +467,152 @@ class TestServerCreate(TestServer): self.assertEqual(self.columns, columns) self.assertEqual(self.datalist(), data) + def test_server_create_with_auto_network(self): + arglist = [ + '--image', 'image1', + '--flavor', 'flavor1', + '--nic', 'auto', + self.new_server.name, + ] + verifylist = [ + ('image', 'image1'), + ('flavor', 'flavor1'), + ('nic', ['auto']), + ('config_drive', False), + ('server_name', self.new_server.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = dict( + meta=None, + files={}, + reservation_id=None, + min_count=1, + max_count=1, + security_groups=[], + userdata=None, + key_name=None, + availability_zone=None, + block_device_mapping_v2=[], + nics='auto', + scheduler_hints={}, + config_drive=None, + ) + # ServerManager.create(name, image, flavor, **kwargs) + self.servers_mock.create.assert_called_with( + self.new_server.name, + self.image, + self.flavor, + **kwargs + ) + + self.assertEqual(self.columns, columns) + self.assertEqual(self.datalist(), data) + + def test_server_create_with_none_network(self): + arglist = [ + '--image', 'image1', + '--flavor', 'flavor1', + '--nic', 'none', + self.new_server.name, + ] + verifylist = [ + ('image', 'image1'), + ('flavor', 'flavor1'), + ('nic', ['none']), + ('config_drive', False), + ('server_name', self.new_server.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = dict( + meta=None, + files={}, + reservation_id=None, + min_count=1, + max_count=1, + security_groups=[], + userdata=None, + key_name=None, + availability_zone=None, + block_device_mapping_v2=[], + nics='none', + scheduler_hints={}, + config_drive=None, + ) + # ServerManager.create(name, image, flavor, **kwargs) + self.servers_mock.create.assert_called_with( + self.new_server.name, + self.image, + self.flavor, + **kwargs + ) + + self.assertEqual(self.columns, columns) + self.assertEqual(self.datalist(), data) + + def test_server_create_with_conflict_network_options(self): + arglist = [ + '--image', 'image1', + '--flavor', 'flavor1', + '--nic', 'none', + '--nic', 'auto', + '--nic', 'port-id=port1', + self.new_server.name, + ] + verifylist = [ + ('image', 'image1'), + ('flavor', 'flavor1'), + ('nic', ['none', 'auto', 'port-id=port1']), + ('config_drive', False), + ('server_name', self.new_server.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + get_endpoints = mock.Mock() + get_endpoints.return_value = {'network': []} + self.app.client_manager.auth_ref = mock.Mock() + self.app.client_manager.auth_ref.service_catalog = mock.Mock() + self.app.client_manager.auth_ref.service_catalog.get_endpoints = ( + get_endpoints) + + find_port = mock.Mock() + network_client = self.app.client_manager.network + network_client.find_port = find_port + port_resource = mock.Mock() + port_resource.id = 'port1_uuid' + find_port.return_value = port_resource + + self.assertRaises(exceptions.CommandError, + self.cmd.take_action, parsed_args) + self.assertNotCalled(self.servers_mock.create) + + def test_server_create_with_invalid_network_options(self): + arglist = [ + '--image', 'image1', + '--flavor', 'flavor1', + '--nic', 'abcdefgh', + self.new_server.name, + ] + verifylist = [ + ('image', 'image1'), + ('flavor', 'flavor1'), + ('nic', ['abcdefgh']), + ('config_drive', False), + ('server_name', self.new_server.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + self.assertRaises(exceptions.CommandError, + self.cmd.take_action, parsed_args) + self.assertNotCalled(self.servers_mock.create) + @mock.patch.object(common_utils, 'wait_for_status', return_value=True) def test_server_create_with_wait_ok(self, mock_wait_for_status): arglist = [ diff --git a/openstackclient/tests/unit/volume/v1/fakes.py b/openstackclient/tests/unit/volume/v1/fakes.py index 78a8227e..fff5181d 100644 --- a/openstackclient/tests/unit/volume/v1/fakes.py +++ b/openstackclient/tests/unit/volume/v1/fakes.py @@ -364,6 +364,9 @@ class FakeVolumev1Client(object): self.qos_specs.resource_class = fakes.FakeResource(None, {}) self.volume_types = mock.Mock() self.volume_types.resource_class = fakes.FakeResource(None, {}) + self.volume_encryption_types = mock.Mock() + self.volume_encryption_types.resource_class = ( + fakes.FakeResource(None, {})) self.transfers = mock.Mock() self.transfers.resource_class = fakes.FakeResource(None, {}) self.volume_snapshots = mock.Mock() @@ -470,6 +473,34 @@ class FakeType(object): return mock.Mock(side_effect=types) + @staticmethod + def create_one_encryption_type(attrs=None): + """Create a fake encryption type. + + :param Dictionary attrs: + A dictionary with all attributes + :return: + A FakeResource object with volume_type_id etc. + """ + attrs = attrs or {} + + # Set default attributes. + encryption_info = { + "volume_type_id": 'type-id-' + uuid.uuid4().hex, + 'provider': 'LuksEncryptor', + 'cipher': None, + 'key_size': None, + 'control_location': 'front-end', + } + + # Overwrite default attributes. + encryption_info.update(attrs) + + encryption_type = fakes.FakeResource( + info=copy.deepcopy(encryption_info), + loaded=True) + return encryption_type + class FakeSnapshot(object): """Fake one or more snapshot.""" diff --git a/openstackclient/tests/unit/volume/v1/test_type.py b/openstackclient/tests/unit/volume/v1/test_type.py index 81ad8301..dcdd3d56 100644 --- a/openstackclient/tests/unit/volume/v1/test_type.py +++ b/openstackclient/tests/unit/volume/v1/test_type.py @@ -31,6 +31,10 @@ class TestType(volume_fakes.TestVolumev1): self.types_mock = self.app.client_manager.volume.volume_types self.types_mock.reset_mock() + self.encryption_types_mock = ( + self.app.client_manager.volume.volume_encryption_types) + self.encryption_types_mock.reset_mock() + class TestTypeCreate(TestType): @@ -75,6 +79,67 @@ class TestTypeCreate(TestType): self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) + def test_type_create_with_encryption(self): + encryption_info = { + 'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': '128', + 'control_location': 'front-end', + } + encryption_type = volume_fakes.FakeType.create_one_encryption_type( + attrs=encryption_info + ) + self.new_volume_type = volume_fakes.FakeType.create_one_type( + attrs={'encryption': encryption_info}) + self.types_mock.create.return_value = self.new_volume_type + self.encryption_types_mock.create.return_value = encryption_type + encryption_columns = ( + 'description', + 'encryption', + 'id', + 'is_public', + 'name', + ) + encryption_data = ( + self.new_volume_type.description, + utils.format_dict(encryption_info), + self.new_volume_type.id, + True, + self.new_volume_type.name, + ) + arglist = [ + '--encryption-provider', 'LuksEncryptor', + '--encryption-cipher', 'aes-xts-plain64', + '--encryption-key-size', '128', + '--encryption-control-location', 'front-end', + self.new_volume_type.name, + ] + verifylist = [ + ('encryption_provider', 'LuksEncryptor'), + ('encryption_cipher', 'aes-xts-plain64'), + ('encryption_key_size', 128), + ('encryption_control_location', 'front-end'), + ('name', self.new_volume_type.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + self.types_mock.create.assert_called_with( + self.new_volume_type.name, + ) + body = { + 'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': 128, + 'control_location': 'front-end', + } + self.encryption_types_mock.create.assert_called_with( + self.new_volume_type, + body, + ) + self.assertEqual(encryption_columns, columns) + self.assertEqual(encryption_data, data) + class TestTypeDelete(TestType): @@ -156,17 +221,17 @@ class TestTypeList(TestType): volume_types = volume_fakes.FakeType.create_types() - columns = ( + columns = [ "ID", "Name", "Is Public", - ) - columns_long = ( + ] + columns_long = [ "ID", "Name", "Is Public", "Properties" - ) + ] data = [] for t in volume_types: @@ -188,6 +253,8 @@ class TestTypeList(TestType): super(TestTypeList, self).setUp() self.types_mock.list.return_value = self.volume_types + self.encryption_types_mock.create.return_value = None + self.encryption_types_mock.update.return_value = None # get the command to test self.cmd = volume_type.ListVolumeType(self.app, None) @@ -195,6 +262,7 @@ class TestTypeList(TestType): arglist = [] verifylist = [ ("long", False), + ("encryption_type", False), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) @@ -217,6 +285,47 @@ class TestTypeList(TestType): self.assertEqual(self.columns_long, columns) self.assertEqual(self.data_long, list(data)) + def test_type_list_with_encryption(self): + encryption_type = volume_fakes.FakeType.create_one_encryption_type( + attrs={'volume_type_id': self.volume_types[0].id}) + encryption_info = { + 'provider': 'LuksEncryptor', + 'cipher': None, + 'key_size': None, + 'control_location': 'front-end', + } + encryption_columns = self.columns + [ + "Encryption", + ] + encryption_data = [] + encryption_data.append(( + self.volume_types[0].id, + self.volume_types[0].name, + self.volume_types[0].is_public, + utils.format_dict(encryption_info), + )) + encryption_data.append(( + self.volume_types[1].id, + self.volume_types[1].name, + self.volume_types[1].is_public, + '-', + )) + + self.encryption_types_mock.list.return_value = [encryption_type] + arglist = [ + "--encryption-type", + ] + verifylist = [ + ("encryption_type", True), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + self.encryption_types_mock.list.assert_called_once_with() + self.types_mock.list.assert_called_once_with() + self.assertEqual(encryption_columns, columns) + self.assertEqual(encryption_data, list(data)) + class TestTypeSet(TestType): @@ -260,6 +369,60 @@ class TestTypeSet(TestType): {'myprop': 'myvalue'}) self.assertIsNone(result) + def test_type_set_new_encryption(self): + arglist = [ + '--encryption-provider', 'LuksEncryptor', + '--encryption-cipher', 'aes-xts-plain64', + '--encryption-key-size', '128', + '--encryption-control-location', 'front-end', + self.volume_type.id, + ] + verifylist = [ + ('encryption_provider', 'LuksEncryptor'), + ('encryption_cipher', 'aes-xts-plain64'), + ('encryption_key_size', 128), + ('encryption_control_location', 'front-end'), + ('volume_type', self.volume_type.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + body = { + 'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': 128, + 'control_location': 'front-end', + } + self.encryption_types_mock.create.assert_called_with( + self.volume_type, + body, + ) + self.assertIsNone(result) + + def test_type_set_new_encryption_without_provider(self): + arglist = [ + '--encryption-cipher', 'aes-xts-plain64', + '--encryption-key-size', '128', + '--encryption-control-location', 'front-end', + self.volume_type.id, + ] + verifylist = [ + ('encryption_cipher', 'aes-xts-plain64'), + ('encryption_key_size', 128), + ('encryption_control_location', 'front-end'), + ('volume_type', self.volume_type.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual("Command Failed: One or more of" + " the operations failed", + str(e)) + self.encryption_types_mock.create.assert_not_called() + self.encryption_types_mock.update.assert_not_called() + class TestTypeShow(TestType): @@ -293,7 +456,8 @@ class TestTypeShow(TestType): self.volume_type.id ] verifylist = [ - ("volume_type", self.volume_type.id) + ("volume_type", self.volume_type.id), + ("encryption_type", False), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) @@ -303,6 +467,50 @@ class TestTypeShow(TestType): self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) + def test_type_show_with_encryption(self): + encryption_type = volume_fakes.FakeType.create_one_encryption_type() + encryption_info = { + 'provider': 'LuksEncryptor', + 'cipher': None, + 'key_size': None, + 'control_location': 'front-end', + } + self.volume_type = volume_fakes.FakeType.create_one_type( + attrs={'encryption': encryption_info}) + self.types_mock.get.return_value = self.volume_type + self.encryption_types_mock.get.return_value = encryption_type + encryption_columns = ( + 'description', + 'encryption', + 'id', + 'is_public', + 'name', + 'properties', + ) + encryption_data = ( + self.volume_type.description, + utils.format_dict(encryption_info), + self.volume_type.id, + True, + self.volume_type.name, + utils.format_dict(self.volume_type.extra_specs) + ) + arglist = [ + '--encryption-type', + self.volume_type.id + ] + verifylist = [ + ('encryption_type', True), + ("volume_type", self.volume_type.id) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + self.types_mock.get.assert_called_with(self.volume_type.id) + self.encryption_types_mock.get.assert_called_with(self.volume_type.id) + self.assertEqual(encryption_columns, columns) + self.assertEqual(encryption_data, data) + class TestTypeUnset(TestType): @@ -317,13 +525,14 @@ class TestTypeUnset(TestType): # Get the command object to test self.cmd = volume_type.UnsetVolumeType(self.app, None) - def test_type_unset(self): + def test_type_unset_property(self): arglist = [ '--property', 'property', '--property', 'multi_property', self.volume_type.id, ] verifylist = [ + ('encryption_type', False), ('property', ['property', 'multi_property']), ('volume_type', self.volume_type.id), ] @@ -333,6 +542,7 @@ class TestTypeUnset(TestType): result = self.cmd.take_action(parsed_args) self.volume_type.unset_keys.assert_called_once_with( ['property', 'multi_property']) + self.encryption_types_mock.delete.assert_not_called() self.assertIsNone(result) def test_type_unset_failed_with_missing_volume_type_argument(self): @@ -362,3 +572,18 @@ class TestTypeUnset(TestType): result = self.cmd.take_action(parsed_args) self.assertIsNone(result) + + def test_type_unset_encryption_type(self): + arglist = [ + '--encryption-type', + self.volume_type.id, + ] + verifylist = [ + ('encryption_type', True), + ('volume_type', self.volume_type.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.encryption_types_mock.delete.assert_called_with(self.volume_type) + self.assertIsNone(result) diff --git a/openstackclient/tests/unit/volume/v2/fakes.py b/openstackclient/tests/unit/volume/v2/fakes.py index a6676403..d54faec7 100644 --- a/openstackclient/tests/unit/volume/v2/fakes.py +++ b/openstackclient/tests/unit/volume/v2/fakes.py @@ -208,6 +208,9 @@ class FakeVolumeClient(object): self.volume_types.resource_class = fakes.FakeResource(None, {}) self.volume_type_access = mock.Mock() self.volume_type_access.resource_class = fakes.FakeResource(None, {}) + self.volume_encryption_types = mock.Mock() + self.volume_encryption_types.resource_class = ( + fakes.FakeResource(None, {})) self.restores = mock.Mock() self.restores.resource_class = fakes.FakeResource(None, {}) self.qos_specs = mock.Mock() @@ -923,3 +926,31 @@ class FakeType(object): types = FakeType.create_types(count) return mock.Mock(side_effect=types) + + @staticmethod + def create_one_encryption_type(attrs=None): + """Create a fake encryption type. + + :param Dictionary attrs: + A dictionary with all attributes + :return: + A FakeResource object with volume_type_id etc. + """ + attrs = attrs or {} + + # Set default attributes. + encryption_info = { + "volume_type_id": 'type-id-' + uuid.uuid4().hex, + 'provider': 'LuksEncryptor', + 'cipher': None, + 'key_size': None, + 'control_location': 'front-end', + } + + # Overwrite default attributes. + encryption_info.update(attrs) + + encryption_type = fakes.FakeResource( + info=copy.deepcopy(encryption_info), + loaded=True) + return encryption_type diff --git a/openstackclient/tests/unit/volume/v2/test_type.py b/openstackclient/tests/unit/volume/v2/test_type.py index cec01bd8..4023d55b 100644 --- a/openstackclient/tests/unit/volume/v2/test_type.py +++ b/openstackclient/tests/unit/volume/v2/test_type.py @@ -36,6 +36,10 @@ class TestType(volume_fakes.TestVolume): self.app.client_manager.volume.volume_type_access) self.types_access_mock.reset_mock() + self.encryption_types_mock = ( + self.app.client_manager.volume.volume_encryption_types) + self.encryption_types_mock.reset_mock() + self.projects_mock = self.app.client_manager.identity.projects self.projects_mock.reset_mock() @@ -131,6 +135,68 @@ class TestTypeCreate(TestType): self.cmd.take_action, parsed_args) + def test_type_create_with_encryption(self): + encryption_info = { + 'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': '128', + 'control_location': 'front-end', + } + encryption_type = volume_fakes.FakeType.create_one_encryption_type( + attrs=encryption_info + ) + self.new_volume_type = volume_fakes.FakeType.create_one_type( + attrs={'encryption': encryption_info}) + self.types_mock.create.return_value = self.new_volume_type + self.encryption_types_mock.create.return_value = encryption_type + encryption_columns = ( + 'description', + 'encryption', + 'id', + 'is_public', + 'name', + ) + encryption_data = ( + self.new_volume_type.description, + utils.format_dict(encryption_info), + self.new_volume_type.id, + True, + self.new_volume_type.name, + ) + arglist = [ + '--encryption-provider', 'LuksEncryptor', + '--encryption-cipher', 'aes-xts-plain64', + '--encryption-key-size', '128', + '--encryption-control-location', 'front-end', + self.new_volume_type.name, + ] + verifylist = [ + ('encryption_provider', 'LuksEncryptor'), + ('encryption_cipher', 'aes-xts-plain64'), + ('encryption_key_size', 128), + ('encryption_control_location', 'front-end'), + ('name', self.new_volume_type.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + self.types_mock.create.assert_called_with( + self.new_volume_type.name, + description=None, + ) + body = { + 'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': 128, + 'control_location': 'front-end', + } + self.encryption_types_mock.create.assert_called_with( + self.new_volume_type, + body, + ) + self.assertEqual(encryption_columns, columns) + self.assertEqual(encryption_data, data) + class TestTypeDelete(TestType): @@ -305,6 +371,7 @@ class TestTypeList(TestType): "--default", ] verifylist = [ + ("encryption_type", False), ("long", False), ("private", False), ("public", False), @@ -317,6 +384,47 @@ class TestTypeList(TestType): self.assertEqual(self.columns, columns) self.assertEqual(self.data_with_default_type, list(data)) + def test_type_list_with_encryption(self): + encryption_type = volume_fakes.FakeType.create_one_encryption_type( + attrs={'volume_type_id': self.volume_types[0].id}) + encryption_info = { + 'provider': 'LuksEncryptor', + 'cipher': None, + 'key_size': None, + 'control_location': 'front-end', + } + encryption_columns = self.columns + [ + "Encryption", + ] + encryption_data = [] + encryption_data.append(( + self.volume_types[0].id, + self.volume_types[0].name, + self.volume_types[0].is_public, + utils.format_dict(encryption_info), + )) + encryption_data.append(( + self.volume_types[1].id, + self.volume_types[1].name, + self.volume_types[1].is_public, + '-', + )) + + self.encryption_types_mock.list.return_value = [encryption_type] + arglist = [ + "--encryption-type", + ] + verifylist = [ + ("encryption_type", True), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + self.encryption_types_mock.list.assert_called_once_with() + self.types_mock.list.assert_called_once_with(is_public=None) + self.assertEqual(encryption_columns, columns) + self.assertEqual(encryption_data, list(data)) + class TestTypeSet(TestType): @@ -331,6 +439,8 @@ class TestTypeSet(TestType): # Return a project self.projects_mock.get.return_value = self.project + self.encryption_types_mock.create.return_value = None + self.encryption_types_mock.update.return_value = None # Get the command object to test self.cmd = volume_type.SetVolumeType(self.app, None) @@ -454,6 +564,107 @@ class TestTypeSet(TestType): self.project.id, ) + def test_type_set_new_encryption(self): + self.encryption_types_mock.update.side_effect = ( + exceptions.NotFound('NotFound')) + arglist = [ + '--encryption-provider', 'LuksEncryptor', + '--encryption-cipher', 'aes-xts-plain64', + '--encryption-key-size', '128', + '--encryption-control-location', 'front-end', + self.volume_type.id, + ] + verifylist = [ + ('encryption_provider', 'LuksEncryptor'), + ('encryption_cipher', 'aes-xts-plain64'), + ('encryption_key_size', 128), + ('encryption_control_location', 'front-end'), + ('volume_type', self.volume_type.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + body = { + 'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': 128, + 'control_location': 'front-end', + } + self.encryption_types_mock.update.assert_called_with( + self.volume_type, + body, + ) + self.encryption_types_mock.create.assert_called_with( + self.volume_type, + body, + ) + self.assertIsNone(result) + + @mock.patch.object(utils, 'find_resource') + def test_type_set_existing_encryption(self, mock_find): + mock_find.side_effect = [self.volume_type, + "existing_encryption_type"] + arglist = [ + '--encryption-provider', 'LuksEncryptor', + '--encryption-cipher', 'aes-xts-plain64', + '--encryption-control-location', 'front-end', + self.volume_type.id, + ] + verifylist = [ + ('encryption_provider', 'LuksEncryptor'), + ('encryption_cipher', 'aes-xts-plain64'), + ('encryption_control_location', 'front-end'), + ('volume_type', self.volume_type.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + body = { + 'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'control_location': 'front-end', + } + self.encryption_types_mock.update.assert_called_with( + self.volume_type, + body, + ) + self.encryption_types_mock.create.assert_not_called() + self.assertIsNone(result) + + def test_type_set_new_encryption_without_provider(self): + self.encryption_types_mock.update.side_effect = ( + exceptions.NotFound('NotFound')) + arglist = [ + '--encryption-cipher', 'aes-xts-plain64', + '--encryption-key-size', '128', + '--encryption-control-location', 'front-end', + self.volume_type.id, + ] + verifylist = [ + ('encryption_cipher', 'aes-xts-plain64'), + ('encryption_key_size', 128), + ('encryption_control_location', 'front-end'), + ('volume_type', self.volume_type.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual("Command Failed: One or more of" + " the operations failed", + str(e)) + body = { + 'cipher': 'aes-xts-plain64', + 'key_size': 128, + 'control_location': 'front-end', + } + self.encryption_types_mock.update.assert_called_with( + self.volume_type, + body, + ) + self.encryption_types_mock.create.assert_not_called() + class TestTypeShow(TestType): @@ -489,6 +700,7 @@ class TestTypeShow(TestType): self.volume_type.id ] verifylist = [ + ("encryption_type", False), ("volume_type", self.volume_type.id) ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) @@ -564,6 +776,52 @@ class TestTypeShow(TestType): ) self.assertEqual(private_type_data, data) + def test_type_show_with_encryption(self): + encryption_type = volume_fakes.FakeType.create_one_encryption_type() + encryption_info = { + 'provider': 'LuksEncryptor', + 'cipher': None, + 'key_size': None, + 'control_location': 'front-end', + } + self.volume_type = volume_fakes.FakeType.create_one_type( + attrs={'encryption': encryption_info}) + self.types_mock.get.return_value = self.volume_type + self.encryption_types_mock.get.return_value = encryption_type + encryption_columns = ( + 'access_project_ids', + 'description', + 'encryption', + 'id', + 'is_public', + 'name', + 'properties', + ) + encryption_data = ( + None, + self.volume_type.description, + utils.format_dict(encryption_info), + self.volume_type.id, + True, + self.volume_type.name, + utils.format_dict(self.volume_type.extra_specs) + ) + arglist = [ + '--encryption-type', + self.volume_type.id + ] + verifylist = [ + ('encryption_type', True), + ("volume_type", self.volume_type.id) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + self.types_mock.get.assert_called_with(self.volume_type.id) + self.encryption_types_mock.get.assert_called_with(self.volume_type.id) + self.assertEqual(encryption_columns, columns) + self.assertEqual(encryption_data, data) + class TestTypeUnset(TestType): @@ -625,6 +883,7 @@ class TestTypeUnset(TestType): self.volume_type.id, ] verifylist = [ + ('encryption_type', False), ('project', ''), ('volume_type', self.volume_type.id), ] @@ -633,7 +892,7 @@ class TestTypeUnset(TestType): result = self.cmd.take_action(parsed_args) self.assertIsNone(result) - + self.encryption_types_mock.delete.assert_not_called() self.assertFalse(self.types_access_mock.remove_project_access.called) def test_type_unset_failed_with_missing_volume_type_argument(self): @@ -649,3 +908,18 @@ class TestTypeUnset(TestType): self.cmd, arglist, verifylist) + + def test_type_unset_encryption_type(self): + arglist = [ + '--encryption-type', + self.volume_type.id, + ] + verifylist = [ + ('encryption_type', True), + ('volume_type', self.volume_type.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.encryption_types_mock.delete.assert_called_with(self.volume_type) + self.assertIsNone(result) diff --git a/openstackclient/tests/unit/volume/v2/test_volume_host.py b/openstackclient/tests/unit/volume/v2/test_volume_host.py index aad7bb0b..b024329a 100644 --- a/openstackclient/tests/unit/volume/v2/test_volume_host.py +++ b/openstackclient/tests/unit/volume/v2/test_volume_host.py @@ -35,6 +35,7 @@ class TestVolumeHostSet(TestVolumeHost): self.host_mock.freeze_host.return_value = None self.host_mock.thaw_host.return_value = None + # Get the command object to mock self.cmd = volume_host.SetVolumeHost(self.app, None) def test_volume_host_set_nothing(self): @@ -84,3 +85,33 @@ class TestVolumeHostSet(TestVolumeHost): self.host_mock.freeze_host.assert_called_with(self.service.host) self.host_mock.thaw_host.assert_not_called() self.assertIsNone(result) + + +class TestVolumeHostFailover(TestVolumeHost): + + service = host_fakes.FakeService.create_one_service() + + def setUp(self): + super(TestVolumeHostFailover, self).setUp() + + self.host_mock.failover_host.return_value = None + + # Get the command object to mock + self.cmd = volume_host.FailoverVolumeHost(self.app, None) + + def test_volume_host_failover(self): + arglist = [ + '--volume-backend', 'backend_test', + self.service.host, + ] + verifylist = [ + ('volume_backend', 'backend_test'), + ('host', self.service.host), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + self.host_mock.failover_host.assert_called_with( + self.service.host, 'backend_test') + self.assertIsNone(result) |