# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import contextlib import datetime import fixtures from unittest import mock import uuid import warnings from openstack.block_storage.v3 import volume from openstack.compute.v2 import _proxy from openstack.compute.v2 import aggregate from openstack.compute.v2 import availability_zone as az from openstack.compute.v2 import console_auth_token from openstack.compute.v2 import extension from openstack.compute.v2 import flavor from openstack.compute.v2 import hypervisor from openstack.compute.v2 import image from openstack.compute.v2 import keypair from openstack.compute.v2 import migration from openstack.compute.v2 import quota_class_set from openstack.compute.v2 import quota_set from openstack.compute.v2 import server from openstack.compute.v2 import server_action from openstack.compute.v2 import server_group from openstack.compute.v2 import server_interface from openstack.compute.v2 import server_ip from openstack.compute.v2 import server_migration from openstack.compute.v2 import server_remote_console from openstack.compute.v2 import service from openstack.compute.v2 import usage from openstack.compute.v2 import volume_attachment from openstack.identity.v3 import project from openstack import proxy as proxy_base from openstack.tests.unit import base from openstack.tests.unit import test_proxy_base from openstack import types from openstack import warnings as os_warnings class TestComputeProxy(test_proxy_base.TestProxyBase): def setUp(self): super().setUp() self.proxy = _proxy.Proxy(self.session) class TestFlavor(TestComputeProxy): def test_flavor_create(self): self.verify_create(self.proxy.create_flavor, flavor.Flavor) def test_flavor_delete(self): self.verify_delete(self.proxy.delete_flavor, flavor.Flavor, False) def test_flavor_update(self): self.verify_update(self.proxy.update_flavor, flavor.Flavor, False) def test_flavor_delete_ignore(self): self.verify_delete(self.proxy.delete_flavor, flavor.Flavor, True) def test_flavor_find(self): self.verify_find(self.proxy.find_flavor, flavor.Flavor) def test_flavor_find_query(self): self.verify_find( self.proxy.find_flavor, flavor.Flavor, method_kwargs={"a": "b"}, expected_kwargs={"a": "b", "ignore_missing": True}, ) def test_flavor_find_fetch_extra(self): """fetch extra_specs is triggered""" with mock.patch( 'openstack.compute.v2.flavor.Flavor.fetch_extra_specs' ) as mocked: res = flavor.Flavor() mocked.return_value = res self._verify( 'openstack.proxy.Proxy._find', self.proxy.find_flavor, method_args=['res', True], method_kwargs={'get_extra_specs': True}, expected_result=res, expected_args=[flavor.Flavor, 'res'], expected_kwargs={'ignore_missing': True}, ) mocked.assert_called_once() def test_flavor_find_skip_fetch_extra(self): """fetch extra_specs not triggered""" with mock.patch( 'openstack.compute.v2.flavor.Flavor.fetch_extra_specs' ) as mocked: res = flavor.Flavor(extra_specs={'a': 'b'}) mocked.return_value = res self._verify( 'openstack.proxy.Proxy._find', self.proxy.find_flavor, method_args=['res', True], expected_result=res, expected_args=[flavor.Flavor, 'res'], expected_kwargs={'ignore_missing': True}, ) mocked.assert_not_called() def test_flavor_get_no_extra(self): """fetch extra_specs not triggered""" with mock.patch( 'openstack.compute.v2.flavor.Flavor.fetch_extra_specs' ) as mocked: res = flavor.Flavor() mocked.return_value = res self._verify( 'openstack.proxy.Proxy._get', self.proxy.get_flavor, method_args=['res'], expected_result=res, expected_args=[flavor.Flavor, 'res'], ) mocked.assert_not_called() def test_flavor_get_fetch_extra(self): """fetch extra_specs is triggered""" with mock.patch( 'openstack.compute.v2.flavor.Flavor.fetch_extra_specs' ) as mocked: res = flavor.Flavor() mocked.return_value = res self._verify( 'openstack.proxy.Proxy._get', self.proxy.get_flavor, method_args=['res', True], expected_result=res, expected_args=[flavor.Flavor, 'res'], ) mocked.assert_called_once() def test_flavor_get_skip_fetch_extra(self): """fetch extra_specs not triggered""" with mock.patch( 'openstack.compute.v2.flavor.Flavor.fetch_extra_specs' ) as mocked: res = flavor.Flavor(extra_specs={'a': 'b'}) mocked.return_value = res self._verify( 'openstack.proxy.Proxy._get', self.proxy.get_flavor, method_args=['res', True], expected_result=res, expected_args=[flavor.Flavor, 'res'], ) mocked.assert_not_called() @mock.patch("openstack.proxy.Proxy._list") @mock.patch("openstack.compute.v2.flavor.Flavor.fetch_extra_specs") def test_flavors_detailed(self, fetch_mock, list_mock): res = self.proxy.flavors(details=True) for r in res: self.assertIsNotNone(r) fetch_mock.assert_not_called() list_mock.assert_called_with( flavor.Flavor, base_path="/flavors/detail" ) @mock.patch("openstack.proxy.Proxy._list") @mock.patch("openstack.compute.v2.flavor.Flavor.fetch_extra_specs") def test_flavors_not_detailed(self, fetch_mock, list_mock): res = self.proxy.flavors(details=False) for r in res: self.assertIsNotNone(r) fetch_mock.assert_not_called() list_mock.assert_called_with(flavor.Flavor, base_path="/flavors") @mock.patch("openstack.proxy.Proxy._list") @mock.patch("openstack.compute.v2.flavor.Flavor.fetch_extra_specs") def test_flavors_query(self, fetch_mock, list_mock): res = self.proxy.flavors(details=False, get_extra_specs=True, a="b") for r in res: fetch_mock.assert_called_with(self.proxy) list_mock.assert_called_with( flavor.Flavor, base_path="/flavors", a="b" ) @mock.patch("openstack.proxy.Proxy._list") @mock.patch("openstack.compute.v2.flavor.Flavor.fetch_extra_specs") def test_flavors_get_extra(self, fetch_mock, list_mock): res = self.proxy.flavors(details=False, get_extra_specs=True) for r in res: fetch_mock.assert_called_with(self.proxy) list_mock.assert_called_with(flavor.Flavor, base_path="/flavors") def test_flavor_get_access(self): self._verify( "openstack.compute.v2.flavor.Flavor.get_access", self.proxy.get_flavor_access, method_args=["value"], expected_args=[self.proxy], ) def test_flavor_add_tenant_access(self): self._verify( "openstack.compute.v2.flavor.Flavor.add_tenant_access", self.proxy.flavor_add_tenant_access, method_args=["value", "fake-tenant"], expected_args=[self.proxy, "fake-tenant"], ) def test_flavor_remove_tenant_access(self): self._verify( "openstack.compute.v2.flavor.Flavor.remove_tenant_access", self.proxy.flavor_remove_tenant_access, method_args=["value", "fake-tenant"], expected_args=[self.proxy, "fake-tenant"], ) def test_flavor_fetch_extra_specs(self): self._verify( "openstack.compute.v2.flavor.Flavor.fetch_extra_specs", self.proxy.fetch_flavor_extra_specs, method_args=["value"], expected_args=[self.proxy], ) def test_create_flavor_extra_specs(self): self._verify( "openstack.compute.v2.flavor.Flavor.create_extra_specs", self.proxy.create_flavor_extra_specs, method_args=["value", {'a': 'b'}], expected_args=[self.proxy], expected_kwargs={"specs": {'a': 'b'}}, ) def test_get_flavor_extra_specs_prop(self): self._verify( "openstack.compute.v2.flavor.Flavor.get_extra_specs_property", self.proxy.get_flavor_extra_specs_property, method_args=["value", "prop"], expected_args=[self.proxy, "prop"], ) def test_update_flavor_extra_specs_prop(self): self._verify( "openstack.compute.v2.flavor.Flavor.update_extra_specs_property", self.proxy.update_flavor_extra_specs_property, method_args=["value", "prop", "val"], expected_args=[self.proxy, "prop", "val"], ) def test_delete_flavor_extra_specs_prop(self): self._verify( "openstack.compute.v2.flavor.Flavor.delete_extra_specs_property", self.proxy.delete_flavor_extra_specs_property, method_args=["value", "prop"], expected_args=[self.proxy, "prop"], ) class TestKeyPair(TestComputeProxy): def test_keypair_create(self): self.verify_create(self.proxy.create_keypair, keypair.Keypair) def test_keypair_delete(self): self._verify( "openstack.compute.v2.keypair.Keypair.delete", self.proxy.delete_keypair, method_args=["value"], expected_args=[self.proxy], expected_kwargs={"params": {}}, ) def test_keypair_delete_ignore(self): self._verify( "openstack.compute.v2.keypair.Keypair.delete", self.proxy.delete_keypair, method_args=["value", True], method_kwargs={"user_id": "fake_user"}, expected_args=[self.proxy], expected_kwargs={"params": {"user_id": "fake_user"}}, ) def test_keypair_delete_user_id(self): self._verify( "openstack.compute.v2.keypair.Keypair.delete", self.proxy.delete_keypair, method_args=["value"], method_kwargs={"user_id": "fake_user"}, expected_args=[self.proxy], expected_kwargs={"params": {"user_id": "fake_user"}}, ) def test_keypair_find(self): self.verify_find(self.proxy.find_keypair, keypair.Keypair) def test_keypair_find_user_id(self): self.verify_find( self.proxy.find_keypair, keypair.Keypair, method_kwargs={'user_id': 'fake_user'}, expected_kwargs={'user_id': 'fake_user'}, ) def test_keypair_get(self): self._verify( "openstack.compute.v2.keypair.Keypair.fetch", self.proxy.get_keypair, method_args=["value"], method_kwargs={}, expected_args=[self.proxy], expected_kwargs={ "error_message": "No Keypair found for value", }, ) def test_keypair_get_user_id(self): self._verify( "openstack.compute.v2.keypair.Keypair.fetch", self.proxy.get_keypair, method_args=["value"], method_kwargs={"user_id": "fake_user"}, expected_args=[self.proxy], expected_kwargs={ "error_message": "No Keypair found for value", "user_id": "fake_user", }, ) def test_keypairs(self): self.verify_list(self.proxy.keypairs, keypair.Keypair) def test_keypairs_user_id(self): self.verify_list( self.proxy.keypairs, keypair.Keypair, method_kwargs={'user_id': 'fake_user'}, expected_kwargs={'user_id': 'fake_user'}, ) class TestKeyPairUrl(base.TestCase): def setUp(self): super().setUp() self.useFixture( fixtures.MonkeyPatch( "openstack.utils.maximum_supported_microversion", lambda *args, **kwargs: "2.10", ) ) def test_keypair_find_user_id(self): self.register_uris( [ dict( method="GET", uri=self.get_mock_url( "compute", "public", append=["os-keypairs", "fake_keypair"], qs_elements=["user_id=fake_user"], ), ), ] ) self.cloud.compute.find_keypair("fake_keypair", user_id="fake_user") def test_keypair_get_user_id(self): self.register_uris( [ dict( method="GET", uri=self.get_mock_url( "compute", "public", append=["os-keypairs", "fake_keypair"], qs_elements=["user_id=fake_user"], ), ), ] ) self.cloud.compute.get_keypair("fake_keypair", user_id="fake_user") def test_keypair_delete_user_id(self): self.register_uris( [ dict( method="DELETE", uri=self.get_mock_url( "compute", "public", append=["os-keypairs", "fake_keypair"], qs_elements=["user_id=fake_user"], ), ), ] ) self.cloud.compute.delete_keypair("fake_keypair", user_id="fake_user") class TestAggregate(TestComputeProxy): def test_aggregate_create(self): self.verify_create(self.proxy.create_aggregate, aggregate.Aggregate) def test_aggregate_delete(self): self.verify_delete( self.proxy.delete_aggregate, aggregate.Aggregate, False ) def test_aggregate_delete_ignore(self): self.verify_delete( self.proxy.delete_aggregate, aggregate.Aggregate, True ) def test_aggregate_find(self): self.verify_find(self.proxy.find_aggregate, aggregate.Aggregate) def test_aggregates(self): self.verify_list(self.proxy.aggregates, aggregate.Aggregate) def test_aggregate_get(self): self.verify_get(self.proxy.get_aggregate, aggregate.Aggregate) def test_aggregate_update(self): self.verify_update(self.proxy.update_aggregate, aggregate.Aggregate) def test_aggregate_add_host(self): self._verify( "openstack.compute.v2.aggregate.Aggregate.add_host", self.proxy.add_host_to_aggregate, method_args=["value", "host"], expected_args=[self.proxy, "host"], ) def test_aggregate_remove_host(self): self._verify( "openstack.compute.v2.aggregate.Aggregate.remove_host", self.proxy.remove_host_from_aggregate, method_args=["value", "host"], expected_args=[self.proxy, "host"], ) def test_aggregate_set_metadata(self): self._verify( "openstack.compute.v2.aggregate.Aggregate.set_metadata", self.proxy.set_aggregate_metadata, method_args=["value", {'a': 'b'}], expected_args=[self.proxy, {'a': 'b'}], ) def test_aggregate_precache_image(self): self._verify( "openstack.compute.v2.aggregate.Aggregate.precache_images", self.proxy.aggregate_precache_images, method_args=["value", '1'], expected_args=[self.proxy, [{'id': '1'}]], ) def test_aggregate_precache_images(self): self._verify( "openstack.compute.v2.aggregate.Aggregate.precache_images", self.proxy.aggregate_precache_images, method_args=["value", ['1', '2']], expected_args=[self.proxy, [{'id': '1'}, {'id': '2'}]], ) class TestService(TestComputeProxy): def test_services(self): self.verify_list(self.proxy.services, service.Service) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=False, ) def test_enable_service_252(self, mv_mock): self._verify( 'openstack.compute.v2.service.Service.enable', self.proxy.enable_service, method_args=["value", "host1", "nova-compute"], expected_args=[self.proxy, "host1", "nova-compute"], ) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=True, ) def test_enable_service_253(self, mv_mock): self._verify( 'openstack.proxy.Proxy._update', self.proxy.enable_service, method_args=["value"], method_kwargs={}, expected_args=[service.Service, "value"], expected_kwargs={'status': 'enabled'}, ) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=False, ) def test_disable_service_252(self, mv_mock): self._verify( 'openstack.compute.v2.service.Service.disable', self.proxy.disable_service, method_args=["value", "host1", "nova-compute"], expected_args=[self.proxy, "host1", "nova-compute", None], ) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=True, ) def test_disable_service_253(self, mv_mock): self._verify( 'openstack.proxy.Proxy._update', self.proxy.disable_service, method_args=["value"], method_kwargs={'disabled_reason': 'some_reason'}, expected_args=[service.Service, "value"], expected_kwargs={ 'status': 'disabled', 'disabled_reason': 'some_reason', }, ) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=False, ) def test_force_service_down_252(self, mv_mock): self._verify( 'openstack.compute.v2.service.Service.set_forced_down', self.proxy.update_service_forced_down, method_args=["value", "host1", "nova-compute"], expected_args=[self.proxy, "host1", "nova-compute", True], ) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=False, ) def test_force_service_down_252_empty_vals(self, mv_mock): self.assertRaises( ValueError, self.proxy.update_service_forced_down, "value", None, None, ) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=False, ) def test_force_service_down_252_empty_vals_svc(self, mv_mock): self._verify( 'openstack.compute.v2.service.Service.set_forced_down', self.proxy.update_service_forced_down, method_args=[{'host': 'a', 'binary': 'b'}, None, None], expected_args=[self.proxy, None, None, True], ) def test_find_service(self): self.verify_find( self.proxy.find_service, service.Service, ) def test_find_service_args(self): self.verify_find( self.proxy.find_service, service.Service, method_kwargs={'host': 'h1'}, expected_kwargs={'host': 'h1'}, ) class TestVolumeAttachment(TestComputeProxy): def test_volume_attachment_create(self): self.verify_create( self.proxy.create_volume_attachment, volume_attachment.VolumeAttachment, method_kwargs={'server': 'server_id', 'volume': 'volume_id'}, expected_kwargs={ 'server_id': 'server_id', 'volume_id': 'volume_id', }, ) def test_volume_attachment_create__legacy_parameters(self): with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') self.verify_create( self.proxy.create_volume_attachment, volume_attachment.VolumeAttachment, method_kwargs={'server': 'server_id', 'volumeId': 'volume_id'}, expected_kwargs={ 'server_id': 'server_id', 'volume_id': 'volume_id', }, ) self.assertEqual(1, len(w)) self.assertEqual( os_warnings.RemovedInSDK50Warning, w[-1].category, ) self.assertIn( 'This method was called with a volume_id or volumeId argument', str(w[-1]), ) def test_volume_attachment_create__missing_parameters(self): exc = self.assertRaises( TypeError, self.proxy.create_volume_attachment, 'server_id', ) self.assertIn( 'create_volume_attachment() missing 1 required positional argument: volume', # noqa: E501 str(exc), ) def test_volume_attachment_update(self): self.verify_update( self.proxy.update_volume_attachment, volume_attachment.VolumeAttachment, method_args=[], method_kwargs={'server': 'server_id', 'volume': 'volume_id'}, expected_args=[None], expected_kwargs={ 'id': 'volume_id', 'server_id': 'server_id', 'volume_id': 'volume_id', }, ) def test_volume_attachment_delete(self): # We pass objects to avoid the lookup that's done as part of the # handling of legacy option order. We test that legacy path separately. fake_server = server.Server(id=str(uuid.uuid4())) fake_volume = volume.Volume(id=str(uuid.uuid4())) self.verify_delete( self.proxy.delete_volume_attachment, volume_attachment.VolumeAttachment, ignore_missing=False, method_args=[fake_server, fake_volume], method_kwargs={}, expected_args=[None], expected_kwargs={ 'id': fake_volume.id, 'server_id': fake_server.id, }, ) def test_volume_attachment_delete__ignore(self): # We pass objects to avoid the lookup that's done as part of the # handling of legacy option order. We test that legacy path separately. fake_server = server.Server(id=str(uuid.uuid4())) fake_volume = volume.Volume(id=str(uuid.uuid4())) self.verify_delete( self.proxy.delete_volume_attachment, volume_attachment.VolumeAttachment, ignore_missing=True, method_args=[fake_server, fake_volume], method_kwargs={}, expected_args=[None], expected_kwargs={ 'id': fake_volume.id, 'server_id': fake_server.id, }, ) def test_volume_attachment_delete__legacy_parameters(self): fake_server = server.Server(id=str(uuid.uuid4())) fake_volume = volume.Volume(id=str(uuid.uuid4())) with mock.patch.object( self.proxy, 'find_server', return_value=None, ) as mock_find_server: # we are calling the method with volume and server ID arguments as # strings and in the wrong order, which results in a query as we # attempt to match the server ID to an actual server before we # switch the argument order once we realize we can't do this self.verify_delete( self.proxy.delete_volume_attachment, volume_attachment.VolumeAttachment, ignore_missing=False, method_args=[fake_volume.id, fake_server.id], method_kwargs={}, expected_args=[None], expected_kwargs={ 'id': fake_volume.id, 'server_id': fake_server.id, }, ) # note that we attempted to call the server with the volume ID but # this was mocked to return None (as would happen in the real # world) mock_find_server.assert_called_once_with( fake_volume.id, ignore_missing=True, ) def test_volume_attachment_get(self): self.verify_get( self.proxy.get_volume_attachment, volume_attachment.VolumeAttachment, method_args=[], method_kwargs={'server': 'server_id', 'volume': 'volume_id'}, expected_kwargs={ 'id': 'volume_id', 'server_id': 'server_id', }, ) def test_volume_attachments(self): self.verify_list( self.proxy.volume_attachments, volume_attachment.VolumeAttachment, method_kwargs={'server': 'server_id'}, expected_kwargs={'server_id': 'server_id'}, ) class TestHypervisor(TestComputeProxy): def test_hypervisors_not_detailed(self): self.verify_list( self.proxy.hypervisors, hypervisor.Hypervisor, method_kwargs={"details": False}, expected_kwargs={}, ) def test_hypervisors_detailed(self): self.verify_list( self.proxy.hypervisors, hypervisor.HypervisorDetail, method_kwargs={"details": True}, expected_kwargs={}, ) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=False, ) def test_hypervisors_search_before_253_no_qp(self, sm): self.verify_list( self.proxy.hypervisors, hypervisor.Hypervisor, base_path='/os-hypervisors/detail', method_kwargs={'details': True}, expected_kwargs={}, ) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=False, ) def test_hypervisors_search_before_253(self, sm): self.verify_list( self.proxy.hypervisors, hypervisor.Hypervisor, base_path='/os-hypervisors/substring/search', method_kwargs={'hypervisor_hostname_pattern': 'substring'}, expected_kwargs={}, ) @mock.patch( 'openstack.utils.supports_microversion', autospec=True, return_value=True, ) def test_hypervisors_search_after_253(self, sm): self.verify_list( self.proxy.hypervisors, hypervisor.Hypervisor, method_kwargs={'hypervisor_hostname_pattern': 'substring'}, base_path=None, expected_kwargs={'hypervisor_hostname_pattern': 'substring'}, ) def test_find_hypervisor_detail(self): self.verify_find( self.proxy.find_hypervisor, hypervisor.Hypervisor, expected_kwargs={ 'list_base_path': '/os-hypervisors/detail', 'ignore_missing': True, }, ) def test_find_hypervisor_no_detail(self): self.verify_find( self.proxy.find_hypervisor, hypervisor.Hypervisor, method_kwargs={'details': False}, expected_kwargs={'list_base_path': None, 'ignore_missing': True}, ) def test_get_hypervisor(self): self.verify_get(self.proxy.get_hypervisor, hypervisor.Hypervisor) def test_get_hypervisor_uptime(self): self._verify( "openstack.compute.v2.hypervisor.Hypervisor.get_uptime", self.proxy.get_hypervisor_uptime, method_args=["value"], expected_args=[self.proxy], ) class TestCompute(TestComputeProxy): def test_extension_find(self): self.verify_find(self.proxy.find_extension, extension.Extension) def test_extensions(self): self.verify_list(self.proxy.extensions, extension.Extension) @contextlib.contextmanager def _check_image_proxy_deprecation_warning(self): with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") yield self.assertEqual(1, len(w)) self.assertTrue(issubclass(w[-1].category, DeprecationWarning)) self.assertIn( "This API is a proxy to the image service ", str(w[-1].message), ) def test_image_delete(self): with self._check_image_proxy_deprecation_warning(): self.verify_delete(self.proxy.delete_image, image.Image, False) def test_image_delete_ignore(self): with self._check_image_proxy_deprecation_warning(): self.verify_delete(self.proxy.delete_image, image.Image, True) def test_image_find(self): with self._check_image_proxy_deprecation_warning(): self.verify_find(self.proxy.find_image, image.Image) def test_image_get(self): with self._check_image_proxy_deprecation_warning(): self.verify_get(self.proxy.get_image, image.Image) def test_images_detailed(self): with self._check_image_proxy_deprecation_warning(): self.verify_list( self.proxy.images, image.ImageDetail, method_kwargs={"details": True, "query": 1}, expected_kwargs={"query": 1}, ) def test_images_not_detailed(self): with self._check_image_proxy_deprecation_warning(): self.verify_list( self.proxy.images, image.Image, method_kwargs={"details": False, "query": 1}, expected_kwargs={"query": 1}, ) def test_limits_get(self): self._verify( "openstack.compute.v2.limits.Limits.fetch", self.proxy.get_limits, method_args=[], method_kwargs={"a": "b"}, expected_args=[self.proxy], expected_kwargs={"a": "b"}, ) def test_server_interface_create(self): self.verify_create( self.proxy.create_server_interface, server_interface.ServerInterface, method_kwargs={"server": "test_id"}, expected_kwargs={"server_id": "test_id"}, ) def test_server_interface_delete(self): self.proxy._get_uri_attribute = lambda *args: args[1] interface_id = "test_interface_id" server_id = "test_server_id" test_interface = server_interface.ServerInterface(id=interface_id) test_interface.server_id = server_id # Case1: ServerInterface instance is provided as value self._verify( "openstack.proxy.Proxy._delete", self.proxy.delete_server_interface, method_args=[test_interface], method_kwargs={"server": server_id}, expected_args=[server_interface.ServerInterface, interface_id], expected_kwargs={"server_id": server_id, "ignore_missing": True}, ) # Case2: ServerInterface ID is provided as value self._verify( "openstack.proxy.Proxy._delete", self.proxy.delete_server_interface, method_args=[interface_id], method_kwargs={"server": server_id}, expected_args=[server_interface.ServerInterface, interface_id], expected_kwargs={"server_id": server_id, "ignore_missing": True}, ) def test_server_interface_delete_ignore(self): self.proxy._get_uri_attribute = lambda *args: args[1] self.verify_delete( self.proxy.delete_server_interface, server_interface.ServerInterface, True, method_kwargs={"server": "test_id"}, expected_kwargs={"server_id": "test_id"}, ) def test_server_interface_get(self): self.proxy._get_uri_attribute = lambda *args: args[1] interface_id = "test_interface_id" server_id = "test_server_id" test_interface = server_interface.ServerInterface(id=interface_id) test_interface.server_id = server_id # Case1: ServerInterface instance is provided as value self._verify( 'openstack.proxy.Proxy._get', self.proxy.get_server_interface, method_args=[test_interface], method_kwargs={"server": server_id}, expected_args=[server_interface.ServerInterface], expected_kwargs={"port_id": interface_id, "server_id": server_id}, ) # Case2: ServerInterface ID is provided as value self._verify( 'openstack.proxy.Proxy._get', self.proxy.get_server_interface, method_args=[interface_id], method_kwargs={"server": server_id}, expected_args=[server_interface.ServerInterface], expected_kwargs={"port_id": interface_id, "server_id": server_id}, ) def test_server_interfaces(self): self.verify_list( self.proxy.server_interfaces, server_interface.ServerInterface, method_args=["test_id"], expected_args=[], expected_kwargs={"server_id": "test_id"}, ) def test_server_ips_with_network_label(self): self.verify_list( self.proxy.server_ips, server_ip.ServerIP, method_args=["test_id"], method_kwargs={"network_label": "test_label"}, expected_args=[], expected_kwargs={ "server_id": "test_id", "network_label": "test_label", }, ) def test_server_ips_without_network_label(self): self.verify_list( self.proxy.server_ips, server_ip.ServerIP, method_args=["test_id"], expected_args=[], expected_kwargs={"server_id": "test_id", "network_label": None}, ) def test_server_create_attrs(self): self.verify_create(self.proxy.create_server, server.Server) def test_server_delete(self): self.verify_delete(self.proxy.delete_server, server.Server, False) def test_server_delete_ignore(self): self.verify_delete(self.proxy.delete_server, server.Server, True) def test_server_force_delete(self): self._verify( "openstack.compute.v2.server.Server.force_delete", self.proxy.delete_server, method_args=["value", False, True], expected_args=[self.proxy], ) def test_server_find(self): self.verify_find( self.proxy.find_server, server.Server, method_kwargs={'all_projects': True}, expected_kwargs={ 'list_base_path': '/servers/detail', 'all_projects': True, }, ) def test_server_get(self): self.verify_get(self.proxy.get_server, server.Server) def test_servers_detailed(self): self.verify_list( self.proxy.servers, server.Server, method_kwargs={"details": True, "changes_since": 1, "image": 2}, expected_kwargs={ "changes_since": 1, "image": 2, "base_path": "/servers/detail", }, ) def test_servers_not_detailed(self): self.verify_list( self.proxy.servers, server.Server, method_kwargs={"details": False, "changes_since": 1, "image": 2}, expected_kwargs={"changes_since": 1, "image": 2}, ) def test_server_update(self): self.verify_update(self.proxy.update_server, server.Server) def test_server_change_password(self): self._verify( "openstack.compute.v2.server.Server.change_password", self.proxy.change_server_password, method_args=["value", "password"], expected_args=[self.proxy, "password"], ) def test_server_get_password(self): self._verify( "openstack.compute.v2.server.Server.get_password", self.proxy.get_server_password, method_args=["value"], expected_args=[self.proxy], ) def test_server_clear_password(self): self._verify( "openstack.compute.v2.server.Server.clear_password", self.proxy.clear_server_password, method_args=["value"], expected_args=[self.proxy], ) def test_server_wait_for(self): value = server.Server(id='1234') self.verify_wait_for_status( self.proxy.wait_for_server, method_args=[value], expected_args=[self.proxy, value, 'ACTIVE', ['ERROR'], 2, 120], expected_kwargs={'callback': None}, ) def test_server_resize(self): self._verify( "openstack.compute.v2.server.Server.resize", self.proxy.resize_server, method_args=["value", "test-flavor"], expected_args=[self.proxy, "test-flavor"], ) def test_server_confirm_resize(self): self._verify( "openstack.compute.v2.server.Server.confirm_resize", self.proxy.confirm_server_resize, method_args=["value"], expected_args=[self.proxy], ) def test_server_revert_resize(self): self._verify( "openstack.compute.v2.server.Server.revert_resize", self.proxy.revert_server_resize, method_args=["value"], expected_args=[self.proxy], ) def test_server_rebuild(self): id = 'test_image_id' image_obj = image.Image(id='test_image_id') # Case1: image object is provided # NOTE: Inside of Server.rebuild is where image_obj gets converted # to an ID instead of object. self._verify( 'openstack.compute.v2.server.Server.rebuild', self.proxy.rebuild_server, method_args=["value"], method_kwargs={ "name": "test_server", "admin_password": "test_pass", "metadata": {"k1": "v1"}, "image": image_obj, }, expected_args=[self.proxy], expected_kwargs={ "name": "test_server", "admin_password": "test_pass", "metadata": {"k1": "v1"}, "image": image_obj, }, ) # Case2: image name or id is provided self._verify( 'openstack.compute.v2.server.Server.rebuild', self.proxy.rebuild_server, method_args=["value"], method_kwargs={ "name": "test_server", "admin_password": "test_pass", "metadata": {"k1": "v1"}, "image": id, }, expected_args=[self.proxy], expected_kwargs={ "name": "test_server", "admin_password": "test_pass", "metadata": {"k1": "v1"}, "image": id, }, ) def test_add_fixed_ip_to_server(self): self._verify( "openstack.compute.v2.server.Server.add_fixed_ip", self.proxy.add_fixed_ip_to_server, method_args=["value", "network-id"], expected_args=[self.proxy, "network-id"], ) def test_fixed_ip_from_server(self): self._verify( "openstack.compute.v2.server.Server.remove_fixed_ip", self.proxy.remove_fixed_ip_from_server, method_args=["value", "address"], expected_args=[self.proxy, "address"], ) def test_floating_ip_to_server(self): self._verify( "openstack.compute.v2.server.Server.add_floating_ip", self.proxy.add_floating_ip_to_server, method_args=["value", "floating-ip"], expected_args=[self.proxy, "floating-ip"], expected_kwargs={'fixed_address': None}, ) def test_add_floating_ip_to_server_with_fixed_addr(self): self._verify( "openstack.compute.v2.server.Server.add_floating_ip", self.proxy.add_floating_ip_to_server, method_args=["value", "floating-ip", 'fixed-addr'], expected_args=[self.proxy, "floating-ip"], expected_kwargs={'fixed_address': 'fixed-addr'}, ) def test_remove_floating_ip_from_server(self): self._verify( "openstack.compute.v2.server.Server.remove_floating_ip", self.proxy.remove_floating_ip_from_server, method_args=["value", "address"], expected_args=[self.proxy, "address"], ) def test_server_backup(self): self._verify( "openstack.compute.v2.server.Server.backup", self.proxy.backup_server, method_args=["value", "name", "daily", 1], expected_args=[self.proxy, "name", "daily", 1], ) def test_server_pause(self): self._verify( "openstack.compute.v2.server.Server.pause", self.proxy.pause_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_unpause(self): self._verify( "openstack.compute.v2.server.Server.unpause", self.proxy.unpause_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_suspend(self): self._verify( "openstack.compute.v2.server.Server.suspend", self.proxy.suspend_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_resume(self): self._verify( "openstack.compute.v2.server.Server.resume", self.proxy.resume_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_lock(self): self._verify( "openstack.compute.v2.server.Server.lock", self.proxy.lock_server, method_args=["value"], expected_args=[self.proxy], expected_kwargs={"locked_reason": None}, ) def test_server_lock_with_options(self): self._verify( "openstack.compute.v2.server.Server.lock", self.proxy.lock_server, method_args=["value"], method_kwargs={"locked_reason": "Because why not"}, expected_args=[self.proxy], expected_kwargs={"locked_reason": "Because why not"}, ) def test_server_unlock(self): self._verify( "openstack.compute.v2.server.Server.unlock", self.proxy.unlock_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_rescue(self): self._verify( "openstack.compute.v2.server.Server.rescue", self.proxy.rescue_server, method_args=["value"], expected_args=[self.proxy], expected_kwargs={"admin_pass": None, "image_ref": None}, ) def test_server_rescue_with_options(self): self._verify( "openstack.compute.v2.server.Server.rescue", self.proxy.rescue_server, method_args=["value", 'PASS', 'IMG'], expected_args=[self.proxy], expected_kwargs={"admin_pass": 'PASS', "image_ref": 'IMG'}, ) def test_server_unrescue(self): self._verify( "openstack.compute.v2.server.Server.unrescue", self.proxy.unrescue_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_evacuate(self): self._verify( "openstack.compute.v2.server.Server.evacuate", self.proxy.evacuate_server, method_args=["value"], expected_args=[self.proxy], expected_kwargs={ "host": None, "admin_pass": None, "force": None, "on_shared_storage": None, }, ) def test_server_evacuate_with_options(self): self._verify( "openstack.compute.v2.server.Server.evacuate", self.proxy.evacuate_server, method_args=["value", 'HOST2', 'NEW_PASS', True], method_kwargs={'on_shared_storage': False}, expected_args=[self.proxy], expected_kwargs={ "host": "HOST2", "admin_pass": 'NEW_PASS', "force": True, "on_shared_storage": False, }, ) def test_server_start(self): self._verify( "openstack.compute.v2.server.Server.start", self.proxy.start_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_stop(self): self._verify( "openstack.compute.v2.server.Server.stop", self.proxy.stop_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_restore(self): self._verify( "openstack.compute.v2.server.Server.restore", self.proxy.restore_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_shelve(self): self._verify( "openstack.compute.v2.server.Server.shelve", self.proxy.shelve_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_shelve_offload(self): self._verify( "openstack.compute.v2.server.Server.shelve_offload", self.proxy.shelve_offload_server, method_args=["value"], expected_args=[self.proxy], ) def test_server_unshelve(self): self._verify( "openstack.compute.v2.server.Server.unshelve", self.proxy.unshelve_server, method_args=["value"], expected_args=[self.proxy], expected_kwargs={ "host": None, "availability_zone": types.UNSET, }, ) def test_server_unshelve_with_options(self): self._verify( "openstack.compute.v2.server.Server.unshelve", self.proxy.unshelve_server, method_args=["value"], method_kwargs={"host": "HOST2", "availability_zone": "AZ2"}, expected_args=[self.proxy], expected_kwargs={"host": "HOST2", "availability_zone": "AZ2"}, ) def test_server_trigger_dump(self): self._verify( "openstack.compute.v2.server.Server.trigger_crash_dump", self.proxy.trigger_server_crash_dump, method_args=["value"], expected_args=[self.proxy], ) def test_server_add_tag(self): self._verify( "openstack.compute.v2.server.Server.add_tag", self.proxy.add_tag_to_server, method_args=["value", "tag"], expected_args=[self.proxy, "tag"], ) def test_server_remove_tag(self): self._verify( "openstack.compute.v2.server.Server.remove_tag", self.proxy.remove_tag_from_server, method_args=["value", "tag"], expected_args=[self.proxy, "tag"], ) def test_server_remove_tags(self): self._verify( "openstack.compute.v2.server.Server.remove_all_tags", self.proxy.remove_tags_from_server, method_args=["value"], expected_args=[self.proxy], ) def test_get_server_output(self): self._verify( "openstack.compute.v2.server.Server.get_console_output", self.proxy.get_server_console_output, method_args=["value"], expected_args=[self.proxy], expected_kwargs={"length": None}, ) self._verify( "openstack.compute.v2.server.Server.get_console_output", self.proxy.get_server_console_output, method_args=["value", 1], expected_args=[self.proxy], expected_kwargs={"length": 1}, ) def test_availability_zones_not_detailed(self): self.verify_list( self.proxy.availability_zones, az.AvailabilityZone, method_kwargs={"details": False}, expected_kwargs={}, ) def test_availability_zones_detailed(self): self.verify_list( self.proxy.availability_zones, az.AvailabilityZoneDetail, method_kwargs={"details": True}, expected_kwargs={}, ) def test_get_all_server_metadata(self): self._verify( "openstack.compute.v2.server.Server.fetch_metadata", self.proxy.get_server_metadata, method_args=["value"], expected_args=[self.proxy], expected_result=server.Server(id="value", metadata={}), ) def test_set_server_metadata(self): kwargs = {"a": "1", "b": "2"} id = "an_id" self._verify( "openstack.compute.v2.server.Server.set_metadata", self.proxy.set_server_metadata, method_args=[id], method_kwargs=kwargs, method_result=server.Server.existing(id=id, metadata=kwargs), expected_args=[self.proxy], expected_kwargs={'metadata': kwargs}, expected_result=server.Server.existing(id=id, metadata=kwargs), ) def test_delete_server_metadata(self): self._verify( "openstack.compute.v2.server.Server.delete_metadata_item", self.proxy.delete_server_metadata, expected_result=None, method_args=["value", ["key"]], expected_args=[self.proxy, "key"], ) def test_create_image(self): metadata = {'k1': 'v1'} with mock.patch( 'openstack.compute.v2.server.Server.create_image' ) as ci_mock: ci_mock.return_value = 'image_id' connection_mock = mock.Mock() connection_mock.get_image = mock.Mock(return_value='image') connection_mock.wait_for_image = mock.Mock() self.proxy._connection = connection_mock rsp = self.proxy.create_server_image( 'server', 'image_name', metadata, wait=True, timeout=1 ) ci_mock.assert_called_with(self.proxy, 'image_name', metadata) self.proxy._connection.get_image.assert_called_with('image_id') self.proxy._connection.wait_for_image.assert_called_with( 'image', timeout=1 ) self.assertEqual(connection_mock.wait_for_image(), rsp) def test_server_group_create(self): self.verify_create( self.proxy.create_server_group, server_group.ServerGroup ) def test_server_group_delete(self): self.verify_delete( self.proxy.delete_server_group, server_group.ServerGroup, False ) def test_server_group_delete_ignore(self): self.verify_delete( self.proxy.delete_server_group, server_group.ServerGroup, True ) def test_server_group_find(self): self.verify_find( self.proxy.find_server_group, server_group.ServerGroup, method_kwargs={'all_projects': True}, expected_kwargs={'all_projects': True}, ) def test_server_group_get(self): self.verify_get(self.proxy.get_server_group, server_group.ServerGroup) def test_server_groups(self): self.verify_list(self.proxy.server_groups, server_group.ServerGroup) def test_live_migrate_server(self): self._verify( 'openstack.compute.v2.server.Server.live_migrate', self.proxy.live_migrate_server, method_args=["value"], method_kwargs={'host': 'host1', 'force': False}, expected_args=[self.proxy], expected_kwargs={ 'host': 'host1', 'force': False, 'block_migration': None, 'disk_over_commit': None, }, ) def test_abort_server_migration(self): self._verify( 'openstack.proxy.Proxy._delete', self.proxy.abort_server_migration, method_args=['server_migration', 'server'], expected_args=[ server_migration.ServerMigration, 'server_migration', ], expected_kwargs={ 'server_id': 'server', 'ignore_missing': True, }, ) def test_force_complete_server_migration(self): self._verify( 'openstack.compute.v2.server_migration.ServerMigration.force_complete', # noqa: E501 self.proxy.force_complete_server_migration, method_args=['server_migration', 'server'], expected_args=[self.proxy], ) def test_get_server_migration(self): self._verify( 'openstack.proxy.Proxy._get', self.proxy.get_server_migration, method_args=['server_migration', 'server'], expected_args=[ server_migration.ServerMigration, 'server_migration', ], expected_kwargs={ 'server_id': 'server', 'ignore_missing': True, }, ) def test_server_migrations(self): self._verify( 'openstack.proxy.Proxy._list', self.proxy.server_migrations, method_args=['server'], expected_args=[server_migration.ServerMigration], expected_kwargs={'server_id': 'server'}, ) def test_migrations(self): self.verify_list(self.proxy.migrations, migration.Migration) def test_migrations_kwargs(self): self.verify_list( self.proxy.migrations, migration.Migration, method_kwargs={'host': 'h1'}, expected_kwargs={'host': 'h1'}, ) def test_fetch_security_groups(self): self._verify( 'openstack.compute.v2.server.Server.fetch_security_groups', self.proxy.fetch_server_security_groups, method_args=["value"], expected_args=[self.proxy], ) def test_add_security_groups(self): self._verify( 'openstack.compute.v2.server.Server.add_security_group', self.proxy.add_security_group_to_server, method_args=["value", 'sg'], expected_args=[self.proxy, 'sg'], ) def test_remove_security_groups(self): self._verify( 'openstack.compute.v2.server.Server.remove_security_group', self.proxy.remove_security_group_from_server, method_args=["value", 'sg'], expected_args=[self.proxy, 'sg'], ) def test_usages(self): self.verify_list(self.proxy.usages, usage.Usage) def test_usages__with_kwargs(self): now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) start = now - datetime.timedelta(weeks=4) end = end = now + datetime.timedelta(days=1) self.verify_list( self.proxy.usages, usage.Usage, method_kwargs={'start': start, 'end': end}, expected_kwargs={ 'start': start.isoformat(), 'end': end.isoformat(), }, ) def test_get_usage(self): self._verify( "openstack.compute.v2.usage.Usage.fetch", self.proxy.get_usage, method_args=['value'], method_kwargs={}, expected_args=[self.proxy], expected_kwargs={}, ) def test_get_usage__with_kwargs(self): now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) start = now - datetime.timedelta(weeks=4) end = end = now + datetime.timedelta(days=1) self._verify( "openstack.compute.v2.usage.Usage.fetch", self.proxy.get_usage, method_args=['value'], method_kwargs={'start': start, 'end': end}, expected_args=[self.proxy], expected_kwargs={ 'start': start.isoformat(), 'end': end.isoformat(), }, ) def test_create_server_remote_console(self): self.verify_create( self.proxy.create_server_remote_console, server_remote_console.ServerRemoteConsole, method_kwargs={"server": "test_id", "type": "fake"}, expected_kwargs={"server_id": "test_id", "type": "fake"}, ) def test_get_console_url(self): self._verify( 'openstack.compute.v2.server.Server.get_console_url', self.proxy.get_server_console_url, method_args=["value", "console_type"], expected_args=[self.proxy, "console_type"], ) @mock.patch('openstack.utils.supports_microversion', autospec=True) @mock.patch('openstack.compute.v2._proxy.Proxy._create', autospec=True) @mock.patch( 'openstack.compute.v2.server.Server.get_console_url', autospec=True ) def test_create_console_mv_old(self, sgc, rcc, smv): console_fake = {'url': 'a', 'type': 'b', 'protocol': 'c'} smv.return_value = False sgc.return_value = console_fake ret = self.proxy.create_console('fake_server', 'fake_type') smv.assert_called_once_with(self.proxy, '2.6') rcc.assert_not_called() sgc.assert_called_with(mock.ANY, self.proxy, 'fake_type') self.assertDictEqual(console_fake, ret) @mock.patch('openstack.utils.supports_microversion', autospec=True) @mock.patch('openstack.compute.v2._proxy.Proxy._create', autospec=True) @mock.patch( 'openstack.compute.v2.server.Server.get_console_url', autospec=True ) def test_create_console_mv_2_6(self, sgc, rcc, smv): console_fake = {'url': 'a', 'type': 'b', 'protocol': 'c'} # Test server_remote_console is triggered when mv>=2.6 smv.return_value = True rcc.return_value = server_remote_console.ServerRemoteConsole( **console_fake ) ret = self.proxy.create_console('fake_server', 'fake_type') smv.assert_called_once_with(self.proxy, '2.6') sgc.assert_not_called() rcc.assert_called_with( mock.ANY, server_remote_console.ServerRemoteConsole, server_id='fake_server', type='fake_type', protocol=None, ) self.assertEqual(console_fake['url'], ret['url']) class TestQuotaClassSet(TestComputeProxy): def test_quota_class_set_get(self): self.verify_get( self.proxy.get_quota_class_set, quota_class_set.QuotaClassSet ) def test_quota_class_set_update(self): self.verify_update( self.proxy.update_quota_class_set, quota_class_set.QuotaClassSet, False, ) class TestQuotaSet(TestComputeProxy): def test_quota_set_get(self): self._verify( 'openstack.resource.Resource.fetch', self.proxy.get_quota_set, method_args=['prj'], expected_args=[ self.proxy, False, None, None, False, ], expected_kwargs={ 'microversion': None, 'resource_response_key': None, }, method_result=quota_set.QuotaSet(), expected_result=quota_set.QuotaSet(), ) def test_quota_set_get_query(self): self._verify( 'openstack.resource.Resource.fetch', self.proxy.get_quota_set, method_args=['prj'], method_kwargs={'usage': True, 'user_id': 'uid'}, expected_args=[ self.proxy, False, '/os-quota-sets/%(project_id)s/detail', None, False, ], expected_kwargs={ 'microversion': None, 'resource_response_key': None, 'user_id': 'uid', }, ) def test_quota_set_get_defaults(self): self._verify( 'openstack.resource.Resource.fetch', self.proxy.get_quota_set_defaults, method_args=['prj'], expected_args=[ self.proxy, False, '/os-quota-sets/%(project_id)s/defaults', None, False, ], expected_kwargs={ 'microversion': None, 'resource_response_key': None, }, ) def test_quota_set_reset(self): self._verify( 'openstack.resource.Resource.delete', self.proxy.revert_quota_set, method_args=['prj'], method_kwargs={'user_id': 'uid'}, expected_args=[self.proxy], expected_kwargs={'user_id': 'uid'}, ) @mock.patch.object(proxy_base.Proxy, "_get_resource") def test_quota_set_update(self, mock_get): fake_project = project.Project(id='prj') fake_quota_set = quota_set.QuotaSet(project_id='prj') mock_get.side_effect = [fake_project, fake_quota_set] self._verify( 'openstack.resource.Resource.commit', self.proxy.update_quota_set, method_args=['prj'], method_kwargs={'ram': 123}, expected_args=[self.proxy], expected_kwargs={}, ) mock_get.assert_has_calls( [ mock.call(project.Project, 'prj'), mock.call(quota_set.QuotaSet, None, project_id='prj', ram=123), ] ) @mock.patch.object(proxy_base.Proxy, "_get_resource") def test_quota_set_update__legacy(self, mock_get): fake_quota_set = quota_set.QuotaSet(project_id='prj') mock_get.side_effect = [fake_quota_set] with warnings.catch_warnings(record=True) as w: warnings.simplefilter('always') self._verify( 'openstack.resource.Resource.commit', self.proxy.update_quota_set, method_args=[fake_quota_set], method_kwargs={'ram': 123}, expected_args=[self.proxy], expected_kwargs={}, ) self.assertEqual(1, len(w)) self.assertEqual( os_warnings.RemovedInSDK50Warning, w[-1].category, ) self.assertIn( "The signature of 'update_quota_set' has changed ", str(w[-1]), ) class TestServerAction(TestComputeProxy): def test_server_action_get(self): self._verify( 'openstack.proxy.Proxy._get', self.proxy.get_server_action, method_args=['request_id'], method_kwargs={'server': 'server_id'}, expected_args=[server_action.ServerAction], expected_kwargs={ 'request_id': 'request_id', 'server_id': 'server_id', }, ) def test_server_actions(self): self.verify_list( self.proxy.server_actions, server_action.ServerAction, method_kwargs={'server': 'server_a'}, expected_kwargs={'server_id': 'server_a'}, ) class TestValidateConsoleAuthToken(TestComputeProxy): def test_validate_console_auth_token(self): self.verify_get( self.proxy.validate_console_auth_token, console_auth_token.ConsoleAuthToken, )