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 1669 additions and 457 deletions
djangoldp/tests/models.py
View file @ e97d33ad
from django.conf import settings
from django.contrib.auth.models import AbstractUser
from django.contrib.auth.models import AbstractUser, Group
from django.db import models
from django.db.models import BinaryField, DateField
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.utils.datetime_safe import date
from djangoldp.fields import LDPUrlField
from djangoldp.models import Model
from djangoldp.permissions import LDPPermissions
from djangoldp.models import Model, DynamicNestedField
from djangoldp.permissions import ACLPermissions, AuthenticatedOnly, ReadOnly, \
ReadAndCreate, AnonymousReadOnly, OwnerPermissions, InheritPermissions
from .permissions import Only2WordsForToto, ReadOnlyStartsWithA
class User(AbstractUser, Model):
class User(AbstractUser, Model):
class Meta(AbstractUser.Meta, Model.Meta):
ordering = ['pk']
serializer_fields = ['@id', 'username', 'first_name', 'last_name', 'email', 'userprofile',
'conversation_set', 'circle_set', 'projects']
anonymous_perms = ['view', 'add']
authenticated_perms = ['inherit', 'change']
owner_perms = ['inherit']
'conversation_set','groups', 'projects', 'owned_circles']
permission_classes = [ReadAndCreate|OwnerPermissions]
rdf_type = 'foaf:user'
nested_fields = ['owned_circles']
class Skill(Model):
......@@ -31,11 +30,12 @@ class Skill(Model):
return self.joboffer_set.filter(date__gte=date.today())
class Meta(Model.Meta):
anonymous_perms = ['view']
authenticated_perms = ['inherit', 'add']
owner_perms = ['inherit', 'change', 'delete', 'control']
serializer_fields = ["@id", "title", "recent_jobs", "slug"]
ordering = ['pk']
permission_classes = [AnonymousReadOnly,ReadAndCreate|OwnerPermissions]
serializer_fields = ["@id", "title", "recent_jobs", "slug", "obligatoire"]
nested_fields = ['joboffer_set']
lookup_field = 'slug'
rdf_type = 'hd:skill'
class JobOffer(Model):
......@@ -51,12 +51,15 @@ class JobOffer(Model):
return self.skills.all().first()
class Meta(Model.Meta):
anonymous_perms = ['view']
authenticated_perms = ['inherit', 'change', 'add']
owner_perms = ['inherit', 'delete', 'control']
ordering = ['pk']
permission_classes = [AnonymousReadOnly, ReadOnly|OwnerPermissions]
serializer_fields = ["@id", "title", "skills", "recent_skills", "resources", "slug", "some_skill", "urlid"]
nested_fields = ['skills', 'resources', 'recent_skills']
container_path = "job-offers/"
lookup_field = 'slug'
rdf_type = 'hd:joboffer'
JobOffer.recent_skills.field = DynamicNestedField(Skill, 'recent_skills')
class Conversation(models.Model):
......@@ -64,11 +67,12 @@ class Conversation(models.Model):
author_user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.DO_NOTHING)
peer_user = models.ForeignKey(settings.AUTH_USER_MODEL, blank=True, null=True, related_name="peers_conv",
on_delete=models.DO_NOTHING)
observers = models.ManyToManyField(settings.AUTH_USER_MODEL, blank=True, related_name='observed_conversations')
class Meta(Model.Meta):
anonymous_perms = ['view']
authenticated_perms = ['inherit', 'add']
owner_perms = ['inherit', 'change', 'delete', 'control']
ordering = ['pk']
permission_classes = [AnonymousReadOnly,ReadAndCreate|OwnerPermissions]
nested_fields=["message_set", "observers"]
owner_field = 'author_user'
......@@ -77,33 +81,99 @@ class Resource(Model):
description = models.CharField(max_length=255)
class Meta(Model.Meta):
anonymous_perms = ['view', 'add', 'delete', 'change', 'control']
authenticated_perms = ['inherit']
owner_perms = ['inherit']
ordering = ['pk']
serializer_fields = ["@id", "joboffers"]
nested_fields = ['joboffers']
depth = 1
rdf_type = 'hd:Resource'
# a resource in which only the owner has permissions (for testing owner permissions)
class OwnedResource(Model):
description = models.CharField(max_length=255, blank=True, null=True)
user = models.ForeignKey(settings.AUTH_USER_MODEL, blank=True, null=True, related_name="owned_resources",
on_delete=models.CASCADE)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [OwnerPermissions]
owner_field = 'user'
serializer_fields = ['@id', 'description', 'user']
nested_fields = ['owned_resources']
depth = 1
class OwnedResourceVariant(Model):
description = models.CharField(max_length=255, blank=True, null=True)
user = models.ForeignKey(settings.AUTH_USER_MODEL, blank=True, null=True, related_name="owned_variant_resources",
on_delete=models.CASCADE)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [ReadOnly|OwnerPermissions]
owner_field = 'user'
serializer_fields = ['@id', 'description', 'user']
depth = 1
class OwnedResourceNestedOwnership(Model):
description = models.CharField(max_length=255, blank=True, null=True)
parent = models.ForeignKey(OwnedResource, blank=True, null=True, related_name="owned_resources",
on_delete=models.CASCADE)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [OwnerPermissions]
owner_field = 'parent__user'
serializer_fields = ['@id', 'description', 'parent']
nested_fields = ['owned_resources']
depth = 1
class OwnedResourceTwiceNestedOwnership(Model):
description = models.CharField(max_length=255, blank=True, null=True)
parent = models.ForeignKey(OwnedResourceNestedOwnership, blank=True, null=True, related_name="owned_resources",
on_delete=models.CASCADE)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [OwnerPermissions]
owner_field = 'parent__parent__user'
serializer_fields = ['@id', 'description', 'parent']
depth = 1
class UserProfile(Model):
description = models.CharField(max_length=255, blank=True, null=True)
user = models.OneToOneField(settings.AUTH_USER_MODEL, related_name='userprofile', on_delete=models.CASCADE)
slug = models.SlugField(blank=True, null=True, unique=True)
class Meta(Model.Meta):
anonymous_perms = ['view']
authenticated_perms = ['inherit']
owner_perms = ['inherit', 'change', 'control']
ordering = ['pk']
permission_classes = [AuthenticatedOnly,ReadOnly|OwnerPermissions]
owner_field = 'user'
lookup_field = 'slug'
serializer_fields = ['@id', 'description', 'settings', 'user']
depth = 1
class NotificationSetting(Model):
user = models.OneToOneField(UserProfile, on_delete=models.CASCADE, related_name="settings")
receiveMail = models.BooleanField(default=True)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [ReadAndCreate|OwnerPermissions]
class Message(models.Model):
text = models.CharField(max_length=255, blank=True, null=True)
conversation = models.ForeignKey(Conversation, on_delete=models.DO_NOTHING)
author_user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.DO_NOTHING)
class Meta(Model.Meta):
anonymous_perms = ['view']
authenticated_perms = ['inherit', 'add']
owner_perms = ['inherit', 'change', 'delete', 'control']
ordering = ['pk']
permission_classes = [AnonymousReadOnly,ReadAndCreate|OwnerPermissions]
class Dummy(models.Model):
......@@ -111,46 +181,74 @@ class Dummy(models.Model):
slug = models.SlugField(blank=True, null=True, unique=True)
class Meta(Model.Meta):
anonymous_perms = ['view']
authenticated_perms = ['inherit', 'add']
owner_perms = ['inherit', 'change', 'delete', 'control']
ordering = ['pk']
permission_classes = [AnonymousReadOnly,ReadAndCreate|OwnerPermissions]
class LDPDummy(Model):
some = models.CharField(max_length=255, blank=True, null=True)
class Meta(Model.Meta):
anonymous_perms = ['view']
authenticated_perms = ['inherit', 'add']
owner_perms = ['inherit', 'change', 'delete', 'control']
ordering = ['pk']
permission_classes = [AnonymousReadOnly,ReadAndCreate|OwnerPermissions]
nested_fields = ['anons']
# model used in django-guardian permission tests (no anonymous etc permissions set)
# model used in django-guardian permission tests (no permission to anyone except suuperusers)
class PermissionlessDummy(Model):
some = models.CharField(max_length=255, blank=True, null=True)
slug = models.SlugField(blank=True, null=True, unique=True)
parent = models.ForeignKey(LDPDummy, on_delete=models.DO_NOTHING, related_name="anons", blank=True, null=True)
class Meta(Model.Meta):
anonymous_perms = []
authenticated_perms = []
owner_perms = []
permissions = (
('custom_permission_permissionlessdummy', 'Custom Permission'),
)
ordering = ['pk']
permission_classes = [ACLPermissions]
lookup_field='slug'
permissions = (('custom_permission_permissionlessdummy', 'Custom Permission'),)
class Post(Model):
content = models.CharField(max_length=255)
author = models.ForeignKey(UserProfile, blank=True, null=True, on_delete=models.SET_NULL)
author = models.ForeignKey(User, blank=True, null=True, on_delete=models.SET_NULL)
peer_user = models.ForeignKey(settings.AUTH_USER_MODEL, blank=True, null=True, related_name="peers_post",
on_delete=models.SET_NULL)
class Meta(Model.Meta):
ordering = ['pk']
auto_author = 'author'
auto_author_field = 'userprofile'
anonymous_perms = ['view', 'add', 'delete', 'add', 'change', 'control']
authenticated_perms = ['inherit']
owner_perms = ['inherit']
rdf_type = 'hd:post'
class AnonymousReadOnlyPost(Model):
content = models.CharField(max_length=255)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [AnonymousReadOnly]
class AuthenticatedOnlyPost(Model):
content = models.CharField(max_length=255)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [AuthenticatedOnly]
class ReadOnlyPost(Model):
content = models.CharField(max_length=255)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [ReadOnly]
class ReadAndCreatePost(Model):
content = models.CharField(max_length=255)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [ReadAndCreate]
class ANDPermissionsDummy(Model):
title = models.CharField(max_length=255)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [ReadOnlyStartsWithA&Only2WordsForToto]
class ORPermissionsDummy(Model):
title = models.CharField(max_length=255)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [ReadOnlyStartsWithA|Only2WordsForToto]
class Invoice(Model):
......@@ -158,22 +256,73 @@ class Invoice(Model):
date = models.DateField(blank=True, null=True)
class Meta(Model.Meta):
ordering = ['pk']
depth = 2
anonymous_perms = ['view']
authenticated_perms = ['inherit', 'add']
owner_perms = ['inherit', 'change', 'delete', 'control']
permission_classes = [AnonymousReadOnly,ReadAndCreate|OwnerPermissions]
nested_fields = ['batches']
class Circle(Model):
name = models.CharField(max_length=255, blank=True)
description = models.CharField(max_length=255, blank=True)
team = models.ManyToManyField(settings.AUTH_USER_MODEL, through="CircleMember", blank=True)
owner = models.ForeignKey(settings.AUTH_USER_MODEL, related_name="owned_circles", on_delete=models.DO_NOTHING, null=True, blank=True)
members = models.OneToOneField(Group, related_name="circle", on_delete=models.SET_NULL, null=True, blank=True)
admins = models.OneToOneField(Group, related_name="admin_circle", on_delete=models.SET_NULL, null=True, blank=True)
class Meta(Model.Meta):
ordering = ['pk']
auto_author = 'owner'
depth = 1
permission_classes = [AnonymousReadOnly,ReadAndCreate|OwnerPermissions|ACLPermissions]
permission_roles = {
'members': {'perms': ['view'], 'add_author': True},
'admins': {'perms': ['view', 'change', 'control'], 'add_author': True},
}
serializer_fields = ['@id', 'name', 'description', 'members', 'owner', 'space']
rdf_type = 'hd:circle'
Group._meta.inherit_permissions += ['circle','admin_circle']
Group._meta.serializer_fields += ['circle', 'admin_circle']
class RestrictedCircle(Model):
name = models.CharField(max_length=255, blank=True)
description = models.CharField(max_length=255, blank=True)
owner = models.ForeignKey(settings.AUTH_USER_MODEL, related_name="owned_restrictedcircles", on_delete=models.DO_NOTHING, null=True, blank=True)
members = models.ForeignKey(Group, related_name="restrictedcircles", on_delete=models.SET_NULL, null=True, blank=True)
admins = models.ForeignKey(Group, related_name="admin_restrictedcircles", on_delete=models.SET_NULL, null=True, blank=True)
class Meta(Model.Meta):
anonymous_perms = ['view', 'add', 'delete', 'add', 'change', 'control']
authenticated_perms = ["inherit"]
ordering = ['pk']
auto_author = 'owner'
permission_classes = [ACLPermissions]
permission_roles = {
'members': {'perms': ['view'], 'add_author': True},
'admins': {'perms': ['view', 'change', 'control'], 'add_author': True},
}
rdf_type = 'hd:circle'
class RestrictedResource(Model):
content = models.CharField(max_length=255, blank=True)
circle = models.ForeignKey(RestrictedCircle, on_delete=models.CASCADE)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [InheritPermissions]
inherit_permissions = ['circle']
class DoubleInheritModel(Model):
content = models.CharField(max_length=255, blank=True)
ro_ancestor = models.ForeignKey(ReadOnlyPost, on_delete=models.CASCADE, null=True, blank=True)
circle = models.ForeignKey(RestrictedCircle, on_delete=models.CASCADE, null=True, blank=True)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [InheritPermissions]
inherit_permissions = ['circle', 'ro_ancestor']
class Space(Model):
name = models.CharField(max_length=255, blank=True)
circle = models.OneToOneField(to=Circle, null=True, blank=True, on_delete=models.CASCADE, related_name='space')
class Meta(Model.Meta):
ordering = ['pk']
class Batch(Model):
......@@ -181,24 +330,11 @@ class Batch(Model):
title = models.CharField(max_length=255, blank=True, null=True)
class Meta(Model.Meta):
ordering = ['pk']
serializer_fields = ['@id', 'title', 'invoice', 'tasks']
anonymous_perms = ['view', 'add']
authenticated_perms = ['inherit', 'add']
owner_perms = ['inherit', 'change', 'delete', 'control']
permission_classes = [ReadAndCreate|OwnerPermissions]
depth = 1
class CircleMember(Model):
circle = models.ForeignKey(Circle, on_delete=models.CASCADE, related_name='members')
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="circles")
is_admin = models.BooleanField(default=False)
class Meta(Model.Meta):
container_path = "circle-members/"
anonymous_perms = ['view', 'add', 'delete', 'add', 'change', 'control']
authenticated_perms = ['inherit']
unique_together = ['user', 'circle']
rdf_type = 'hd:circlemember'
rdf_type = 'hd:batch'
class Task(models.Model):
......@@ -206,20 +342,30 @@ class Task(models.Model):
title = models.CharField(max_length=255)
class Meta(Model.Meta):
ordering = ['pk']
serializer_fields = ['@id', 'title', 'batch']
anonymous_perms = ['view']
authenticated_perms = ['inherit', 'add']
owner_perms = ['inherit', 'change', 'delete', 'control']
permission_classes = [AnonymousReadOnly,ReadAndCreate|OwnerPermissions]
class ModelTask(Model, Task):
class Meta(Model.Meta):
ordering = ['pk']
STATUS_CHOICES = [
('Public', 'Public'),
('Private', 'Private'),
('Archived', 'Archived'),
]
class Project(Model):
description = models.CharField(max_length=255, null=True, blank=False)
team = models.ManyToManyField(settings.AUTH_USER_MODEL, blank=True, related_name='projects')
status = models.CharField(max_length=8, choices=STATUS_CHOICES, default='Private', null=True, blank=True)
members = models.ManyToManyField(settings.AUTH_USER_MODEL, blank=True, related_name='projects')
class Meta(Model.Meta):
anonymous_perms = ['view', 'add', 'delete', 'add', 'change', 'control']
authenticated_perms = ["inherit"]
ordering = ['pk']
rdf_type = 'hd:project'
nested_fields = ['members']
class DateModel(Model):
......@@ -227,6 +373,7 @@ class DateModel(Model):
value = models.DateField()
class Meta(Model.Meta):
ordering = ['pk']
rdf_type = "hd:date"
serializer_fields_exclude = ['excluded']
......@@ -235,9 +382,21 @@ class DateChild(Model):
parent = models.ForeignKey(DateModel, on_delete=models.CASCADE, related_name='children')
class Meta(Model.Meta):
ordering = ['pk']
rdf_type = 'hd:datechild'
@receiver(post_save, sender=User)
def update_perms(sender, instance, created, **kwargs):
LDPPermissions.invalidate_cache()
class MyAbstractModel(Model):
defaultsomething = models.CharField(max_length=255, blank=True)
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [ACLPermissions]
abstract = True
rdf_type = "wow:defaultrdftype"
class NoSuperUsersAllowedModel(Model):
class Meta(Model.Meta):
ordering = ['pk']
permission_classes = [ACLPermissions]
\ No newline at end of file
djangoldp/tests/perf_result.csv
View file @ e97d33ad
......@@ -46,3 +46,6 @@ jbl+AC0-T440p,Oct 09 2020 11:56:19,True,True,100,0.003119325637817,0.00560247182
jbl+AC0-T440p,Oct 09 2020 11:58:22,True,True,100,0.003008058071136,0.005401248931885,0.010658957958222,0.003909242153168,0.000718443393707,0.301162958145142,TRUE,3
jbl+AC0-T440p,Oct 09 2020 11:59:16,True,True,100,0.003015418052673,0.005526115894318,0.010740044116974,0.00400491476059,0.000724492073059,0.313828229904175,TRUE,4
jbl+AC0-T440p,Oct 09 2020 12:00:32,True,True,100,0.002969658374786,0.005434756278992,0.018136837482452,0.003030817508698,0.000726938247681,0.320115327835083,TRUE,0
jbl+AC0-T440p,Oct 09 2020 12:21:00,True,True,100,0.003493466377258,0.006103293895721,0.01923253774643,0.003091294765472,0.000737550258636,0.369867086410522,TRUE,0
jbl+AC0-T440p,Oct 15 2020 22:00:10,True,True,100,0.003004941940308,0.00546817779541,0.018348352909088,0.003068554401398,0.000729415416718,0.320573329925537,TRUE,0
jbl+AC0-T440p,Oct 15 2020 22:15:26,True,True,100,0.003350086212158,0.005898218154907,0.011625332832337,0.004264788627625,0.000795011520386,0.319289922714233,TRUE,1
djangoldp/tests/performance_runner.py
View file @ e97d33ad
import sys
import yaml
import django
from djangoldp.tests import settings_default
from django.conf import settings
from django.conf import settings as django_settings
from djangoldp.conf.ldpsettings import LDPSettings
from djangoldp.tests.server_settings import yaml_config
# configure settings not to use pagination
settings.configure(default_settings=settings_default,
REST_FRAMEWORK = {
'DEFAULT_PAGINATION_CLASS': None
})
# load test config
config = yaml.safe_load(yaml_config)
ldpsettings = LDPSettings(config)
django_settings.configure(ldpsettings,
REST_FRAMEWORK = {
'DEFAULT_PAGINATION_CLASS': None
},
ANONYMOUS_USER_NAME=None)
django.setup()
from django.test.runner import DiscoverRunner
......
djangoldp/tests/permissions.py 0 → 100644
View file @ e97d33ad
from djangoldp.filters import BaseFilterBackend
from djangoldp.permissions import LDPBasePermission
class StartsWithAFilter(BaseFilterBackend):
"""Only objects whose title starts in A get through"""
def filter_queryset(self, request, queryset, view):
return queryset.filter(title__startswith='A')
class ReadOnlyStartsWithA(LDPBasePermission):
"""Only gives read-only access and only to objects which title starts with A"""
filter_backend = StartsWithAFilter
permissions = {'view', 'list'}
def check_perms(self, obj):
return getattr(obj, 'title', '').startswith('A')
def has_object_permission(self, request, view, obj=None):
return self.check_perms(obj)
def get_permissions(self, user, model, obj=None):
return self.permissions if self.check_perms(obj) else set()
class ContainsSpace(BaseFilterBackend):
"""Only objects whose title contains a space get through"""
def filter_queryset(self, request, queryset, view):
if request.user.username != 'toto':
return queryset.none()
return queryset.filter(title__contains=' ')
class Only2WordsForToto(LDPBasePermission):
"""Only gives access if the user's username is toto and only to objects whose title has two words (contains space)"""
filter_backend = ContainsSpace
def has_permission(self, request, view):
return request.user.username == 'toto'
def check_perms(self, obj):
return ' ' in getattr(obj, 'title', '')
def has_object_permission(self, request, view, obj=None):
return self.check_perms(obj)
def get_permissions(self, user, model, obj=None):
return self.permissions if self.check_perms(obj) else set()
\ No newline at end of file
djangoldp/tests/runner.py
View file @ e97d33ad
import sys
import yaml
import django
from djangoldp.tests import settings_default
from django.conf import settings
from django.conf import settings as django_settings
from djangoldp.conf.ldpsettings import LDPSettings
from djangoldp.tests.server_settings import yaml_config
settings.configure(default_settings=settings_default)
# load test config
config = yaml.safe_load(yaml_config)
ldpsettings = LDPSettings(config)
django_settings.configure(ldpsettings)
django.setup()
from django.test.runner import DiscoverRunner
......@@ -12,11 +17,15 @@ from django.test.runner import DiscoverRunner
test_runner = DiscoverRunner(verbosity=1)
failures = test_runner.run_tests([
'djangoldp.tests.tests_settings',
'djangoldp.tests.tests_ldp_model',
'djangoldp.tests.tests_save',
'djangoldp.tests.tests_model_serializer',
'djangoldp.tests.tests_ldp_viewset',
'djangoldp.tests.tests_user_permissions',
'djangoldp.tests.tests_guardian',
'djangoldp.tests.tests_anonymous_permissions',
'djangoldp.tests.tests_permissions',
'djangoldp.tests.tests_post',
'djangoldp.tests.tests_update',
'djangoldp.tests.tests_auto_author',
'djangoldp.tests.tests_get',
......@@ -24,8 +33,8 @@ failures = test_runner.run_tests([
'djangoldp.tests.tests_sources',
'djangoldp.tests.tests_pagination',
'djangoldp.tests.tests_inbox',
'djangoldp.tests.tests_backlinks_service'
'djangoldp.tests.tests_backlinks_service',
'djangoldp.tests.tests_cache'
])
if failures:
sys.exit(failures)
djangoldp/tests/scripts/__init__.py 0 → 100644
View file @ e97d33ad
djangoldp/tests/scripts/prod_data_generator.py 0 → 100644
View file @ e97d33ad
import json
import argparse
from pathlib import Path
from datetime import datetime
from utils import generate_users, generate_projects, generate_skills
'''
A script which generates and outputs random production data, into a parameterised file (json), which can be used as
a Django fixture or imported into a live database
e.g. python manage.py loaddata fixture.json
for help run python prod_data_generator.py -h
'''
# starting from offset ensures that existing users etc are not disturbed
parser = argparse.ArgumentParser(description='generates and outputs random test data, into a file used by the performance unit tests')
parser.add_argument(dest='count', metavar='N', type=int, help='the number of users (and projects) to generate')
parser.add_argument('--offset', dest='offset', type=int, default=100, help='an offset to start primary keys at (should be larger than the largest pre-existing project/user primary key)')
parser.add_argument('-f', dest='file_dest', type=str, default="../fixtures/live.json", help='the file destination to write to')
parser.add_argument('-s', dest='generate_skills', type=bool, default=False, help='Do you want to generate skills too ?')
args = parser.parse_args()
count = args.count
OFFSET = args.offset
user_template = {
'model': 'djangoldp_account.ldpuser',
'pk': 0,
'fields': {
'username': 'john',
'email': 'jlennon@c.coop',
'password':'glassonion',
'first_name': 'John',
'last_name': 'Lennon'
}
}
project_template = {
'model': 'djangoldp_project.project',
'pk': 0,
'fields': {
'description': 'Test',
'status': 'Public',
'creationDate': str(datetime.date(datetime.now()))
}
}
skill_template = {
'model': 'djangoldp_skill.skill',
'pk': 0,
'fields': {
'name': 'PHP',
}
}
fixture = generate_users(count, user_template, offset=OFFSET)
fixture = generate_projects(count, project_template, fixture=fixture, offset=OFFSET)
if args.generate_skills:
fixture = generate_skills(count, skill_template, fixture=fixture, offset=OFFSET)
with open(Path(__file__).parent / args.file_dest, 'w') as output:
json.dump(fixture, output)
print(str(count))
djangoldp/tests/scripts/skills.json 0 → 100644
View file @ e97d33ad
source diff could not be displayed: it is too large. Options to address this: view the blob.
djangoldp/tests/scripts/test_data_generator.py
View file @ e97d33ad
import uuid
import json
import sys
import random
import argparse
from pathlib import Path
from copy import deepcopy
from utils import generate_users, generate_projects
'''
A script which generates and outputs random test data, into a file used by the performance unit tests
usage: python test_data_generator.py [number_rows]
e.g. python test_data_generator.py 2000
for help run python test_data_generator.py -h
'''
count = int(sys.argv[1])
fixture = list()
parser = argparse.ArgumentParser(description='generates and outputs random test data, into a file used by the performance unit tests')
parser.add_argument(dest='count', metavar='N', type=int, help='the number of users (and projects) to generate')
parser.add_argument('-f', dest='file_dest', type=str, default="../fixtures/test.json", help='the file destination to write to')
args = parser.parse_args()
count = args.count
user_template = {
'model': 'tests.user',
......@@ -20,7 +21,9 @@ user_template = {
'fields': {
'username': 'john',
'email': 'jlennon@c.coop',
'password':'glass onion'
'password':'glassonion',
'first_name': 'John',
'last_name': 'Lennon'
}
}
......@@ -28,38 +31,15 @@ project_template = {
'model': 'tests.project',
'pk': 0,
'fields': {
'status': 'Public',
'description': 'Test'
}
}
def generate_user(i):
user = deepcopy(user_template)
user['pk'] = i
user['fields']['username'] = str(uuid.uuid4())
user['fields']['email'] = user['fields']['username'] + "@c.coop"
return user
def generate_project(i):
project = deepcopy(project_template)
project['pk'] = i
project['fields']['team'] = list()
# append random number of users, max 10 for a single project
for j in range(random.randint(1, 10)):
project['fields']['team'].append(random.randint(1, count-1))
return project
# create N users
for i in range(count):
user = generate_user(i)
fixture.append(user)
# create N projects
for i in range(count):
project = generate_project(i)
fixture.append(project)
fixture = generate_users(count, user_template)
fixture = generate_projects(count, project_template, fixture=fixture, production=False)
with open(Path(__file__).parent / "../fixtures/test.json", 'w') as output:
with open(Path(__file__).parent / args.file_dest, 'w') as output:
json.dump(fixture, output)
print(str(count))
djangoldp/tests/scripts/utils.py 0 → 100644
View file @ e97d33ad
from copy import deepcopy
import random
from faker import Faker
'''
Contains definitions used in common by multiple scripts within this directory
'''
def generate_user(i, user_template):
myFactory = Faker()
user = deepcopy(user_template)
email = myFactory.unique.email().split('@')
email.insert(1, str(random.randint(0, 5)))
email.insert(2, "@")
email_str = "".join(email)
user['pk'] = i
user['fields']['username'] = myFactory.unique.user_name() + str(random.randint(0, 100))
user['fields']['email'] = email_str
user['fields']['first_name'] = myFactory.first_name()
user['fields']['last_name'] = myFactory.last_name()
return user
def generate_users(count, user_template, fixture=None, offset=0):
if fixture is None:
fixture = list()
for i in range(count):
j = offset + i
user = generate_user(j, user_template)
fixture.append(user)
return fixture
def generate_project_member_and_user_pks(project_pk, offset, total_users, max_members_per_project):
'''
returns a generator of tuples (new project member PKs and selected user PKs)
raises error if there are not enough users
'''
# we want to select a handful of random users
# to save time we just select a random user within a safe range and then grab a bunch of adjacent users
start_user_pk = random.randint(max(offset, 1), offset + (total_users - (max_members_per_project + 1)))
if start_user_pk < offset:
raise IndexError('not enough users!')
for i in range(random.randint(1, max_members_per_project)):
j = offset + (i + (project_pk * max_members_per_project)) # generate a unique integer id
user_pk = start_user_pk + i # select the next user
yield (j, user_pk)
def generate_project_members(project_pk, fixture, offset, total_users):
max_members_per_project = 10
def generate_project_member(i, user_pk):
return {
'model': 'djangoldp_project.member',
'pk': i,
'fields': {
'project': project_pk,
'user': user_pk
}
}
for (j, user_pk) in generate_project_member_and_user_pks(project_pk, offset, total_users, max_members_per_project):
fixture.append(generate_project_member(j, user_pk))
return fixture
def generate_skill(i, skill_template):
myFactory = Faker()
skill = deepcopy(skill_template)
skill['pk'] = i
skill['fields']['name'] = myFactory.unique.job()
return skill
def generate_user_pks(skill_pk, offset, total_users, max_users_per_skill):
'''
returns a generator of tuples ()
raises error if there are not enough users
'''
# we want to select a handful of random users
# to save time we just select a random user within a safe range and then grab a bunch of adjacent users
start_user_pk = random.randint(0, offset + (total_users - (max_users_per_skill + 1)))
# if start_user_pk < offset:
# raise IndexError('not enough users!')
for i in range(random.randint(1, max_users_per_skill)):
j = offset + (i + (skill_pk * max_users_per_skill)) # generate a unique integer id
user_pk = start_user_pk + i # select the next user
yield (j, user_pk)
def append_users_to_skill(skill, offset, total_users):
max_users_per_skill = 250
skill['fields']['users'] = []
for (j, user_pk) in generate_user_pks(skill['pk'], offset, total_users, max_users_per_skill):
skill['fields']['users'].append(user_pk)
return skill
def generate_skills(count, skill_template, fixture=None, offset=0):
if fixture is None:
fixture = list()
for i in range(count):
j = offset + i
skill = generate_skill(j, skill_template)
skill = append_users_to_skill(skill, offset, count)
fixture.append(skill)
return fixture
def generate_project(i, project_template):
project = deepcopy(project_template)
project['pk'] = i
return project
def append_members_to_project(project, offset, total_users):
max_members_per_project = 10
project['members'] = []
for (j, user_pk) in generate_project_member_and_user_pks(project['pk'], offset, total_users, max_members_per_project):
project['members'].append(user_pk)
return project
def generate_projects(count, project_template, fixture=None, offset=0, production=True):
if fixture is None:
fixture = list()
for i in range(count):
j = offset + i
project = generate_project(j, project_template)
# project members using direct ManyToMany field. Generate them as a field on project
if not production:
project = append_members_to_project(project, offset, count)
fixture.append(project)
# project members using Member through model, generate them as separate in the fixture
if production:
# append random number of project members, max 10 for a single project
generate_project_members(j, fixture, offset, count)
return fixture
djangoldp/tests/server_settings.py 0 → 100644
View file @ e97d33ad
"""
This module contains the YAML configuration for a testing djangoldp server.
"""
yaml_config = """
dependencies:
ldppackages:
- djangoldp.tests # fetch 'djangoldp.tests.djangoldp_settings'
- djangoldp.tests.dummy.apps.DummyConfig # already declared in 'djangoldp.tests'
server:
ALLOWED_HOSTS:
- '*'
AUTH_USER_MODEL: tests.User
EMAIL_HOST: somewhere
ANONYMOUS_USER_NAME: None
ROOT_URLCONF: djangoldp.urls
SEND_BACKLINKS: false
SITE_URL: http://happy-dev.fr
BASE_URL: http://happy-dev.fr
REST_FRAMEWORK:
DEFAULT_PAGINATION_CLASS: djangoldp.pagination.LDPPagination
PAGE_SIZE: 5
USE_TZ: false
SEND_BACKLINKS: false
GUARDIAN_AUTO_PREFETCH: true
SERIALIZER_CACHE: false
STORE_ACTIVITIES: VERBOSE
"""
djangoldp/tests/settings_default.py deleted 100644 → 0
View file @ 58ae8934
from django.conf.global_settings import *
# defines default settings for testing DjangoLDP. You can use this in your own packages by following the example in
# runner.py
DEBUG=False
ALLOWED_HOSTS=["*"]
SITE_URL='http://happy-dev.fr'
BASE_URL='http://happy-dev.fr'
DJANGOLDP_PACKAGES=['djangoldp.tests']
INSTALLED_APPS=('django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.admin',
'django.contrib.messages',
'django.contrib.staticfiles',
'guardian',
'djangoldp',
'djangoldp.tests',
)
DATABASES={
'default': {
'ENGINE': 'django.db.backends.sqlite3',
}
}
REST_FRAMEWORK = {
'DEFAULT_PAGINATION_CLASS': 'djangoldp.pagination.LDPPagination',
'PAGE_SIZE': 5
}
AUTH_USER_MODEL='tests.User'
ANONYMOUS_USER_NAME = None
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
AUTHENTICATION_BACKENDS=(
'django.contrib.auth.backends.ModelBackend', 'guardian.backends.ObjectPermissionBackend')
ROOT_URLCONF='djangoldp.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
LDP_RDF_CONTEXT={
"@context": {
"@vocab": "http://happy-dev.fr/owl/#",
"foaf": "http://xmlns.com/foaf/0.1/",
"doap": "http://usefulinc.com/ns/doap#",
"ldp": "http://www.w3.org/ns/ldp#",
"rdfs": "http://www.w3.org/2000/01/rdf-schema#",
"rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
"xsd": "http://www.w3.org/2001/XMLSchema#",
"geo": "http://www.w3.org/2003/01/geo/wgs84_pos#",
"acl": "http://www.w3.org/ns/auth/acl#",
"name": "rdfs:label",
"website": "foaf:homepage",
"deadline": "xsd:dateTime",
"lat": "geo:lat",
"lng": "geo:long",
"jabberID": "foaf:jabberID",
"permissions": "acl:accessControl",
"mode": "acl:mode",
"view": "acl:Read",
"change": "acl:Write",
"add": "acl:Append",
"delete": "acl:Delete",
"control": "acl:Control"
}
}
SEND_BACKLINKS=False
GUARDIAN_AUTO_PREFETCH = True
djangoldp/tests/tests_anonymous_permissions.py
View file @ e97d33ad
......@@ -3,9 +3,7 @@ import json
from django.test import TestCase
from rest_framework.test import APIClient
from djangoldp.permissions import LDPPermissions
from djangoldp.tests.models import JobOffer
from djangoldp.views import LDPViewSet
class TestAnonymousUserPermissions(TestCase):
......@@ -26,6 +24,8 @@ class TestAnonymousUserPermissions(TestCase):
response = self.client.post('/job-offers/', data=json.dumps(post), content_type='application/ld+json')
self.assertEqual(response.status_code, 403)
# TODO: test POST request for anonymous user where it's allowed
def test_put_request_for_anonymousUser(self):
body = {'title':"job_updated"}
response = self.client.put('/job-offers/{}/'.format(self.job.pk), data=json.dumps(body),
......
djangoldp/tests/tests_auto_author.py
View file @ e97d33ad
import json
from django.contrib.auth import get_user_model
from rest_framework.test import APIRequestFactory, APIClient, APITestCase
from rest_framework.test import APIClient, APIRequestFactory, APITestCase
from djangoldp.tests.models import UserProfile
......@@ -18,18 +19,8 @@ class TestAutoAuthor(APITestCase):
def test_save_with_anonymous_user(self):
post = {
'@graph': [{'http://happy-dev.fr/owl/#content': "post content"}]}
response = self.client.post('/posts/', data=json.dumps(post), content_type='application/ld+json')
self.assertEqual(response.status_code, 201)
self.assertEquals(response.data['content'], "post content")
def test_auto_author_field(self):
self.client.force_authenticate(user=self.user)
post = {
'@graph': [{'http://happy-dev.fr/owl/#content': "post content"}]}
'@graph': [{'https://cdn.startinblox.com/owl#content': "post content"}]}
response = self.client.post('/posts/', data=json.dumps(post), content_type='application/ld+json')
self.assertEqual(response.status_code, 201)
self.assertEquals(response.data['content'], "post content")
self.assertIsNotNone(response.data['author'])
self.assertEqual(response.data['content'], "post content")
\ No newline at end of file
djangoldp/tests/tests_backlinks_service.py
View file @ e97d33ad
import json
import copy
import time
import uuid
from django.contrib.auth import get_user_model
from django.db import IntegrityError
from django.test import override_settings
from rest_framework.test import APIClient, APITestCase
from djangoldp.tests.models import Circle, CircleMember, Project, UserProfile, DateModel, DateChild
from djangoldp.models import Activity, Follower
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):
......@@ -22,7 +26,9 @@ class TestsBacklinksService(APITestCase):
urlid = 'https://distant.com/users/' + username
return get_user_model().objects.create_user(username=username, email=email, password='test', urlid=urlid)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# 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')
......@@ -54,24 +60,24 @@ class TestsBacklinksService(APITestCase):
local_circle.delete()
self.assertEqual(Activity.objects.all().count(), 4)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
@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.team.add(external_a)
project.team.add(external_b)
project.team.add(external_c)
project.members.add(external_a)
project.members.add(external_b)
project.members.add(external_c)
self.assertEqual(Activity.objects.all().count(), 3)
# remove one individual
project.team.remove(external_a)
project.members.remove(external_a)
self.assertEqual(Activity.objects.all().count(), 4)
# clear the rest
project.team.clear()
project.members.clear()
self.assertEqual(Activity.objects.all().count(), 6)
prior_count = Activity.objects.all().count()
......@@ -79,12 +85,241 @@ class TestsBacklinksService(APITestCase):
project.delete()
self.assertEqual(Activity.objects.all().count(), prior_count)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
@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.team.add(external_a)
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_get.py
View file @ e97d33ad
from djangoldp.serializers import LDListMixin, LDPSerializer
from rest_framework.test import APIRequestFactory, APIClient, APITestCase
from datetime import datetime
from rest_framework.test import APIRequestFactory, APIClient, APITestCase
from djangoldp.tests.models import Post, Invoice, JobOffer, Skill, Batch, DateModel
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient, APIRequestFactory, APITestCase
from djangoldp.serializers import GLOBAL_SERIALIZER_CACHE
from djangoldp.tests.models import (Batch, Circle, Conversation, DateModel,
Invoice, JobOffer, Message, Post, Skill,
User, UserProfile)
class TestGET(APITestCase):
......@@ -11,18 +14,37 @@ class TestGET(APITestCase):
def setUp(self):
self.factory = APIRequestFactory()
self.client = APIClient()
LDListMixin.to_representation_cache.invalidate_cache()
LDPSerializer.to_representation_cache.invalidate_cache()
self.ordered_fields = ['@context', '@type', '@id']
setattr(Circle._meta, 'depth', 0)
setattr(Circle._meta, 'empty_containers', [])
def tearDown(self):
pass
GLOBAL_SERIALIZER_CACHE.reset()
def test_get_resource(self):
post = Post.objects.create(content="content")
response = self.client.get('/posts/{}/'.format(post.pk), content_type='application/ld+json')
response = self.client.get('/posts/{}/'.format(post.pk), content_type='application/ld+json', HTTP_ORIGIN='http://localhost:8080/test/')
self.assertEqual(response.status_code, 200)
self.assertEquals(response.data['content'], "content")
self.assertEqual(response.data['content'], "content")
self.assertIn('author', response.data)
self.assertIn('@type', response.data)
# test headers returned
self.assertEqual(response['Content-Type'], 'application/ld+json')
self.assertEqual(response['Accept-Post'], 'application/ld+json')
self.assertEqual(response['Allow'], 'GET, PUT, PATCH, DELETE, HEAD, OPTIONS')
self.assertEqual(response['Access-Control-Allow-Origin'], 'http://localhost:8080/test/')
self.assertIn('DPoP', response['Access-Control-Allow-Headers'])
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")
......@@ -30,26 +52,26 @@ class TestGET(APITestCase):
Post.objects.create(content="federated", urlid="https://external.com/posts/1/")
response = self.client.get('/posts/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertIn('permissions', response.data)
self.assertEquals(1, len(response.data['ldp:contains']))
self.assertEquals(2, len(response.data['permissions'])) # read and add
self.assertEqual(1, len(response.data['ldp:contains']))
self.assertIn('@type', response.data)
self.assertIn('@type', response.data['ldp:contains'][0])
self.assertNotIn('permissions', response.data['ldp:contains'][0])
Invoice.objects.create(title="content")
response = self.client.get('/invoices/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertIn('permissions', response.data)
self.assertEquals(1, len(response.data['permissions'])) # read only
self.assertEqual(1, len(response.data['permissions'])) # read only
def test_get_empty_container(self):
Post.objects.all().delete()
response = self.client.get('/posts/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEquals(0, len(response.data['ldp:contains']))
self.assertEqual(0, len(response.data['ldp:contains']))
def test_get_filtered_fields(self):
skill = Skill.objects.create(title="Java", obligatoire="ok", slug="1")
skill2 = Skill.objects.create(title="Java", obligatoire="ok", slug="2")
skill3 = Skill.objects.create(urlid="http://external/skills/1")
skill3 = Skill.objects.create(urlid="http://happy-dev.hubl.fr/skills/1")
job = JobOffer.objects.create(title="job", slug="1")
job.skills.add(skill)
job.skills.add(skill2)
......@@ -59,7 +81,11 @@ class TestGET(APITestCase):
self.assertEqual(response.status_code, 200)
self.assertIn('recent_skills', response.data)
self.assertEqual(response.data['recent_skills']['@id'], "http://happy-dev.fr/job-offers/1/recent_skills/")
self.assertEqual(response.data['skills']['ldp:contains'][2]['@id'], "http://external/skills/1")
# the external resource should be serialized with its @id and @type.. and only these fields
self.assertEqual(response.data['skills']['ldp:contains'][2]['@id'], "http://happy-dev.hubl.fr/skills/1")
self.assertIn('@type', response.data['skills']['ldp:contains'][1])
self.assertIn('@type', response.data['skills']['ldp:contains'][2])
self.assertEqual(len(response.data['skills']['ldp:contains'][2].items()), 2)
def test_get_reverse_filtered_fields(self):
skill = Skill.objects.create(title="Java", obligatoire="ok", slug="1")
......@@ -81,7 +107,10 @@ class TestGET(APITestCase):
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'], "http://testserver/skills/1/")
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")
......@@ -89,19 +118,122 @@ class TestGET(APITestCase):
distant_batch = Batch.objects.create(invoice=invoice, title="distant", urlid="https://external.com/batch/1/")
response = self.client.get('/invoices/{}/batches/'.format(invoice.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEquals(response.data['@id'], 'http://happy-dev.fr/invoices/{}/batches/'.format(invoice.pk))
self.assertEquals(len(response.data['ldp:contains']), 2)
self.assertEquals(response.data['ldp:contains'][0]['invoice']['@id'], 'http://happy-dev.fr/invoices/{}/'.format(invoice.pk))
self.assertEqual(response.data['@id'], 'http://happy-dev.fr/invoices/{}/batches/'.format(invoice.pk))
self.assertIn('permissions', response.data)
self.assertEqual(len(response.data['ldp:contains']), 2)
self.assertIn('@type', response.data['ldp:contains'][0])
self.assertIn('@type', response.data['ldp:contains'][1])
self.assertNotIn('permissions', response.data['ldp:contains'][0])
self.assertNotIn('permissions', response.data['ldp:contains'][1])
self.assertEqual(response.data['ldp:contains'][0]['invoice']['@id'], invoice.urlid)
self.assertEqual(response.data['ldp:contains'][1]['@id'], distant_batch.urlid)
def test_get_nested_without_related_name(self):
user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com', password='glass onion')
conversation = Conversation.objects.create(author_user=user)
message = Message.objects.create(conversation=conversation, author_user=user)
response = self.client.get('/conversations/{}/message_set/'.format(conversation.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['@id'], 'http://happy-dev.fr/conversations/{}/message_set/'.format(conversation.pk))
self.assertEqual(len(response.data['ldp:contains']), 1)
# TODO: https://git.startinblox.com/djangoldp-packages/djangoldp/issues/335
# test getting a route with multiple nested fields (/job-offers/X/skills/Y/)
'''def test_get_twice_nested(self):
job = JobOffer.objects.create(title="job", slug="slug1")
skill = Skill.objects.create(title='old', obligatoire='old', slug='skill1')
job.skills.add(skill)
self.assertEqual(job.skills.count(), 1)
response = self.client.get('/job-offers/{}/skills/{}/'.format(job.slug, skill.slug))
self.assertEqual(response.status_code, 200)'''
def test_serializer_excludes(self):
date = DateModel.objects.create(excluded='test', value=datetime.now())
response = self.client.get('/dates/{}/'.format(date.pk), content_type='application/ld+json')
response = self.client.get('/datemodels/{}/'.format(date.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertNotIn('excluded', response.data.keys())
def test_serializer_excludes_serializer_fields_set_also(self):
setattr(DateModel._meta, 'serializer_fields', ['value', 'excluded'])
date = DateModel.objects.create(excluded='test', value=datetime.now())
response = self.client.get('/dates/{}/'.format(date.pk), content_type='application/ld+json')
response = self.client.get('/datemodels/{}/'.format(date.pk), content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertNotIn('excluded', response.data.keys())
def _set_up_circle_and_user(self):
user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com',
password='glass onion')
circle = Circle.objects.create(name='test', description='test', owner=user)
self.client.force_authenticate(user)
circle.members.user_set.add(user)
return user
# tests for functionality allowing me to set containers to be serialized without content\
# test for normal functioning (without setting)
def test_empty_container_serialization_nested_serializer_no_empty(self):
setattr(Circle._meta, 'depth', 1)
self._set_up_circle_and_user()
response = self.client.get('/circles/', content_type='application/ld+json')
self.assertEqual(response.data['@type'], 'ldp:Container')
self.assertIn('@id', response.data)
self.assertIn('permissions', response.data)
self.assertIn('members', response.data['ldp:contains'][0])
self.assertEqual(response.data['ldp:contains'][0]['members']['@type'], 'foaf:Group')
self.assertIn('@id', response.data['ldp:contains'][0]['members'])
self.assertEqual(len(response.data['ldp:contains'][0]['members']['user_set']), 1)
# test for functioning with setting
def test_empty_container_serialization_nested_serializer_empty(self):
setattr(User._meta, 'depth', 1)
setattr(User._meta, 'empty_containers', ['owned_circles'])
self._set_up_circle_and_user()
response = self.client.get('/users/', content_type='application/ld+json')
self.assertEqual(response.data['@type'], 'ldp:Container')
self.assertIn('owned_circles', response.data['ldp:contains'][0])
self.assertIn('@id', response.data['ldp:contains'][0]['owned_circles'])
self.assertNotIn('permissions', response.data['ldp:contains'][0]['owned_circles'])
self.assertNotIn('ldp:contains', response.data['ldp:contains'][0]['owned_circles'])
# should serialize as normal on the nested viewset (directly asking for the container)
# test for normal functioning (without setting)
def test_empty_container_serialization_nested_viewset_no_empty(self):
user = self._set_up_circle_and_user()
response = self.client.get(f'/users/{user.pk}/owned_circles/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['@type'], 'ldp:Container')
self.assertIn('@id', response.data)
self.assertIn('ldp:contains', response.data)
self.assertIn('permissions', response.data)
self.assertIn('owner', response.data['ldp:contains'][0])
# test for functioning with setting
def test_empty_container_serialization_nested_viewset_empty(self):
setattr(User._meta, 'empty_containers', ['owned_circles'])
user = self._set_up_circle_and_user()
response = self.client.get(f'/users/{user.pk}/owned_circles/', content_type='application/ld+json')
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data['@type'], 'ldp:Container')
self.assertIn('@id', response.data)
self.assertIn('ldp:contains', response.data)
self.assertIn('permissions', response.data)
self.assertIn('owner', response.data['ldp:contains'][0])
# # test for checking fields ordering
# def test_ordered_field(self):
# self._set_up_circle_and_user()
# response = self.client.get('/users/', content_type='application/ld+json')
# fields_to_test = [
# response.data.keys(),
# response.data['ldp:contains'][-1],
# response.data['ldp:contains'][-1]['circle_set']
# ]
# for test_fields in fields_to_test:
# test_fields = list(test_fields)
# o_f = [field for field in self.ordered_fields if field in test_fields]
# self.assertEqual(o_f, test_fields[:len(o_f)])
djangoldp/tests/tests_guardian.py
View file @ e97d33ad
import json
import uuid
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group
from rest_framework.test import APIClient, APITestCase
from guardian.shortcuts import assign_perm
from .models import PermissionlessDummy, Dummy
from djangoldp.permissions import LDPPermissions
from .models import PermissionlessDummy, Dummy, LDPDummy
class TestsGuardian(APITestCase):
......@@ -15,45 +16,116 @@ class TestsGuardian(APITestCase):
def setUpLoggedInUser(self):
self.user = get_user_model().objects.create_user(username='john', email='jlennon@beatles.com',
password='glass onion')
self.group = Group.objects.create(name='Test')
self.user.groups.add(self.group)
self.user.save()
self.client.force_authenticate(user=self.user)
# optional setup for testing PermissionlessDummy model with parameterised perms
def setUpGuardianDummyWithPerms(self, perms=[]):
self.dummy = PermissionlessDummy.objects.create(some='test', slug='test')
def _get_dummy_with_perms(self, perms=None, parent=None, group=False):
if perms is None:
perms = []
dummy = PermissionlessDummy.objects.create(some='test', slug=uuid.uuid4(), parent=parent)
model_name = PermissionlessDummy._meta.model_name
for perm in perms:
assign_perm(perm + '_' + model_name, self.user, self.dummy)
perm = perm + '_' + model_name
if group:
#assigns container-level and object-level perms
assign_perm('tests.'+perm, self.group)
assign_perm(perm, self.group, dummy)
else:
assign_perm('tests.'+perm, self.user)
assign_perm(perm, self.user, dummy)
return dummy
# optional setup for testing PermissionlessDummy model with parameterised perms
def setUpGuardianDummyWithPerms(self, perms=None, parent=None, group=False):
self.dummy = self._get_dummy_with_perms(perms, parent, group)
# auxiliary function converts permission format for test
def _unpack_permissions(self, perms_from_response):
return perms_from_response
# test that dummy with no permissions set returns no results
def test_get_dummy_no_permissions(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms()
response = self.client.get('/permissionless-dummys/{}/'.format(self.dummy.slug))
self.assertEqual(response.status_code, 403)
self.assertEqual(response.status_code, 404)
# test with anonymous user
def test_get_dummy_anonymous_user(self):
self.setUpGuardianDummyWithPerms()
response = self.client.get('/permissionless-dummys/')
# I have no object permissions - I should receive a 403
self.assertEqual(response.status_code, 403)
def test_list_dummy_exception(self):
self.setUpLoggedInUser()
# I have permission on a permissionless dummy, but not in general
dummy_a = self._get_dummy_with_perms()
dummy_b = self._get_dummy_with_perms(['view'])
response = self.client.get('/permissionless-dummys/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
containees = [d['@id'] for d in response.data['ldp:contains']]
self.assertNotIn(dummy_a.urlid, containees)
self.assertIn(dummy_b.urlid, containees)
def test_list_dummy_group_exception(self):
self.setUpLoggedInUser()
dummy_a = self._get_dummy_with_perms()
dummy_b = self._get_dummy_with_perms(['view'], group=True)
response = self.client.get('/permissionless-dummys/')
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
containees = [d['@id'] for d in response.data['ldp:contains']]
self.assertNotIn(dummy_a.urlid, containees)
self.assertIn(dummy_b.urlid, containees)
def test_list_dummy_exception_nested_view(self):
self.setUpLoggedInUser()
parent = LDPDummy.objects.create(some="test")
# two dummies, one I have permission to view and one I don't
dummy_a = self._get_dummy_with_perms(parent=parent)
dummy_b = self._get_dummy_with_perms(['view'], parent)
response = self.client.get('/ldpdummys/{}/anons/'.format(parent.pk))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['ldp:contains']), 1)
def test_list_dummy_exception_nested_serializer(self):
self.setUpLoggedInUser()
parent = LDPDummy.objects.create(some="test")
# two dummies, one I have permission to view and one I don't
dummy_a = self._get_dummy_with_perms(parent=parent)
dummy_b = self._get_dummy_with_perms(['view'], parent)
response = self.client.get('/ldpdummys/{}/'.format(parent.pk))
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response.data['anons']['ldp:contains']), 1)
def test_get_dummy_permission_granted(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['view'])
response = self.client.get('/permissionless-dummys/{}/'.format(self.dummy.slug))
self.assertEqual(response.status_code, 200)
def test_get_dummy_group_permission_granted(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['view'], group=True)
response = self.client.get('/permissionless-dummys/{}/'.format(self.dummy.slug))
self.assertEqual(response.status_code, 200)
def test_get_dummy_permission_rejected(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['view'])
dummy_without = PermissionlessDummy.objects.create(some='test2', slug='test2')
response = self.client.get('/permissionless-dummys/{}/'.format(dummy_without.slug))
self.assertEqual(response.status_code, 403)
self.assertEqual(response.status_code, 404)
def test_patch_dummy_permission_granted(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['change'])
self.setUpGuardianDummyWithPerms(['view', 'change'])
body = {'some': "some_new"}
response = self.client.patch('/permissionless-dummys/{}/'.format(self.dummy.slug), data=json.dumps(body),
content_type='application/ld+json')
......@@ -66,16 +138,17 @@ class TestsGuardian(APITestCase):
body = {'some': "some_new"}
response = self.client.patch('/permissionless-dummys/{}/'.format(dummy_without.slug), data=json.dumps(body),
content_type='application/ld+json')
self.assertEqual(response.status_code, 403)
self.assertEqual(response.status_code, 404)
# TODO: PUT container of many objects approved on specific resource for which I do not have _model_ permissions
# test that custom permissions are returned on a model
def test_custom_permissions(self):
self.setUpLoggedInUser()
self.setUpGuardianDummyWithPerms(['custom_permission'])
self.setUpGuardianDummyWithPerms(['custom_permission', 'view'])
permissions = LDPPermissions()
result = permissions.user_permissions(self.user, self.dummy)
self.assertIn('custom_permission', result)
response = self.client.get('/permissionless-dummys/{}/'.format(self.dummy.slug))
self.assertIn('custom_permission', self._unpack_permissions(response.data['permissions']))
# test that duplicate permissions aren't returned
def test_no_duplicate_permissions(self):
......@@ -85,6 +158,11 @@ class TestsGuardian(APITestCase):
assign_perm('view_' + model_name, self.user, dummy)
permissions = LDPPermissions()
result = permissions.user_permissions(self.user, dummy)
self.assertEqual(result.count('view'), 1)
response = self.client.get('/dummys/{}/'.format(dummy.slug))
self.assertEqual(response.status_code, 200)
perms = self._unpack_permissions(response.data['permissions'])
self.assertIn('view', perms)
view_perms = [perm for perm in perms if perm == 'view']
self.assertEqual(len(view_perms), 1)
# TODO: attempting to migrate my object permissions by changing FK reference
djangoldp/tests/tests_inbox.py
View file @ e97d33ad
import json
from django.contrib.auth import get_user_model
from django.conf import settings
from django.db import IntegrityError
from django.contrib.auth import get_user_model
from django.test import override_settings
from rest_framework.test import APIClient, APITestCase
from djangoldp.tests.models import Circle, CircleMember, Project, UserProfile, DateModel, DateChild
from djangoldp.models import Activity, Follower
from djangoldp.tests.models import Circle, DateChild, DateModel, Project
class TestsInbox(APITestCase):
......@@ -19,7 +20,7 @@ class TestsInbox(APITestCase):
res = {
"@context": [
"https://www.w3.org/ns/activitystreams",
{"hd": "http://happy-dev.fr/owl/#"}
{"hd": "https://cdn.startinblox.com/owl#"}
],
"summary": "Something happened",
"type": type,
......@@ -49,7 +50,7 @@ class TestsInbox(APITestCase):
def _assert_activity_created(self, response, activity_len=1):
'''Auxiliary function asserts that the activity was created and returned correctly'''
activities = Activity.objects.all()
self.assertEquals(len(activities), activity_len)
self.assertEqual(len(activities), activity_len)
self.assertIn(response["Location"], activities.values_list('urlid', flat=True))
def _assert_follower_created(self, local_urlid, external_urlid):
......@@ -77,18 +78,49 @@ class TestsInbox(APITestCase):
# assert that the circle was created and the user associated as owner
circles = Circle.objects.all()
self.assertEquals(len(circles), 1)
self.assertEqual(len(circles), 1)
self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
self.assertEqual(circles[0].owner, self.user)
self._assert_activity_created(response)
# assert external circle member now following local user
self.assertEquals(Follower.objects.count(), 1)
self.assertEqual(Follower.objects.count(), 1)
self._assert_follower_created(self.user.urlid, "https://distant.com/circles/1/")
# # tests creation, and tests that consequential creation also happens
# # i.e. that I pass it an external circle which it doesn't know about, and it creates that too
# def test_create_activity_circle_member(self):
# obj = {
# "@type": "hd:circlemember",
# "@id": "https://distant.com/circlemembers/1/",
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": "https://distant.com/circles/1/"
# }
# }
# payload = self._get_activity_request_template("Create", obj)
# response = self.client.post('/inbox/',
# data=json.dumps(payload), content_type='application/ld+json')
# self.assertEqual(response.status_code, 201)
# # assert that the circle was created and the user associated as member
# circles = Circle.objects.all()
# self.assertEqual(len(circles), 1)
# self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
# self.assertTrue(circles[0].members.filter(user=self.user).exists())
# self._assert_activity_created(response)
# # assert external circle member now following local user
# self._assert_follower_created(self.user.urlid, "https://distant.com/circlemembers/1/")
# sender has sent a circle with a local user that doesn't exist
def test_create_activity_circle_local(self):
urlid = '{}{}'.format(settings.SITE_URL, 'someonewhodoesntexist')
urlid = '{}/{}'.format(settings.SITE_URL, 'someonewhodoesntexist')
obj = {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/",
......@@ -106,8 +138,8 @@ class TestsInbox(APITestCase):
self.assertEqual(response.status_code, 404)
# assert that the circle was not created neither a backlinked user
self.assertEquals(Circle.objects.count(), 0)
self.assertEquals(get_user_model().objects.count(), prior_users_length)
self.assertEqual(Circle.objects.count(), 0)
self.assertEqual(get_user_model().objects.count(), prior_users_length)
#
# ADD ACTIVITIES
......@@ -128,92 +160,93 @@ class TestsInbox(APITestCase):
# assert that the project backlink(s) & activity were created
projects = Project.objects.all()
user_projects = self.user.projects.all()
self.assertEquals(len(projects), 1)
self.assertEquals(len(user_projects), 1)
self.assertEqual(len(projects), 1)
self.assertEqual(len(user_projects), 1)
self.assertIn("https://distant.com/projects/1/", projects.values_list('urlid', flat=True))
self.assertIn("https://distant.com/projects/1/", user_projects.values_list('urlid', flat=True))
self._assert_activity_created(response)
# assert external circle member now following local user
self.assertEquals(Follower.objects.count(), 1)
self.assertEqual(Follower.objects.count(), 1)
self._assert_follower_created(self.user.urlid, "https://distant.com/projects/1/")
# circle model has a many-to-many with user, through an intermediate model
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
def test_add_activity_circle(self):
ext_circlemember_urlid = "https://distant.com/circle-members/1/"
ext_circle_urlid = "https://distant.com/circles/1/"
obj = {
"@type": "hd:circlemember",
"@id": ext_circlemember_urlid,
"user": {
"@type": "foaf:user",
"@id": self.user.urlid
},
"circle": {
"@type": "hd:circle",
"@id": ext_circle_urlid
}
}
payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
response = self.client.post('/inbox/',
data=json.dumps(payload), content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
self.assertEqual(response.status_code, 201)
# assert that the circle backlink(s) & activity were created
circles = Circle.objects.all()
user_circles = self.user.circles.all()
self.assertEquals(len(circles), 1)
self.assertEquals(len(user_circles), 1)
self.assertIn(ext_circle_urlid, circles.values_list('urlid', flat=True))
self.assertIn(ext_circlemember_urlid, user_circles.values_list('urlid', flat=True))
self._assert_activity_created(response)
# assert external circle member now following local user
self.assertEquals(Follower.objects.count(), 1)
self._assert_follower_created(self.user.urlid, ext_circlemember_urlid)
#TODO: write a new test for the new circle architecture
# # circle model has a many-to-many with user, through an intermediate model
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_circle(self):
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# ext_circle_urlid = "https://distant.com/circles/1/"
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": ext_circle_urlid
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload), content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self.assertEqual(response.status_code, 201)
# # assert that the circle backlink(s) & activity were created
# circles = Circle.objects.all()
# user_circles = self.user.circles.all()
# self.assertEqual(len(circles), 1)
# self.assertEqual(len(user_circles), 1)
# self.assertIn(ext_circle_urlid, circles.values_list('urlid', flat=True))
# self.assertIn(ext_circlemember_urlid, user_circles.values_list('urlid', flat=True))
# self._assert_activity_created(response)
# # assert external circle member now following local user
# self.assertEqual(Follower.objects.count(), 1)
# self._assert_follower_created(self.user.urlid, ext_circlemember_urlid)
# test sending an add activity when the backlink already exists
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
def test_add_activity_object_already_added(self):
circle = Circle.objects.create(urlid="https://distant.com/circles/1/")
cm = CircleMember.objects.create(urlid="https://distant.com/circle-members/1/", circle=circle, user=self.user)
obj = {
"@type": "hd:circlemember",
"@id": "https://distant.com/circle-members/1/",
"user": {
"@type": "foaf:user",
"@id": self.user.urlid
},
"circle": {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/"
}
}
payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
prior_count = Activity.objects.count()
response = self.client.post('/inbox/',
data=json.dumps(payload),
content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
self.assertEqual(response.status_code, 201)
# assert that the circle backlink(s) & activity were created
circles = Circle.objects.all()
user_circles = self.user.circles.all()
self.assertEquals(len(circles), 1)
self.assertEquals(len(user_circles), 1)
self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
self.assertIn("https://distant.com/circle-members/1/", user_circles.values_list('urlid', flat=True))
self._assert_activity_created(response)
self.assertEqual(Activity.objects.count(), prior_count + 1)
# assert that followers exist for the external urlids
self.assertEquals(Follower.objects.count(), 1)
self._assert_follower_created(self.user.urlid, cm.urlid)
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_object_already_added(self):
# circle = Circle.objects.create(urlid="https://distant.com/circles/1/")
# circle.members.user_set.add(self.user)
# obj = {
# "@type": "hd:circlemember",
# "@id": "https://distant.com/circle-members/1/",
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": "https://distant.com/circles/1/"
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# prior_count = Activity.objects.count()
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self.assertEqual(response.status_code, 201)
# # assert that the circle backlink(s) & activity were created
# circles = Circle.objects.all()
# user_circles = self.user.circles.all()
# self.assertEqual(len(circles), 1)
# self.assertEqual(len(user_circles), 1)
# self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
# self.assertIn("https://distant.com/circle-members/1/", user_circles.values_list('urlid', flat=True))
# self._assert_activity_created(response)
# self.assertEqual(Activity.objects.count(), prior_count + 1)
# # assert that followers exist for the external urlids
# self.assertEqual(Follower.objects.count(), 1)
# self._assert_follower_created(self.user.urlid, '') #TODO: replace with an existing model
# TODO: https://git.startinblox.com/djangoldp-packages/djangoldp/issues/250
def test_add_activity_str_parameter(self):
......@@ -224,6 +257,8 @@ class TestsInbox(APITestCase):
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 = {
......@@ -235,11 +270,144 @@ class TestsInbox(APITestCase):
data=json.dumps(payload), content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
self.assertEqual(response.status_code, 404)
def _test_fail_behaviour(self, response, status_code=400):
self.assertEqual(response.status_code, 400)
# assert that nothing was created
self.assertEqual(Circle.objects.count(), 0)
self.assertEqual(self.user.owned_circles.count(), 0)
self.assertEqual(Activity.objects.count(), 0)
self.assertEqual(Follower.objects.count(), 0)
# # error behaviour - invalid url
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_empty_url(self):
# # an invalid url
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# ext_circle_urlid = ""
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": ext_circle_urlid
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self._test_fail_behaviour(response, 400)
# # error behaviour - invalid url
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_invalid_url(self):
# # an invalid url
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# ext_circle_urlid = "not$valid$url"
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": ext_circle_urlid
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self._test_fail_behaviour(response, 400)
# # # error behaviour - None url
# # @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# # def test_add_activity_none_url(self):
# # # an invalid url
# # ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# # ext_circle_urlid = None
# # obj = {
# # "@type": "hd:circlemember",
# # "@id": ext_circlemember_urlid,
# # "user": {
# # "@type": "foaf:user",
# # "@id": self.user.urlid
# # },
# # "circle": {
# # "@type": "hd:circle",
# # "@id": ext_circle_urlid
# # }
# # }
# # payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# # response = self.client.post('/inbox/',
# # data=json.dumps(payload),
# # content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# # self._test_fail_behaviour(response, 400)
# # missing @id on a sub-object
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_no_id(self):
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle"
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self._test_fail_behaviour(response, 400)
# # missing @type on a sub-object
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_add_activity_no_type(self):
# ext_circlemember_urlid = "https://distant.com/circle-members/1/"
# obj = {
# "@type": "hd:circlemember",
# "@id": ext_circlemember_urlid,
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@id": "https://distant.com/circles/1/"
# }
# }
# payload = self._get_activity_request_template("Add", obj, self._build_target_from_user(self.user))
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self._test_fail_behaviour(response, 404)
def test_invalid_activity_missing_actor(self):
payload = {
"@context": [
"https://www.w3.org/ns/activitystreams",
{"hd": "http://happy-dev.fr/owl/#"}
{"hd": "https://cdn.startinblox.com/owl#"}
],
"summary": "Test was added to Test Circle",
"type": "Add",
......@@ -274,8 +442,8 @@ class TestsInbox(APITestCase):
# assert that the project backlink(s) & activity were created
projects = Project.objects.all()
user_projects = self.user.projects.all()
self.assertEquals(len(projects), 1)
self.assertEquals(len(user_projects), 1)
self.assertEqual(len(projects), 1)
self.assertEqual(len(user_projects), 1)
self.assertIn("https://distant.com/projects/1/", projects.values_list('urlid', flat=True))
self.assertIn("https://distant.com/projects/1/", user_projects.values_list('urlid', flat=True))
self._assert_activity_created(response)
......@@ -283,6 +451,7 @@ class TestsInbox(APITestCase):
self.assertNotEqual(backlink.pk, 100)
def test_missing_not_null_field_activity(self):
# TODO: catch the warning
# DateChild must not have a null reference to parent
# and parent must not have a null field 'date', which here is missing
obj = {
......@@ -327,13 +496,13 @@ class TestsInbox(APITestCase):
# assert that the circle backlink(s) were removed & activity were created
projects = Project.objects.all()
user_projects = self.user.projects.all()
self.assertEquals(len(projects), 1)
self.assertEquals(len(user_projects), 0)
self.assertEqual(len(projects), 1)
self.assertEqual(len(user_projects), 0)
self.assertIn("https://distant.com/projects/1/", projects.values_list('urlid', flat=True))
self._assert_activity_created(response, prior_activity_count + 2)
self._assert_activity_created(response, prior_activity_count + 1)
self.assertEqual(Follower.objects.count(), 0)
# TODO: test_remove_activity_project_using_target
# 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):
......@@ -350,7 +519,7 @@ class TestsInbox(APITestCase):
self.assertEqual(response.status_code, 201)
self._assert_activity_created(response)
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
@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)
......@@ -378,41 +547,41 @@ class TestsInbox(APITestCase):
# just received, did not send
self.assertEqual(Activity.objects.all().count(), prior_count + 1)
# Delete CircleMember
@override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
def test_delete_activity_circle_using_origin(self):
circle = Circle.objects.create(urlid="https://distant.com/circles/1/", allow_create_backlink=False)
cm = CircleMember.objects.create(urlid="https://distant.com/circle-members/1/",circle=circle, user=self.user)
Follower.objects.create(object=self.user.urlid, inbox='https://distant.com/inbox/',
follower=cm.urlid, is_backlink=True)
obj = {
"@type": "hd:circlemember",
"@id": "https://distant.com/circle-members/1/",
"user": {
"@type": "foaf:user",
"@id": self.user.urlid
},
"circle": {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/"
}
}
payload = self._get_activity_request_template("Delete", obj)
response = self.client.post('/inbox/',
data=json.dumps(payload),
content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
self.assertEqual(response.status_code, 201)
# assert that the CircleMember was deleted and activity was created
circles = Circle.objects.all()
user_circles = self.user.circles.all()
self.assertEquals(len(circles), 1)
self.assertEquals(CircleMember.objects.count(), 0)
self.assertEquals(len(user_circles), 0)
self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
self._assert_activity_created(response)
self.assertEqual(Follower.objects.count(), 0)
# # Delete CircleMember
# @override_settings(SEND_BACKLINKS=True, DISABLE_OUTBOX=True)
# def test_delete_activity_circle_using_origin(self):
# circle = Circle.objects.create(urlid="https://distant.com/circles/1/", allow_create_backlink=False)
# circle.members.user_set.add(self.user)
# Follower.objects.create(object=self.user.urlid, inbox='https://distant.com/inbox/',
# follower=circle.urlid, is_backlink=True)
# obj = {
# "@type": "hd:circlemember",
# "@id": "https://distant.com/circle-members/1/",
# "user": {
# "@type": "foaf:user",
# "@id": self.user.urlid
# },
# "circle": {
# "@type": "hd:circle",
# "@id": "https://distant.com/circles/1/"
# }
# }
# payload = self._get_activity_request_template("Delete", obj)
# response = self.client.post('/inbox/',
# data=json.dumps(payload),
# content_type='application/ld+json;profile="https://www.w3.org/ns/activitystreams"')
# self.assertEqual(response.status_code, 201)
# # assert that the Circle member was removed and activity was created
# circles = Circle.objects.all()
# user_circles = self.user.circles.all()
# self.assertEqual(len(circles), 1)
# self.assertEqual(circle.members.count(), 0)
# self.assertEqual(len(user_circles), 0)
# self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
# self._assert_activity_created(response)
# self.assertEqual(Follower.objects.count(), 0)
# TODO: test_delete_activity_circle_using_target
......@@ -423,6 +592,8 @@ class TestsInbox(APITestCase):
circle = Circle.objects.create(urlid="https://distant.com/circles/1/", owner=self.user)
self.assertEqual(circle.owner, self.user)
prior_user_count = get_user_model().objects.count()
obj = {
"@type": "hd:circle",
"@id": "https://distant.com/circles/1/",
......@@ -439,8 +610,8 @@ class TestsInbox(APITestCase):
# assert that the circle was created and the user associated as owner
circles = Circle.objects.all()
users = get_user_model().objects.all()
self.assertEquals(len(circles), 1)
self.assertEquals(len(users), 2)
self.assertEqual(len(circles), 1)
self.assertEqual(len(users), prior_user_count + 1)
distant_user = get_user_model().objects.get(urlid="https://distant.com/users/1/")
self.assertIn("https://distant.com/circles/1/", circles.values_list('urlid', flat=True))
self.assertEqual(circles[0].owner, distant_user)
......@@ -463,7 +634,7 @@ class TestsInbox(APITestCase):
# assert that Follower was created with correct values
followers = Follower.objects.all()
self.assertEquals(len(followers), 1)
self.assertEqual(len(followers), 1)
self._assert_activity_created(response)
follower = followers[0]
self.assertEqual("http://127.0.0.1:8000/inbox/", follower.inbox)
......@@ -474,10 +645,10 @@ class TestsInbox(APITestCase):
circle = Circle.objects.create(description='Test Description')
Follower.objects.create(object=circle.urlid, inbox="http://127.0.0.1:8000/inbox/")
followers = Follower.objects.all()
self.assertEquals(len(followers), 1)
self.assertEqual(len(followers), 1)
circle.delete()
followers = Follower.objects.all()
self.assertEquals(len(followers), 0)
self.assertEqual(len(followers), 0)
#
# GET Inbox
......@@ -485,3 +656,6 @@ class TestsInbox(APITestCase):
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, Circle, CircleMember
from djangoldp.tests.models import (Dummy, JobOffer, LDPDummy,
NoSuperUsersAllowedModel)
class LDPModelTest(TestCase):
def test_class_not_inheriting_ldp_model(self):
dummy = Dummy.objects.create(some="text")
self.assertEquals("/dummys/", Model.container_id(dummy))
self.assertEquals("/dummys/{}/".format(dummy.slug), Model.resource_id(dummy))
self.assertEqual("/dummys/", Model.container_id(dummy))
self.assertEqual("/dummys/{}/".format(dummy.slug), Model.resource_id(dummy))
def test_class_inheriting_ldp_model(self):
dummy = LDPDummy.objects.create(some="text")
self.assertEquals("/ldpdummys/", dummy.get_container_id())
self.assertEquals("http://happy-dev.fr/ldpdummys/{}/".format(dummy.pk), dummy.get_absolute_url())
self.assertEquals("/ldpdummys/", Model.container_id(dummy))
self.assertEquals("/ldpdummys/{}/".format(dummy.pk), Model.resource_id(dummy))
self.assertEqual("/ldpdummys/", dummy.get_container_id())
self.assertEqual("http://happy-dev.fr/ldpdummys/{}/".format(dummy.pk), dummy.get_absolute_url())
self.assertEqual("/ldpdummys/", Model.container_id(dummy))
self.assertEqual("/ldpdummys/{}/".format(dummy.pk), Model.resource_id(dummy))
def test_from_resolve_id(self):
saved_instance = Dummy.objects.create(some="text", slug="someid")
result = Model.resolve_id("/dummys/{}/".format(saved_instance.slug))
self.assertEquals(saved_instance, result)
self.assertEqual(saved_instance, result)
def test_resolve_container(self):
result = Model.resolve_container("/dummys/")
self.assertEquals(Dummy, result)
self.assertEqual(Dummy, result)
def test_auto_url(self):
from django.urls import get_resolver
......@@ -35,7 +34,7 @@ class LDPModelTest(TestCase):
view_name = '{}-list'.format(dummy._meta.object_name.lower())
path = 'http://happy-dev.fr/{}{}/'.format(get_resolver().reverse_dict[view_name][0][0][0], dummy.pk)
self.assertEquals(path, dummy.get_absolute_url())
self.assertEqual(path, dummy.get_absolute_url())
def test_ldp_manager_local_objects(self):
local = LDPDummy.objects.create(some='text')
......@@ -44,21 +43,4 @@ class LDPModelTest(TestCase):
local_queryset = LDPDummy.objects.local()
self.assertEqual(local_queryset.count(), 1)
self.assertIn(local, local_queryset)
self.assertNotIn(external, local_queryset)
def test_ldp_manager_nested_fields_auto(self):
nested_fields = Circle.objects.nested_fields()
expected_nested_fields = ['team', 'members']
self.assertEqual(len(nested_fields), len(expected_nested_fields))
for expected in expected_nested_fields:
self.assertIn(expected, nested_fields)
nested_fields = CircleMember.objects.nested_fields()
expected_nested_fields = []
self.assertEqual(nested_fields, expected_nested_fields)
def test_ldp_manager_nested_fields_exclude(self):
setattr(Circle.Meta, 'nested_fields_exclude', ['team'])
nested_fields = Circle.objects.nested_fields()
expected_nested_fields = ['members']
self.assertEqual(nested_fields, expected_nested_fields)
self.assertNotIn(external, local_queryset)
\ No newline at end of file