Coverage for manila/tests/test_exception.py: 98%
415 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright 2010 United States Government as represented by the
2# Administrator of the National Aeronautics and Space Administration.
3# Copyright 2014 Mirantis, Inc.
4# All Rights Reserved.
5#
6# Licensed under the Apache License, Version 2.0 (the "License"); you may
7# not use this file except in compliance with the License. You may obtain
8# a copy of the License at
9#
10# http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15# License for the specific language governing permissions and limitations
16# under the License.
18import ddt
20from manila import exception
21from manila import test
24class FakeNotifier(object):
25 """Acts like the manila.openstack.common.notifier.api module."""
26 ERROR = 88
28 def __init__(self):
29 self.provided_publisher = None
30 self.provided_event = None
31 self.provided_priority = None
32 self.provided_payload = None
34 def notify(self, context, publisher, event, priority, payload):
35 self.provided_publisher = publisher
36 self.provided_event = event
37 self.provided_priority = priority
38 self.provided_payload = payload
41@ddt.ddt
42class ManilaExceptionTestCase(test.TestCase):
43 def test_default_error_msg(self):
44 class FakeManilaException(exception.ManilaException):
45 message = "default message"
47 exc = FakeManilaException()
48 self.assertEqual('default message', str(exc))
50 def test_error_msg(self):
51 self.assertEqual('test',
52 str(exception.ManilaException('test')))
54 def test_default_error_msg_with_kwargs(self):
55 class FakeManilaException(exception.ManilaException):
56 message = "default message: %(code)s"
58 exc = FakeManilaException(code=500)
59 self.assertEqual('default message: 500', str(exc))
61 def test_error_msg_exception_with_kwargs(self):
62 # NOTE(dprince): disable format errors for this test
63 self.flags(fatal_exception_format_errors=False)
65 class FakeManilaException(exception.ManilaException):
66 message = "default message: %(misspelled_code)s"
68 exc = FakeManilaException(code=500)
69 self.assertEqual('default message: %(misspelled_code)s',
70 str(exc))
72 def test_default_error_code(self):
73 class FakeManilaException(exception.ManilaException):
74 code = 404
76 exc = FakeManilaException()
77 self.assertEqual(404, exc.kwargs['code'])
79 def test_error_code_from_kwarg(self):
80 class FakeManilaException(exception.ManilaException):
81 code = 500
83 exc = FakeManilaException(code=404)
84 self.assertEqual(404, exc.kwargs['code'])
86 def test_error_msg_is_exception_to_string(self):
87 msg = 'test message'
88 exc1 = Exception(msg)
89 exc2 = exception.ManilaException(exc1)
90 self.assertEqual(msg, exc2.msg)
92 def test_exception_kwargs_to_string(self):
93 msg = 'test message'
94 exc1 = Exception(msg)
95 exc2 = exception.ManilaException(kwarg1=exc1)
96 self.assertEqual(msg, exc2.kwargs['kwarg1'])
98 def test_exception_multi_kwargs_to_string(self):
99 exc = exception.ManilaException(
100 'fake_msg', foo=Exception('foo_msg'), bar=Exception('bar_msg'))
101 self.assertEqual('fake_msg', exc.msg)
102 self.assertEqual('foo_msg', exc.kwargs['foo'])
103 self.assertEqual('bar_msg', exc.kwargs['bar'])
104 self.assertNotIn('fake_msg', exc.kwargs)
105 self.assertNotIn('foo_msg', exc.kwargs)
106 self.assertNotIn('bar_msg', exc.kwargs)
108 @ddt.data("test message.", "test message....", ".")
109 def test_exception_not_redundant_period(self, msg):
110 exc1 = Exception(msg)
111 exc2 = exception.ManilaException(exc1)
112 self.assertEqual(msg, exc2.msg)
114 def test_exception_redundant_period(self):
115 msg = "test message.."
116 exc1 = Exception(msg)
117 exc2 = exception.ManilaException(exc1)
118 self.assertEqual("test message.", exc2.msg)
120 def test_replication_exception(self):
121 # Verify response code for exception.ReplicationException
122 reason = "Something bad happened."
123 e = exception.ReplicationException(reason=reason)
124 self.assertEqual(500, e.code)
125 self.assertIn(reason, e.msg)
127 def test_snapshot_access_already_exists(self):
128 # Verify response code for exception.ShareSnapshotAccessExists
129 access_type = "fake_type"
130 access = "fake_access"
131 e = exception.ShareSnapshotAccessExists(access_type=access_type,
132 access=access)
133 self.assertEqual(400, e.code)
134 self.assertIn(access_type, e.msg)
135 self.assertIn(access, e.msg)
137 def test_manage_share_server_error(self):
138 # Verify response code for exception.ManageShareServerError
139 reason = 'Invalid share server id.'
140 share_server_id = 'fake'
141 e = exception.ManageShareServerError(reason=reason,
142 share_server_id=share_server_id)
144 self.assertEqual(500, e.code)
145 self.assertIn(reason, e.msg)
147 def test_ip_address_generation_failure(self):
148 # verify response code for exception.IpAddressGenerationFailureClient
149 e = exception.IpAddressGenerationFailureClient()
150 self.assertEqual(500, e.code)
153class ManilaExceptionResponseCode400(test.TestCase):
155 def test_invalid(self):
156 # Verify response code for exception.Invalid
157 e = exception.Invalid()
158 self.assertEqual(400, e.code)
160 def test_invalid_input(self):
161 # Verify response code for exception.InvalidInput
162 reason = "fake_reason"
163 e = exception.InvalidInput(reason=reason)
164 self.assertEqual(400, e.code)
165 self.assertIn(reason, e.msg)
167 def test_invalid_request(self):
168 # Verify response code for exception.InvalidRequest
169 e = exception.InvalidRequest()
170 self.assertEqual(400, e.code)
172 def test_invalid_results(self):
173 # Verify response code for exception.InvalidResults
174 e = exception.InvalidResults()
175 self.assertEqual(400, e.code)
177 def test_invalid_uuid(self):
178 # Verify response code for exception.InvalidUUID
179 uuid = "fake_uuid"
180 e = exception.InvalidUUID(uuid=uuid)
181 self.assertEqual(400, e.code)
182 self.assertIn(uuid, e.msg)
184 def test_invalid_content_type(self):
185 # Verify response code for exception.InvalidContentType
186 content_type = "fake_content_type"
187 e = exception.InvalidContentType(content_type=content_type)
188 self.assertEqual(400, e.code)
189 self.assertIn(content_type, e.msg)
191 def test_invalid_parameter_value(self):
192 # Verify response code for exception.InvalidParameterValue
193 err = "fake_err"
194 e = exception.InvalidParameterValue(err=err)
195 self.assertEqual(400, e.code)
196 self.assertIn(err, e.msg)
198 def test_invalid_reservation_expiration(self):
199 # Verify response code for exception.InvalidReservationExpiration
200 expire = "fake_expire"
201 e = exception.InvalidReservationExpiration(expire=expire)
202 self.assertEqual(400, e.code)
203 self.assertIn(expire, e.msg)
205 def test_invalid_quota_value(self):
206 # Verify response code for exception.InvalidQuotaValue
207 unders = '-1'
208 e = exception.InvalidQuotaValue(unders=unders)
209 self.assertEqual(400, e.code)
211 def test_invalid_share(self):
212 # Verify response code for exception.InvalidShare
213 reason = "fake_reason"
214 e = exception.InvalidShare(reason=reason)
215 self.assertEqual(400, e.code)
216 self.assertIn(reason, e.msg)
218 def test_invalid_share_access(self):
219 # Verify response code for exception.InvalidShareAccess
220 reason = "fake_reason"
221 e = exception.InvalidShareAccess(reason=reason)
222 self.assertEqual(400, e.code)
223 self.assertIn(reason, e.msg)
225 def test_invalid_share_snapshot(self):
226 # Verify response code for exception.InvalidShareSnapshot
227 reason = "fake_reason"
228 e = exception.InvalidShareSnapshot(reason=reason)
229 self.assertEqual(400, e.code)
230 self.assertIn(reason, e.msg)
232 def test_invalid_share_metadata(self):
233 # Verify response code for exception.InvalidMetadata
234 e = exception.InvalidMetadata()
235 self.assertEqual(400, e.code)
237 def test_invalid_share_metadata_size(self):
238 # Verify response code for exception.InvalidMetadataSize
239 e = exception.InvalidMetadataSize()
240 self.assertEqual(400, e.code)
242 def test_invalid_volume(self):
243 # Verify response code for exception.InvalidVolume
244 e = exception.InvalidVolume()
245 self.assertEqual(400, e.code)
247 def test_invalid_share_type(self):
248 # Verify response code for exception.InvalidShareType
249 reason = "fake_reason"
250 e = exception.InvalidShareType(reason=reason)
251 self.assertEqual(400, e.code)
252 self.assertIn(reason, e.msg)
254 def test_manage_invalid_share_snapshot(self):
255 # Verify response code for exception.ManageInvalidShareSnapshot
256 reason = "fake_reason"
257 e = exception.ManageInvalidShareSnapshot(reason=reason)
258 self.assertEqual(400, e.code)
259 self.assertIn(reason, e.msg)
261 def test_unmanage_invalid_share_snapshot(self):
262 # Verify response code for exception.UnmanageInvalidShareSnapshot
263 reason = "fake_reason"
264 e = exception.UnmanageInvalidShareSnapshot(reason=reason)
265 self.assertEqual(400, e.code)
266 self.assertIn(reason, e.msg)
268 def test_invalid_share_snapshot_instance(self):
269 # Verify response code for exception.InvalidShareSnapshotInstance
270 reason = "fake_reason"
271 e = exception.InvalidShareSnapshotInstance(reason=reason)
272 self.assertEqual(400, e.code)
273 self.assertIn(reason, e.msg)
276class ManilaExceptionResponseCode403(test.TestCase):
278 def test_not_authorized(self):
279 # Verify response code for exception.NotAuthorized
280 e = exception.NotAuthorized()
281 self.assertEqual(403, e.code)
283 def test_admin_required(self):
284 # Verify response code for exception.AdminRequired
285 e = exception.AdminRequired()
286 self.assertEqual(403, e.code)
288 def test_policy_not_authorized(self):
289 # Verify response code for exception.PolicyNotAuthorized
290 action = "fake_action"
291 e = exception.PolicyNotAuthorized(action=action)
292 self.assertEqual(403, e.code)
293 self.assertIn(action, e.msg)
296class ManilaExceptionResponseCode404(test.TestCase):
298 def test_not_found(self):
299 # Verify response code for exception.NotFound
300 e = exception.NotFound()
301 self.assertEqual(404, e.code)
303 def test_share_network_not_found(self):
304 # Verify response code for exception.ShareNetworkNotFound
305 share_network_id = "fake_share_network_id"
306 e = exception.ShareNetworkNotFound(share_network_id=share_network_id)
307 self.assertEqual(404, e.code)
308 self.assertIn(share_network_id, e.msg)
310 def test_share_network_subnet_not_found(self):
311 # Verify response code for exception.ShareNetworkSubnetNotFound
312 share_network_subnet_id = "fake_share_network_subnet_id"
313 e = exception.ShareNetworkSubnetNotFound(
314 share_network_subnet_id=share_network_subnet_id)
315 self.assertEqual(404, e.code)
316 self.assertIn(share_network_subnet_id, e.msg)
318 def test_share_server_not_found(self):
319 # Verify response code for exception.ShareServerNotFound
320 share_server_id = "fake_share_server_id"
321 e = exception.ShareServerNotFound(share_server_id=share_server_id)
322 self.assertEqual(404, e.code)
323 self.assertIn(share_server_id, e.msg)
325 def test_share_server_not_found_by_filters(self):
326 # Verify response code for exception.ShareServerNotFoundByFilters
327 filters_description = "host = fakeHost"
328 e = exception.ShareServerNotFoundByFilters(
329 filters_description=filters_description)
330 self.assertEqual(404, e.code)
331 self.assertIn(filters_description, e.msg)
333 def test_service_not_found(self):
334 # Verify response code for exception.ServiceNotFound
335 service_id = "fake_service_id"
336 e = exception.ServiceNotFound(service_id=service_id)
337 self.assertEqual(404, e.code)
338 self.assertIn(service_id, e.msg)
340 def test_host_not_found(self):
341 # Verify response code for exception.HostNotFound
342 host = "fake_host"
343 e = exception.HostNotFound(host=host)
344 self.assertEqual(404, e.code)
345 self.assertIn(host, e.msg)
347 def test_scheduler_host_filter_not_found(self):
348 # Verify response code for exception.SchedulerHostFilterNotFound
349 filter_name = "fake_filter_name"
350 e = exception.SchedulerHostFilterNotFound(filter_name=filter_name)
351 self.assertEqual(404, e.code)
352 self.assertIn(filter_name, e.msg)
354 def test_scheduler_host_weigher_not_found(self):
355 # Verify response code for exception.SchedulerHostWeigherNotFound
356 weigher_name = "fake_weigher_name"
357 e = exception.SchedulerHostWeigherNotFound(weigher_name=weigher_name)
358 self.assertEqual(404, e.code)
359 self.assertIn(weigher_name, e.msg)
361 def test_host_binary_not_found(self):
362 # Verify response code for exception.HostBinaryNotFound
363 host = "fake_host"
364 binary = "fake_binary"
365 e = exception.HostBinaryNotFound(binary=binary, host=host)
366 self.assertEqual(404, e.code)
367 self.assertIn(binary, e.msg)
368 self.assertIn(host, e.msg)
370 def test_quota_not_found(self):
371 # Verify response code for exception.QuotaNotFound
372 e = exception.QuotaNotFound()
373 self.assertEqual(404, e.code)
375 def test_quota_resource_unknown(self):
376 # Verify response code for exception.QuotaResourceUnknown
377 unknown = "fake_quota_resource"
378 e = exception.QuotaResourceUnknown(unknown=unknown)
379 self.assertEqual(404, e.code)
381 def test_project_quota_not_found(self):
382 # Verify response code for exception.ProjectQuotaNotFound
383 project_id = "fake_tenant_id"
384 e = exception.ProjectQuotaNotFound(project_id=project_id)
385 self.assertEqual(404, e.code)
387 def test_quota_class_not_found(self):
388 # Verify response code for exception.QuotaClassNotFound
389 class_name = "FakeQuotaClass"
390 e = exception.QuotaClassNotFound(class_name=class_name)
391 self.assertEqual(404, e.code)
393 def test_quota_usage_not_found(self):
394 # Verify response code for exception.QuotaUsageNotFound
395 project_id = "fake_tenant_id"
396 e = exception.QuotaUsageNotFound(project_id=project_id)
397 self.assertEqual(404, e.code)
399 def test_reservation_not_found(self):
400 # Verify response code for exception.ReservationNotFound
401 uuid = "fake_uuid"
402 e = exception.ReservationNotFound(uuid=uuid)
403 self.assertEqual(404, e.code)
405 def test_migration_not_found(self):
406 # Verify response code for exception.MigrationNotFound
407 migration_id = "fake_migration_id"
408 e = exception.MigrationNotFound(migration_id=migration_id)
409 self.assertEqual(404, e.code)
410 self.assertIn(migration_id, e.msg)
412 def test_migration_not_found_by_status(self):
413 # Verify response code for exception.MigrationNotFoundByStatus
414 status = "fake_status"
415 instance_id = "fake_instance_id"
416 e = exception.MigrationNotFoundByStatus(status=status,
417 instance_id=instance_id)
418 self.assertEqual(404, e.code)
419 self.assertIn(status, e.msg)
420 self.assertIn(instance_id, e.msg)
422 def test_file_not_found(self):
423 # Verify response code for exception.FileNotFound
424 file_path = "fake_file_path"
425 e = exception.FileNotFound(file_path=file_path)
426 self.assertEqual(404, e.code)
427 self.assertIn(file_path, e.msg)
429 def test_config_not_found(self):
430 # Verify response code for exception.ConfigNotFound
431 path = "fake_path"
432 e = exception.ConfigNotFound(path=path)
433 self.assertEqual(404, e.code)
434 self.assertIn(path, e.msg)
436 def test_paste_app_not_found(self):
437 # Verify response code for exception.PasteAppNotFound
438 name = "fake_name"
439 path = "fake_path"
440 e = exception.PasteAppNotFound(name=name, path=path)
441 self.assertEqual(404, e.code)
442 self.assertIn(name, e.msg)
443 self.assertIn(path, e.msg)
445 def test_share_snapshot_not_found(self):
446 # Verify response code for exception.ShareSnapshotNotFound
447 snapshot_id = "fake_snapshot_id"
448 e = exception.ShareSnapshotNotFound(snapshot_id=snapshot_id)
449 self.assertEqual(404, e.code)
450 self.assertIn(snapshot_id, e.msg)
452 def test_metadata_item_not_found(self):
453 # verify response code for exception.MetadataItemNotFound
454 e = exception.MetadataItemNotFound()
455 self.assertEqual(404, e.code)
457 def test_security_service_not_found(self):
458 # verify response code for exception.SecurityServiceNotFound
459 security_service_id = "fake_security_service_id"
460 e = exception.SecurityServiceNotFound(
461 security_service_id=security_service_id)
462 self.assertEqual(404, e.code)
463 self.assertIn(security_service_id, e.msg)
465 def test_volume_not_found(self):
466 # verify response code for exception.VolumeNotFound
467 volume_id = "fake_volume_id"
468 e = exception.VolumeNotFound(volume_id=volume_id)
469 self.assertEqual(404, e.code)
470 self.assertIn(volume_id, e.msg)
472 def test_volume_snapshot_not_found(self):
473 # verify response code for exception.VolumeSnapshotNotFound
474 snapshot_id = "fake_snapshot_id"
475 e = exception.VolumeSnapshotNotFound(snapshot_id=snapshot_id)
476 self.assertEqual(404, e.code)
477 self.assertIn(snapshot_id, e.msg)
479 def test_share_type_not_found(self):
480 # verify response code for exception.ShareTypeNotFound
481 share_type_id = "fake_share_type_id"
482 e = exception.ShareTypeNotFound(share_type_id=share_type_id)
483 self.assertEqual(404, e.code)
484 self.assertIn(share_type_id, e.msg)
486 def test_share_type_not_found_by_name(self):
487 # verify response code for exception.ShareTypeNotFoundByName
488 share_type_name = "fake_share_type_name"
489 e = exception.ShareTypeNotFoundByName(
490 share_type_name=share_type_name)
491 self.assertEqual(404, e.code)
492 self.assertIn(share_type_name, e.msg)
494 def test_share_type_does_not_exist(self):
495 # verify response code for exception.ShareTypeDoesNotExist
496 share_type = "fake_share_type_1234"
497 e = exception.ShareTypeDoesNotExist(share_type=share_type)
498 self.assertEqual(404, e.code)
499 self.assertIn(share_type, e.msg)
501 def test_share_type_extra_specs_not_found(self):
502 # verify response code for exception.ShareTypeExtraSpecsNotFound
503 share_type_id = "fake_share_type_id"
504 extra_specs_key = "fake_extra_specs_key"
505 e = exception.ShareTypeExtraSpecsNotFound(
506 share_type_id=share_type_id, extra_specs_key=extra_specs_key)
507 self.assertEqual(404, e.code)
508 self.assertIn(share_type_id, e.msg)
509 self.assertIn(extra_specs_key, e.msg)
511 def test_default_share_type_not_configured(self):
512 # Verify response code for exception.DefaultShareTypeNotConfigured
513 e = exception.DefaultShareTypeNotConfigured()
514 self.assertEqual(404, e.code)
516 def test_instance_not_found(self):
517 # verify response code for exception.InstanceNotFound
518 instance_id = "fake_instance_id"
519 e = exception.InstanceNotFound(instance_id=instance_id)
520 self.assertEqual(404, e.code)
521 self.assertIn(instance_id, e.msg)
523 def test_share_replica_not_found_exception(self):
524 # Verify response code for exception.ShareReplicaNotFound
525 replica_id = "FAKE_REPLICA_ID"
526 e = exception.ShareReplicaNotFound(replica_id=replica_id)
527 self.assertEqual(404, e.code)
528 self.assertIn(replica_id, e.msg)
530 def test_storage_resource_not_found(self):
531 # verify response code for exception.StorageResourceNotFound
532 name = "fake_name"
533 e = exception.StorageResourceNotFound(name=name)
534 self.assertEqual(404, e.code)
535 self.assertIn(name, e.msg)
537 def test_snapshot_resource_not_found(self):
538 # verify response code for exception.SnapshotResourceNotFound
539 name = "fake_name"
540 e = exception.SnapshotResourceNotFound(name=name)
541 self.assertEqual(404, e.code)
542 self.assertIn(name, e.msg)
544 def test_snapshot_instance_not_found(self):
545 # verify response code for exception.ShareSnapshotInstanceNotFound
546 instance_id = 'fake_instance_id'
547 e = exception.ShareSnapshotInstanceNotFound(instance_id=instance_id)
548 self.assertEqual(404, e.code)
549 self.assertIn(instance_id, e.msg)
551 def test_export_location_not_found(self):
552 # verify response code for exception.ExportLocationNotFound
553 uuid = "fake-export-location-uuid"
554 e = exception.ExportLocationNotFound(uuid=uuid)
555 self.assertEqual(404, e.code)
556 self.assertIn(uuid, e.msg)
558 def test_share_resource_not_found(self):
559 # verify response code for exception.ShareResourceNotFound
560 share_id = "fake_share_id"
561 e = exception.ShareResourceNotFound(share_id=share_id)
562 self.assertEqual(404, e.code)
563 self.assertIn(share_id, e.msg)
565 def test_share_not_found(self):
566 # verify response code for exception.ShareNotFound
567 share_id = "fake_share_id"
568 e = exception.ShareNotFound(share_id=share_id)
569 self.assertEqual(404, e.code)
570 self.assertIn(share_id, e.msg)
572 def test_resource_lock_not_found(self):
573 # verify response code for exception.ResourceLockNotFound
574 lock_id = "fake_lock_id"
575 e = exception.ResourceLockNotFound(lock_id=lock_id)
576 self.assertEqual(404, e.code)
577 self.assertIn(lock_id, e.msg)
580class ManilaExceptionResponseCode413(test.TestCase):
582 def test_quota_error(self):
583 # verify response code for exception.QuotaError
584 e = exception.QuotaError()
585 self.assertEqual(413, e.code)
587 def test_share_size_exceeds_available_quota(self):
588 # verify response code for exception.ShareSizeExceedsAvailableQuota
589 e = exception.ShareSizeExceedsAvailableQuota()
590 self.assertEqual(413, e.code)
592 def test_share_limit_exceeded(self):
593 # verify response code for exception.ShareLimitExceeded
594 allowed = 776 # amount of allowed shares
595 e = exception.ShareLimitExceeded(allowed=allowed)
596 self.assertEqual(413, e.code)
597 self.assertIn(str(allowed), e.msg)
599 def test_snapshot_limit_exceeded(self):
600 # verify response code for exception.SnapshotLimitExceeded
601 allowed = 777 # amount of allowed snapshots
602 e = exception.SnapshotLimitExceeded(allowed=allowed)
603 self.assertEqual(413, e.code)
604 self.assertIn(str(allowed), e.msg)
606 def test_share_networks_limit_exceeded(self):
607 # verify response code for exception.ShareNetworksLimitExceeded
608 allowed = 778 # amount of allowed share networks
609 e = exception.ShareNetworksLimitExceeded(allowed=allowed)
610 self.assertEqual(413, e.code)
611 self.assertIn(str(allowed), e.msg)
613 def test_port_limit_exceeded(self):
614 # verify response code for exception.PortLimitExceeded
615 e = exception.PortLimitExceeded()
616 self.assertEqual(413, e.code)
618 def test_per_share_limit_exceeded(self):
619 # verify response code for exception.ShareSizeExceedsLimit
620 size = 779 # amount of share size
621 limit = 775 # amount of allowed share size limit
622 e = exception.ShareSizeExceedsLimit(size=size, limit=limit)
623 self.assertEqual(413, e.code)