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
python类HttpRequest()的实例源码
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')
