Python django.http 模块,HttpRequest() 实例源码

我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用django.http.HttpRequest()

项目:Bitpoll    作者:fsinfuhh    | 项目源码 | 文件源码
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
项目:Bitpoll    作者:fsinfuhh    | 项目源码 | 文件源码
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
项目:drf-metadata    作者:night-crawler    | 项目源码 | 文件源码
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
                }
            ]
        }
项目:drf-metadata    作者:night-crawler    | 项目源码 | 文件源码
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
项目:drf-metadata    作者:night-crawler    | 项目源码 | 文件源码
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
                }
            ]
        }
项目:byro    作者:byro    | 项目源码 | 文件源码
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('/')
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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.")
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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")
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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()
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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()
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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()
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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.")
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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')
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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.")
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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')
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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']])
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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))
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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)
项目:mos-horizon    作者:Mirantis    | 项目源码 | 文件源码
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))