我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用django.http.HttpRequest()。
def can_vote(self, user: BitpollUser, request: HttpRequest, is_edit: bool=False) -> bool: """ Determine if the user is allowed to vote :param is_edit: if the vote is an edit :param user: :param request: :return: """ has_voted = self.has_voted(user) if self.one_vote_per_user and has_voted and not is_edit: messages.error(request, _("It is only one vote allowed. You have already voted.")) return False elif self.require_login and not user.is_authenticated: messages.error(request, _("Login required to vote.")) return False elif self.require_invitation and (not user.is_authenticated or user not in self.invitation_set.all().values('user')): messages.error(request, _("You are not allowed to vote in this poll. You have to be invited")) return False return True
def can_edit(self, user: BitpollUser, request: HttpRequest=None) -> bool: """ check if the user can edit this Poll :param user: The user to edit the Poll :param request: The request object, if this is set a error message will be emitted via the django.messages framework :return: """ has_owner = self.group or self.user is_owner = self.is_owner(user) can_edit = ((not has_owner) or is_owner) and user.is_authenticated or not has_owner if request and not can_edit: messages.error( request, _("You are not allowed to edit this Poll.") ) return can_edit
def test__determine_metadata_plain(self): _metadata = AuthorMetaData().determine_metadata(HttpRequest(), MyAPIView()) metadata = force_evaluate(_metadata) assert metadata == { 'title': 'author', 'description': 'description', 'fields': [ { 'type': 'CharField', 'name': 'name', 'verbose_name': 'name', 'help_text': '', 'blank': False, 'null': False, 'editable': True, 'max_length': 255, 'required': True }, { 'type': 'DateField', 'name': 'birth', 'verbose_name': 'birth date', 'help_text': '', 'blank': False, 'null': False, 'editable': True, 'required': True } ] }
def test_init(self): _metadata = ImpersonateMetadata().determine_metadata(HttpRequest(), MyAPIView()) metadata = force_evaluate(_metadata) assert metadata == { 'title': 'View site as another user', 'action_name': 'Impersonate', 'description': 'description', 'fields': [ { 'data': '/data/', 'name': 'user_id', 'verbose_name': 'User', 'required': True, 'help_text': '', 'blank': True, 'null': False, 'type': 'ForeignKey' } ] } # noinspection PyPep8Naming
def test__get_field_NAME(self): class CustomImpersonateMetadata(ImpersonateMetadata): def get_field_user_id(self, field_name, request): return { 'lol': 1 } _metadata = CustomImpersonateMetadata().determine_metadata(HttpRequest(), MyAPIView()) metadata = force_evaluate(_metadata) assert metadata == { 'title': 'View site as another user', 'action_name': 'Impersonate', 'description': 'description', 'fields': [ { 'type': 'ForeignKey', 'name': 'user_id', 'verbose_name': 'User', 'required': True, 'blank': True, 'null': False, 'help_text': '', 'data': '/data/', 'lol': 1 } ] }
def post(self, request: HttpRequest, *args, **kwargs) -> HttpResponseRedirect: username = request.POST.get('username') password = request.POST.get('password') user = authenticate(username=username, password=password) if user is None: messages.error(request, _('No user account matches the entered credentials.')) return redirect('common:login') if not user.is_active: messages.error(request, _('User account is deactivated.')) return redirect('common:login') login(request, user) url = urllib.parse.unquote(request.GET.get('next', '')) if url and is_safe_url(url, request.get_host()): return redirect(url) return redirect('/')
def test_delete_container_nonempty(self): container = self.containers.first() objects = self.objects.list() api.swift.swift_get_objects(IsA(http.HttpRequest), container.name).AndReturn([objects, False]) self.mox.ReplayAll() action_string = u"containers__delete__%s" % container.name form_data = {"action": action_string} req = self.factory.post(CONTAINER_INDEX_URL, form_data) req.META['HTTP_REFERER'] = '%s/%s' % (CONTAINER_INDEX_URL, container.name) table = tables.ContainersTable(req, self.containers.list()) handled = table.maybe_handle() self.assertEqual(handled.status_code, 302) self.assertEqual(six.text_type(list(req._messages)[0].message), u"The container cannot be deleted " u"since it is not empty.")
def test_create_container_post(self): for container in self.containers.list(): self.mox.ResetAll() # mandatory in a for loop api.swift.swift_create_container(IsA(http.HttpRequest), container.name, metadata=({'is_public': False})) self.mox.ReplayAll() formData = {'name': container.name, 'access': "private", 'method': forms.CreateContainer.__name__} res = self.client.post( reverse('horizon:project:containers:create'), formData) args = (utils.wrap_delimiter(container.name),) url = reverse('horizon:project:containers:index', args=args) self.assertRedirectsNoFollow(res, url)
def test_delete_pseudo_folder(self): container = self.containers.first() folder = self.folder.first() args = (utils.wrap_delimiter(container.name),) index_url = reverse('horizon:project:containers:index', args=args) api.swift.swift_delete_object(IsA(http.HttpRequest), container.name, folder.name + '/') self.mox.ReplayAll() action_string = "objects__delete_object__%s/%s" % (container.name, folder.name) form_data = {"action": action_string} req = self.factory.post(index_url, form_data) kwargs = {"container_name": container.name} table = tables.ObjectsTable(req, self.folder.list(), **kwargs) handled = table.maybe_handle() self.assertEqual(handled['location'], index_url)
def test_copy(self): container_1 = self.containers.get(name=CONTAINER_NAME_1) container_2 = self.containers.get(name=CONTAINER_NAME_2) obj = self.objects.first() ret = (self.containers.list(), False) api.swift.swift_get_containers(IsA(http.HttpRequest)).AndReturn(ret) api.swift.swift_copy_object(IsA(http.HttpRequest), container_1.name, obj.name, container_2.name, obj.name) self.mox.ReplayAll() formData = {'method': forms.CopyObject.__name__, 'new_container_name': container_2.name, 'new_object_name': obj.name, 'orig_container_name': container_1.name, 'orig_object_name': obj.name} copy_url = reverse('horizon:project:containers:object_copy', args=[container_1.name, obj.name]) res = self.client.post(copy_url, formData) args = (utils.wrap_delimiter(container_2.name),) index_url = reverse('horizon:project:containers:index', args=args) self.assertRedirectsNoFollow(res, index_url)
def test_copy_get(self): original_name = u"test folder%\u6346/test.txt" copy_name = u"test.copy.txt" container = self.containers.first() obj = self.objects.get(name=original_name) ret = (self.containers.list(), False) api.swift.swift_get_containers(IsA(http.HttpRequest)).AndReturn(ret) self.mox.ReplayAll() copy_url = reverse('horizon:project:containers:object_copy', args=[container.name, obj.name]) res = self.client.get(copy_url) # The copy's name must appear in initial data pattern = ('<input id="id_new_object_name" value="%s" ' 'name="new_object_name" type="text" ' 'class="form-control" maxlength="255" />' % copy_name) self.assertContains(res, pattern, html=True)
def test_view_container(self): for container in self.containers.list(): self.mox.ResetAll() # mandatory in a for loop api.swift.swift_get_container(IsA(http.HttpRequest), container.name, with_data=False) \ .AndReturn(container) self.mox.ReplayAll() view_url = reverse('horizon:project:containers:container_detail', args=[container.name]) res = self.client.get(view_url) self.assertTemplateUsed(res, 'project/containers/container_detail.html') self.assertContains(res, container.name, 1, 200) self.assertNotContains(res, INVALID_CONTAINER_NAME_1) self.assertNotContains(res, INVALID_CONTAINER_NAME_2)
def test_view_object(self): for container in self.containers.list(): for obj in self.objects.list(): self.mox.ResetAll() # mandatory in a for loop api.swift.swift_get_object(IsA(http.HttpRequest), container.name, obj.name, with_data=False) \ .AndReturn(obj) self.mox.ReplayAll() view_url = reverse('horizon:project:containers:object_detail', args=[container.name, obj.name]) res = self.client.get(view_url) self.assertTemplateUsed( res, 'project/containers/object_detail.html') self.assertContains(res, obj.name, 1, 200) self._test_invalid_paths(res)
def _test_new_button_disabled_when_quota_exceeded( self, expected_string, networks_quota=10, routers_quota=10, instances_quota=10): quota_data = self.quota_usages.first() quota_data['networks']['available'] = networks_quota quota_data['routers']['available'] = routers_quota quota_data['instances']['available'] = instances_quota quotas.tenant_quota_usages( IsA(http.HttpRequest)) \ .MultipleTimes().AndReturn(quota_data) self.mox.ReplayAll() res = self.client.get(INDEX_URL) self.assertTemplateUsed(res, 'project/network_topology/index.html') self.assertContains(res, expected_string, html=True, msg_prefix="The create button is not disabled")
def test_edit_attachments_auto_device_name(self): volume = self.cinder_volumes.first() servers = [s for s in self.servers.list() if s.tenant_id == self.request.user.tenant_id] volume.attachments = [{'id': volume.id, 'volume_id': volume.id, 'volume_name': volume.name, 'instance': servers[0], 'device': '', 'server_id': servers[0].id}] cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) api.nova.server_list(IsA(http.HttpRequest)).AndReturn([servers, False]) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:attach', args=[volume.id]) res = self.client.get(url) form = res.context['form'] self.assertIsInstance(form.fields['device'].widget, widgets.TextInput) self.assertFalse(form.fields['device'].required)
def test_edit_attachments_cannot_set_mount_point(self): volume = self.cinder_volumes.first() servers = [s for s in self.servers.list() if s.tenant_id == self.request.user.tenant_id] cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) api.nova.server_list(IsA(http.HttpRequest)).AndReturn([servers, False]) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:attach', args=[volume.id]) res = self.client.get(url) # Assert the device field is hidden. form = res.context['form'] self.assertIsInstance(form.fields['device'].widget, widgets.HiddenInput)
def test_edit_attachments_attached_volume(self): servers = [s for s in self.servers.list() if s.tenant_id == self.request.user.tenant_id] server = servers[0] volume = self.cinder_volumes.list()[0] cinder.volume_get(IsA(http.HttpRequest), volume.id) \ .AndReturn(volume) api.nova.server_list(IsA(http.HttpRequest)) \ .AndReturn([servers, False]) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:attach', args=[volume.id]) res = self.client.get(url) self.assertEqual(res.context['form'].fields['instance']._choices[0][1], "Select an instance") self.assertEqual(len(res.context['form'].fields['instance'].choices), 2) self.assertEqual(res.context['form'].fields['instance']._choices[1][0], server.id) self.assertEqual(res.status_code, 200)
def test_create_snapshot_button_attributes(self): limits = {'maxTotalSnapshots': 2} limits['totalSnapshotsUsed'] = 1 volume = self.cinder_volumes.first() cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.tenant_absolute_limits(IsA(http.HttpRequest)).AndReturn(limits) self.mox.ReplayAll() res_url = (VOLUME_INDEX_URL + "?action=row_update&table=volumes&obj_id=" + volume.id) res = self.client.get(res_url, {}, HTTP_X_REQUESTED_WITH='XMLHttpRequest') snapshot_action = self._get_volume_row_action_from_ajax( res, 'snapshots', volume.id) self.assertEqual('horizon:project:volumes:volumes:create_snapshot', snapshot_action.url) self.assertEqual(set(['ajax-modal']), set(snapshot_action.classes)) self.assertEqual('Create Snapshot', six.text_type(snapshot_action.verbose_name)) self.assertEqual((('volume', 'volume:create_snapshot'),), snapshot_action.policy_rules)
def test_detail_view(self): volume = self.cinder_volumes.first() server = self.servers.first() snapshots = self.cinder_volume_snapshots.list() volume.attachments = [{"server_id": server.id}] cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.volume_snapshot_list(IsA(http.HttpRequest), search_opts={'volume_id': volume.id})\ .AndReturn(snapshots) api.nova.server_get(IsA(http.HttpRequest), server.id).AndReturn(server) cinder.tenant_absolute_limits(IsA(http.HttpRequest))\ .AndReturn(self.cinder_limits['absolute']) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:detail', args=[volume.id]) res = self.client.get(url) self.assertTemplateUsed(res, 'horizon/common/_detail.html') self.assertEqual(res.context['volume'].id, volume.id) self.assertNoMessages()
def test_encryption_detail_view_encrypted(self): enc_meta = self.cinder_volume_encryption.first() volume = self.cinder_volumes.get(name='my_volume2') cinder.volume_get_encryption_metadata( IsA(http.HttpRequest), volume.id).AndReturn(enc_meta) cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:encryption_detail', args=[volume.id]) res = self.client.get(url) self.assertContains(res, "Volume Encryption Details: %s" % volume.name, 1, 200) self.assertContains(res, "<dd>%s</dd>" % volume.volume_type, 1, 200) self.assertContains(res, "<dd>%s</dd>" % enc_meta.provider, 1, 200) self.assertContains(res, "<dd>%s</dd>" % enc_meta.control_location, 1, 200) self.assertContains(res, "<dd>%s</dd>" % enc_meta.cipher, 1, 200) self.assertContains(res, "<dd>%s</dd>" % enc_meta.key_size, 1, 200) self.assertNoMessages()
def test_encryption_detail_view_unencrypted(self): enc_meta = self.cinder_volume_encryption.list()[1] volume = self.cinder_volumes.get(name='my_volume2') cinder.volume_get_encryption_metadata( IsA(http.HttpRequest), volume.id).AndReturn(enc_meta) cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:encryption_detail', args=[volume.id]) res = self.client.get(url) self.assertContains(res, "Volume Encryption Details: %s" % volume.name, 1, 200) self.assertContains(res, "<h3>Volume is Unencrypted</h3>", 1, 200) self.assertNoMessages()
def test_detail_view_with_exception(self): volume = self.cinder_volumes.first() server = self.servers.first() volume.attachments = [{"server_id": server.id}] cinder.volume_get(IsA(http.HttpRequest), volume.id).\ AndRaise(self.exceptions.cinder) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:detail', args=[volume.id]) res = self.client.get(url) self.assertRedirectsNoFollow(res, VOLUME_INDEX_URL)
def test_update_volume(self): volume = self.cinder_volumes.get(name="my_volume") cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.volume_update(IsA(http.HttpRequest), volume.id, volume.name, volume.description) cinder.volume_set_bootable(IsA(http.HttpRequest), volume.id, False) self.mox.ReplayAll() formData = {'method': 'UpdateForm', 'name': volume.name, 'description': volume.description, 'bootable': False} url = reverse('horizon:project:volumes:volumes:update', args=[volume.id]) res = self.client.post(url, formData) self.assertRedirectsNoFollow(res, VOLUME_INDEX_URL)
def test_update_volume_without_name(self): volume = self.cinder_volumes.get(name="my_volume") cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.volume_update(IsA(http.HttpRequest), volume.id, '', volume.description) cinder.volume_set_bootable(IsA(http.HttpRequest), volume.id, False) self.mox.ReplayAll() formData = {'method': 'UpdateForm', 'name': '', 'description': volume.description, 'bootable': False} url = reverse('horizon:project:volumes:volumes:update', args=[volume.id]) res = self.client.post(url, formData) self.assertRedirectsNoFollow(res, VOLUME_INDEX_URL)
def test_update_volume_bootable_flag(self): volume = self.cinder_bootable_volumes.get(name="my_volume") cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.volume_update(IsA(http.HttpRequest), volume.id, volume.name, 'update bootable flag') cinder.volume_set_bootable(IsA(http.HttpRequest), volume.id, True) self.mox.ReplayAll() formData = {'method': 'UpdateForm', 'name': volume.name, 'description': 'update bootable flag', 'bootable': True} url = reverse('horizon:project:volumes:volumes:update', args=[volume.id]) res = self.client.post(url, formData) self.assertRedirectsNoFollow(res, VOLUME_INDEX_URL)
def test_extend_volume_with_wrong_size(self): volume = self.cinder_volumes.first() formData = {'name': u'A Volume I Am Making', 'orig_size': volume.size, 'new_size': 10} cinder.volume_get(IsA(http.HttpRequest), volume.id).\ AndReturn(self.cinder_volumes.first()) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:extend', args=[volume.id]) res = self.client.post(url, formData) self.assertFormErrors(res, 1, "New size must be greater than " "current size.")
def test_retype_volume_supported_action_item(self): volume = self.cinder_volumes.get(name='v2_volume') limits = self.cinder_limits['absolute'] cinder.volume_get(IsA(http.HttpRequest), volume.id).AndReturn(volume) cinder.tenant_absolute_limits(IsA(http.HttpRequest))\ .MultipleTimes('limits').AndReturn(limits) self.mox.ReplayAll() url = (VOLUME_INDEX_URL + "?action=row_update&table=volumes&obj_id=" + volume.id) res = self.client.get(url, {}, HTTP_X_REQUESTED_WITH='XMLHttpRequest') self.assertEqual(res.status_code, 200) self.assertContains(res, 'Change Volume Type') self.assertContains(res, 'retype')
def test_extend_volume_with_size_out_of_quota(self): volume = self.volumes.first() usage_limit = {'maxTotalVolumeGigabytes': 100, 'gigabytesUsed': 20, 'volumesUsed': len(self.volumes.list()), 'maxTotalVolumes': 6} formData = {'name': u'A Volume I Am Making', 'orig_size': volume.size, 'new_size': 1000} quotas.tenant_limit_usages(IsA(http.HttpRequest)).\ AndReturn(usage_limit) cinder.volume_get(IsA(http.HttpRequest), volume.id).\ AndReturn(self.volumes.first()) self.mox.ReplayAll() url = reverse('horizon:project:volumes:volumes:extend', args=[volume.id]) res = self.client.post(url, formData) self.assertFormError(res, "form", "new_size", "Volume cannot be extended to 1000GiB as you " "only have 80GiB of your quota available.")
def test_create_transfer(self): volumes = self.volumes.list() volToTransfer = [v for v in volumes if v.status == 'available'][0] formData = {'volume_id': volToTransfer.id, 'name': u'any transfer name'} cinder.transfer_create(IsA(http.HttpRequest), formData['volume_id'], formData['name']).AndReturn( self.cinder_volume_transfers.first()) self.mox.ReplayAll() # Create a transfer for the first available volume url = reverse('horizon:project:volumes:volumes:create_transfer', args=[volToTransfer.id]) res = self.client.post(url, formData) self.assertNoFormErrors(res)
def test_create_snapshot_get(self): volume = self.cinder_volumes.first() cinder.volume_get(IsA(http.HttpRequest), volume.id) \ .AndReturn(volume) snapshot_used = len(self.cinder_volume_snapshots.list()) usage_limit = {'maxTotalVolumeGigabytes': 250, 'gigabytesUsed': 20, 'snapshotsUsed': snapshot_used, 'maxTotalSnapshots': 6} quotas.tenant_limit_usages(IsA(http.HttpRequest)).\ AndReturn(usage_limit) self.mox.ReplayAll() url = reverse('horizon:project:volumes:' 'volumes:create_snapshot', args=[volume.id]) res = self.client.get(url) self.assertTemplateUsed(res, 'project/volumes/volumes/' 'create_snapshot.html')
def test_create_snapshot_post(self): volume = self.cinder_volumes.first() snapshot = self.cinder_volume_snapshots.first() cinder.volume_get(IsA(http.HttpRequest), volume.id) \ .AndReturn(volume) cinder.volume_snapshot_create(IsA(http.HttpRequest), volume.id, snapshot.name, snapshot.description, force=False) \ .AndReturn(snapshot) self.mox.ReplayAll() formData = {'method': 'CreateSnapshotForm', 'tenant_id': self.tenant.id, 'volume_id': volume.id, 'name': snapshot.name, 'description': snapshot.description} url = reverse('horizon:project:volumes:volumes:create_snapshot', args=[volume.id]) res = self.client.post(url, formData) self.assertRedirectsNoFollow(res, VOLUME_SNAPSHOTS_TAB_URL)
def test_delete_volume_snapshot(self): vol_snapshots = self.cinder_volume_snapshots.list() volumes = self.cinder_volumes.list() snapshot = self.cinder_volume_snapshots.first() api.cinder.volume_backup_supported(IsA(http.HttpRequest)). \ MultipleTimes().AndReturn(True) api.cinder.volume_snapshot_list_paged( IsA(http.HttpRequest), paginate=True, marker=None, sort_dir='desc').AndReturn([vol_snapshots, False, False]) api.cinder.volume_list(IsA(http.HttpRequest)). \ AndReturn(volumes) api.cinder.volume_snapshot_delete(IsA(http.HttpRequest), snapshot.id) self.mox.ReplayAll() formData = {'action': 'volume_snapshots__delete__%s' % snapshot.id} res = self.client.post(VOLUME_SNAPSHOTS_TAB_URL, formData) self.assertRedirectsNoFollow(res, VOLUME_SNAPSHOTS_TAB_URL) self.assertMessageCount(success=1)
def test_volume_snapshot_detail_get(self): volume = self.cinder_volumes.first() snapshot = self.cinder_volume_snapshots.first() api.cinder.volume_get(IsA(http.HttpRequest), volume.id). \ AndReturn(volume) api.cinder.volume_snapshot_get(IsA(http.HttpRequest), snapshot.id). \ AndReturn(snapshot) self.mox.ReplayAll() url = reverse('horizon:project:volumes:snapshots:detail', args=[snapshot.id]) res = self.client.get(url) self.assertTemplateUsed(res, 'horizon/common/_detail.html') self.assertEqual(res.context['snapshot'].id, snapshot.id)
def test_volume_snapshot_detail_with_volume_get_exception(self): # Test to verify redirect if get volume fails volume = self.cinder_volumes.first() snapshot = self.cinder_volume_snapshots.first() api.cinder.volume_get(IsA(http.HttpRequest), volume.id). \ AndRaise(self.exceptions.cinder) api.cinder.volume_snapshot_get(IsA(http.HttpRequest), snapshot.id). \ AndReturn(snapshot) self.mox.ReplayAll() url = reverse('horizon:project:volumes:snapshots:detail', args=[snapshot.id]) res = self.client.get(url) self.assertRedirectsNoFollow(res, INDEX_URL)
def test_update_snapshot(self): snapshot = self.cinder_volume_snapshots.first() cinder.volume_snapshot_get(IsA(http.HttpRequest), snapshot.id) \ .AndReturn(snapshot) cinder.volume_snapshot_update(IsA(http.HttpRequest), snapshot.id, snapshot.name, snapshot.description) \ .AndReturn(snapshot) self.mox.ReplayAll() formData = {'method': 'UpdateSnapshotForm', 'name': snapshot.name, 'description': snapshot.description} url = reverse(('horizon:project:volumes:snapshots:update'), args=[snapshot.id]) res = self.client.post(url, formData) self.assertRedirectsNoFollow(res, INDEX_URL)
def test_delete_cgroup(self): cgroups = self.cinder_consistencygroups.list() cgroup = self.cinder_consistencygroups.first() cinder.volume_cgroup_list_with_vol_type_names(IsA(http.HttpRequest)).\ AndReturn(cgroups) cinder.volume_cgroup_delete(IsA(http.HttpRequest), cgroup.id, force=False) if django.VERSION < (1, 9): cinder.volume_cgroup_list_with_vol_type_names( IsA(http.HttpRequest)).AndReturn(cgroups) self.mox.ReplayAll() formData = {'action': 'volume_cgroups__deletecg__%s' % cgroup.id} res = self.client.post(VOLUME_CGROUPS_TAB_URL, formData, follow=True) self.assertIn("Scheduled deletion of Consistency Group: cg_1", [m.message for m in res.context['messages']])
def test_update_cgroup_add_vol(self): cgroup = self.cinder_consistencygroups.first() volume = self.cinder_volumes.first() formData = {'volume_types': '1', 'name': 'test CG', 'description': 'test desc'} cinder.volume_cgroup_get(IsA( http.HttpRequest), cgroup.id).\ AndReturn(cgroup) cinder.volume_cgroup_update( IsA(http.HttpRequest), formData['name'], formData['description'], add_vols=volume)\ .AndReturn(cgroup) self.mox.ReplayAll() url = reverse('horizon:project:volumes:cgroups:update', args=[cgroup.id]) res = self.client.post(url, formData) self.assertNoFormErrors(res)
def test_update_cgroup_remove_vol(self): cgroup = self.cinder_consistencygroups.first() volume = self.cinder_volumes.first() formData = {'volume_types': '1', 'name': 'test CG', 'description': 'test desc'} cinder.volume_cgroup_get(IsA( http.HttpRequest), cgroup.id).\ AndReturn(cgroup) cinder.volume_cgroup_update( IsA(http.HttpRequest), formData['name'], formData['description'], remove_vols=volume)\ .AndReturn(cgroup) self.mox.ReplayAll() url = reverse('horizon:project:volumes:cgroups:update', args=[cgroup.id]) res = self.client.post(url, formData) self.assertNoFormErrors(res)
def test_update_cgroup_name_and_description(self): cgroup = self.cinder_consistencygroups.first() formData = {'volume_types': '1', 'name': 'test CG-new', 'description': 'test desc-new'} cinder.volume_cgroup_get(IsA( http.HttpRequest), cgroup.id).\ AndReturn(cgroup) cinder.volume_cgroup_update( IsA(http.HttpRequest), formData['name'], formData['description'])\ .AndReturn(cgroup) self.mox.ReplayAll() url = reverse('horizon:project:volumes:cgroups:update', args=[cgroup.id]) res = self.client.post(url, formData) self.assertNoFormErrors(res)
def test_create_backup_post(self): volume = self.volumes.first() backup = self.cinder_volume_backups.first() api.cinder.volume_backup_create(IsA(http.HttpRequest), volume.id, backup.container_name, backup.name, backup.description) \ .AndReturn(backup) self.mox.ReplayAll() formData = {'method': 'CreateBackupForm', 'tenant_id': self.tenant.id, 'volume_id': volume.id, 'container_name': backup.container_name, 'name': backup.name, 'description': backup.description} url = reverse('horizon:project:volumes:volumes:create_backup', args=[volume.id]) res = self.client.post(url, formData) self.assertNoFormErrors(res) self.assertMessageCount(error=0, warning=0) self.assertRedirectsNoFollow(res, VOLUME_BACKUPS_TAB_URL)
def test_delete_volume_backup(self): vol_backups = self.cinder_volume_backups.list() volumes = self.cinder_volumes.list() backup = self.cinder_volume_backups.first() api.cinder.volume_backup_supported(IsA(http.HttpRequest)). \ MultipleTimes().AndReturn(True) api.cinder.volume_backup_list_paged( IsA(http.HttpRequest), marker=None, sort_dir='desc', paginate=True).AndReturn([vol_backups, False, False]) api.cinder.volume_list(IsA(http.HttpRequest)). \ AndReturn(volumes) api.cinder.volume_backup_delete(IsA(http.HttpRequest), backup.id) self.mox.ReplayAll() formData = {'action': 'volume_backups__delete__%s' % backup.id} res = self.client.post(INDEX_URL + "?tab=volumes_and_snapshots__backups_tab", formData) self.assertRedirectsNoFollow(res, INDEX_URL + "?tab=volumes_and_snapshots__backups_tab") self.assertMessageCount(success=1)
def test_volume_backup_detail_get(self): backup = self.cinder_volume_backups.first() volume = self.cinder_volumes.get(id=backup.volume_id) api.cinder.volume_backup_get(IsA(http.HttpRequest), backup.id). \ AndReturn(backup) api.cinder.volume_get(IsA(http.HttpRequest), backup.volume_id). \ AndReturn(volume) self.mox.ReplayAll() url = reverse('horizon:project:volumes:backups:detail', args=[backup.id]) res = self.client.get(url) self.assertTemplateUsed(res, 'horizon/common/_detail.html') self.assertEqual(res.context['backup'].id, backup.id)
def test_restore_backup(self): backup = self.cinder_volume_backups.first() volumes = self.cinder_volumes.list() api.cinder.volume_list(IsA(http.HttpRequest)). \ AndReturn(volumes) api.cinder.volume_backup_restore(IsA(http.HttpRequest), backup.id, backup.volume_id). \ AndReturn(backup) self.mox.ReplayAll() formData = {'method': 'RestoreBackupForm', 'backup_id': backup.id, 'backup_name': backup.name, 'volume_id': backup.volume_id} url = reverse('horizon:project:volumes:backups:restore', args=[backup.id]) url += '?%s' % urlencode({'backup_name': backup.name, 'volume_id': backup.volume_id}) res = self.client.post(url, formData) self.assertNoFormErrors(res) self.assertMessageCount(info=1) self.assertRedirectsNoFollow(res, INDEX_URL)
def _test_index_paginated(self, marker, sort_dir, volumes, url, has_more, has_prev): backup_supported = True vol_snaps = self.cinder_volume_snapshots.list() api.cinder.volume_backup_supported(IsA(http.HttpRequest)).\ MultipleTimes().AndReturn(backup_supported) api.cinder.volume_list_paged(IsA(http.HttpRequest), marker=marker, sort_dir=sort_dir, search_opts=None, paginate=True).\ AndReturn([volumes, has_more, has_prev]) api.cinder.volume_snapshot_list( IsA(http.HttpRequest), search_opts=None).AndReturn(vol_snaps) api.nova.server_list(IsA(http.HttpRequest), search_opts=None).\ AndReturn([self.servers.list(), False]) api.cinder.tenant_absolute_limits(IsA(http.HttpRequest)).MultipleTimes().\ AndReturn(self.cinder_limits['absolute']) self.mox.ReplayAll() res = self.client.get(urlunquote(url)) self.assertEqual(res.status_code, 200) self.assertTemplateUsed(res, 'project/volumes/index.html') self.mox.UnsetStubs() return res
def _test_snapshots_index_paginated(self, marker, sort_dir, snapshots, url, has_more, has_prev): backup_supported = True api.cinder.volume_backup_supported(IsA(http.HttpRequest)).\ MultipleTimes().AndReturn(backup_supported) api.cinder.volume_snapshot_list_paged( IsA(http.HttpRequest), marker=marker, sort_dir=sort_dir, paginate=True).AndReturn([snapshots, has_more, has_prev]) api.cinder.volume_list(IsA(http.HttpRequest)).AndReturn( self.cinder_volumes.list()) self.mox.ReplayAll() res = self.client.get(urlunquote(url)) self.assertEqual(res.status_code, 200) self.assertTemplateUsed(res, 'project/volumes/index.html') self.mox.UnsetStubs() return res
def _test_backups_index_paginated(self, marker, sort_dir, backups, url, has_more, has_prev): backup_supported = True api.cinder.volume_backup_supported(IsA(http.HttpRequest)).\ MultipleTimes().AndReturn(backup_supported) api.cinder.volume_backup_list_paged( IsA(http.HttpRequest), marker=marker, sort_dir=sort_dir, paginate=True).AndReturn([backups, has_more, has_prev]) api.cinder.volume_list(IsA(http.HttpRequest)).AndReturn( self.cinder_volumes.list()) self.mox.ReplayAll() res = self.client.get(urlunquote(url)) self.assertEqual(res.status_code, 200) self.assertTemplateUsed(res, 'project/volumes/index.html') self.mox.UnsetStubs() return res
def test_add_vpnservice_get(self): networks = [{'subnets': [self.subnets.first(), ]}, ] routers = self.routers.list() api.neutron.network_list_for_tenant( IsA(http.HttpRequest), self.tenant.id).AndReturn(networks) api.neutron.router_list( IsA(http.HttpRequest), tenant_id=self.tenant.id).AndReturn(routers) self.mox.ReplayAll() res = self.client.get(reverse(self.ADDVPNSERVICE_PATH)) workflow = res.context['workflow'] self.assertTemplateUsed(res, views.WorkflowView.template_name) self.assertEqual(workflow.name, workflows.AddVPNService.name) expected_objs = ['<AddVPNServiceStep: addvpnserviceaction>', ] self.assertQuerysetEqual(workflow.steps, expected_objs)
def test_add_vpnservice_post(self): vpnservice = self.vpnservices.first() networks = [{'subnets': [self.subnets.first(), ]}, ] routers = self.routers.list() api.neutron.network_list_for_tenant( IsA(http.HttpRequest), self.tenant.id).AndReturn(networks) api.neutron.router_list( IsA(http.HttpRequest), tenant_id=self.tenant.id).AndReturn(routers) form_data = {'name': vpnservice['name'], 'description': vpnservice['description'], 'subnet_id': vpnservice['subnet_id'], 'router_id': vpnservice['router_id'], 'admin_state_up': vpnservice['admin_state_up']} api.vpn.vpnservice_create( IsA(http.HttpRequest), **form_data).AndReturn(vpnservice) self.mox.ReplayAll() res = self.client.post(reverse(self.ADDVPNSERVICE_PATH), form_data) self.assertNoFormErrors(res) self.assertRedirectsNoFollow(res, str(self.INDEX_URL))
def test_add_vpnservice_post_error(self): vpnservice = self.vpnservices.first() networks = [{'subnets': [self.subnets.first(), ]}, ] routers = self.routers.list() api.neutron.network_list_for_tenant( IsA(http.HttpRequest), self.tenant.id).AndReturn(networks) api.neutron.router_list( IsA(http.HttpRequest), tenant_id=self.tenant.id).AndReturn(routers) self.mox.ReplayAll() form_data = {'name': vpnservice['name'], 'description': vpnservice['description'], 'subnet_id': '', 'router_id': '', 'admin_state_up': vpnservice['admin_state_up']} res = self.client.post(reverse(self.ADDVPNSERVICE_PATH), form_data) self.assertFormErrors(res, 2)
def test_add_ikepolicy_post(self): ikepolicy = self.ikepolicies.first() form_data = {'name': ikepolicy['name'], 'description': ikepolicy['description'], 'auth_algorithm': ikepolicy['auth_algorithm'], 'encryption_algorithm': ikepolicy[ 'encryption_algorithm'], 'ike_version': ikepolicy['ike_version'], 'lifetime_units': ikepolicy['lifetime']['units'], 'lifetime_value': ikepolicy['lifetime']['value'], 'phase1_negotiation_mode': ikepolicy[ 'phase1_negotiation_mode'], 'pfs': ikepolicy['pfs']} api.vpn.ikepolicy_create( IsA(http.HttpRequest), **form_data).AndReturn(ikepolicy) self.mox.ReplayAll() res = self.client.post(reverse(self.ADDIKEPOLICY_PATH), form_data) self.assertNoFormErrors(res) self.assertRedirectsNoFollow(res, str(self.INDEX_URL))