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 2318 additions and 69 deletions
djangoldp/tests/perf_result.csv 0 → 100644
View file @ e97d33ad
Machine,Date,Auth,WithPermsCache,volume,test+AF8-get+AF8-resource,test+AF8-get+AF8-container,test+AF8-get+AF8-filtered+AF8-fields,test+AF8-get+AF8-reverse+AF8-filtered+AF8-fields,test+AF8-get+AF8-nested,test+AF8-get+AF8-users+AF8-container,Prefetch,default depth
jbl+AC0-T440p,Sep 22 2020 10:50:51,False,False,200,0.003339644670486,0.006944504976273,0.038935685157776,0.024031536579132,0.000708421468735,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:51:46,False,False,200,0.0035072016716,0.006944673061371,0.039835988283157,0.025360778570175,0.000757339000702,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:52:42,False,False,200,0.003284044265747,0.006942090988159,0.038870732784271,0.023859632015228,0.000705161094666,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:53:16,False,False,100,0.003656179904938,0.005776383876801,0.025797350406647,0.01539302110672,0.000770201683044,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:53:33,False,False,100,0.003554759025574,0.005703027248383,0.024777753353119,0.015221126079559,0.000770528316498,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:53:49,False,False,100,0.003367004394531,0.005602278709412,0.023594326972962,0.014168989658356,0.000726828575134,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:54:03,False,False,50,0.003355793952942,0.005232772827148,0.016062431335449,0.009248399734497,0.000776686668396,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:54:09,False,False,50,0.003454508781433,0.005315055847168,0.016247057914734,0.009447617530823,0.00073832988739,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:54:15,False,False,50,0.003420171737671,0.005717425346375,0.016275815963745,0.009424614906311,0.001325125694275,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:57:41,False,False,300,0.003357520103455,0.009047945340474,0.055130259990692,0.033688295682271,0.000706691741943,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 10:59:35,False,False,300,0.003680046399434,0.009138919512431,0.056478141943614,0.0363059147199,0.000769446690877,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 11:01:29,False,False,300,0.003643860816956,0.008885918458303,0.059775860309601,0.035221153100332,0.000756018956502,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 11:47:40,False,False,100,0.003384988307953,0.006034939289093,0.024095425605774,0.014140074253082,0.000722093582153,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 11:47:57,False,False,100,0.003611071109772,0.005775241851807,0.023724327087402,0.014749829769135,0.000745611190796,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 11:48:15,False,False,100,0.003316740989685,0.005551462173462,0.023505146503449,0.014274184703827,0.000737235546112,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 11:51:06,False,False,200,0.003252120018005,0.006922056674957,0.038872839212418,0.025012502670288,0.000715854167938,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 11:52:07,False,False,200,0.003315222263336,0.007173013687134,0.039467182159424,0.0239526450634,0.000736322402954,,FALSE,0
jbl+AC0-T440p,Sep 22 2020 11:53:59,False,False,200,0.003276619911194,0.006898198127747,0.038627609014511,0.023467609882355,0.000708512067795,,FALSE,0
jbl+AC0-T440p,Sep 23 2020 15:19:39,True,False,100,0.006617827415466,0.245147013664246,0.261345520019531,0.209938230514526,0.001274492740631,0.01203465461731,TRUE,0
jbl+AC0-T440p,Sep 23 2020 15:23:58,True,False,100,0.006518981456757,0.263970799446106,0.139407794475555,0.113074848651886,0.001235642433167,0.642078399658203,TRUE,0
jbl+AC0-T440p,Sep 23 2020 15:25:33,True,False,100,0.006539282798767,0.263329057693481,0.143536510467529,0.115545327663422,0.00125715970993,0.520937442779541,TRUE,0
jbl+AC0-T440p,Sep 23 2020 15:47:31,True,True,100,0.003864502906799,0.019258742332459,0.01636646270752,0.008915212154388,0.000782597064972,0.354249000549316,TRUE,0
jbl+AC0-T440p,Sep 23 2020 15:48:30,True,True,100,0.003590517044067,0.019879310131073,0.016916983127594,0.009615495204926,0.000798478126526,0.364127635955811,TRUE,0
jbl+AC0-T440p,Sep 23 2020 15:49:19,True,True,100,0.003716588020325,0.023860175609589,0.016380727291107,0.009003615379334,0.000774307250977,0.35044002532959,TRUE,0
jbl+AC0-T440p,Sep 23 2020 16:56:57,True,True,100,0.004425497055054,0.019956090450287,0.017114706039429,0.010911240577698,0.000832149982452,0.336694478988647,TRUE,0
jbl+AC0-T440p,Sep 23 2020 16:57:23,True,True,100,0.004397692680359,0.021222379207611,0.01826550245285,0.010625832080841,0.000812151432037,0.344640016555786,TRUE,0
jbl+AC0-T440p,Sep 23 2020 16:57:45,True,True,100,0.004602279663086,0.020297501087189,0.017129955291748,0.010850946903229,0.000865814685822,0.339866161346436,TRUE,0
jbl+AC0-T440p,Sep 23 2020 16:58:37,True,True,200,0.004463336467743,0.036689649820328,0.025502071380615,0.015932221412659,0.000898872613907,0.681740045547485,TRUE,0
jbl+AC0-T440p,Sep 23 2020 16:59:37,True,True,200,0.004517335891724,0.036278907060623,0.025654143095017,0.01576028585434,0.000849905014038,0.660790681838989,TRUE,0
jbl+AC0-T440p,Sep 23 2020 17:14:05,True,False,100,0.006808481216431,0.252511320114136,0.139744215011597,0.111351528167725,0.001188087463379,0.564764976501465,TRUE,0
jbl+AC0-T440p,Sep 23 2020 17:16:58,True,False,100,0.006502165794373,0.242799952030182,0.137602522373199,0.108403618335724,0.001143708229065,0.556174516677856,TRUE,0
jbl+AC0-T440p,Sep 24 2020 06:53:53,True,False,100,0.007479875087738,0.252197952270508,0.141312582492828,0.109222292900085,0.001601278781891,0.52592396736145,TRUE,0
jbl+AC0-T440p,Sep 24 2020 06:56:06,True,False,100,0.020340206623077,1.31586099863052,0.729812262058258,0.577438371181488,0.00078241109848,1.78533124923706,FALSE,0
jbl+AC0-T440p,Sep 24 2020 07:04:00,True,False,100,0.006233677864075,0.242916750907898,0.135480484962463,0.10392139673233,0.000762076377869,0.569819927215576,TRUE,0
jbl+AC0-T440p,Sep 24 2020 07:05:19,True,True,100,0.006471273899078,0.023659512996674,0.020732533931732,0.015365273952484,0.000769484043121,0.549034357070923,FALSE,0
jbl+AC0-T440p,Sep 24 2020 07:05:50,True,True,100,0.005183663368225,0.021180493831635,0.016473467350006,0.010494797229767,0.000771188735962,0.321053028106689,TRUE,0
jbl+AC0-T440p,Sep 24 2020 07:33:51,True,True,100,0.004018228054047,0.019896368980408,0.024588730335236,0.015463829040527,0.000797684192657,0.375835657119751,TRUE,0
jbl+AC0-T440p,Oct 08 2020 15:01:01,True,True,100,0.004086337089539,0.022444577217102,1.26581408977509,0.014725821018219,0.002466685771942,0.310548305511475,TRUE,1
jbl+AC0-T440p,Oct 08 2020 15:25:42,True,True,100,0.004928917884827,0.020494124889374,1.19440255403519,0.01545866727829,0.000807287693024,0.304153442382812,TRUE,1
jbl+AC0-T440p,Oct 08 2020 15:44:13,True,True,100,0.00410740852356,0.020648219585419,1.24338629007339,0.01569117307663,0.00077033996582,0.33369255065918,TRUE,1
jbl+AC0-T440p,Oct 08 2020 16:19:05,True,True,100,0.004798595905304,0.022070643901825,1.24563392400742,0.015214123725891,0.000787632465363,0.333659410476685,TRUE,1
jbl+AC0-T440p,Oct 09 2020 11:23:54,True,True,100,0.004018263816834,0.020824022293091,1.16614150524139,0.015614166259766,0.000755190849304,0.34318208694458,TRUE,1
jbl+AC0-T440p,Oct 09 2020 11:54:15,True,True,100,0.003045120239258,0.005557940006256,0.009205477237701,0.003398790359497,0.00073746919632,0.356267929077148,TRUE,1
jbl+AC0-T440p,Oct 09 2020 11:56:19,True,True,100,0.003119325637817,0.005602471828461,0.009082851409912,0.003396863937378,0.000744948387146,0.303446769714356,TRUE,2
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+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 0 → 100644
View file @ e97d33ad
import sys
import yaml
import django
from django.conf import settings as django_settings
from djangoldp.conf.ldpsettings import LDPSettings
from djangoldp.tests.server_settings import yaml_config
# 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
test_runner = DiscoverRunner(verbosity=1)
failures = test_runner.run_tests([
# 'djangoldp.tests.tests_performance',
'djangoldp.tests.tests_perf_get'
])
if failures:
sys.exit(failures)
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 django
import sys
from django.conf import settings
import yaml
settings.configure(DEBUG=True,
DATABASES={
'default': {
'ENGINE': 'django.db.backends.sqlite3',
}
},
ROOT_URLCONF='djangoldp.tests.urls',
DJANGOLDP_PACKAGES=['djangoldp.tests'],
INSTALLED_APPS=('django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.admin',
'guardian',
'djangoldp',
'djangoldp.tests',
))
import django
from django.conf import settings as django_settings
from djangoldp.conf.ldpsettings import LDPSettings
from djangoldp.tests.server_settings import yaml_config
# load test config
config = yaml.safe_load(yaml_config)
ldpsettings = LDPSettings(config)
django_settings.configure(ldpsettings)
django.setup()
from django.test.runner import DiscoverRunner
......@@ -26,11 +17,24 @@ 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_update'])
'djangoldp.tests.tests_permissions',
'djangoldp.tests.tests_post',
'djangoldp.tests.tests_update',
'djangoldp.tests.tests_auto_author',
'djangoldp.tests.tests_get',
'djangoldp.tests.tests_delete',
'djangoldp.tests.tests_sources',
'djangoldp.tests.tests_pagination',
'djangoldp.tests.tests_inbox',
'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 0 → 100644
View file @ e97d33ad
import json
import argparse
from pathlib import Path
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
for help run python test_data_generator.py -h
'''
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',
'pk': 0,
'fields': {
'username': 'john',
'email': 'jlennon@c.coop',
'password':'glassonion',
'first_name': 'John',
'last_name': 'Lennon'
}
}
project_template = {
'model': 'tests.project',
'pk': 0,
'fields': {
'status': 'Public',
'description': 'Test'
}
}
fixture = generate_users(count, user_template)
fixture = generate_projects(count, project_template, fixture=fixture, production=False)
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/tests_anonymous_permissions.py
View file @ e97d33ad
from django.contrib.auth.models import AnonymousUser
from django.test import TestCase, RequestFactory
import json
from django.test import TestCase
from rest_framework.test import APIClient
from djangoldp.permissions import AnonymousReadOnly
from djangoldp.tests.models import JobOffer
from djangoldp.views import LDPViewSet
class TestAnonymousUserPermissions(TestCase):
def setUp(self):
self.factory = RequestFactory()
# self.c = Client()
self.user = AnonymousUser
def test_get_request_with_anonymousUser(self):
request = self.factory.get("/job-offers/")
request.user = self.user
my_view = LDPViewSet.as_view({'get': 'list'},
model=JobOffer,
nested_fields=["skills"],
permission_classes=[AnonymousReadOnly])
response = my_view(request)
self.client = APIClient(enforce_csrf_checks=True)
self.job = JobOffer.objects.create(title="job", slug=1)
def test_get_request_for_anonymousUser(self):
response = self.client.get('/job-offers/')
self.assertEqual(response.status_code, 200)
def test_get_1_request_for_anonymousUser(self):
response = self.client.get('/job-offers/1/')
self.assertEqual(response.status_code, 200)
def test_request_options_create_with_anonymousUser(self):
request = self.factory.options("/job-offers/")
request.user = self.user
my_view = LDPViewSet.as_view({'options': 'create'},
model=JobOffer,
nested_fields=["skills"],
permission_classes=[AnonymousReadOnly])
response = my_view(request)
def test_post_request_for_anonymousUser(self):
post = {'title': "job_created"}
response = self.client.post('/job-offers/', data=json.dumps(post), content_type='application/ld+json')
self.assertEqual(response.status_code, 403)
def test_request_options_update_with_anonymousUser(self):
request = self.factory.options("/job-offers/")
request.user = self.user
my_view = LDPViewSet.as_view({'options': 'update'},
model=JobOffer,
nested_fields=["skills"],
permission_classes=[AnonymousReadOnly])
response = my_view(request)
# 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),
content_type='application/ld+json')
self.assertEqual(response.status_code, 403)
def test_patch_request_for_anonymousUser(self):
response = self.client.patch('/job-offers/' + str(self.job.pk) + "/",
content_type='application/ld+json')
self.assertEqual(response.status_code, 403)
djangoldp/tests/tests_auto_author.py 0 → 100644
View file @ e97d33ad
import json
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient, APIRequestFactory, APITestCase
from djangoldp.tests.models import UserProfile
class TestAutoAuthor(APITestCase):
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')
UserProfile.objects.create(user=self.user)
def tearDown(self):
self.user.delete()
def test_save_with_anonymous_user(self):
post = {
'@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.assertEqual(response.data['content'], "post content")
\ No newline at end of file
djangoldp/tests/tests_backlinks_service.py 0 → 100644
View file @ e97d33ad
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.activities.services import (BACKLINKS_ACTOR, ActivityPubService,
ActivityQueueService)
from djangoldp.models import Activity, ScheduledActivity
from djangoldp.tests.models import Circle, Project
class TestsBacklinksService(APITestCase):
def setUp(self):
self.client = APIClient(enforce_csrf_checks=True)
self.local_user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com',
password='glass onion')
def _get_random_external_user(self):
'''Auxiliary function creates a user with random external urlid and returns it'''
username = str(uuid.uuid4())
email = username + '@test.com'
urlid = 'https://distant.com/users/' + username
return get_user_model().objects.create_user(username=username, email=email, password='test', urlid=urlid)
# TODO: inbox discovery (https://git.startinblox.com/djangoldp-packages/djangoldp/issues/233)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_local_object_with_distant_foreign_key(self):
# a local Circle with a distant owner
local_circle = Circle.objects.create(description='Test')
external_user = self._get_random_external_user()
local_circle.owner = external_user
local_circle.save()
# assert that a activity was sent
self.assertEqual(Activity.objects.all().count(), 1)
# reset to a local user, another (update) activity should be sent
local_circle.owner = self.local_user
local_circle.save()
self.assertEqual(Activity.objects.all().count(), 2)
# external user should no longer be following the object. A further update should not send an activity
# TODO: https://git.startinblox.com/djangoldp-packages/djangoldp/issues/257
'''another_user = get_user_model().objects.create_user(username='test', email='test@test.com',
password='glass onion')
local_circle.owner = another_user
local_circle.save()
self.assertEqual(Activity.objects.all().count(), 2)'''
# re-add the external user as owner
local_circle.owner = external_user
local_circle.save()
# delete parent
local_circle.delete()
self.assertEqual(Activity.objects.all().count(), 4)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_local_object_with_external_m2m_join_leave(self):
# a local project with three distant users
project = Project.objects.create(description='Test')
external_a = self._get_random_external_user()
external_b = self._get_random_external_user()
external_c = self._get_random_external_user()
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.members.remove(external_a)
self.assertEqual(Activity.objects.all().count(), 4)
# clear the rest
project.members.clear()
self.assertEqual(Activity.objects.all().count(), 6)
prior_count = Activity.objects.all().count()
# once removed I should not be following the object anymore
project.delete()
self.assertEqual(Activity.objects.all().count(), prior_count)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_local_object_with_external_m2m_delete_parent(self):
project = Project.objects.create(description='Test')
external_a = self._get_random_external_user()
project.members.add(external_a)
prior_count = Activity.objects.all().count()
project.delete()
self.assertEqual(Activity.objects.all().count(), prior_count + 1)
# test that older ScheduledActivity is discarded for newer ScheduledActivity
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_old_invalid_scheduled_activity_discarded(self):
def send_two_activities_and_assert_old_discarded(obj):
# there are two scheduled activities with the same object, (and different time stamps)
old_activity = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='old')
old_scheduled = ActivityQueueService._save_sent_activity(old_activity, ScheduledActivity)
new_activity = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Update', summary='new')
new_scheduled = ActivityQueueService._save_sent_activity(new_activity, ScheduledActivity)
# both are sent to the ActivityQueueService
ActivityQueueService._activity_queue_worker('http://127.0.0.1:8001/idontexist/', old_scheduled)
time.sleep(0.1)
ActivityQueueService._activity_queue_worker('http://127.0.0.1:8001/idontexist/', new_scheduled)
time.sleep(0.1)
# assert that all scheduled activities were cleaned up
self.assertEqual(ScheduledActivity.objects.count(), 0)
# assert that ONLY the newly scheduled activity was sent
activities = Activity.objects.all()
self.assertEqual(Activity.objects.count(), 1)
astream = activities[0].to_activitystream()
self.assertEqual(astream['summary'], new_activity['summary'])
activities[0].delete()
# variation using expanded syntax
obj = {
'@id': 'https://test.com/users/test/'
}
send_two_activities_and_assert_old_discarded(obj)
# variation using id-only syntax
obj = 'https://test.com/users/test/'
send_two_activities_and_assert_old_discarded(obj)
# test that older ScheduledActivity is still sent if it's on a different object
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_old_valid_scheduled_activity_sent(self):
# there are two scheduled activities with different objects
obj = 'https://test.com/users/test1/'
activity_a = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='A')
scheduled_a = ActivityQueueService._save_sent_activity(activity_a, ScheduledActivity)
obj = 'https://test.com/users/test2/'
activity_b = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='B')
scheduled_b = ActivityQueueService._save_sent_activity(activity_b, ScheduledActivity)
# both are sent to the same inbox
ActivityQueueService._activity_queue_worker('http://127.0.0.1:8001/idontexist/', scheduled_a)
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.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')
def test_old_valid_scheduled_activity_sent_same_object(self):
obj = 'https://test.com/users/test1/'
target = {'@type': 'hd:skill', '@id': 'https://api.test1.startinblox.com/skills/4/'}
activity_a = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Add', summary='A', target=target)
scheduled_a = ActivityQueueService._save_sent_activity(activity_a, ScheduledActivity)
obj = 'https://test.com/users/test1/'
target = {'@type': 'hd:joboffer', '@id': 'https://api.test1.startinblox.com/job-offers/1/'}
activity_b = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Add', summary='B', target=target)
scheduled_b = ActivityQueueService._save_sent_activity(activity_b, ScheduledActivity)
# both are sent to the same inbox
ActivityQueueService._activity_queue_worker('http://127.0.0.1:8001/idontexist/', scheduled_a)
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.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
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_matching_origin_and_target_not_sent(self):
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/'}}
scheduled_a = ActivityQueueService._save_sent_activity(a, ScheduledActivity)
b = {'type': 'Remove', 'actor': {'type': 'Service', 'name': 'Backlinks Service'},
'object': {'@type': 'foaf:user', '@id': 'https://api.test2.startinblox.com/users/calum/'},
'origin': {'@type': 'hd:skill', '@id': 'https://api.test1.startinblox.com/skills/3/'}}
scheduled_b = ActivityQueueService._save_sent_activity(b, ScheduledActivity)
# both are sent to the same inbox
ActivityQueueService._activity_queue_worker('http://127.0.0.1:8001/idontexist/', scheduled_a)
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.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):
# an add activity was sent previously
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_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',
external_id='https://distant.com/inbox/')
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.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):
# an remove activity was sent previously
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_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',
external_id='https://distant.com/inbox/')
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.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):
# a remove activity was sent previously
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_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',
external_id='https://distant.com/inbox/')
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled_b)
# assert that both activities sent, and the scheduled activites cleaned up
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):
# no activity has been sent with this target, before this add
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/'}}
scheduled = ActivityQueueService._save_sent_activity(a, ScheduledActivity, success=True, type='add',
external_id='https://distant.com/inbox/')
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled)
# assert that the activity was sent, and the scheduled activites cleaned up
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')
def test_unnecessary_update_not_sent(self):
# an object was sent in one activity
obj = {
'@type': 'hd:circle',
'@id': 'https://test.com/circles/8/',
'owner': {'@id': 'https://distant.com/users/john/',
'@type': 'foaf:user'}
}
activity_a = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='A')
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')
scheduled_b = ActivityQueueService._save_sent_activity(activity_b, ScheduledActivity, type='update',
external_id='https://distant.com/inbox/')
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.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):
# an object was sent in one activity
obj = {
'@type': 'hd:circle',
'@id': 'https://test.com/circles/8/',
'owner': {'@id': 'https://distant.com/users/john/',
'@type': 'foaf:user'}
}
activity_a = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='A')
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/'
activity_b = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='B')
scheduled_b = ActivityQueueService._save_sent_activity(activity_b, ScheduledActivity, type='update',
external_id='https://distant.com/inbox/')
ActivityQueueService._activity_queue_worker('https://distant.com/inbox/', scheduled_b)
# assert that both activities were sent
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):
# no prior activity was sent for this object - should send
obj = {
'@type': 'hd:circle',
'@id': 'https://test.com/circles/8/',
'owner': {'@id': 'https://distant.com/users/john/',
'@type': 'foaf:user'}
}
activity = ActivityPubService.build_activity(BACKLINKS_ACTOR, obj, activity_type='Create', summary='A')
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.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_delete.py 0 → 100644
View file @ e97d33ad
from django.test import TestCase
from rest_framework.test import APIRequestFactory, APIClient
from djangoldp.tests.models import Post
class TestDelete(TestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.client = APIClient()
def tearDown(self):
pass
def test_delete(self):
post = Post.objects.create(content="content")
response = self.client.delete('/posts/{}/'.format(post.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 204)
self.assertEqual(Post.objects.filter(pk=post.pk).count(), 0)
djangoldp/tests/tests_get.py 0 → 100644
View file @ e97d33ad
from datetime import datetime
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):
def setUp(self):
self.factory = APIRequestFactory()
self.client = APIClient()
self.ordered_fields = ['@context', '@type', '@id']
setattr(Circle._meta, 'depth', 0)
setattr(Circle._meta, 'empty_containers', [])
def tearDown(self):
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', HTTP_ORIGIN='http://localhost:8080/test/')
self.assertEqual(response.status_code, 200)
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'])
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)
response = self.client.get('/posts/{}/'.format(post.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['content'], "content")
self.assertEqual(response.data['author']['@id'], user.urlid)
def test_get_container(self):
Post.objects.create(content="content")
# federated object - should not be returned in the container view
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.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.assertEqual(1, len(response.data['permissions'])) # read only
def test_get_empty_container(self):
response = self.client.get('/posts/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
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://happy-dev.hubl.fr/skills/1")
job = JobOffer.objects.create(title="job", slug="1")
job.skills.add(skill)
job.skills.add(skill2)
job.skills.add(skill3)
job.save()
response = self.client.get('/job-offers/{}/'.format(job.slug), content_type='application/ld+json')
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/")
# 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")
job = JobOffer.objects.create(title="job", slug="1")
job.skills.add(skill)
job.save()
response = self.client.get('/skills/{}/'.format(skill.slug), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertIn('recent_jobs', response.data)
self.assertEqual(response.data['recent_jobs']['@id'], "http://happy-dev.fr/skills/1/recent_jobs/")
def test_get_virtual_field(self):
skill = Skill.objects.create(title="Java", obligatoire="ok", slug="1")
skill2 = Skill.objects.create(title="Java", obligatoire="ok", slug="2")
job = JobOffer.objects.create(title="job", slug="1")
job.skills.add(skill)
job.skills.add(skill2)
job.save()
response = self.client.get('/job-offers/{}/'.format(job.slug), content_type='application/ld+json')
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")
batch = Batch.objects.create(invoice=invoice, title="batch")
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.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('/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('/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 0 → 100644
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, LDPDummy
class TestsGuardian(APITestCase):
def setUp(self):
self.client = APIClient(enforce_csrf_checks=True)
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)
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:
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, 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, 404)
def test_patch_dummy_permission_granted(self):
self.setUpLoggedInUser()
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')
self.assertEqual(response.status_code, 200)
def test_patch_dummy_permission_rejected(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['change'])
dummy_without = PermissionlessDummy.objects.create(some='test2', slug='test2')
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, 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', 'view'])
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):
self.setUpLoggedInUser()
dummy = Dummy.objects.create(some='test', slug='test')
model_name = Dummy._meta.model_name
assign_perm('view_' + model_name, self.user, dummy)
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 0 → 100644
View file @ e97d33ad
import json
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.models import Activity, Follower
from djangoldp.tests.models import Circle, DateChild, DateModel, Project
class TestsInbox(APITestCase):
def setUp(self):
self.client = APIClient(enforce_csrf_checks=True)
self.user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com',
password='glass onion')
def _get_activity_request_template(self, type, obj, target=None, origin=None):
res = {
"@context": [
"https://www.w3.org/ns/activitystreams",
{"hd": "https://cdn.startinblox.com/owl#"}
],
"summary": "Something happened",
"type": type,
"actor": {
"type": "Service",
"name": "Backlinks Service",
"inbox": "http://127.0.0.1:8000/inbox/"
},
"object": obj
}
if target is not None:
res.update({"target": target})
if origin is not None:
res.update({"origin": origin})
return res
def _build_target_from_user(self, user):
return {
"@type": "foaf:user",
"name": user.get_full_name(),
"@id": user.urlid
}
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.assertEqual(len(activities), activity_len)
self.assertIn(response["Location"], activities.values_list('urlid', flat=True))
def _assert_follower_created(self, local_urlid, external_urlid):
existing_followers = Follower.objects.filter(object=local_urlid).values_list('follower', flat=True)
self.assertTrue(external_urlid in existing_followers)
#
# CREATE ACTIVITY
#
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
def test_create_activity_circle(self):
obj = {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/",
"owner": {
"@type": "foaf:user",
"@id": self.user.urlid
}
}
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 owner
circles = Circle.objects.all()
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.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')
obj = {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/",
"owner": {
"@type": "foaf:user",
"@id": urlid
}
}
payload = self._get_activity_request_template("Create", obj)
prior_users_length = get_user_model().objects.count()
response = self.client.post('/inbox/',
data=json.dumps(payload), content_type='application/ld+json')
self.assertEqual(response.status_code, 404)
# assert that the circle was not created neither a backlinked user
self.assertEqual(Circle.objects.count(), 0)
self.assertEqual(get_user_model().objects.count(), prior_users_length)
#
# ADD ACTIVITIES
#
# project model has a direct many-to-many with User
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
def test_add_activity_project(self):
obj = {
"@type": "hd:project",
"@id": "https://distant.com/projects/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')
self.assertEqual(response.status_code, 201)
# assert that the project backlink(s) & activity were created
projects = Project.objects.all()
user_projects = self.user.projects.all()
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.assertEqual(Follower.objects.count(), 1)
self._assert_follower_created(self.user.urlid, "https://distant.com/projects/1/")
#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/")
# 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):
payload = self._get_activity_request_template("Add", "https://distant.com/somethingunknown/1/",
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, 400)
# TODO: may pass an object without an explicit urlid e.g. Person actor, or Collection target
# error behaviour - unknown model
def test_add_activity_unknown(self):
obj = {
"@type": "hd:somethingunknown",
"@id": "https://distant.com/somethingunknown/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.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": "https://cdn.startinblox.com/owl#"}
],
"summary": "Test was added to Test Circle",
"type": "Add",
"object": {
"@type": "hd:somethingunknown",
"@id": "https://distant.com/somethingunknown/1/"
},
"target": {
"@type": "foaf:user",
"@id": self.user.urlid
}
}
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, 400)
# test activity setting unsafe fields in object
def test_unsafe_fields_in_activity(self):
obj = {
"@type": "hd:project",
"@id": "https://distant.com/projects/1/",
"pk": 100,
"id": 100
}
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')
self.assertEqual(response.status_code, 201)
# assert that the project backlink(s) & activity were created
projects = Project.objects.all()
user_projects = self.user.projects.all()
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)
backlink = Project.objects.get(urlid="https://distant.com/projects/1/")
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 = {
"@type": "hd:datechild",
"@id": "https://distant.com/datechilds/1/",
"parent": {
"@type": "hd:date",
"@id": "https://distant.com/dates/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, 200)
dates = DateModel.objects.all()
date_children = DateChild.objects.all()
self.assertEqual(len(dates), 0)
self.assertEqual(len(date_children), 0)
#
# REMOVE & DELETE ACTIVITIES
#
# project model has a direct many-to-many with User
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
def test_remove_activity_project_using_origin(self):
project = Project.objects.create(urlid="https://distant.com/projects/1/")
self.user.projects.add(project)
Follower.objects.create(object=self.user.urlid, inbox='https://distant.com/inbox/',
follower=project.urlid, is_backlink=True)
prior_activity_count = Activity.objects.count()
obj = {
"@type": "hd:project",
"@id": "https://distant.com/projects/1/"
}
payload = self._get_activity_request_template("Remove", obj, origin=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) were removed & activity were created
projects = Project.objects.all()
user_projects = self.user.projects.all()
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)
# TODO: test_remove_activity_project_using_target (https://git.startinblox.com/djangoldp-packages/djangoldp/issues/231)
# error behaviour - project does not exist on user
def test_remove_activity_nonexistent_project(self):
Project.objects.create(urlid="https://distant.com/projects/1/")
obj = {
"@type": "hd:project",
"@id": "https://distant.com/projects/1/"
}
payload = self._get_activity_request_template("Remove", obj, origin=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)
self._assert_activity_created(response)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX='DEBUG')
def test_removing_object_twice(self):
project = Project.objects.create(urlid="https://distant.com/projects/1/")
self.user.projects.add(project)
prior_count = Activity.objects.all().count()
# remove once via activity
obj = {
"@type": "hd:project",
"@id": "https://distant.com/projects/1/"
}
payload = self._get_activity_request_template("Remove", obj, origin=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)
# received and then sent
self.assertEqual(Activity.objects.all().count(), prior_count + 2)
prior_count = Activity.objects.all().count()
# sending remove activity again
payload = self._get_activity_request_template("Remove", obj, origin=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"')
# 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)
# 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
#
# UPDATE Activities
#
def test_update_activity_circle(self):
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/",
"owner": {
"@type": "foaf:user",
"@id": "https://distant.com/users/1/"
}
}
payload = self._get_activity_request_template("Update", 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 owner
circles = Circle.objects.all()
users = get_user_model().objects.all()
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)
self._assert_activity_created(response)
#
# FOLLOW activities
#
def test_follow_activity(self):
circle = Circle.objects.create(description='Test Description')
obj = {
"@type": "hd:circle",
"@id": circle.urlid
}
payload = self._get_activity_request_template("Follow", obj)
response = self.client.post('/inbox/',
data=json.dumps(payload), content_type='application/ld+json')
self.assertEqual(response.status_code, 201)
# assert that Follower was created with correct values
followers = Follower.objects.all()
self.assertEqual(len(followers), 1)
self._assert_activity_created(response)
follower = followers[0]
self.assertEqual("http://127.0.0.1:8000/inbox/", follower.inbox)
self.assertEqual(circle.urlid, follower.object)
# test Followers are auto-deleted when the object they're following is deleted
def test_follower_auto_delete(self):
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.assertEqual(len(followers), 1)
circle.delete()
followers = Follower.objects.all()
self.assertEqual(len(followers), 0)
#
# GET Inbox
#
def test_get_inbox(self):
response = self.client.get('/inbox/')
self.assertEqual(response.status_code, 405)
# TODO: GET inbox for specific resource - should return a list of activities sent to this inbox
# TODO: view to access outbox (https://git.startinblox.com/djangoldp-packages/djangoldp/issues/284)
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
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.pk), 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("/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")
result = Model.resolve_id("/dummys/{}".format(saved_instance.pk))
self.assertEquals(saved_instance, result)
saved_instance = Dummy.objects.create(some="text", slug="someid")
result = Model.resolve_id("/dummys/{}/".format(saved_instance.slug))
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
dummy = LDPDummy.objects.create(some="text")
view_name = '{}-list'.format(dummy._meta.object_name.lower())
path = '/{}{}'.format(get_resolver().reverse_dict[view_name][0][0][0], dummy.pk)
self.assertEquals(path, dummy.get_absolute_url())
path = 'http://happy-dev.fr/{}{}/'.format(get_resolver().reverse_dict[view_name][0][0][0], dummy.pk)
self.assertEqual(path, dummy.get_absolute_url())
def test_ldp_manager_local_objects(self):
local = LDPDummy.objects.create(some='text')
external = LDPDummy.objects.create(some='text', urlid='https://distant.com/ldpdummys/1/')
self.assertEqual(LDPDummy.objects.count(), 2)
local_queryset = LDPDummy.objects.local()
self.assertEqual(local_queryset.count(), 1)
self.assertIn(local, local_queryset)
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)