Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • djangoldp-packages/djangoldp
  • decentral1se/djangoldp
  • femmefaytale/djangoldp
  • jvtrudel/djangoldp
4 results
Show changes
Hide whitespace changes
Inline Side-by-side
Showing
with 1327 additions and 407 deletions
djangoldp/tests/perf_result.csv
View file @ e97d33ad
......@@ -46,4 +46,6 @@ jbl+AC0-T440p,Oct 09 2020 11:56:19,True,True,100,0.003119325637817,0.00560247182
jbl+AC0-T440p,Oct 09 2020 11:58:22,True,True,100,0.003008058071136,0.005401248931885,0.010658957958222,0.003909242153168,0.000718443393707,0.301162958145142,TRUE,3
jbl+AC0-T440p,Oct 09 2020 11:59:16,True,True,100,0.003015418052673,0.005526115894318,0.010740044116974,0.00400491476059,0.000724492073059,0.313828229904175,TRUE,4
jbl+AC0-T440p,Oct 09 2020 12:00:32,True,True,100,0.002969658374786,0.005434756278992,0.018136837482452,0.003030817508698,0.000726938247681,0.320115327835083,TRUE,0
jbl-T440p,Oct 09 2020 12:21:00,True,True,100,0.0034934663772583007,0.0061032938957214355,0.019232537746429443,0.003091294765472412,0.0007375502586364747,0.36986708641052246,N/A
jbl+AC0-T440p,Oct 09 2020 12:21:00,True,True,100,0.003493466377258,0.006103293895721,0.01923253774643,0.003091294765472,0.000737550258636,0.369867086410522,TRUE,0
jbl+AC0-T440p,Oct 15 2020 22:00:10,True,True,100,0.003004941940308,0.00546817779541,0.018348352909088,0.003068554401398,0.000729415416718,0.320573329925537,TRUE,0
jbl+AC0-T440p,Oct 15 2020 22:15:26,True,True,100,0.003350086212158,0.005898218154907,0.011625332832337,0.004264788627625,0.000795011520386,0.319289922714233,TRUE,1
djangoldp/tests/performance_runner.py
View file @ e97d33ad
import sys
import yaml
import django
from djangoldp.tests import settings_default
from django.conf import settings
from django.conf import settings as django_settings
from djangoldp.conf.ldpsettings import LDPSettings
from djangoldp.tests.server_settings import yaml_config
# configure settings not to use pagination
settings.configure(default_settings=settings_default,
REST_FRAMEWORK = {
'DEFAULT_PAGINATION_CLASS': None
})
# load test config
config = yaml.safe_load(yaml_config)
ldpsettings = LDPSettings(config)
django_settings.configure(ldpsettings,
REST_FRAMEWORK = {
'DEFAULT_PAGINATION_CLASS': None
},
ANONYMOUS_USER_NAME=None)
django.setup()
from django.test.runner import DiscoverRunner
......
djangoldp/tests/permissions.py 0 → 100644
View file @ e97d33ad
from djangoldp.filters import BaseFilterBackend
from djangoldp.permissions import LDPBasePermission
class StartsWithAFilter(BaseFilterBackend):
"""Only objects whose title starts in A get through"""
def filter_queryset(self, request, queryset, view):
return queryset.filter(title__startswith='A')
class ReadOnlyStartsWithA(LDPBasePermission):
"""Only gives read-only access and only to objects which title starts with A"""
filter_backend = StartsWithAFilter
permissions = {'view', 'list'}
def check_perms(self, obj):
return getattr(obj, 'title', '').startswith('A')
def has_object_permission(self, request, view, obj=None):
return self.check_perms(obj)
def get_permissions(self, user, model, obj=None):
return self.permissions if self.check_perms(obj) else set()
class ContainsSpace(BaseFilterBackend):
"""Only objects whose title contains a space get through"""
def filter_queryset(self, request, queryset, view):
if request.user.username != 'toto':
return queryset.none()
return queryset.filter(title__contains=' ')
class Only2WordsForToto(LDPBasePermission):
"""Only gives access if the user's username is toto and only to objects whose title has two words (contains space)"""
filter_backend = ContainsSpace
def has_permission(self, request, view):
return request.user.username == 'toto'
def check_perms(self, obj):
return ' ' in getattr(obj, 'title', '')
def has_object_permission(self, request, view, obj=None):
return self.check_perms(obj)
def get_permissions(self, user, model, obj=None):
return self.permissions if self.check_perms(obj) else set()
\ No newline at end of file
djangoldp/tests/runner.py
View file @ e97d33ad
import sys
import yaml
import django
from djangoldp.tests import settings_default
from django.conf import settings
from django.conf import settings as django_settings
from djangoldp.conf.ldpsettings import LDPSettings
from djangoldp.tests.server_settings import yaml_config
settings.configure(default_settings=settings_default)
# load test config
config = yaml.safe_load(yaml_config)
ldpsettings = LDPSettings(config)
django_settings.configure(ldpsettings)
django.setup()
from django.test.runner import DiscoverRunner
......@@ -12,11 +17,15 @@ from django.test.runner import DiscoverRunner
test_runner = DiscoverRunner(verbosity=1)
failures = test_runner.run_tests([
'djangoldp.tests.tests_settings',
'djangoldp.tests.tests_ldp_model',
'djangoldp.tests.tests_save',
'djangoldp.tests.tests_model_serializer',
'djangoldp.tests.tests_ldp_viewset',
'djangoldp.tests.tests_user_permissions',
'djangoldp.tests.tests_guardian',
'djangoldp.tests.tests_anonymous_permissions',
'djangoldp.tests.tests_permissions',
'djangoldp.tests.tests_post',
'djangoldp.tests.tests_update',
'djangoldp.tests.tests_auto_author',
'djangoldp.tests.tests_get',
......@@ -24,8 +33,8 @@ failures = test_runner.run_tests([
'djangoldp.tests.tests_sources',
'djangoldp.tests.tests_pagination',
'djangoldp.tests.tests_inbox',
'djangoldp.tests.tests_backlinks_service'
'djangoldp.tests.tests_backlinks_service',
'djangoldp.tests.tests_cache'
])
if failures:
sys.exit(failures)
djangoldp/tests/scripts/__init__.py 0 → 100644
View file @ e97d33ad
djangoldp/tests/scripts/prod_data_generator.py 0 → 100644
View file @ e97d33ad
import json
import argparse
from pathlib import Path
from datetime import datetime
from utils import generate_users, generate_projects, generate_skills
'''
A script which generates and outputs random production data, into a parameterised file (json), which can be used as
a Django fixture or imported into a live database
e.g. python manage.py loaddata fixture.json
for help run python prod_data_generator.py -h
'''
# starting from offset ensures that existing users etc are not disturbed
parser = argparse.ArgumentParser(description='generates and outputs random test data, into a file used by the performance unit tests')
parser.add_argument(dest='count', metavar='N', type=int, help='the number of users (and projects) to generate')
parser.add_argument('--offset', dest='offset', type=int, default=100, help='an offset to start primary keys at (should be larger than the largest pre-existing project/user primary key)')
parser.add_argument('-f', dest='file_dest', type=str, default="../fixtures/live.json", help='the file destination to write to')
parser.add_argument('-s', dest='generate_skills', type=bool, default=False, help='Do you want to generate skills too ?')
args = parser.parse_args()
count = args.count
OFFSET = args.offset
user_template = {
'model': 'djangoldp_account.ldpuser',
'pk': 0,
'fields': {
'username': 'john',
'email': 'jlennon@c.coop',
'password':'glassonion',
'first_name': 'John',
'last_name': 'Lennon'
}
}
project_template = {
'model': 'djangoldp_project.project',
'pk': 0,
'fields': {
'description': 'Test',
'status': 'Public',
'creationDate': str(datetime.date(datetime.now()))
}
}
skill_template = {
'model': 'djangoldp_skill.skill',
'pk': 0,
'fields': {
'name': 'PHP',
}
}
fixture = generate_users(count, user_template, offset=OFFSET)
fixture = generate_projects(count, project_template, fixture=fixture, offset=OFFSET)
if args.generate_skills:
fixture = generate_skills(count, skill_template, fixture=fixture, offset=OFFSET)
with open(Path(__file__).parent / args.file_dest, 'w') as output:
json.dump(fixture, output)
print(str(count))
djangoldp/tests/scripts/skills.json 0 → 100644
View file @ e97d33ad
source diff could not be displayed: it is too large. Options to address this: view the blob.
djangoldp/tests/scripts/test_data_generator.py
View file @ e97d33ad
import uuid
import json
import sys
import random
import argparse
from pathlib import Path
from copy import deepcopy
from utils import generate_users, generate_projects
'''
A script which generates and outputs random test data, into a file used by the performance unit tests
usage: python test_data_generator.py [number_rows]
e.g. python test_data_generator.py 2000
for help run python test_data_generator.py -h
'''
count = int(sys.argv[1])
fixture = list()
parser = argparse.ArgumentParser(description='generates and outputs random test data, into a file used by the performance unit tests')
parser.add_argument(dest='count', metavar='N', type=int, help='the number of users (and projects) to generate')
parser.add_argument('-f', dest='file_dest', type=str, default="../fixtures/test.json", help='the file destination to write to')
args = parser.parse_args()
count = args.count
user_template = {
'model': 'tests.user',
......@@ -20,7 +21,9 @@ user_template = {
'fields': {
'username': 'john',
'email': 'jlennon@c.coop',
'password':'glass onion'
'password':'glassonion',
'first_name': 'John',
'last_name': 'Lennon'
}
}
......@@ -28,38 +31,15 @@ project_template = {
'model': 'tests.project',
'pk': 0,
'fields': {
'status': 'Public',
'description': 'Test'
}
}
def generate_user(i):
user = deepcopy(user_template)
user['pk'] = i
user['fields']['username'] = str(uuid.uuid4())
user['fields']['email'] = user['fields']['username'] + "@c.coop"
return user
def generate_project(i):
project = deepcopy(project_template)
project['pk'] = i
project['fields']['team'] = list()
# append random number of users, max 10 for a single project
for j in range(random.randint(1, 10)):
project['fields']['team'].append(random.randint(1, count-1))
return project
# create N users
for i in range(count):
user = generate_user(i)
fixture.append(user)
# create N projects
for i in range(count):
project = generate_project(i)
fixture.append(project)
fixture = generate_users(count, user_template)
fixture = generate_projects(count, project_template, fixture=fixture, production=False)
with open(Path(__file__).parent / "../fixtures/test.json", 'w') as output:
with open(Path(__file__).parent / args.file_dest, 'w') as output:
json.dump(fixture, output)
print(str(count))
djangoldp/tests/scripts/utils.py 0 → 100644
View file @ e97d33ad
from copy import deepcopy
import random
from faker import Faker
'''
Contains definitions used in common by multiple scripts within this directory
'''
def generate_user(i, user_template):
myFactory = Faker()
user = deepcopy(user_template)
email = myFactory.unique.email().split('@')
email.insert(1, str(random.randint(0, 5)))
email.insert(2, "@")
email_str = "".join(email)
user['pk'] = i
user['fields']['username'] = myFactory.unique.user_name() + str(random.randint(0, 100))
user['fields']['email'] = email_str
user['fields']['first_name'] = myFactory.first_name()
user['fields']['last_name'] = myFactory.last_name()
return user
def generate_users(count, user_template, fixture=None, offset=0):
if fixture is None:
fixture = list()
for i in range(count):
j = offset + i
user = generate_user(j, user_template)
fixture.append(user)
return fixture
def generate_project_member_and_user_pks(project_pk, offset, total_users, max_members_per_project):
'''
returns a generator of tuples (new project member PKs and selected user PKs)
raises error if there are not enough users
'''
# we want to select a handful of random users
# to save time we just select a random user within a safe range and then grab a bunch of adjacent users
start_user_pk = random.randint(max(offset, 1), offset + (total_users - (max_members_per_project + 1)))
if start_user_pk < offset:
raise IndexError('not enough users!')
for i in range(random.randint(1, max_members_per_project)):
j = offset + (i + (project_pk * max_members_per_project)) # generate a unique integer id
user_pk = start_user_pk + i # select the next user
yield (j, user_pk)
def generate_project_members(project_pk, fixture, offset, total_users):
max_members_per_project = 10
def generate_project_member(i, user_pk):
return {
'model': 'djangoldp_project.member',
'pk': i,
'fields': {
'project': project_pk,
'user': user_pk
}
}
for (j, user_pk) in generate_project_member_and_user_pks(project_pk, offset, total_users, max_members_per_project):
fixture.append(generate_project_member(j, user_pk))
return fixture
def generate_skill(i, skill_template):
myFactory = Faker()
skill = deepcopy(skill_template)
skill['pk'] = i
skill['fields']['name'] = myFactory.unique.job()
return skill
def generate_user_pks(skill_pk, offset, total_users, max_users_per_skill):
'''
returns a generator of tuples ()
raises error if there are not enough users
'''
# we want to select a handful of random users
# to save time we just select a random user within a safe range and then grab a bunch of adjacent users
start_user_pk = random.randint(0, offset + (total_users - (max_users_per_skill + 1)))
# if start_user_pk < offset:
# raise IndexError('not enough users!')
for i in range(random.randint(1, max_users_per_skill)):
j = offset + (i + (skill_pk * max_users_per_skill)) # generate a unique integer id
user_pk = start_user_pk + i # select the next user
yield (j, user_pk)
def append_users_to_skill(skill, offset, total_users):
max_users_per_skill = 250
skill['fields']['users'] = []
for (j, user_pk) in generate_user_pks(skill['pk'], offset, total_users, max_users_per_skill):
skill['fields']['users'].append(user_pk)
return skill
def generate_skills(count, skill_template, fixture=None, offset=0):
if fixture is None:
fixture = list()
for i in range(count):
j = offset + i
skill = generate_skill(j, skill_template)
skill = append_users_to_skill(skill, offset, count)
fixture.append(skill)
return fixture
def generate_project(i, project_template):
project = deepcopy(project_template)
project['pk'] = i
return project
def append_members_to_project(project, offset, total_users):
max_members_per_project = 10
project['members'] = []
for (j, user_pk) in generate_project_member_and_user_pks(project['pk'], offset, total_users, max_members_per_project):
project['members'].append(user_pk)
return project
def generate_projects(count, project_template, fixture=None, offset=0, production=True):
if fixture is None:
fixture = list()
for i in range(count):
j = offset + i
project = generate_project(j, project_template)
# project members using direct ManyToMany field. Generate them as a field on project
if not production:
project = append_members_to_project(project, offset, count)
fixture.append(project)
# project members using Member through model, generate them as separate in the fixture
if production:
# append random number of project members, max 10 for a single project
generate_project_members(j, fixture, offset, count)
return fixture
djangoldp/tests/server_settings.py 0 → 100644
View file @ e97d33ad
"""
This module contains the YAML configuration for a testing djangoldp server.
"""
yaml_config = """
dependencies:
ldppackages:
- djangoldp.tests # fetch 'djangoldp.tests.djangoldp_settings'
- djangoldp.tests.dummy.apps.DummyConfig # already declared in 'djangoldp.tests'
server:
ALLOWED_HOSTS:
- '*'
AUTH_USER_MODEL: tests.User
EMAIL_HOST: somewhere
ANONYMOUS_USER_NAME: None
ROOT_URLCONF: djangoldp.urls
SEND_BACKLINKS: false
SITE_URL: http://happy-dev.fr
BASE_URL: http://happy-dev.fr
REST_FRAMEWORK:
DEFAULT_PAGINATION_CLASS: djangoldp.pagination.LDPPagination
PAGE_SIZE: 5
USE_TZ: false
SEND_BACKLINKS: false
GUARDIAN_AUTO_PREFETCH: true
SERIALIZER_CACHE: false
STORE_ACTIVITIES: VERBOSE
"""
djangoldp/tests/settings_default.py deleted 100644 → 0
View file @ 84febabb
from django.conf.global_settings import *
# defines default settings for testing DjangoLDP. You can use this in your own packages by following the example in
# runner.py
DEBUG=False
ALLOWED_HOSTS=["*"]
SITE_URL='http://happy-dev.fr'
BASE_URL='http://happy-dev.fr'
DJANGOLDP_PACKAGES=['djangoldp.tests']
INSTALLED_APPS=('django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.admin',
'django.contrib.messages',
'django.contrib.staticfiles',
'guardian',
'djangoldp',
'djangoldp.tests',
)
DATABASES={
'default': {
'ENGINE': 'django.db.backends.sqlite3',
}
}
REST_FRAMEWORK = {
'DEFAULT_PAGINATION_CLASS': 'djangoldp.pagination.LDPPagination',
'PAGE_SIZE': 5
}
AUTH_USER_MODEL='tests.User'
ANONYMOUS_USER_NAME = None
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
AUTHENTICATION_BACKENDS=(
'django.contrib.auth.backends.ModelBackend', 'guardian.backends.ObjectPermissionBackend')
ROOT_URLCONF='djangoldp.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
LDP_RDF_CONTEXT={
"@context": {
"@vocab": "http://happy-dev.fr/owl/#",
"foaf": "http://xmlns.com/foaf/0.1/",
"doap": "http://usefulinc.com/ns/doap#",
"ldp": "http://www.w3.org/ns/ldp#",
"rdfs": "http://www.w3.org/2000/01/rdf-schema#",
"rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
"xsd": "http://www.w3.org/2001/XMLSchema#",
"geo": "http://www.w3.org/2003/01/geo/wgs84_pos#",
"acl": "http://www.w3.org/ns/auth/acl#",
"name": "rdfs:label",
"website": "foaf:homepage",
"deadline": "xsd:dateTime",
"lat": "geo:lat",
"lng": "geo:long",
"jabberID": "foaf:jabberID",
"permissions": "acl:accessControl",
"mode": "acl:mode",
"view": "acl:Read",
"change": "acl:Write",
"add": "acl:Append",
"delete": "acl:Delete",
"control": "acl:Control"
}
}
SEND_BACKLINKS=False
GUARDIAN_AUTO_PREFETCH = True
djangoldp/tests/tests_anonymous_permissions.py
View file @ e97d33ad
......@@ -3,9 +3,7 @@ import json
from django.test import TestCase
from rest_framework.test import APIClient
from djangoldp.permissions import LDPPermissions
from djangoldp.tests.models import JobOffer
from djangoldp.views import LDPViewSet
class TestAnonymousUserPermissions(TestCase):
......@@ -26,6 +24,8 @@ class TestAnonymousUserPermissions(TestCase):
response = self.client.post('/job-offers/', data=json.dumps(post), content_type='application/ld+json')
self.assertEqual(response.status_code, 403)
# TODO: test POST request for anonymous user where it's allowed
def test_put_request_for_anonymousUser(self):
body = {'title':"job_updated"}
response = self.client.put('/job-offers/{}/'.format(self.job.pk), data=json.dumps(body),
......
djangoldp/tests/tests_auto_author.py
View file @ e97d33ad
import json
from django.contrib.auth import get_user_model
from rest_framework.test import APIRequestFactory, APIClient, APITestCase
from rest_framework.test import APIClient, APIRequestFactory, APITestCase
from djangoldp.tests.models import UserProfile
......@@ -18,18 +19,8 @@ class TestAutoAuthor(APITestCase):
def test_save_with_anonymous_user(self):
post = {
'@graph': [{'http://happy-dev.fr/owl/#content': "post content"}]}
response = self.client.post('/posts/', data=json.dumps(post), content_type='application/ld+json')
self.assertEqual(response.status_code, 201)
self.assertEquals(response.data['content'], "post content")
def test_auto_author_field(self):
self.client.force_authenticate(user=self.user)
post = {
'@graph': [{'http://happy-dev.fr/owl/#content': "post content"}]}
'@graph': [{'https://cdn.startinblox.com/owl#content': "post content"}]}
response = self.client.post('/posts/', data=json.dumps(post), content_type='application/ld+json')
self.assertEqual(response.status_code, 201)
self.assertEquals(response.data['content'], "post content")
self.assertIsNotNone(response.data['author'])
self.assertEqual(response.data['content'], "post content")
\ No newline at end of file
djangoldp/tests/tests_backlinks_service.py
View file @ e97d33ad
import uuid
import copy
import time
import uuid
from django.contrib.auth import get_user_model
from django.test import override_settings
from rest_framework.test import APIClient, APITestCase
from djangoldp.tests.models import Circle, Project
from djangoldp.activities.services import (BACKLINKS_ACTOR, ActivityPubService,
ActivityQueueService)
from djangoldp.models import Activity, ScheduledActivity
from djangoldp.activities.services import BACKLINKS_ACTOR, ActivityPubService, ActivityQueueService
from djangoldp.tests.models import Circle, Project
class TestsBacklinksService(APITestCase):
......@@ -63,17 +67,17 @@ class TestsBacklinksService(APITestCase):
external_a = self._get_random_external_user()
external_b = self._get_random_external_user()
external_c = self._get_random_external_user()
project.team.add(external_a)
project.team.add(external_b)
project.team.add(external_c)
project.members.add(external_a)
project.members.add(external_b)
project.members.add(external_c)
self.assertEqual(Activity.objects.all().count(), 3)
# remove one individual
project.team.remove(external_a)
project.members.remove(external_a)
self.assertEqual(Activity.objects.all().count(), 4)
# clear the rest
project.team.clear()
project.members.clear()
self.assertEqual(Activity.objects.all().count(), 6)
prior_count = Activity.objects.all().count()
......@@ -85,7 +89,7 @@ class TestsBacklinksService(APITestCase):
def test_local_object_with_external_m2m_delete_parent(self):
project = Project.objects.create(description='Test')
external_a = self._get_random_external_user()
project.team.add(external_a)
project.members.add(external_a)
prior_count = Activity.objects.all().count()
project.delete()
......@@ -110,13 +114,13 @@ class TestsBacklinksService(APITestCase):
time.sleep(0.1)
# assert that all scheduled activities were cleaned up
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEqual(ScheduledActivity.objects.count(), 0)
# assert that ONLY the newly scheduled activity was sent
activities = Activity.objects.all()
self.assertEquals(Activity.objects.count(), 1)
self.assertEqual(Activity.objects.count(), 1)
astream = activities[0].to_activitystream()
self.assertEquals(astream['summary'], new_activity['summary'])
self.assertEqual(astream['summary'], new_activity['summary'])
activities[0].delete()
# variation using expanded syntax
......@@ -146,8 +150,8 @@ class TestsBacklinksService(APITestCase):
ActivityQueueService._activity_queue_worker('http://127.0.0.1:8001/idontexist/', scheduled_b)
# assert that both scheduled activities were sent, and the scheduled activities were cleaned up
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 2)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 2)
# variation on the previous test where the two activities are working on different models (using the same object)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
......@@ -167,8 +171,8 @@ class TestsBacklinksService(APITestCase):
ActivityQueueService._activity_queue_worker('http://127.0.0.1:8001/idontexist/', scheduled_b)
# assert that both scheduled activities were sent, and the scheduled activities were cleaned up
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 2)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 2)
# variation using an Add and a Remove (one defines target, the other origin)
# also tests that an unnecessary add is not sent
......@@ -188,8 +192,8 @@ class TestsBacklinksService(APITestCase):
ActivityQueueService._activity_queue_worker('http://127.0.0.1:8001/idontexist/', scheduled_b)
# assert that both scheduled activities were sent, and the scheduled activities were cleaned up
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 1)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 1)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_unnecessary_add_not_sent(self):
......@@ -197,8 +201,7 @@ class TestsBacklinksService(APITestCase):
a = {'type': 'Add', 'actor': {'type': 'Service', 'name': 'Backlinks Service'},
'object': {'@type': 'foaf:user', '@id': 'https://api.test2.startinblox.com/users/calum/'},
'target': {'@type': 'hd:skill', '@id': 'https://api.test1.startinblox.com/skills/3/'}}
ActivityQueueService._save_sent_activity(a, Activity, success=True, type='add',
external_id='https://distant.com/inbox/')
ActivityQueueService._save_activity_from_response({'status_code': '201'}, 'https://distant.com/inbox/', a)
# no remove has since been sent, but a new Add is scheduled
scheduled_b = ActivityQueueService._save_sent_activity(a, ScheduledActivity, success=False, type='add',
......@@ -206,8 +209,8 @@ class TestsBacklinksService(APITestCase):
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled_b)
# assert that only the previous activity was sent, and the scheduled activites cleaned up
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 1)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 1)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_unnecessary_remove_not_sent(self):
......@@ -215,8 +218,7 @@ class TestsBacklinksService(APITestCase):
a = {'type': 'Remove', 'actor': {'type': 'Service', 'name': 'Backlinks Service'},
'object': {'@type': 'foaf:user', '@id': 'https://api.test2.startinblox.com/users/calum/'},
'target': {'@type': 'hd:skill', '@id': 'https://api.test1.startinblox.com/skills/3/'}}
ActivityQueueService._save_sent_activity(a, Activity, success=True, type='remove',
external_id='https://distant.com/inbox/')
ActivityQueueService._save_activity_from_response({'status_code': '201'}, 'https://distant.com/inbox/', a)
# no add has since been sent, but a new Remove is scheduled
scheduled_b = ActivityQueueService._save_sent_activity(a, ScheduledActivity, success=False, type='remove',
......@@ -224,8 +226,8 @@ class TestsBacklinksService(APITestCase):
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled_b)
# assert that only the previous activity was sent, and the scheduled activites cleaned up
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 1)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 1)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_necessary_add_sent(self):
......@@ -233,8 +235,7 @@ class TestsBacklinksService(APITestCase):
a = {'type': 'Remove', 'actor': {'type': 'Service', 'name': 'Backlinks Service'},
'object': {'@type': 'foaf:user', '@id': 'https://api.test2.startinblox.com/users/calum/'},
'target': {'@type': 'hd:skill', '@id': 'https://api.test1.startinblox.com/skills/3/'}}
ActivityQueueService._save_sent_activity(a, Activity, success=True, type='remove',
external_id='https://distant.com/inbox/')
ActivityQueueService._save_activity_from_response({'status_code': '201'}, 'https://distant.com/inbox/', a)
# an add is now being sent
scheduled_b = ActivityQueueService._save_sent_activity(a, ScheduledActivity, type='add',
......@@ -242,8 +243,8 @@ class TestsBacklinksService(APITestCase):
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled_b)
# assert that both activities sent, and the scheduled activites cleaned up
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 2)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 2)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_first_add_sent(self):
......@@ -256,8 +257,8 @@ class TestsBacklinksService(APITestCase):
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled)
# assert that the activity was sent, and the scheduled activites cleaned up
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 1)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 1)
# validate Update activity objects have new info before sending the notification
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
......@@ -270,8 +271,7 @@ class TestsBacklinksService(APITestCase):
'@type': 'foaf:user'}
}
activity_a = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='A')
ActivityQueueService._save_sent_activity(activity_a, Activity, success=True, type='create',
external_id='https://distant.com/inbox/')
ActivityQueueService._save_activity_from_response({'status_code': '201'}, 'https://distant.com/inbox/', activity_a)
# now I'm sending an update, which doesn't change anything about the object
activity_b = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='B')
......@@ -281,8 +281,8 @@ class TestsBacklinksService(APITestCase):
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled_b)
# assert that only the previous activity was sent, and the scheduled activites cleaned up
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 1)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 1)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_necessary_update_is_sent(self):
......@@ -294,8 +294,7 @@ class TestsBacklinksService(APITestCase):
'@type': 'foaf:user'}
}
activity_a = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='A')
ActivityQueueService._save_sent_activity(activity_a, Activity, success=True, type='create',
external_id='https://distant.com/inbox/')
ActivityQueueService._save_activity_from_response({'status_code': '201'}, 'https://distant.com/inbox/', activity_a)
# now I'm sending an update, which changes the owner of the circle
obj['owner']['@id'] = 'https://distant.com/users/mark/'
......@@ -306,8 +305,8 @@ class TestsBacklinksService(APITestCase):
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled_b)
# assert that both activities were sent
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 2)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 2)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_first_update_is_sent(self):
......@@ -322,5 +321,5 @@ class TestsBacklinksService(APITestCase):
scheduled = ActivityQueueService._save_sent_activity(activity, ScheduledActivity, type='update',
external_id='https://distant.com/inbox/')
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled)
self.assertEquals(ScheduledActivity.objects.count(), 0)
self.assertEquals(Activity.objects.count(), 1)
self.assertEqual(ScheduledActivity.objects.count(), 0)
self.assertEqual(Activity.objects.count(), 1)
djangoldp/tests/tests_cache.py 0 → 100644
View file @ e97d33ad
from django.contrib.auth import get_user_model
from django.test import TestCase, override_settings
from rest_framework.test import APIClient, APIRequestFactory
from rest_framework.utils import json
from djangoldp.tests.models import Circle, Conversation, Project
class TestCache(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.client = APIClient()
self.user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com',
password='glass onion', first_name='John')
self.client.force_authenticate(self.user)
def tearDown(self):
setattr(Circle._meta, 'depth', 0)
setattr(Circle._meta, 'empty_containers', [])
# test container cache after new resource added
@override_settings(SERIALIZER_CACHE=True)
def test_save_fk_graph_with_nested(self):
response = self.client.get('/batchs/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
post = {
'@graph': [
{
'https://cdn.startinblox.com/owl#title': "title",
'https://cdn.startinblox.com/owl#invoice': {
'@id': "_.123"
}
},
{
'@id': "_.123",
'https://cdn.startinblox.com/owl#title': "title 2"
}
]
}
response = self.client.post('/batchs/', data=json.dumps(post), content_type='application/ld+json')
self.assertEqual(response.status_code, 201)
response = self.client.get('/batchs/', content_type='application/ld+json')
self.assertIn('ldp:contains', response.data)
self.assertEqual(response.data['ldp:contains'][0]['title'], "title")
self.assertEqual(response.data['ldp:contains'][0]['invoice']['title'], "title 2")
# test resource cache after it is updated
@override_settings(SERIALIZER_CACHE=True)
def test_update_with_new_fk_relation(self):
conversation = Conversation.objects.create(author_user=self.user, description="conversation description")
response = self.client.get('/conversations/{}/'.format(conversation.pk), content_type='application/ld+json')
body = [
{
'@id': "/conversations/{}/".format(conversation.pk),
'https://cdn.startinblox.com/owl#description': "conversation update",
'https://cdn.startinblox.com/owl#peer_user': {
'@id': self.user.urlid,
}
}
]
response = self.client.put('/conversations/{}/'.format(conversation.pk), data=json.dumps(body),
content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
response = self.client.get('/conversations/{}/'.format(conversation.pk), content_type='application/ld+json')
self.assertIn('peer_user', response.data)
self.assertEqual('conversation update', response.data['description'])
self.assertEqual(response.data['peer_user']['@id'], self.user.urlid)
self.assertIn('@type', response.data['peer_user'])
# test resource cache after it is updated - external resource
@override_settings(SERIALIZER_CACHE=True)
def test_update_with_new_fk_relation_external(self):
conversation = Conversation.objects.create(author_user=self.user, description="conversation description")
response = self.client.get('/conversations/{}/'.format(conversation.pk), content_type='application/ld+json')
external_user = get_user_model().objects.create_user(username='external', email='jlennon@beatles.com',
password='glass onion', urlid='https://external.com/users/external/')
body = [
{
'@id': "/conversations/{}/".format(conversation.pk),
'https://cdn.startinblox.com/owl#description': "conversation update",
'https://cdn.startinblox.com/owl#peer_user': {
'@id': external_user.urlid,
}
}
]
response = self.client.put('/conversations/{}/'.format(conversation.pk), data=json.dumps(body),
content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
response = self.client.get('/conversations/{}/'.format(conversation.pk), content_type='application/ld+json')
self.assertIn('peer_user', response.data)
self.assertEqual('conversation update', response.data['description'])
self.assertIn('@id', response.data['peer_user'])
# serialize external id and only external id
self.assertEqual(response.data['peer_user']['@id'], external_user.urlid)
self.assertIn('@type', response.data['peer_user'])
self.assertEqual(len(response.data['peer_user']), 2)
# test container cache after member is deleted by view
@override_settings(SERIALIZER_CACHE=True)
def test_cached_container_deleted_resource_view(self):
conversation = Conversation.objects.create(author_user=self.user, description="conversation description")
response = self.client.get('/conversations/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
response = self.client.delete('/conversations/{}/'.format(conversation.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 204)
response = self.client.get('/conversations/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 0)
# test container cache after member is deleted manually
@override_settings(SERIALIZER_CACHE=True)
def test_cached_container_deleted_resource_manual(self):
conversation = Conversation.objects.create(author_user=self.user, description="conversation description")
response = self.client.get('/conversations/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
conversation.delete()
response = self.client.get('/conversations/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 0)
# test resource cache after it is deleted manually
@override_settings(SERIALIZER_CACHE=True)
def test_cached_resource_deleted_resource_manual(self):
conversation = Conversation.objects.create(author_user=self.user, description="conversation description")
response = self.client.get('/conversations/{}/'.format(conversation.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
conversation.delete()
response = self.client.get('/conversations/{}/'.format(conversation.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 404)
# test container cache following m2m_changed - Project (which inherits from djangoldp.models.Model)
@override_settings(SERIALIZER_CACHE=True)
def test_cached_container_m2m_changed_project(self):
project = Project.objects.create(description='Test')
response = self.client.get('/projects/{}/members/'.format(project.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 0)
project.members.add(self.user)
response = self.client.get('/projects/{}/members/'.format(project.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
project.members.remove(self.user)
response = self.client.get('/projects/{}/members/'.format(project.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 0)
project.members.add(self.user)
project.members.clear()
response = self.client.get('/projects/{}/members/'.format(project.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 0)
# test container cache following m2m_changed - Conversation (which does not inherit from djangoldp.models.Model)
@override_settings(SERIALIZER_CACHE=True)
def test_cached_container_m2m_changed_conversation(self):
conversation = Conversation.objects.create(author_user=self.user, description="conversation description")
response = self.client.get('/conversations/{}/observers/'.format(conversation.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 0)
conversation.observers.add(self.user)
response = self.client.get('/conversations/{}/observers/'.format(conversation.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
conversation.observers.remove(self.user)
response = self.client.get('/conversations/{}/observers/'.format(conversation.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 0)
conversation.observers.add(self.user)
conversation.observers.clear()
response = self.client.get('/conversations/{}/observers/'.format(conversation.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 0)
# test cache working inside of the nested field (serializer) of another object
@override_settings(SERIALIZER_CACHE=True)
def test_cached_container_serializer_nested_field(self):
project = Project.objects.create(description='Test')
response = self.client.get('/projects/{}/'.format(project.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['members']['ldp:contains']), 0)
project.members.add(self.user)
response = self.client.get('/projects/{}/'.format(project.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['members']['ldp:contains']), 1)
project.members.remove(self.user)
response = self.client.get('/projects/{}/'.format(project.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['members']['ldp:contains']), 0)
project.members.add(self.user)
project.members.clear()
response = self.client.get('/projects/{}/'.format(project.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['members']['ldp:contains']), 0)
# test cache working on a serialized nested field at higher depth
@override_settings(SERIALIZER_CACHE=True)
def test_cache_depth_2(self):
setattr(Circle._meta, 'depth', 2)
circle = Circle.objects.create(description='Test')
response = self.client.get('/circles/{}/'.format(circle.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['members']['user_set']), 0)
circle.members.user_set.add(self.user)
response = self.client.get('/circles/{}/'.format(circle.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['members']['user_set']), 1)
# assert the depth is applied
self.assertIn('first_name', response.data['members']['user_set'][0])
self.assertEqual(response.data['members']['user_set'][0]['first_name'], self.user.first_name)
# make a change to the _user_
self.user.first_name = "Alan"
self.user.save()
# assert that the use under the circles members has been updated
response = self.client.get('/circles/{}/'.format(circle.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['members']['user_set']), 1)
self.assertIn('first_name', response.data['members']['user_set'][0])
self.assertEqual(response.data['members']['user_set'][0]['first_name'], self.user.first_name)
# test the cache behaviour when empty_containers is an active setting
@override_settings(SERIALIZER_CACHE=True)
def test_cache_empty_container(self):
setattr(Circle._meta, 'depth', 1)
setattr(Circle._meta, 'empty_containers', ['members'])
circle = Circle.objects.create(name='test', description='test', owner=self.user)
circle.members.user_set.add(self.user)
# make one call on the parent
response = self.client.get('/circles/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertIn('ldp:contains', response.data)
self.assertEqual(len(response.data['ldp:contains']), 1)
self.assertIn('members', response.data['ldp:contains'][0])
self.assertIn('@id', response.data['ldp:contains'][0]['members'])
self.assertIn('user_set', response.data['ldp:contains'][0]['members'])
self.assertEqual(len(response.data['ldp:contains'][0]['members']['user_set']), 1)
self.assertIn('@id', response.data['ldp:contains'][0]['members']['user_set'][0])
self.assertEqual(response.data['ldp:contains'][0]['members']['user_set'][0]['@type'], 'foaf:user')
# and a second on the child
response = self.client.get(response.data['ldp:contains'][0]['members']['@id'], content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertIn('@id', response.data)
self.assertEqual(len(response.data['user_set']), 1)
djangoldp/tests/tests_get.py
View file @ e97d33ad
from djangoldp.serializers import LDListMixin, LDPSerializer
from django.contrib.auth import get_user_model
from datetime import datetime
from rest_framework.test import APIRequestFactory, APIClient, APITestCase
from djangoldp.tests.models import Post, Invoice, JobOffer, Skill, Batch, DateModel, UserProfile
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient, APIRequestFactory, APITestCase
from djangoldp.serializers import GLOBAL_SERIALIZER_CACHE
from djangoldp.tests.models import (Batch, Circle, Conversation, DateModel,
Invoice, JobOffer, Message, Post, Skill,
User, UserProfile)
class TestGET(APITestCase):
......@@ -11,29 +14,37 @@ class TestGET(APITestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.client = APIClient()
LDListMixin.to_representation_cache.invalidate_cache()
LDPSerializer.to_representation_cache.invalidate_cache()
self.ordered_fields = ['@context', '@type', '@id']
setattr(Circle._meta, 'depth', 0)
setattr(Circle._meta, 'empty_containers', [])
def tearDown(self):
pass
GLOBAL_SERIALIZER_CACHE.reset()
def test_get_resource(self):
post = Post.objects.create(content="content")
response = self.client.get('/posts/{}/'.format(post.pk), content_type='application/ld+json')
response = self.client.get('/posts/{}/'.format(post.pk), content_type='application/ld+json', HTTP_ORIGIN='http://localhost:8080/test/')
self.assertEqual(response.status_code, 200)
self.assertEquals(response.data['content'], "content")
self.assertEqual(response.data['content'], "content")
self.assertIn('author', response.data)
self.assertIn('@type', response.data)
# test headers returned
self.assertEqual(response['Content-Type'], 'application/ld+json')
self.assertEqual(response['Accept-Post'], 'application/ld+json')
self.assertEqual(response['Allow'], 'GET, PUT, PATCH, DELETE, HEAD, OPTIONS')
self.assertEqual(response['Access-Control-Allow-Origin'], 'http://localhost:8080/test/')
self.assertIn('DPoP', response['Access-Control-Allow-Headers'])
# TODO: https://git.startinblox.com/djangoldp-packages/djangoldp/issues/293
'''def test_get_resource_urlid(self):
def test_get_resource_urlid(self):
user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com',
password='glass onion')
UserProfile.objects.create(user=user)
post = Post.objects.create(content="content", author=user.userprofile)
post = Post.objects.create(content="content", author=user)
response = self.client.get('/posts/{}/'.format(post.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEquals(response.data['content'], "content")
self.assertEqual(response.data['author'], user.userprofile.urlid)'''
self.assertEqual(response.data['content'], "content")
self.assertEqual(response.data['author']['@id'], user.urlid)
def test_get_container(self):
Post.objects.create(content="content")
......@@ -41,26 +52,26 @@ class TestGET(APITestCase):
Post.objects.create(content="federated", urlid="https://external.com/posts/1/")
response = self.client.get('/posts/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertIn('permissions', response.data)
self.assertEquals(1, len(response.data['ldp:contains']))
self.assertEquals(2, len(response.data['permissions'])) # read and add
self.assertEqual(1, len(response.data['ldp:contains']))
self.assertIn('@type', response.data)
self.assertIn('@type', response.data['ldp:contains'][0])
self.assertNotIn('permissions', response.data['ldp:contains'][0])
Invoice.objects.create(title="content")
response = self.client.get('/invoices/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertIn('permissions', response.data)
self.assertEquals(1, len(response.data['permissions'])) # read only
self.assertEqual(1, len(response.data['permissions'])) # read only
def test_get_empty_container(self):
Post.objects.all().delete()
response = self.client.get('/posts/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEquals(0, len(response.data['ldp:contains']))
self.assertEqual(0, len(response.data['ldp:contains']))
def test_get_filtered_fields(self):
skill = Skill.objects.create(title="Java", obligatoire="ok", slug="1")
skill2 = Skill.objects.create(title="Java", obligatoire="ok", slug="2")
skill3 = Skill.objects.create(urlid="http://external/skills/1")
skill3 = Skill.objects.create(urlid="http://happy-dev.hubl.fr/skills/1")
job = JobOffer.objects.create(title="job", slug="1")
job.skills.add(skill)
job.skills.add(skill2)
......@@ -70,7 +81,11 @@ class TestGET(APITestCase):
self.assertEqual(response.status_code, 200)
self.assertIn('recent_skills', response.data)
self.assertEqual(response.data['recent_skills']['@id'], "http://happy-dev.fr/job-offers/1/recent_skills/")
self.assertEqual(response.data['skills']['ldp:contains'][2]['@id'], "http://external/skills/1")
# the external resource should be serialized with its @id and @type.. and only these fields
self.assertEqual(response.data['skills']['ldp:contains'][2]['@id'], "http://happy-dev.hubl.fr/skills/1")
self.assertIn('@type', response.data['skills']['ldp:contains'][1])
self.assertIn('@type', response.data['skills']['ldp:contains'][2])
self.assertEqual(len(response.data['skills']['ldp:contains'][2].items()), 2)
def test_get_reverse_filtered_fields(self):
skill = Skill.objects.create(title="Java", obligatoire="ok", slug="1")
......@@ -93,6 +108,9 @@ class TestGET(APITestCase):
self.assertEqual(response.status_code, 200)
self.assertIn('some_skill', response.data)
self.assertEqual(response.data['some_skill']['@id'], skill.urlid)
response = self.client.get('/job-offers/{}/recent_skills/'.format(job.slug), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 2)
def test_get_nested(self):
invoice = Invoice.objects.create(title="invoice")
......@@ -100,19 +118,122 @@ class TestGET(APITestCase):
distant_batch = Batch.objects.create(invoice=invoice, title="distant", urlid="https://external.com/batch/1/")
response = self.client.get('/invoices/{}/batches/'.format(invoice.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEquals(response.data['@id'], 'http://happy-dev.fr/invoices/{}/batches/'.format(invoice.pk))
self.assertEquals(len(response.data['ldp:contains']), 2)
self.assertEquals(response.data['ldp:contains'][0]['invoice']['@id'], 'http://happy-dev.fr/invoices/{}/'.format(invoice.pk))
self.assertEqual(response.data['@id'], 'http://happy-dev.fr/invoices/{}/batches/'.format(invoice.pk))
self.assertIn('permissions', response.data)
self.assertEqual(len(response.data['ldp:contains']), 2)
self.assertIn('@type', response.data['ldp:contains'][0])
self.assertIn('@type', response.data['ldp:contains'][1])
self.assertNotIn('permissions', response.data['ldp:contains'][0])
self.assertNotIn('permissions', response.data['ldp:contains'][1])
self.assertEqual(response.data['ldp:contains'][0]['invoice']['@id'], invoice.urlid)
self.assertEqual(response.data['ldp:contains'][1]['@id'], distant_batch.urlid)
def test_get_nested_without_related_name(self):
user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com', password='glass onion')
conversation = Conversation.objects.create(author_user=user)
message = Message.objects.create(conversation=conversation, author_user=user)
response = self.client.get('/conversations/{}/message_set/'.format(conversation.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['@id'], 'http://happy-dev.fr/conversations/{}/message_set/'.format(conversation.pk))
self.assertEqual(len(response.data['ldp:contains']), 1)
# TODO: https://git.startinblox.com/djangoldp-packages/djangoldp/issues/335
# test getting a route with multiple nested fields (/job-offers/X/skills/Y/)
'''def test_get_twice_nested(self):
job = JobOffer.objects.create(title="job", slug="slug1")
skill = Skill.objects.create(title='old', obligatoire='old', slug='skill1')
job.skills.add(skill)
self.assertEqual(job.skills.count(), 1)
response = self.client.get('/job-offers/{}/skills/{}/'.format(job.slug, skill.slug))
self.assertEqual(response.status_code, 200)'''
def test_serializer_excludes(self):
date = DateModel.objects.create(excluded='test', value=datetime.now())
response = self.client.get('/dates/{}/'.format(date.pk), content_type='application/ld+json')
response = self.client.get('/datemodels/{}/'.format(date.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertNotIn('excluded', response.data.keys())
def test_serializer_excludes_serializer_fields_set_also(self):
setattr(DateModel._meta, 'serializer_fields', ['value', 'excluded'])
date = DateModel.objects.create(excluded='test', value=datetime.now())
response = self.client.get('/dates/{}/'.format(date.pk), content_type='application/ld+json')
response = self.client.get('/datemodels/{}/'.format(date.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertNotIn('excluded', response.data.keys())
def _set_up_circle_and_user(self):
user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com',
password='glass onion')
circle = Circle.objects.create(name='test', description='test', owner=user)
self.client.force_authenticate(user)
circle.members.user_set.add(user)
return user
# tests for functionality allowing me to set containers to be serialized without content\
# test for normal functioning (without setting)
def test_empty_container_serialization_nested_serializer_no_empty(self):
setattr(Circle._meta, 'depth', 1)
self._set_up_circle_and_user()
response = self.client.get('/circles/', content_type='application/ld+json')
self.assertEqual(response.data['@type'], 'ldp:Container')
self.assertIn('@id', response.data)
self.assertIn('permissions', response.data)
self.assertIn('members', response.data['ldp:contains'][0])
self.assertEqual(response.data['ldp:contains'][0]['members']['@type'], 'foaf:Group')
self.assertIn('@id', response.data['ldp:contains'][0]['members'])
self.assertEqual(len(response.data['ldp:contains'][0]['members']['user_set']), 1)
# test for functioning with setting
def test_empty_container_serialization_nested_serializer_empty(self):
setattr(User._meta, 'depth', 1)
setattr(User._meta, 'empty_containers', ['owned_circles'])
self._set_up_circle_and_user()
response = self.client.get('/users/', content_type='application/ld+json')
self.assertEqual(response.data['@type'], 'ldp:Container')
self.assertIn('owned_circles', response.data['ldp:contains'][0])
self.assertIn('@id', response.data['ldp:contains'][0]['owned_circles'])
self.assertNotIn('permissions', response.data['ldp:contains'][0]['owned_circles'])
self.assertNotIn('ldp:contains', response.data['ldp:contains'][0]['owned_circles'])
# should serialize as normal on the nested viewset (directly asking for the container)
# test for normal functioning (without setting)
def test_empty_container_serialization_nested_viewset_no_empty(self):
user = self._set_up_circle_and_user()
response = self.client.get(f'/users/{user.pk}/owned_circles/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['@type'], 'ldp:Container')
self.assertIn('@id', response.data)
self.assertIn('ldp:contains', response.data)
self.assertIn('permissions', response.data)
self.assertIn('owner', response.data['ldp:contains'][0])
# test for functioning with setting
def test_empty_container_serialization_nested_viewset_empty(self):
setattr(User._meta, 'empty_containers', ['owned_circles'])
user = self._set_up_circle_and_user()
response = self.client.get(f'/users/{user.pk}/owned_circles/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['@type'], 'ldp:Container')
self.assertIn('@id', response.data)
self.assertIn('ldp:contains', response.data)
self.assertIn('permissions', response.data)
self.assertIn('owner', response.data['ldp:contains'][0])
# # test for checking fields ordering
# def test_ordered_field(self):
# self._set_up_circle_and_user()
# response = self.client.get('/users/', content_type='application/ld+json')
# fields_to_test = [
# response.data.keys(),
# response.data['ldp:contains'][-1],
# response.data['ldp:contains'][-1]['circle_set']
# ]
# for test_fields in fields_to_test:
# test_fields = list(test_fields)
# o_f = [field for field in self.ordered_fields if field in test_fields]
# self.assertEqual(o_f, test_fields[:len(o_f)])
djangoldp/tests/tests_guardian.py
View file @ e97d33ad
import json
import uuid
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from rest_framework.test import APIClient, APITestCase
from guardian.shortcuts import assign_perm
from .models import PermissionlessDummy, Dummy
from djangoldp.permissions import LDPPermissions
from .models import PermissionlessDummy, Dummy, LDPDummy
class TestsGuardian(APITestCase):
......@@ -15,45 +16,116 @@ class TestsGuardian(APITestCase):
def setUpLoggedInUser(self):
self.user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com',
password='glass onion')
self.group = Group.objects.create(name='Test')
self.user.groups.add(self.group)
self.user.save()
self.client.force_authenticate(user=self.user)
# optional setup for testing PermissionlessDummy model with parameterised perms
def setUpGuardianDummyWithPerms(self, perms=[]):
self.dummy = PermissionlessDummy.objects.create(some='test', slug='test')
def _get_dummy_with_perms(self, perms=None, parent=None, group=False):
if perms is None:
perms = []
dummy = PermissionlessDummy.objects.create(some='test', slug=uuid.uuid4(), parent=parent)
model_name = PermissionlessDummy._meta.model_name
for perm in perms:
assign_perm(perm + '_' + model_name, self.user, self.dummy)
perm = perm + '_' + model_name
if group:
#assigns container-level and object-level perms
assign_perm('tests.'+perm, self.group)
assign_perm(perm, self.group, dummy)
else:
assign_perm('tests.'+perm, self.user)
assign_perm(perm, self.user, dummy)
return dummy
# optional setup for testing PermissionlessDummy model with parameterised perms
def setUpGuardianDummyWithPerms(self, perms=None, parent=None, group=False):
self.dummy = self._get_dummy_with_perms(perms, parent, group)
# auxiliary function converts permission format for test
def _unpack_permissions(self, perms_from_response):
return perms_from_response
# test that dummy with no permissions set returns no results
def test_get_dummy_no_permissions(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms()
response = self.client.get('/permissionless-dummys/{}/'.format(self.dummy.slug))
self.assertEqual(response.status_code, 403)
self.assertEqual(response.status_code, 404)
# test with anonymous user
def test_get_dummy_anonymous_user(self):
self.setUpGuardianDummyWithPerms()
response = self.client.get('/permissionless-dummys/')
# I have no object permissions - I should receive a 403
self.assertEqual(response.status_code, 403)
def test_list_dummy_exception(self):
self.setUpLoggedInUser()
# I have permission on a permissionless dummy, but not in general
dummy_a = self._get_dummy_with_perms()
dummy_b = self._get_dummy_with_perms(['view'])
response = self.client.get('/permissionless-dummys/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
containees = [d['@id'] for d in response.data['ldp:contains']]
self.assertNotIn(dummy_a.urlid, containees)
self.assertIn(dummy_b.urlid, containees)
def test_list_dummy_group_exception(self):
self.setUpLoggedInUser()
dummy_a = self._get_dummy_with_perms()
dummy_b = self._get_dummy_with_perms(['view'], group=True)
response = self.client.get('/permissionless-dummys/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
containees = [d['@id'] for d in response.data['ldp:contains']]
self.assertNotIn(dummy_a.urlid, containees)
self.assertIn(dummy_b.urlid, containees)
def test_list_dummy_exception_nested_view(self):
self.setUpLoggedInUser()
parent = LDPDummy.objects.create(some="test")
# two dummies, one I have permission to view and one I don't
dummy_a = self._get_dummy_with_perms(parent=parent)
dummy_b = self._get_dummy_with_perms(['view'], parent)
response = self.client.get('/ldpdummys/{}/anons/'.format(parent.pk))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
def test_list_dummy_exception_nested_serializer(self):
self.setUpLoggedInUser()
parent = LDPDummy.objects.create(some="test")
# two dummies, one I have permission to view and one I don't
dummy_a = self._get_dummy_with_perms(parent=parent)
dummy_b = self._get_dummy_with_perms(['view'], parent)
response = self.client.get('/ldpdummys/{}/'.format(parent.pk))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['anons']['ldp:contains']), 1)
def test_get_dummy_permission_granted(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['view'])
response = self.client.get('/permissionless-dummys/{}/'.format(self.dummy.slug))
self.assertEqual(response.status_code, 200)
def test_get_dummy_group_permission_granted(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['view'], group=True)
response = self.client.get('/permissionless-dummys/{}/'.format(self.dummy.slug))
self.assertEqual(response.status_code, 200)
def test_get_dummy_permission_rejected(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['view'])
dummy_without = PermissionlessDummy.objects.create(some='test2', slug='test2')
response = self.client.get('/permissionless-dummys/{}/'.format(dummy_without.slug))
self.assertEqual(response.status_code, 403)
self.assertEqual(response.status_code, 404)
def test_patch_dummy_permission_granted(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['change'])
self.setUpGuardianDummyWithPerms(['view', 'change'])
body = {'some': "some_new"}
response = self.client.patch('/permissionless-dummys/{}/'.format(self.dummy.slug), data=json.dumps(body),
content_type='application/ld+json')
......@@ -66,16 +138,17 @@ class TestsGuardian(APITestCase):
body = {'some': "some_new"}
response = self.client.patch('/permissionless-dummys/{}/'.format(dummy_without.slug), data=json.dumps(body),
content_type='application/ld+json')
self.assertEqual(response.status_code, 403)
self.assertEqual(response.status_code, 404)
# TODO: PUT container of many objects approved on specific resource for which I do not have _model_ permissions
# test that custom permissions are returned on a model
def test_custom_permissions(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['custom_permission'])
self.setUpGuardianDummyWithPerms(['custom_permission', 'view'])
permissions = LDPPermissions()
result = permissions.user_permissions(self.user, self.dummy)
self.assertIn('custom_permission', result)
response = self.client.get('/permissionless-dummys/{}/'.format(self.dummy.slug))
self.assertIn('custom_permission', self._unpack_permissions(response.data['permissions']))
# test that duplicate permissions aren't returned
def test_no_duplicate_permissions(self):
......@@ -85,6 +158,11 @@ class TestsGuardian(APITestCase):
assign_perm('view_' + model_name, self.user, dummy)
permissions = LDPPermissions()
result = permissions.user_permissions(self.user, dummy)
self.assertEqual(result.count('view'), 1)
response = self.client.get('/dummys/{}/'.format(dummy.slug))
self.assertEqual(response.status_code, 200)
perms = self._unpack_permissions(response.data['permissions'])
self.assertIn('view', perms)
view_perms = [perm for perm in perms if perm == 'view']
self.assertEqual(len(view_perms), 1)
# TODO: attempting to migrate my object permissions by changing FK reference
djangoldp/tests/tests_inbox.py
View file @ e97d33ad
import json
from django.contrib.auth import get_user_model
from django.conf import settings
from django.contrib.auth import get_user_model
from django.test import override_settings
from rest_framework.test import APIClient, APITestCase
from djangoldp.tests.models import Circle, CircleMember, Project, DateModel, DateChild
from djangoldp.models import Activity, Follower
from djangoldp.tests.models import Circle, DateChild, DateModel, Project
class TestsInbox(APITestCase):
......@@ -18,7 +20,7 @@ class TestsInbox(APITestCase):
res = {
"@context": [
"https://www.w3.org/ns/activitystreams",
{"hd": "http://happy-dev.fr/owl/#"}
{"hd": "https://cdn.startinblox.com/owl#"}
],
"summary": "Something happened",
"type": type,
......@@ -48,7 +50,7 @@ class TestsInbox(APITestCase):
def _assert_activity_created(self, response, activity_len=1):
'''Auxiliary function asserts that the activity was created and returned correctly'''
activities = Activity.objects.all()
self.assertEquals(len(activities), activity_len)
self.assertEqual(len(activities), activity_len)
self.assertIn(response["Location"], activities.values_list('urlid', flat=True))
def _assert_follower_created(self, local_urlid, external_urlid):
......@@ -76,18 +78,49 @@ class TestsInbox(APITestCase):
# assert that the circle was created and the user associated as owner
circles = Circle.objects.all()
self.assertEquals(len(circles), 1)
self.assertEqual(len(circles), 1)
self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
self.assertEqual(circles[0].owner, self.user)
self._assert_activity_created(response)
# assert external circle member now following local user
self.assertEquals(Follower.objects.count(), 1)
self.assertEqual(Follower.objects.count(), 1)
self._assert_follower_created(self.user.urlid, "https://distant.com/circles/1/")
# # tests creation, and tests that consequential creation also happens
# # i.e. that I pass it an external circle which it doesn't know about, and it creates that too
# def test_create_activity_circle_member(self):
# obj = {
# "@type": "hd:circlemember",
# "@id": "https://distant.com/circlemembers/1/",
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": "https://distant.com/circles/1/"
# }
# }
# payload = self._get_activity_request_template("Create", obj)
# response = self.client.post('/inbox/',
# data=json.dumps(payload), content_type='application/ld+json')
# self.assertEqual(response.status_code, 201)
# # assert that the circle was created and the user associated as member
# circles = Circle.objects.all()
# self.assertEqual(len(circles), 1)
# self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
# self.assertTrue(circles[0].members.filter(user=self.user).exists())
# self._assert_activity_created(response)
# # assert external circle member now following local user
# self._assert_follower_created(self.user.urlid, "https://distant.com/circlemembers/1/")
# sender has sent a circle with a local user that doesn't exist
def test_create_activity_circle_local(self):
urlid = '{}{}'.format(settings.SITE_URL, 'someonewhodoesntexist')
urlid = '{}/{}'.format(settings.SITE_URL, 'someonewhodoesntexist')
obj = {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/",
......@@ -105,8 +138,8 @@ class TestsInbox(APITestCase):
self.assertEqual(response.status_code, 404)
# assert that the circle was not created neither a backlinked user
self.assertEquals(Circle.objects.count(), 0)
self.assertEquals(get_user_model().objects.count(), prior_users_length)
self.assertEqual(Circle.objects.count(), 0)
self.assertEqual(get_user_model().objects.count(), prior_users_length)
#
# ADD ACTIVITIES
......@@ -127,92 +160,93 @@ class TestsInbox(APITestCase):
# assert that the project backlink(s) & activity were created
projects = Project.objects.all()
user_projects = self.user.projects.all()
self.assertEquals(len(projects), 1)
self.assertEquals(len(user_projects), 1)
self.assertEqual(len(projects), 1)
self.assertEqual(len(user_projects), 1)
self.assertIn("https://distant.com/projects/1/", projects.values_list('urlid', flat=True))
self.assertIn("https://distant.com/projects/1/", user_projects.values_list('urlid', flat=True))
self._assert_activity_created(response)
# assert external circle member now following local user
self.assertEquals(Follower.objects.count(), 1)
self.assertEqual(Follower.objects.count(), 1)
self._assert_follower_created(self.user.urlid, "https://distant.com/projects/1/")
# circle model has a many-to-many with user, through an intermediate model
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
def test_add_activity_circle(self):
ext_circlemember_urlid = "https://distant.com/circle-members/1/"
ext_circle_urlid = "https://distant.com/circles/1/"
obj = {
"@type": "hd:circlemember",
"@id": ext_circlemember_urlid,
"user": {
"@type": "foaf:user",
"@id": self.user.urlid
},
"circle": {
"@type": "hd:circle",
"@id": ext_circle_urlid
}
}
payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
response = self.client.post('/inbox/',
data=json.dumps(payload), content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
self.assertEqual(response.status_code, 201)
# assert that the circle backlink(s) & activity were created
circles = Circle.objects.all()
user_circles = self.user.circles.all()
self.assertEquals(len(circles), 1)
self.assertEquals(len(user_circles), 1)
self.assertIn(ext_circle_urlid, circles.values_list('urlid', flat=True))
self.assertIn(ext_circlemember_urlid, user_circles.values_list('urlid', flat=True))
self._assert_activity_created(response)
# assert external circle member now following local user
self.assertEquals(Follower.objects.count(), 1)
self._assert_follower_created(self.user.urlid, ext_circlemember_urlid)
#TODO: write a new test for the new circle architecture
# # circle model has a many-to-many with user, through an intermediate model
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_circle(self):
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# ext_circle_urlid = "https://distant.com/circles/1/"
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": ext_circle_urlid
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload), content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self.assertEqual(response.status_code, 201)
# # assert that the circle backlink(s) & activity were created
# circles = Circle.objects.all()
# user_circles = self.user.circles.all()
# self.assertEqual(len(circles), 1)
# self.assertEqual(len(user_circles), 1)
# self.assertIn(ext_circle_urlid, circles.values_list('urlid', flat=True))
# self.assertIn(ext_circlemember_urlid, user_circles.values_list('urlid', flat=True))
# self._assert_activity_created(response)
# # assert external circle member now following local user
# self.assertEqual(Follower.objects.count(), 1)
# self._assert_follower_created(self.user.urlid, ext_circlemember_urlid)
# test sending an add activity when the backlink already exists
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
def test_add_activity_object_already_added(self):
circle = Circle.objects.create(urlid="https://distant.com/circles/1/")
cm = CircleMember.objects.create(urlid="https://distant.com/circle-members/1/", circle=circle, user=self.user)
obj = {
"@type": "hd:circlemember",
"@id": "https://distant.com/circle-members/1/",
"user": {
"@type": "foaf:user",
"@id": self.user.urlid
},
"circle": {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/"
}
}
payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
prior_count = Activity.objects.count()
response = self.client.post('/inbox/',
data=json.dumps(payload),
content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
self.assertEqual(response.status_code, 201)
# assert that the circle backlink(s) & activity were created
circles = Circle.objects.all()
user_circles = self.user.circles.all()
self.assertEquals(len(circles), 1)
self.assertEquals(len(user_circles), 1)
self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
self.assertIn("https://distant.com/circle-members/1/", user_circles.values_list('urlid', flat=True))
self._assert_activity_created(response)
self.assertEqual(Activity.objects.count(), prior_count + 1)
# assert that followers exist for the external urlids
self.assertEquals(Follower.objects.count(), 1)
self._assert_follower_created(self.user.urlid, cm.urlid)
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_object_already_added(self):
# circle = Circle.objects.create(urlid="https://distant.com/circles/1/")
# circle.members.user_set.add(self.user)
# obj = {
# "@type": "hd:circlemember",
# "@id": "https://distant.com/circle-members/1/",
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": "https://distant.com/circles/1/"
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# prior_count = Activity.objects.count()
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self.assertEqual(response.status_code, 201)
# # assert that the circle backlink(s) & activity were created
# circles = Circle.objects.all()
# user_circles = self.user.circles.all()
# self.assertEqual(len(circles), 1)
# self.assertEqual(len(user_circles), 1)
# self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
# self.assertIn("https://distant.com/circle-members/1/", user_circles.values_list('urlid', flat=True))
# self._assert_activity_created(response)
# self.assertEqual(Activity.objects.count(), prior_count + 1)
# # assert that followers exist for the external urlids
# self.assertEqual(Follower.objects.count(), 1)
# self._assert_follower_created(self.user.urlid, '') #TODO: replace with an existing model
# TODO: https://git.startinblox.com/djangoldp-packages/djangoldp/issues/250
def test_add_activity_str_parameter(self):
......@@ -236,11 +270,144 @@ class TestsInbox(APITestCase):
data=json.dumps(payload), content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
self.assertEqual(response.status_code, 404)
def _test_fail_behaviour(self, response, status_code=400):
self.assertEqual(response.status_code, 400)
# assert that nothing was created
self.assertEqual(Circle.objects.count(), 0)
self.assertEqual(self.user.owned_circles.count(), 0)
self.assertEqual(Activity.objects.count(), 0)
self.assertEqual(Follower.objects.count(), 0)
# # error behaviour - invalid url
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_empty_url(self):
# # an invalid url
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# ext_circle_urlid = ""
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": ext_circle_urlid
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self._test_fail_behaviour(response, 400)
# # error behaviour - invalid url
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_invalid_url(self):
# # an invalid url
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# ext_circle_urlid = "not$valid$url"
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": ext_circle_urlid
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self._test_fail_behaviour(response, 400)
# # # error behaviour - None url
# # @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# # def test_add_activity_none_url(self):
# # # an invalid url
# # ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# # ext_circle_urlid = None
# # obj = {
# # "@type": "hd:circlemember",
# # "@id": ext_circlemember_urlid,
# # "user": {
# # "@type": "foaf:user",
# # "@id": self.user.urlid
# # },
# # "circle": {
# # "@type": "hd:circle",
# # "@id": ext_circle_urlid
# # }
# # }
# # payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# # response = self.client.post('/inbox/',
# # data=json.dumps(payload),
# # content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# # self._test_fail_behaviour(response, 400)
# # missing @id on a sub-object
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_no_id(self):
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle"
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self._test_fail_behaviour(response, 400)
# # missing @type on a sub-object
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_no_type(self):
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@id": "https://distant.com/circles/1/"
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self._test_fail_behaviour(response, 404)
def test_invalid_activity_missing_actor(self):
payload = {
"@context": [
"https://www.w3.org/ns/activitystreams",
{"hd": "http://happy-dev.fr/owl/#"}
{"hd": "https://cdn.startinblox.com/owl#"}
],
"summary": "Test was added to Test Circle",
"type": "Add",
......@@ -275,8 +442,8 @@ class TestsInbox(APITestCase):
# assert that the project backlink(s) & activity were created
projects = Project.objects.all()
user_projects = self.user.projects.all()
self.assertEquals(len(projects), 1)
self.assertEquals(len(user_projects), 1)
self.assertEqual(len(projects), 1)
self.assertEqual(len(user_projects), 1)
self.assertIn("https://distant.com/projects/1/", projects.values_list('urlid', flat=True))
self.assertIn("https://distant.com/projects/1/", user_projects.values_list('urlid', flat=True))
self._assert_activity_created(response)
......@@ -284,6 +451,7 @@ class TestsInbox(APITestCase):
self.assertNotEqual(backlink.pk, 100)
def test_missing_not_null_field_activity(self):
# TODO: catch the warning
# DateChild must not have a null reference to parent
# and parent must not have a null field 'date', which here is missing
obj = {
......@@ -328,8 +496,8 @@ class TestsInbox(APITestCase):
# assert that the circle backlink(s) were removed & activity were created
projects = Project.objects.all()
user_projects = self.user.projects.all()
self.assertEquals(len(projects), 1)
self.assertEquals(len(user_projects), 0)
self.assertEqual(len(projects), 1)
self.assertEqual(len(user_projects), 0)
self.assertIn("https://distant.com/projects/1/", projects.values_list('urlid', flat=True))
self._assert_activity_created(response, prior_activity_count + 1)
self.assertEqual(Follower.objects.count(), 0)
......@@ -379,41 +547,41 @@ class TestsInbox(APITestCase):
# just received, did not send
self.assertEqual(Activity.objects.all().count(), prior_count + 1)
# Delete CircleMember
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
def test_delete_activity_circle_using_origin(self):
circle = Circle.objects.create(urlid="https://distant.com/circles/1/", allow_create_backlink=False)
cm = CircleMember.objects.create(urlid="https://distant.com/circle-members/1/",circle=circle, user=self.user)
Follower.objects.create(object=self.user.urlid, inbox='https://distant.com/inbox/',
follower=cm.urlid, is_backlink=True)
obj = {
"@type": "hd:circlemember",
"@id": "https://distant.com/circle-members/1/",
"user": {
"@type": "foaf:user",
"@id": self.user.urlid
},
"circle": {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/"
}
}
payload = self._get_activity_request_template("Delete", obj)
response = self.client.post('/inbox/',
data=json.dumps(payload),
content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
self.assertEqual(response.status_code, 201)
# assert that the CircleMember was deleted and activity was created
circles = Circle.objects.all()
user_circles = self.user.circles.all()
self.assertEquals(len(circles), 1)
self.assertEquals(CircleMember.objects.count(), 0)
self.assertEquals(len(user_circles), 0)
self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
self._assert_activity_created(response)
self.assertEqual(Follower.objects.count(), 0)
# # Delete CircleMember
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_delete_activity_circle_using_origin(self):
# circle = Circle.objects.create(urlid="https://distant.com/circles/1/", allow_create_backlink=False)
# circle.members.user_set.add(self.user)
# Follower.objects.create(object=self.user.urlid, inbox='https://distant.com/inbox/',
# follower=circle.urlid, is_backlink=True)
# obj = {
# "@type": "hd:circlemember",
# "@id": "https://distant.com/circle-members/1/",
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": "https://distant.com/circles/1/"
# }
# }
# payload = self._get_activity_request_template("Delete", obj)
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self.assertEqual(response.status_code, 201)
# # assert that the Circle member was removed and activity was created
# circles = Circle.objects.all()
# user_circles = self.user.circles.all()
# self.assertEqual(len(circles), 1)
# self.assertEqual(circle.members.count(), 0)
# self.assertEqual(len(user_circles), 0)
# self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
# self._assert_activity_created(response)
# self.assertEqual(Follower.objects.count(), 0)
# TODO: test_delete_activity_circle_using_target
......@@ -424,6 +592,8 @@ class TestsInbox(APITestCase):
circle = Circle.objects.create(urlid="https://distant.com/circles/1/", owner=self.user)
self.assertEqual(circle.owner, self.user)
prior_user_count = get_user_model().objects.count()
obj = {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/",
......@@ -440,8 +610,8 @@ class TestsInbox(APITestCase):
# assert that the circle was created and the user associated as owner
circles = Circle.objects.all()
users = get_user_model().objects.all()
self.assertEquals(len(circles), 1)
self.assertEquals(len(users), 2)
self.assertEqual(len(circles), 1)
self.assertEqual(len(users), prior_user_count + 1)
distant_user = get_user_model().objects.get(urlid="https://distant.com/users/1/")
self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
self.assertEqual(circles[0].owner, distant_user)
......@@ -464,7 +634,7 @@ class TestsInbox(APITestCase):
# assert that Follower was created with correct values
followers = Follower.objects.all()
self.assertEquals(len(followers), 1)
self.assertEqual(len(followers), 1)
self._assert_activity_created(response)
follower = followers[0]
self.assertEqual("http://127.0.0.1:8000/inbox/", follower.inbox)
......@@ -475,10 +645,10 @@ class TestsInbox(APITestCase):
circle = Circle.objects.create(description='Test Description')
Follower.objects.create(object=circle.urlid, inbox="http://127.0.0.1:8000/inbox/")
followers = Follower.objects.all()
self.assertEquals(len(followers), 1)
self.assertEqual(len(followers), 1)
circle.delete()
followers = Follower.objects.all()
self.assertEquals(len(followers), 0)
self.assertEqual(len(followers), 0)
#
# GET Inbox
......
djangoldp/tests/tests_ldp_model.py
View file @ e97d33ad
import unittest
from django.test import TestCase
from djangoldp.models import Model
from djangoldp.tests.models import Dummy, LDPDummy, Circle, CircleMember
from djangoldp.tests.models import (Dummy, JobOffer, LDPDummy,
NoSuperUsersAllowedModel)
class LDPModelTest(TestCase):
def test_class_not_inheriting_ldp_model(self):
dummy = Dummy.objects.create(some="text")
self.assertEquals("/dummys/", Model.container_id(dummy))
self.assertEquals("/dummys/{}/".format(dummy.slug), Model.resource_id(dummy))
self.assertEqual("/dummys/", Model.container_id(dummy))
self.assertEqual("/dummys/{}/".format(dummy.slug), Model.resource_id(dummy))
def test_class_inheriting_ldp_model(self):
dummy = LDPDummy.objects.create(some="text")
self.assertEquals("/ldpdummys/", dummy.get_container_id())
self.assertEquals("http://happy-dev.fr/ldpdummys/{}/".format(dummy.pk), dummy.get_absolute_url())
self.assertEquals("/ldpdummys/", Model.container_id(dummy))
self.assertEquals("/ldpdummys/{}/".format(dummy.pk), Model.resource_id(dummy))
self.assertEqual("/ldpdummys/", dummy.get_container_id())
self.assertEqual("http://happy-dev.fr/ldpdummys/{}/".format(dummy.pk), dummy.get_absolute_url())
self.assertEqual("/ldpdummys/", Model.container_id(dummy))
self.assertEqual("/ldpdummys/{}/".format(dummy.pk), Model.resource_id(dummy))
def test_from_resolve_id(self):
saved_instance = Dummy.objects.create(some="text", slug="someid")
result = Model.resolve_id("/dummys/{}/".format(saved_instance.slug))
self.assertEquals(saved_instance, result)
self.assertEqual(saved_instance, result)
def test_resolve_container(self):
result = Model.resolve_container("/dummys/")
self.assertEquals(Dummy, result)
self.assertEqual(Dummy, result)
def test_auto_url(self):
from django.urls import get_resolver
......@@ -35,7 +34,7 @@ class LDPModelTest(TestCase):
view_name = '{}-list'.format(dummy._meta.object_name.lower())
path = 'http://happy-dev.fr/{}{}/'.format(get_resolver().reverse_dict[view_name][0][0][0], dummy.pk)
self.assertEquals(path, dummy.get_absolute_url())
self.assertEqual(path, dummy.get_absolute_url())
def test_ldp_manager_local_objects(self):
local = LDPDummy.objects.create(some='text')
......@@ -44,21 +43,4 @@ class LDPModelTest(TestCase):
local_queryset = LDPDummy.objects.local()
self.assertEqual(local_queryset.count(), 1)
self.assertIn(local, local_queryset)
self.assertNotIn(external, local_queryset)
def test_ldp_manager_nested_fields_auto(self):
nested_fields = Circle.objects.nested_fields()
expected_nested_fields = ['team', 'members']
self.assertEqual(len(nested_fields), len(expected_nested_fields))
for expected in expected_nested_fields:
self.assertIn(expected, nested_fields)
nested_fields = CircleMember.objects.nested_fields()
expected_nested_fields = []
self.assertEqual(nested_fields, expected_nested_fields)
def test_ldp_manager_nested_fields_exclude(self):
setattr(Circle.Meta, 'nested_fields_exclude', ['team'])
nested_fields = Circle.objects.nested_fields()
expected_nested_fields = ['members']
self.assertEqual(nested_fields, expected_nested_fields)
self.assertNotIn(external, local_queryset)
\ No newline at end of file
djangoldp/tests/tests_ldp_viewset.py 0 → 100644
View file @ e97d33ad
from django.contrib.auth import get_user_model
from rest_framework import status
from rest_framework.test import APIRequestFactory, APIClient, APITestCase
from djangoldp.tests.models import User, Circle, Project
from djangoldp.serializers import LDPSerializer
from djangoldp.related import get_prefetch_fields
class LDPViewSet(APITestCase):
user_serializer_fields = ['@id', 'username', 'first_name', 'last_name', 'email', 'userprofile', 'conversation_set', 'projects']
user_expected_fields = {'userprofile', 'conversation_set', 'projects', 'conversation_set__author_user', 'conversation_set__peer_user'}
project_serializer_fields = ['@id', 'description', 'members']
project_expected_fields = {'members', 'members__userprofile'}
def setUpLoggedInUser(self):
self.factory = APIRequestFactory()
self.client = APIClient()
self.user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com',
password='glass onion', first_name='John')
self.client.force_authenticate(self.user)
def _get_serializer(self, model, depth, fields):
meta_args = {'model': model, 'depth': depth, 'fields': fields}
meta_class = type('Meta', (), meta_args)
return (type(LDPSerializer)('TestSerializer', (LDPSerializer,), {'Meta': meta_class}))()
def test_get_prefetch_fields_user(self):
model = User
depth = 0
serializer_fields = self.user_serializer_fields
expected_fields = self.user_expected_fields
serializer = self._get_serializer(model, depth, serializer_fields)
result = get_prefetch_fields(model, serializer, depth)
self.assertEqual(expected_fields, result)
def test_get_prefetch_fields_circle(self):
model = Circle
depth = 0
serializer_fields = ['@id', 'name', 'description', 'owner', 'members']
expected_fields = {'owner', 'members', 'admins', 'space'}
serializer = self._get_serializer(model, depth, serializer_fields)
result = get_prefetch_fields(model, serializer, depth)
self.assertEqual(expected_fields, result)
def test_get_prefetch_fields_project(self):
model = Project
depth = 0
serializer_fields = self.project_serializer_fields
expected_fields = self.project_expected_fields
serializer = self._get_serializer(model, depth, serializer_fields)
result = get_prefetch_fields(model, serializer, depth)
self.assertEqual(expected_fields, result)
# TODO: dynamically generating serializer fields is necessary to retrieve many-to-many fields at depth > 0,
# but the _all_ default has issues detecting reverse many-to-many fields
'''def test_get_prefetch_fields_depth_1(self):
model = Project
depth = 2
serializer_fields = self.project_serializer_fields
user_expected = set(['team__' + x for x in self.user_expected_fields])
expected_fields = self.project_expected_fields.union(user_expected)
serializer = self._get_serializer(model, depth, serializer_fields)
result = get_prefetch_fields(model, serializer, depth)
self.assertEqual(expected_fields, result)'''
def test_get_shape_param(self):
self.setUpLoggedInUser()
circle = Circle.objects.create(name='test circle')
# request id and name only
fields_shape = '["@id", "name"]'
response = self.client.get('/circles/', HTTP_ACCEPT_MODEL_FIELDS=fields_shape)
self.assertEqual(response.status_code, status.HTTP_200_OK)
response_data_keys = response.data['ldp:contains'][0].keys()
self.assertIn(len(response_data_keys), (3,4))
self.assertIn('@id', response_data_keys)
self.assertIn('name', response_data_keys)
self.assertIn('@type', response_data_keys)
if len(response_data_keys)>3:
self.assertIn('@context', response_data_keys)
def test_search_fields_basic(self):
self.setUpLoggedInUser()
lowercase_circle = Circle.objects.create(name='test circle')
uppercase_circle = Circle.objects.create(name='hello world', description='test')
response = self.client.get('/circles/?search-fields=name&search-terms=test&search-method=basic')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['ldp:contains']), 1)
self.assertEqual(response.data['ldp:contains'][0]['name'], lowercase_circle.name)
# test multiple search fields
response = self.client.get('/circles/?search-fields=name,description&search-terms=test&search-method=basic')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['ldp:contains']), 2)
def test_search_fields_ibasic(self):
self.setUpLoggedInUser()
lowercase_circle = Circle.objects.create(name='test circle')
uppercase_circle = Circle.objects.create(name='TEST')
response = self.client.get('/circles/?search-fields=name&search-terms=test&search-method=ibasic')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['ldp:contains']), 2)
def test_search_fields_exact(self):
self.setUpLoggedInUser()
lowercase_circle = Circle.objects.create(name='test circle')
uppercase_circle = Circle.objects.create(name='TEST')
response = self.client.get('/circles/?search-fields=name&search-terms=test&search-method=exact')
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['ldp:contains']), 0)
response = self.client.get('/circles/?search-fields=name&search-terms=test%20circle&search-method=exact')
self.assertEqual(response.data['ldp:contains'][0]['name'], lowercase_circle.name)