Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def dump_models(path=None):
result = serialize("json", sum([list(x) for x in
[get_user_model().objects.all(),
Profile.objects.all(),
Layer.objects.all(),
Map.objects.all(),
Document.objects.all(),
Tag.objects.all(),
TaggedItem.objects.all(),
]], []), indent=2, use_natural_keys=True)
if path is None:
parent, _ = os.path.split(__file__)
path = os.path.join(parent, 'fixtures', 'search_testdata.json')
with open(path, 'w') as f:
f.write(result)
def test_batch_edit(self):
Model = Document
view = 'document_batch_metadata'
resources = Model.objects.all()[:3]
ids = ','.join([str(element.pk) for element in resources])
# test non-admin access
self.client.login(username="bobby", password="bob")
response = self.client.get(reverse(view, args=(ids,)))
self.assertEquals(response.status_code, 401)
# test group change
group = Group.objects.first()
self.client.login(username='admin', password='admin')
response = self.client.post(
reverse(view, args=(ids,)),
data={'group': group.pk},
)
self.assertEquals(response.status_code, 302)
resources = Model.objects.filter(id__in=[r.pk for r in resources])
def dump_models(path=None):
result = serialize("json", sum([list(x) for x in
[get_user_model().objects.all(),
Profile.objects.all(),
Layer.objects.all(),
Map.objects.all(),
Document.objects.all(),
Tag.objects.all(),
TaggedItem.objects.all(),
]], []), indent=2, use_natural_keys=True)
if path is None:
parent, _ = os.path.split(__file__)
path = os.path.join(parent, 'fixtures', 'search_testdata.json')
with open(path, 'w') as f:
f.write(result)
def setUp(self):
if not os.path.exists(os.path.join(os.path.split(__file__)[0], 'test_ogr')):
self.skipTest('Skipping test due to missing test data.')
# These tests require geonode to be running on :80!
self.postgis = db.connections['datastore']
self.postgis_settings = self.postgis.settings_dict
self.username, self.password = self.create_user('admin', 'admin', is_superuser=True)
self.non_admin_username, self.non_admin_password = self.create_user('non_admin', 'non_admin')
self.cat = Catalog(ogc_server_settings.internal_rest, *ogc_server_settings.credentials)
self.workspace = 'geonode'
self.datastore = self.create_datastore(self.postgis, self.cat)
def test_map_json(self):
map_obj = Map.objects.all().first()
map_id = map_obj.id
# Test that saving a map when not logged in gives 401
response = self.client.put(
reverse(
'map_json',
args=(
str(map_id),
)),
data=self.viewer_config,
content_type="text/json")
self.assertEqual(response.status_code, 401)
self.client.login(username=self.user, password=self.passwd)
response = self.client.put(
reverse(
'map_json',
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 test_assign_remove_permissions(self):
# Assing
layer = Layer.objects.all().first()
perm_spec = layer.get_all_level_info()
self.assertNotIn(self.user, perm_spec["users"])
utils.set_layers_permissions("write", None, [self.username], None, None)
layer_after = Layer.objects.get(name=layer.name)
perm_spec = layer_after.get_all_level_info()
for perm in utils.WRITE_PERMISSIONS:
self.assertIn(perm, perm_spec["users"][self.user])
# Remove
utils.set_layers_permissions("write", None, [self.username], None, True)
layer_after = Layer.objects.get(name=layer.name)
perm_spec = layer_after.get_all_level_info()
for perm in utils.WRITE_PERMISSIONS:
self.assertNotIn(perm, perm_spec["users"][self.user])
def test_layer_detail_page_slave_site(self):
"""
Test that the detail page is not found of the resource is on another site
"""
# test that the CA layer detail page, that does not belong to the SlaveSite, is not found
self.client.login(username=self.user, password=self.passwd)
response = self.client.get(reverse('layer_detail', args=[Layer.objects.all()[0].alternate]))
self.assertEqual(response.status_code, 404)
def test_sync_resources_with_guardian_delay_true(self):
with self.settings(DELAYED_SECURITY_SIGNALS=True):
# Set geofence (and so the dirty state)
set_geofence_all(self._l)
# Retrieve the same layer
dirty_layer = Layer.objects.get(pk=self._l.id)
# Check dirty state (True)
self.assertTrue(dirty_layer.dirty_state)
# Call sync resources
sync_resources_with_guardian()
clean_layer = Layer.objects.get(pk=self._l.id)
# Check dirty state
self.assertFalse(clean_layer.dirty_state)
def test_download_url(self):
layer = Layer.objects.all().first()
self.client.login(username='admin', password='admin')
# ... all should be good
response = self.client.get(reverse('download', args=(layer.id,)))
# Espected 404 since there are no files available for this layer
self.assertEqual(response.status_code, 404)
data = response.content
self.assertTrue(
"No files have been found for this resource. Please, contact a system administrator." in data)