Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_resource_permissions(self):
"""
Tests that the client can get and set group permissions through the test_resource_permissions view.
"""
self.assertTrue(self.client.login(username="admin", password="admin"))
layer = Layer.objects.all()[0]
document = Document.objects.all()[0]
map_obj = Map.objects.all()[0]
layer.set_default_permissions()
document.set_default_permissions()
map_obj.set_default_permissions()
objects = layer, document, map_obj
for obj in objects:
response = self.client.get(
reverse(
'resource_permissions',
kwargs=dict(
resource_id=obj.id)))
self.assertEqual(response.status_code, 200)
js = json.loads(response.content)
permissions = js.get('permissions', dict())
def bulk_favorite_objects(self, user):
'get the actual favorite objects for a user as a dict by content_type'
favs = {}
for m in (Document, Map, Layer, get_user_model()):
ct = ContentType.objects.get_for_model(m)
f = self.favorites_for_user(user).filter(content_type=ct)
favs[ct.name] = m.objects.filter(id__in=f.values('object_id'))
return favs
if q is None: return None
if query.kw:
# hard to handle - not supporting at the moment
return Profile.objects.none()
if query.owner:
q = q.filter(user__username__icontains = query.owner)
if query.extent:
q = filter_by_extent(Map, q, query.extent, True) | \
filter_by_extent(Layer, q, query.extent, True)
if query.period:
q = filter_by_period(Map, q, *query.period, user=True) | \
filter_by_period(Layer, q, *query.period, user=True)
if query.added:
q = q.filter(user__date_joined__gt = query.added)
if query.query:
qs = Q(user__username__icontains=query.query) | \
Q(user__first_name__icontains=query.query) | \
Q(user__last_name__icontains=query.query)
for field in extension.owner_query_fields:
qs = qs | Q(**{'%s__icontains' % field: query.query})
q = q.filter(qs)
rules = _rank_rules(User,['username', 10, 5]) + \
_rank_rules(Profile,['organization', 5, 2])
added = extension.owner_rank_rules()
counts = layers.values('storeType').annotate(count=Count('storeType'))
count_dict = dict([(count['storeType'], count['count'])
for count in counts])
facets = {
'raster': count_dict.get('coverageStore', 0),
'vector': count_dict.get('dataStore', 0),
'remote': count_dict.get('remoteStore', 0),
}
# Break early if only_layers is set.
if facet_type == 'layers':
return facets
maps = Map.objects.filter(title__icontains=title_filter)
documents = Document.objects.filter(title__icontains=title_filter)
if not settings.SKIP_PERMS_FILTER:
maps = maps.filter(id__in=authorized)
documents = documents.filter(id__in=authorized)
facets['map'] = maps.count()
facets['document'] = documents.count()
if facet_type == 'home':
facets['user'] = get_user_model().objects.exclude(
username='AnonymousUser').count()
facets['app'] = App.objects.count()
facets['group'] = GroupProfile.objects.exclude(
access="private").count()
facets['layer'] = facets['raster'] + \
signals.pre_delete.connect(
qgis_server_pre_delete,
dispatch_uid='Layer-qgis_server_pre_delete',
sender=Layer)
signals.post_save.connect(
qgis_server_post_save,
dispatch_uid='Layer-qgis_server_post_save',
sender=Layer)
signals.pre_save.connect(
qgis_server_pre_save_maplayer,
dispatch_uid='MapLayer-qgis_server_pre_save_maplayer',
sender=MapLayer)
signals.post_save.connect(
qgis_server_post_save_map,
dispatch_uid='Map-qgis_server_post_save_map',
sender=Map)
signals.post_delete.connect(
qgis_server_layer_post_delete,
dispatch_uid='QGISServerLayer-qgis_server_layer_post_delete',
sender=QGISServerLayer)
def _by_storyteller_pager(req, user, what):
if what == 'maps':
query = models.PublishingStatus.objects.get_public(user, Map)
exclude = req.GET.get('exclude', None) if req else None
if exclude:
query = query.exclude(id=exclude)
elif what == 'layers':
query = models.PublishingStatus.objects.get_public(user, Layer)
for e in settings.LAYER_EXCLUSIONS:
query = query.exclude(name__regex=e)
else:
return HttpResponse(status=400)
return Paginator(query, 5)
def validate_wm_map(sender, instance, **kwargs):
"""
Validate a map instance.
"""
if instance.bbox_y0 > instance.bbox_y1:
y0 = instance.bbox_y1
y1 = instance.bbox_y0
instance.bbox_y0 = y0
instance.bbox_y1 = y1
post_save.connect(save_profile, sender=Profile)
post_save.connect(add_ext_layer, sender=Layer)
post_save.connect(add_ext_map, sender=Map)
post_save.connect(validate_wm_map, sender=Map)
created_verb=_("added a comment"),
target=getattr(instance, 'content_object', None),
updated_verb=_("updated a comment"),
)
action_settings['layer'].update(created_verb=_('uploaded'))
action = action_settings[obj_type]
if created:
# object was created
verb = action.get('created_verb')
raw_action = 'created'
else:
if created is False:
# object was saved.
if not isinstance(instance, Layer) and not isinstance(instance, Map):
verb = action.get('updated_verb')
raw_action = 'updated'
if created is None:
# object was deleted.
verb = action.get('deleted_verb')
raw_action = 'deleted'
action.update(action_object=None,
target=None)
if verb:
try:
activity.send(action.get('actor'),
verb=u"{verb}".format(verb=verb),
action_object=action.get('action_object'),
target=action.get('target', None),
def _create_viewer_config():
DEFAULT_MAP_CONFIG, DEFAULT_BASE_LAYERS = default_map_config()
_map = Map(projection="EPSG:900913", zoom = 1, center_x = 0, center_y = 0)
return json.dumps(_map.viewer_json(*DEFAULT_BASE_LAYERS))
_viewer_config = _create_viewer_config()