Skip to content
Snippets Groups Projects
Commit 47748b0f authored by Tore.Brede's avatar Tore.Brede
Browse files

Merge branch 'master' into GREG-9_audit_log

parents 8e492a0d 96bfda34
Branches
No related tags found
1 merge request!9GREG-9 Audit log
Pipeline #91266 passed
Showing
with 1538 additions and 142 deletions
from django_filters import rest_framework as filters from django_filters.rest_framework import (
BaseInFilter,
BooleanFilter,
FilterSet,
)
from greg.models import Person, PersonRole from greg.models import (
Person,
PersonRole,
PersonIdentity,
)
class PersonRoleFilter(filters.FilterSet): class PersonRoleFilter(FilterSet):
type = filters.BaseInFilter(field_name="role__type", lookup_expr="in") type = BaseInFilter(field_name="role__type", lookup_expr="in")
class Meta: class Meta:
model = PersonRole model = PersonRole
fields = ["type"] fields = ["type"]
class PersonFilter(filters.FilterSet): class PersonFilter(FilterSet):
verified = filters.BooleanFilter( verified = BooleanFilter(
field_name="person__verified_by_id", lookup_expr="isnull", exclude=True field_name="identities__verified_by_id", lookup_expr="isnull", exclude=True
) )
class Meta: class Meta:
model = Person model = Person
fields = ["first_name", "last_name", "verified"] fields = ["first_name", "last_name", "verified"]
class PersonIdentityFilter(FilterSet):
class Meta:
model = PersonIdentity
fields = ["type", "verified_by_id"]
from rest_framework import serializers from rest_framework import serializers
from greg.models import Person, PersonRole, Role from greg.models import Person, PersonRole, Role, PersonIdentity
class PersonSerializer(serializers.ModelSerializer): class PersonSerializer(serializers.ModelSerializer):
...@@ -13,6 +13,11 @@ class PersonSerializer(serializers.ModelSerializer): ...@@ -13,6 +13,11 @@ class PersonSerializer(serializers.ModelSerializer):
"date_of_birth", "date_of_birth",
"email", "email",
"mobile_phone", "mobile_phone",
"email_verified_date",
"mobile_phone",
"mobile_phone_verified_date",
"registration_completed_date",
"token",
] ]
...@@ -31,3 +36,24 @@ class PersonRoleSerializer(serializers.ModelSerializer): ...@@ -31,3 +36,24 @@ class PersonRoleSerializer(serializers.ModelSerializer):
"updated", "updated",
"role", "role",
] ]
class PersonIdentitySerializer(serializers.ModelSerializer):
class Meta:
model = PersonIdentity
fields = "__all__"
def is_duplicate(self, identity_type: str, value: str) -> bool:
# Guests may be verified using another unrecognised identification method,
# which the sponsor is required to elaborate in the value column.
# In this case we cannot assume the union of the identity type and
# the value to be unique across all records.
if identity_type == PersonIdentity.IdentityType.OTHER:
return False
# If the type is a specific ID type, then duplicates are not expected
return (
PersonIdentity.objects.filter(type__like=identity_type)
.filter(value__like=value)
.exists()
)
...@@ -6,4 +6,4 @@ from greg.models import Sponsor ...@@ -6,4 +6,4 @@ from greg.models import Sponsor
class SponsorSerializer(serializers.ModelSerializer): class SponsorSerializer(serializers.ModelSerializer):
class Meta: class Meta:
model = Sponsor model = Sponsor
fields = ["id", "feide_id"] fields = ["id", "feide_id", "first_name", "last_name"]
from django.urls import ( from django.urls import (
path,
re_path, re_path,
) )
from rest_framework.routers import DefaultRouter from rest_framework.routers import DefaultRouter
from drf_spectacular.views import (
SpectacularAPIView,
SpectacularSwaggerView,
)
from greg.api.views.consent import ConsentViewSet from greg.api.views.consent import ConsentViewSet
from greg.api.views.organizational_unit import OrganizationalUnitViewSet from greg.api.views.organizational_unit import OrganizationalUnitViewSet
from greg.api.views.person import ( from greg.api.views.person import (
PersonRoleViewSet, PersonRoleViewSet,
PersonViewSet, PersonViewSet,
PersonIdentityViewSet,
) )
from greg.api.views.role import RoleViewSet from greg.api.views.role import RoleViewSet
from greg.api.views.health import Health from greg.api.views.sponsor import SponsorViewSet, SponsorGuestsViewSet
from greg.api.views.sponsor import SponsorViewSet
router = DefaultRouter() router = DefaultRouter()
router.register(r"persons", PersonViewSet, basename="person") router.register(r"persons", PersonViewSet, basename="person")
...@@ -29,13 +24,6 @@ router.register(r"orgunit", OrganizationalUnitViewSet, basename="orgunit") ...@@ -29,13 +24,6 @@ router.register(r"orgunit", OrganizationalUnitViewSet, basename="orgunit")
urlpatterns = router.urls urlpatterns = router.urls
urlpatterns += [ urlpatterns += [
path("schema/", SpectacularAPIView.as_view(), name="schema"),
path(
"schema/swagger-ui/",
SpectacularSwaggerView.as_view(url_name="schema"),
name="swagger-ui",
),
path("health/", Health.as_view()),
re_path( re_path(
r"^persons/(?P<person_id>[0-9]+)/roles/$", r"^persons/(?P<person_id>[0-9]+)/roles/$",
PersonRoleViewSet.as_view({"get": "list", "post": "create"}), PersonRoleViewSet.as_view({"get": "list", "post": "create"}),
...@@ -43,7 +31,26 @@ urlpatterns += [ ...@@ -43,7 +31,26 @@ urlpatterns += [
), ),
re_path( re_path(
r"^persons/(?P<person_id>[0-9]+)/roles/(?P<id>[0-9]+)/$", r"^persons/(?P<person_id>[0-9]+)/roles/(?P<id>[0-9]+)/$",
PersonRoleViewSet.as_view({"get": "retrieve"}), PersonRoleViewSet.as_view(
{"get": "retrieve", "patch": "partial_update", "delete": "destroy"}
),
name="person_role-detail", name="person_role-detail",
), ),
re_path(
r"^persons/(?P<person_id>[0-9]+)/identities/$",
PersonIdentityViewSet.as_view({"get": "list", "post": "create"}),
name="person_identity-list",
),
re_path(
r"^persons/(?P<person_id>[0-9]+)/identities/(?P<id>[0-9]+)$",
PersonIdentityViewSet.as_view(
{"get": "retrieve", "delete": "destroy", "patch": "partial_update"}
),
name="person_identity-detail",
),
re_path(
r"^sponsors/(?P<sponsor_id>[0-9]+)/guests/$",
SponsorGuestsViewSet.as_view({"get": "list"}),
name="sponsor_guests-list",
),
] ]
from django.core.exceptions import ValidationError from django.core.exceptions import ValidationError
from django_filters import rest_framework as filters from django_filters import rest_framework as filters
from drf_spectacular.utils import extend_schema, OpenApiParameter from drf_spectacular.utils import extend_schema, OpenApiParameter
from rest_framework import viewsets from rest_framework import viewsets, status
from rest_framework.response import Response
from greg.api.filters import PersonFilter, PersonRoleFilter from greg.api.filters import PersonFilter, PersonRoleFilter, PersonIdentityFilter
from greg.api.pagination import PrimaryKeyCursorPagination from greg.api.pagination import PrimaryKeyCursorPagination
from greg.api.serializers.person import PersonSerializer, PersonRoleSerializer from greg.api.serializers.person import (
from greg.models import Person, PersonRole PersonSerializer,
PersonRoleSerializer,
PersonIdentitySerializer,
)
from greg.models import Person, PersonRole, PersonIdentity
class PersonViewSet(viewsets.ModelViewSet): class PersonViewSet(viewsets.ModelViewSet):
...@@ -31,7 +36,7 @@ class PersonViewSet(viewsets.ModelViewSet): ...@@ -31,7 +36,7 @@ class PersonViewSet(viewsets.ModelViewSet):
] ]
) )
def list(self, request, *args, **kwargs): def list(self, request, *args, **kwargs):
return super().list(request) return super().list(request, *args, **kwargs)
class PersonRoleViewSet(viewsets.ModelViewSet): class PersonRoleViewSet(viewsets.ModelViewSet):
...@@ -42,6 +47,7 @@ class PersonRoleViewSet(viewsets.ModelViewSet): ...@@ -42,6 +47,7 @@ class PersonRoleViewSet(viewsets.ModelViewSet):
pagination_class = PrimaryKeyCursorPagination pagination_class = PrimaryKeyCursorPagination
filter_backends = (filters.DjangoFilterBackend,) filter_backends = (filters.DjangoFilterBackend,)
filterset_class = PersonRoleFilter filterset_class = PersonRoleFilter
lookup_field = "id"
def get_queryset(self): def get_queryset(self):
qs = self.queryset qs = self.queryset
...@@ -62,3 +68,49 @@ class PersonRoleViewSet(viewsets.ModelViewSet): ...@@ -62,3 +68,49 @@ class PersonRoleViewSet(viewsets.ModelViewSet):
raise ValidationError("No person id") raise ValidationError("No person id")
serializer.save(person_id=person_id) serializer.save(person_id=person_id)
class PersonIdentityViewSet(viewsets.ModelViewSet):
"""
Person identity API
"""
queryset = PersonIdentity.objects.all().order_by("id")
serializer_class = PersonIdentitySerializer
pagination_class = PrimaryKeyCursorPagination
filter_backends = (filters.DjangoFilterBackend,)
filterset_class = PersonIdentityFilter
# This is set so that the id parameter in the path of the URL is used for looking up objects
lookup_url_kwarg = "id"
def get_queryset(self):
qs = self.queryset
if not self.kwargs:
return qs.none()
person_id = self.kwargs["person_id"]
qs = qs.filter(person_id=person_id)
return qs
def create(self, request, *args, **kwargs):
# Want to get the person id which is part of the API path and then
# include this with the data used to create the identity for the person
person_id = self.kwargs["person_id"]
if person_id is None:
# Should not happen, the person ID is part of the API path
raise ValidationError("No person id")
input_data = request.data.copy()
input_data["person"] = person_id
serializer = self.get_serializer(data=input_data)
if serializer.is_duplicate(input_data["type"], input_data["value"]):
raise ValidationError("Duplicate identity entry exists")
serializer.is_valid(raise_exception=True)
self.perform_create(serializer)
headers = self.get_success_headers(serializer.data)
return Response(
serializer.data, status=status.HTTP_201_CREATED, headers=headers
)
from rest_framework.viewsets import ReadOnlyModelViewSet from drf_spectacular.utils import extend_schema, OpenApiParameter
from rest_framework import mixins
from rest_framework.viewsets import GenericViewSet, ModelViewSet
from greg.api.pagination import PrimaryKeyCursorPagination from greg.api.pagination import PrimaryKeyCursorPagination
from greg.api.serializers import PersonSerializer
from greg.api.serializers.sponsor import SponsorSerializer from greg.api.serializers.sponsor import SponsorSerializer
from greg.models import Sponsor from greg.models import Sponsor, Person
class SponsorViewSet(ReadOnlyModelViewSet): class SponsorViewSet(ModelViewSet):
"""Sponsor API""" """Sponsor API"""
queryset = Sponsor.objects.all().order_by("id") queryset = Sponsor.objects.all().order_by("id")
serializer_class = SponsorSerializer serializer_class = SponsorSerializer
pagination_class = PrimaryKeyCursorPagination pagination_class = PrimaryKeyCursorPagination
lookup_field = "id" lookup_field = "id"
@extend_schema(
parameters=[
OpenApiParameter(
name="sponsor_id",
description="Sponsor ID",
location=OpenApiParameter.PATH,
required=True,
type=int,
)
]
)
class SponsorGuestsViewSet(mixins.ListModelMixin, GenericViewSet):
queryset = Person.objects.all().order_by("id")
serializer_class = PersonSerializer
pagination_class = PrimaryKeyCursorPagination
lookup_field = "id"
def get_queryset(self):
qs = self.queryset
if not self.kwargs:
return qs.none()
sponsor_id = self.kwargs["sponsor_id"]
qs = qs.filter(person_roles__registered_by=sponsor_id).order_by("id")
return qs
import logging
import signal
import sys
from threading import Event
import daemon
import lockfile
from daemon import pidfile
from django.conf import settings
from django.core.management.base import BaseCommand
from greg.schedule import ExpiringRolesNotification
logging.config.dictConfig(settings.LOGGING)
logger = logging.getLogger()
def exception_handler(ex_cls, ex, tb):
logger.critical("Uncaught exception:", exc_info=(ex_cls, ex, tb))
sys.excepthook = exception_handler
class Command(BaseCommand):
"""
This command starts a basic task runner. All tasks it is supposed to
run are given explicitly in code
"""
help = "Start schedule task runner"
def add_arguments(self, parser):
parser.add_argument(
"--detach", action="store_true", help="Run detached as a dæmon"
)
parser.add_argument(
"--use-pidfile", action="store_true", help="Use a PID lockfile"
)
def handle(self, *args, **options):
lock_context = None
if options.get("use_pidfile"):
lock_context = pidfile.TimeoutPIDLockFile(
settings.SCHEDULE_TASKS["daemon"]["pid_file"]
)
try:
exit_event = Event()
def exit_wrapper(*args):
exit_event.set()
with daemon.DaemonContext(
pidfile=lock_context,
files_preserve=[x.stream.fileno() for x in logger.handlers],
stderr=sys.stderr,
stdout=sys.stdout,
signal_map={signal.SIGINT: exit_wrapper, signal.SIGTERM: exit_wrapper},
detach_process=options.get("detach"),
):
logger.info(
"Running %s", "detached" if options.get("detach") else "attached"
)
# For now there is just one task, but the idea is that more can
# be added
expiring_role_notification_task = ExpiringRolesNotification()
while not exit_event.is_set():
expiring_role_notification_task.run()
# The single task is set up for far only needs to run once
# a day, but running every 6 hours just in case a run fails,
# it is up the task it self to figure out if needs to do
# something every time it is called
exit_event.wait(60 * 60 * 6)
except lockfile.AlreadyLocked as e:
logger.warning("Can't start daemon: %s", e)
sys.exit(1)
finally:
logger.info("Stopped")
import os
import signal
from django.conf import settings
from django.core.management.base import BaseCommand
class Command(BaseCommand):
help = "Stop schedule tasks"
def handle(self, *args, **options):
try:
with open(settings.SCHEDULE_TASKS["daemon"]["pid_file"], "r") as f:
pid = int(f.read().strip())
os.kill(pid, signal.SIGINT)
except FileNotFoundError:
pass
# Generated by Django 3.2.5 on 2021-07-15 13:31 # Generated by Django 3.2.5 on 2021-08-23 08:53
import datetime import datetime
import dirtyfields.dirtyfields import dirtyfields.dirtyfields
...@@ -73,10 +73,10 @@ class Migration(migrations.Migration): ...@@ -73,10 +73,10 @@ class Migration(migrations.Migration):
('updated', models.DateTimeField(auto_now=True)), ('updated', models.DateTimeField(auto_now=True)),
('first_name', models.CharField(max_length=256)), ('first_name', models.CharField(max_length=256)),
('last_name', models.CharField(max_length=256)), ('last_name', models.CharField(max_length=256)),
('date_of_birth', models.DateField()), ('date_of_birth', models.DateField(null=True)),
('email', models.EmailField(max_length=254)), ('email', models.EmailField(blank=True, max_length=254)),
('email_verified_date', models.DateField(null=True)), ('email_verified_date', models.DateField(null=True)),
('mobile_phone', models.CharField(max_length=15)), ('mobile_phone', models.CharField(blank=True, max_length=15)),
('mobile_phone_verified_date', models.DateField(null=True)), ('mobile_phone_verified_date', models.DateField(null=True)),
('registration_completed_date', models.DateField(null=True)), ('registration_completed_date', models.DateField(null=True)),
('token', models.CharField(blank=True, max_length=32)), ('token', models.CharField(blank=True, max_length=32)),
...@@ -104,6 +104,14 @@ class Migration(migrations.Migration): ...@@ -104,6 +104,14 @@ class Migration(migrations.Migration):
}, },
bases=(dirtyfields.dirtyfields.DirtyFieldsMixin, models.Model), bases=(dirtyfields.dirtyfields.DirtyFieldsMixin, models.Model),
), ),
migrations.CreateModel(
name='ScheduleTask',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=32)),
('last_completed', models.DateTimeField(null=True)),
],
),
migrations.CreateModel( migrations.CreateModel(
name='Sponsor', name='Sponsor',
fields=[ fields=[
...@@ -111,6 +119,8 @@ class Migration(migrations.Migration): ...@@ -111,6 +119,8 @@ class Migration(migrations.Migration):
('created', models.DateTimeField(auto_now_add=True)), ('created', models.DateTimeField(auto_now_add=True)),
('updated', models.DateTimeField(auto_now=True)), ('updated', models.DateTimeField(auto_now=True)),
('feide_id', models.CharField(max_length=256)), ('feide_id', models.CharField(max_length=256)),
('first_name', models.CharField(max_length=256)),
('last_name', models.CharField(max_length=256)),
], ],
bases=(dirtyfields.dirtyfields.DirtyFieldsMixin, models.Model), bases=(dirtyfields.dirtyfields.DirtyFieldsMixin, models.Model),
), ),
...@@ -137,19 +147,16 @@ class Migration(migrations.Migration): ...@@ -137,19 +147,16 @@ class Migration(migrations.Migration):
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('created', models.DateTimeField(auto_now_add=True)), ('created', models.DateTimeField(auto_now_add=True)),
('updated', models.DateTimeField(auto_now=True)), ('updated', models.DateTimeField(auto_now=True)),
('start_date', models.DateField()), ('start_date', models.DateField(null=True)),
('end_date', models.DateField()), ('end_date', models.DateField()),
('contact_person_unit', models.TextField()), ('contact_person_unit', models.TextField(blank=True)),
('comments', models.TextField(blank=True)), ('comments', models.TextField(blank=True)),
('available_in_search', models.BooleanField(default=False)), ('available_in_search', models.BooleanField(default=False)),
('person', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='person_roles', to='greg.person')), ('person', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='person_roles', to='greg.person')),
('registered_by', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='sponsor_role', to='greg.sponsor')), ('registered_by', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='sponsor_role', to='greg.sponsor')),
('role', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='person_roles', to='greg.role')), ('role', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='person_roles', to='greg.role')),
('unit', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='unit_person_role', to='greg.organizationalunit')), ('unit', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='unit_person_role', to='greg.organizationalunit')),
], ],
options={
'abstract': False,
},
bases=(dirtyfields.dirtyfields.DirtyFieldsMixin, models.Model), bases=(dirtyfields.dirtyfields.DirtyFieldsMixin, models.Model),
), ),
migrations.CreateModel( migrations.CreateModel(
...@@ -158,12 +165,12 @@ class Migration(migrations.Migration): ...@@ -158,12 +165,12 @@ class Migration(migrations.Migration):
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('created', models.DateTimeField(auto_now_add=True)), ('created', models.DateTimeField(auto_now_add=True)),
('updated', models.DateTimeField(auto_now=True)), ('updated', models.DateTimeField(auto_now=True)),
('type', models.CharField(choices=[('PASSPORT_NUMBER', 'Passport Number'), ('FEIDE_ID', 'Feide Id')], max_length=15)), ('type', models.CharField(choices=[('ID_PORTEN', 'Id Porten'), ('FEIDE_ID', 'Feide Id'), ('PASSPORT', 'Passport'), ('DRIVERS_LICENSE', 'Drivers License'), ('NATIONAL_ID_CARD', 'National Id Card'), ('NATIONAL_ID_NUMBER', 'National Id Number'), ('OTHER', 'Other')], max_length=18)),
('source', models.CharField(max_length=256)), ('source', models.CharField(max_length=256)),
('value', models.CharField(max_length=256)), ('value', models.CharField(max_length=256)),
('verified', models.CharField(blank=True, choices=[('AUTOMATIC', 'Automatic'), ('MANUAL', 'Manual')], max_length=9)), ('verified', models.CharField(blank=True, choices=[('AUTOMATIC', 'Automatic'), ('MANUAL', 'Manual')], max_length=9)),
('verified_when', models.DateField(null=True)), ('verified_when', models.DateField(null=True)),
('person', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='person', to='greg.person')), ('person', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='identities', to='greg.person')),
('verified_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.PROTECT, related_name='sponsor', to='greg.sponsor')), ('verified_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.PROTECT, related_name='sponsor', to='greg.sponsor')),
], ],
options={ options={
...@@ -177,9 +184,9 @@ class Migration(migrations.Migration): ...@@ -177,9 +184,9 @@ class Migration(migrations.Migration):
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('created', models.DateTimeField(auto_now_add=True)), ('created', models.DateTimeField(auto_now_add=True)),
('updated', models.DateTimeField(auto_now=True)), ('updated', models.DateTimeField(auto_now=True)),
('consent_given_at', models.DateField()), ('consent_given_at', models.DateField(null=True)),
('consent', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='link_person_consent', to='greg.consent')), ('consent', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='link_person_consent', to='greg.consent')),
('person', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='link_person_consent', to='greg.person')), ('person', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='link_person_consent', to='greg.person')),
], ],
bases=(dirtyfields.dirtyfields.DirtyFieldsMixin, models.Model), bases=(dirtyfields.dirtyfields.DirtyFieldsMixin, models.Model),
), ),
...@@ -201,6 +208,10 @@ class Migration(migrations.Migration): ...@@ -201,6 +208,10 @@ class Migration(migrations.Migration):
model_name='sponsor', model_name='sponsor',
constraint=models.UniqueConstraint(fields=('feide_id',), name='unique_feide_id'), constraint=models.UniqueConstraint(fields=('feide_id',), name='unique_feide_id'),
), ),
migrations.AddConstraint(
model_name='personrole',
constraint=models.UniqueConstraint(fields=('person_id', 'role_id', 'unit_id', 'start_date', 'end_date'), name='person_role_unique'),
),
migrations.AddConstraint( migrations.AddConstraint(
model_name='personconsent', model_name='personconsent',
constraint=models.UniqueConstraint(fields=('person', 'consent'), name='person_consent_unique'), constraint=models.UniqueConstraint(fields=('person', 'consent'), name='person_consent_unique'),
......
from datetime import date from datetime import (
date,
datetime,
)
from dirtyfields import DirtyFieldsMixin from dirtyfields import DirtyFieldsMixin
from django.db import models from django.db import models
...@@ -34,10 +37,10 @@ class Person(BaseModel): ...@@ -34,10 +37,10 @@ class Person(BaseModel):
first_name = models.CharField(max_length=256) first_name = models.CharField(max_length=256)
last_name = models.CharField(max_length=256) last_name = models.CharField(max_length=256)
date_of_birth = models.DateField() date_of_birth = models.DateField(null=True)
email = models.EmailField() email = models.EmailField(blank=True)
email_verified_date = models.DateField(null=True) email_verified_date = models.DateField(null=True)
mobile_phone = models.CharField(max_length=15) mobile_phone = models.CharField(max_length=15, blank=True)
mobile_phone_verified_date = models.DateField(null=True) mobile_phone_verified_date = models.DateField(null=True)
registration_completed_date = models.DateField(null=True) registration_completed_date = models.DateField(null=True)
token = models.CharField(max_length=32, blank=True) token = models.CharField(max_length=32, blank=True)
...@@ -57,6 +60,65 @@ class Person(BaseModel): ...@@ -57,6 +60,65 @@ class Person(BaseModel):
self.last_name, self.last_name,
) )
@property
def is_registered(self) -> bool:
"""
A registered guest is a person who has completed
the registration process.
The registration process requires that the guest has verified
their email address via a token link, filled in the required
personal information along with providing at least one
verification method, and accepted the institution's mandatory
consent forms.
The registered guest may or may not already be verified,
depending on the verification method. However, before a
guest is cleared for account creation at the institution's
IGA, the guest must be both registered (``is_registered``)
and verified (``is_verified``).
"""
# registration_completed_date is set only after accepting consents
return (
self.registration_completed_date is not None
and self.registration_completed_date <= datetime.now()
)
@property
def is_verified(self) -> bool:
"""
A verified guest is a person whom has had their personal
identity verified.
Due to the diversity of guests at a university institution,
there are many ways for guests to identify themselves.
These include Feide ID, passport number, driver's license,
national ID card, or another manual (human) verification.
Some of these methods are implicitly trusted (Feide ID) because
the guest is likely a visitor from another academic institution
who has already been pre-verified. Others are manul, such
as the sponsor vouching for having checked the guest's
personal details against his or her passport.
The verified guest may or may not have completed
the registration process which implies that it is only
the combination of being registered (``is_registered``)
and being verified (``is_verified``) that qualifies for being cleared
for account creation in the IGA.
Note that we do not distinguish between the quality,
authenticity, or trust level of the guest's associated identities.
"""
# the requirement is minimum one personal identity
return (
self.identities.filter(
verified_when__isnull=False,
verified_when__lte=datetime.now(),
).count()
>= 1
)
class Role(BaseModel): class Role(BaseModel):
"""A role variant.""" """A role variant."""
...@@ -85,7 +147,7 @@ class PersonRole(BaseModel): ...@@ -85,7 +147,7 @@ class PersonRole(BaseModel):
"""The relationship between a person and a role.""" """The relationship between a person and a role."""
person = models.ForeignKey( person = models.ForeignKey(
"Person", on_delete=models.PROTECT, related_name="person_roles" "Person", on_delete=models.CASCADE, related_name="person_roles"
) )
role = models.ForeignKey( role = models.ForeignKey(
"Role", on_delete=models.PROTECT, related_name="person_roles" "Role", on_delete=models.PROTECT, related_name="person_roles"
...@@ -93,16 +155,25 @@ class PersonRole(BaseModel): ...@@ -93,16 +155,25 @@ class PersonRole(BaseModel):
unit = models.ForeignKey( unit = models.ForeignKey(
"OrganizationalUnit", on_delete=models.PROTECT, related_name="unit_person_role" "OrganizationalUnit", on_delete=models.PROTECT, related_name="unit_person_role"
) )
start_date = models.DateField() # The start date can be null for people that are already
# attached to the institution but does not have a guest account
start_date = models.DateField(null=True)
end_date = models.DateField() end_date = models.DateField()
# TODO Is this field needed? contact_person_unit = models.TextField(blank=True)
contact_person_unit = models.TextField()
comments = models.TextField(blank=True) comments = models.TextField(blank=True)
available_in_search = models.BooleanField(default=False) available_in_search = models.BooleanField(default=False)
registered_by = models.ForeignKey( registered_by = models.ForeignKey(
"Sponsor", on_delete=models.PROTECT, related_name="sponsor_role" "Sponsor", on_delete=models.PROTECT, related_name="sponsor_role"
) )
class Meta:
constraints = [
models.UniqueConstraint(
fields=["person_id", "role_id", "unit_id", "start_date", "end_date"],
name="person_role_unique",
)
]
def __repr__(self): def __repr__(self):
return "{}(id={!r}, person={!r}, role={!r})".format( return "{}(id={!r}, person={!r}, role={!r})".format(
self.__class__.__name__, self.pk, self.person, self.role self.__class__.__name__, self.pk, self.person, self.role
...@@ -131,19 +202,25 @@ class Notification(BaseModel): ...@@ -131,19 +202,25 @@ class Notification(BaseModel):
class PersonIdentity(BaseModel): class PersonIdentity(BaseModel):
# TODO: Add more types
class IdentityType(models.TextChoices): class IdentityType(models.TextChoices):
PASSPORT_NUMBER = "PASSPORT_NUMBER" ID_PORTEN = "ID_PORTEN"
FEIDE_ID = "FEIDE_ID" FEIDE_ID = "FEIDE_ID"
PASSPORT = "PASSPORT"
DRIVERS_LICENSE = "DRIVERS_LICENSE"
NATIONAL_ID_CARD = "NATIONAL_ID_CARD"
# In Norway this is the foedselsnummer
NATIONAL_ID_NUMBER = "NATIONAL_ID_NUMBER"
# Sponsor writes what is used in the value column
OTHER = "OTHER"
class Verified(models.TextChoices): class Verified(models.TextChoices):
AUTOMATIC = "AUTOMATIC" AUTOMATIC = "AUTOMATIC"
MANUAL = "MANUAL" MANUAL = "MANUAL"
person = models.ForeignKey( person = models.ForeignKey(
"Person", on_delete=models.PROTECT, related_name="person" "Person", on_delete=models.CASCADE, related_name="identities"
) )
type = models.CharField(max_length=15, choices=IdentityType.choices) type = models.CharField(max_length=18, choices=IdentityType.choices)
source = models.CharField(max_length=256) source = models.CharField(max_length=256)
value = models.CharField(max_length=256) value = models.CharField(max_length=256)
verified = models.CharField(max_length=9, choices=Verified.choices, blank=True) verified = models.CharField(max_length=9, choices=Verified.choices, blank=True)
...@@ -153,15 +230,14 @@ class PersonIdentity(BaseModel): ...@@ -153,15 +230,14 @@ class PersonIdentity(BaseModel):
verified_when = models.DateField(null=True) verified_when = models.DateField(null=True)
def __repr__(self): def __repr__(self):
return ( return "{}(id={!r}, type={!r}, source={!r}, value={!r}, verified_by={!r}, verified_when={!r})".format(
"{}(id={!r}, type={!r}, source={!r}, value={!r}, verified_by={!r})".format(
self.__class__.__name__, self.__class__.__name__,
self.pk, self.pk,
self.type, self.type,
self.source, self.source,
self.value, self.value,
self.verified_by, self.verified_by,
) self.verified_when,
) )
...@@ -197,12 +273,13 @@ class PersonConsent(BaseModel): ...@@ -197,12 +273,13 @@ class PersonConsent(BaseModel):
""" """
person = models.ForeignKey( person = models.ForeignKey(
"Person", on_delete=models.PROTECT, related_name="link_person_consent" "Person", on_delete=models.CASCADE, related_name="link_person_consent"
) )
consent = models.ForeignKey( consent = models.ForeignKey(
"Consent", on_delete=models.PROTECT, related_name="link_person_consent" "Consent", on_delete=models.PROTECT, related_name="link_person_consent"
) )
consent_given_at = models.DateField() # If the date is blank it means the person has not given consent yet
consent_given_at = models.DateField(null=True)
class Meta: class Meta:
constraints = [ constraints = [
...@@ -248,6 +325,9 @@ class Sponsor(BaseModel): ...@@ -248,6 +325,9 @@ class Sponsor(BaseModel):
""" """
feide_id = models.CharField(max_length=256) feide_id = models.CharField(max_length=256)
first_name = models.CharField(max_length=256)
last_name = models.CharField(max_length=256)
units = models.ManyToManyField( units = models.ManyToManyField(
"OrganizationalUnit", "OrganizationalUnit",
through="SponsorOrganizationalUnit", through="SponsorOrganizationalUnit",
...@@ -255,8 +335,12 @@ class Sponsor(BaseModel): ...@@ -255,8 +335,12 @@ class Sponsor(BaseModel):
) )
def __repr__(self): def __repr__(self):
return "{}(id={!r}, feide_id={!r})".format( return "{}(id={!r}, feide_id={!r}, first_name={!r}, last_name={!r})".format(
self.__class__.__name__, self.pk, self.feide_id self.__class__.__name__,
self.pk,
self.feide_id,
self.first_name,
self.last_name,
) )
class Meta: class Meta:
...@@ -294,3 +378,17 @@ class SponsorOrganizationalUnit(BaseModel): ...@@ -294,3 +378,17 @@ class SponsorOrganizationalUnit(BaseModel):
self.organizational_unit, self.organizational_unit,
self.hierarchical_access, self.hierarchical_access,
) )
class ScheduleTask(models.Model):
"""
Stores information about a task
"""
name = models.CharField(max_length=32)
last_completed = models.DateTimeField(null=True)
def __repr__(self):
return "{}(id={!r}, name={!r}, last_completed={!r})".format(
self.__class__.__name__, self.pk, self.name, self.last_completed
)
import time
from datetime import date, datetime, timedelta
from typing import Optional
from abc import ABC, abstractmethod
from django.utils import timezone
from greg.models import PersonRole, Notification, ScheduleTask
class BaseSchedule(ABC):
"""
Provides common methods for tasks
"""
task_object: ScheduleTask
def __init__(self, name: str):
self.task_object = ScheduleTask.objects.get(name=name)
def run(self):
self._run_internal()
self.task_object.last_completed = timezone.now()
self.task_object.save()
@abstractmethod
def _run_internal(self):
pass
def get_last_run(self) -> Optional[datetime]:
self.task_object.refresh_from_db()
return self.task_object.last_completed
def _store_notification(
self, identifier, object_type, operation, **kwargs
) -> Notification:
return Notification.objects.create(
identifier=identifier,
object_type=object_type,
operation=operation,
issued_at=int(time.time()),
meta=kwargs,
)
class ExpiringRolesNotification(BaseSchedule):
"""
This task does a simple check for roles that will expire in 30 days
and creates entries in the notification table for them.
There should be an entry in the ScheduleTask-table with name role_expiration
before this task is run.
Some assumptions that are made:
- The task will be run every day, it does
not keep track of which roles it has or has not notified, it only
looks at which roles will expire in exactly 30 days
- If there are roles created that expire in less than 30 days, no
notification is necessary
"""
EXPIRATION_THRESHOLD_DAYS = 30
def __init__(self):
# Will raise exception if there is not exactly one result
super().__init__("role_expiration")
def _run_internal(self):
last_run = self.get_last_run()
# Only run once per day
if last_run is None or last_run.date() != date.today():
check_date = datetime.today() + timedelta(
days=self.EXPIRATION_THRESHOLD_DAYS
)
self.__get_roles_about_to_expire(check_date)
def __get_roles_about_to_expire(self, end_date: date):
roles_about_to_expire = PersonRole.objects.filter(end_date=end_date)
for person_role in roles_about_to_expire:
meta = {"person_id": person_role.person.id, "role_id": person_role.role.id}
self._store_notification(
identifier=person_role.id,
object_type="PersonRole",
operation="expire_reminder",
**meta
)
import time import time
import logging import logging
from typing import Dict
from django.db import models from django.db import models
from django.dispatch import receiver from django.dispatch import receiver
...@@ -9,6 +10,9 @@ from greg.models import ( ...@@ -9,6 +10,9 @@ from greg.models import (
PersonRole, PersonRole,
Role, Role,
Notification, Notification,
PersonIdentity,
PersonConsent,
Consent,
) )
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
...@@ -17,6 +21,8 @@ SUPPORTED_MODELS = ( ...@@ -17,6 +21,8 @@ SUPPORTED_MODELS = (
Person, Person,
PersonRole, PersonRole,
Role, Role,
PersonIdentity,
PersonConsent,
) )
...@@ -30,6 +36,10 @@ def disconnect_notification_signals(*args, **kwargs): ...@@ -30,6 +36,10 @@ def disconnect_notification_signals(*args, **kwargs):
models.signals.pre_save.disconnect(dispatch_uid="add_changed_fields_callback") models.signals.pre_save.disconnect(dispatch_uid="add_changed_fields_callback")
models.signals.post_save.disconnect(dispatch_uid="save_notification_callback") models.signals.post_save.disconnect(dispatch_uid="save_notification_callback")
models.signals.post_delete.disconnect(dispatch_uid="delete_notification_callback") models.signals.post_delete.disconnect(dispatch_uid="delete_notification_callback")
models.signals.m2m_changed.connect(
receiver=m2m_changed_notification_callback,
dispatch_uid="m2m_changed_notification_callback",
)
@receiver(models.signals.post_migrate) @receiver(models.signals.post_migrate)
...@@ -70,7 +80,7 @@ def add_changed_fields_callback(sender, instance, raw, *args, **kwargs): ...@@ -70,7 +80,7 @@ def add_changed_fields_callback(sender, instance, raw, *args, **kwargs):
Makes note of any dirty (changed) fields before they are saved, Makes note of any dirty (changed) fields before they are saved,
stuffing them in the instance for use in any post-save callbacks. stuffing them in the instance for use in any post-save callbacks.
""" """
if not isinstance(instance, (Person, PersonRole)): if not isinstance(instance, SUPPORTED_MODELS):
return return
changed = instance.is_dirty() changed = instance.is_dirty()
if not changed: if not changed:
...@@ -84,11 +94,8 @@ def add_changed_fields_callback(sender, instance, raw, *args, **kwargs): ...@@ -84,11 +94,8 @@ def add_changed_fields_callback(sender, instance, raw, *args, **kwargs):
def save_notification_callback(sender, instance, created, *args, **kwargs): def save_notification_callback(sender, instance, created, *args, **kwargs):
if not isinstance(instance, SUPPORTED_MODELS): if not isinstance(instance, SUPPORTED_MODELS):
return return
meta = {} meta = _create_metadata(instance)
operation = "add" if created else "update" operation = "add" if created else "update"
if isinstance(instance, PersonRole):
meta["person_id"] = instance.person.id
meta["role_id"] = instance.role.id
_store_notification( _store_notification(
identifier=instance.id, identifier=instance.id,
object_type=instance._meta.object_name, object_type=instance._meta.object_name,
...@@ -101,13 +108,78 @@ def save_notification_callback(sender, instance, created, *args, **kwargs): ...@@ -101,13 +108,78 @@ def save_notification_callback(sender, instance, created, *args, **kwargs):
def delete_notification_callback(sender, instance, *args, **kwargs): def delete_notification_callback(sender, instance, *args, **kwargs):
if not isinstance(instance, SUPPORTED_MODELS): if not isinstance(instance, SUPPORTED_MODELS):
return return
meta = {} meta = _create_metadata(instance)
if isinstance(instance, PersonRole):
meta["person_id"] = instance.person.id
meta["role_id"] = instance.role.id
_store_notification( _store_notification(
identifier=instance.id, identifier=instance.id,
object_type=instance._meta.object_name, object_type=instance._meta.object_name,
operation="delete", operation="delete",
**meta **meta
) )
@receiver(models.signals.m2m_changed, dispatch_uid="m2m_changed_notification_callback")
def m2m_changed_notification_callback(
sender, instance, action, *args, model=None, pk_set=None, **kwargs
):
if action not in ("post_add", "post_remove"):
return
if sender not in (PersonConsent, PersonRole, PersonIdentity):
return
operation = "add" if action == "post_add" else "delete"
instance_type = type(instance)
if sender is PersonConsent:
person_consents = []
if instance_type is Person and model is Consent:
person_consents = PersonConsent.objects.filter(
person_id=instance.id, consent_id__in=pk_set
)
elif instance_type is Consent and model is Person:
person_consents = PersonConsent.objects.filter(
consent_id=instance.id, person_id__in=pk_set
)
for pc in person_consents:
meta = _create_metadata(pc)
_store_notification(
identifier=pc.id,
object_type=PersonConsent._meta.object_name,
operation=operation,
**meta
)
elif sender is PersonRole:
person_roles = []
if instance_type is Person and model is Role:
person_roles = PersonRole.objects.filter(
person_id=instance.id, role_id__in=pk_set
)
elif instance_type is Role and model is Person:
person_roles = PersonRole.objects.filter(
role_id=instance.id, person_id__in=pk_set
)
for pr in person_roles:
meta = _create_metadata(pr)
_store_notification(
identifier=pr.id,
object_type=PersonRole._meta.object_name,
operation=operation,
**meta
)
def _create_metadata(instance) -> Dict:
meta = {}
if isinstance(instance, PersonRole):
meta["person_id"] = instance.person.id
meta["role_id"] = instance.role.id
if isinstance(instance, PersonIdentity):
meta["person_id"] = instance.person.id
meta["identity_id"] = instance.id
if isinstance(instance, PersonConsent):
meta["person_id"] = instance.person.id
meta["consent_id"] = instance.consent.id
return meta
...@@ -22,7 +22,7 @@ def consent_foo() -> Consent: ...@@ -22,7 +22,7 @@ def consent_foo() -> Consent:
@pytest.mark.django_db @pytest.mark.django_db
def test_get_consent(client, consent_foo): def test_get_consent(client, consent_foo):
resp = client.get(reverse("consent-detail", kwargs={"id": consent_foo.id})) resp = client.get(reverse("v1:consent-detail", kwargs={"id": consent_foo.id}))
assert resp.status_code == status.HTTP_200_OK assert resp.status_code == status.HTTP_200_OK
data = resp.json() data = resp.json()
assert data.get("id") == consent_foo.id assert data.get("id") == consent_foo.id
......
import pytest from typing import Dict
import pytest
from rest_framework import status from rest_framework import status
from rest_framework.reverse import reverse from rest_framework.reverse import reverse
from rest_framework.status import HTTP_200_OK from rest_framework.status import HTTP_200_OK
from greg.models import Person from django.core.exceptions import ValidationError
from greg.models import (
PersonIdentity,
Sponsor,
Role,
OrganizationalUnit,
Consent,
Person,
PersonRole,
)
@pytest.fixture @pytest.fixture
def person_foo() -> Person: def role_visiting_professor() -> Role:
return Person.objects.create( return Role.objects.create(
first_name="Foo", type="visiting_professor",
last_name="Foo", name_nb="Gjesteprofessor",
date_of_birth="2001-01-27", name_en="Visiting professor",
email="test@example.org", description_nb="Gjesteprofessor",
mobile_phone="123456788", description_en="Visiting professor",
default_duration_days=180,
) )
@pytest.fixture @pytest.fixture
def person_bar() -> Person: def unit_human_resources() -> OrganizationalUnit:
return Person.objects.create( return OrganizationalUnit.objects.create(
first_name="Bar", orgreg_id="org_unit_1", name_nb="Personal", name_en="Human Resources"
last_name="Bar", )
date_of_birth="2000-07-01",
email="test2@example.org",
mobile_phone="123456789", @pytest.fixture()
def sponsor_bar() -> Sponsor:
return Sponsor.objects.create(feide_id="bar")
@pytest.fixture
def role_data_guest(
role_test_guest: Role, sponsor_bar: Sponsor, unit_foo: OrganizationalUnit
) -> Dict:
return {
"role": "Test Guest",
"start_date": "2021-06-10",
"end_date": "2021-08-10",
"registered_by": sponsor_bar.id,
"unit": unit_foo.id,
}
@pytest.fixture
def consent_foo() -> Consent:
return Consent.objects.create(
type="test_consent",
consent_name_en="Test1",
consent_name_nb="Test2",
consent_description_en="Test description",
consent_description_nb="Test beskrivelse",
consent_link_en="https://example.org",
consent_link_nb="https://example.org",
valid_from="2018-01-20",
user_allowed_to_change=True,
) )
@pytest.mark.django_db @pytest.mark.django_db
def test_get_person(client, person_foo): def test_get_person(client, person_foo):
resp = client.get(reverse("person-detail", kwargs={"id": person_foo.id})) resp = client.get(reverse("v1:person-detail", kwargs={"id": person_foo.id}))
assert resp.status_code == HTTP_200_OK assert resp.status_code == HTTP_200_OK
data = resp.json() data = resp.json()
assert data.get("id") == person_foo.id assert data.get("id") == person_foo.id
...@@ -41,44 +82,48 @@ def test_get_person(client, person_foo): ...@@ -41,44 +82,48 @@ def test_get_person(client, person_foo):
@pytest.mark.django_db @pytest.mark.django_db
def test_persons(client, person_foo, person_bar): def test_persons(client, person_foo, person_bar):
resp = client.get(reverse("person-list")) resp = client.get(reverse("v1:person-list"))
assert resp.status_code == HTTP_200_OK assert resp.status_code == HTTP_200_OK
data = resp.json() data = resp.json()
assert len(data["results"]) == 2 assert len(data["results"]) == 2
@pytest.mark.django_db @pytest.mark.django_db
def test_persons_verified_filter_include(client, setup_db_test_data): def test_persons_verified_filter_include(
url = reverse("person-list") client, person_bar, person_foo, person_foo_verified
):
url = reverse("v1:person-list")
response = client.get(url, {"verified": "true"}) response = client.get(url, {"verified": "true"})
results = response.json()["results"] results = response.json()["results"]
assert len(results) == 1 assert len(results) == 1
# The following person will have a verified identity set up for him assert results[0]["first_name"] == "Foo"
# in the test data assert results[0]["last_name"] == "Foo"
assert results[0]["first_name"] == "Christopher"
assert results[0]["last_name"] == "Flores"
@pytest.mark.django_db @pytest.mark.django_db
def test_persons_verified_filter_exclude(client, setup_db_test_data): def test_persons_verified_filter_exclude(
url = reverse("person-list") client, person_bar, person_foo, person_foo_verified
):
url = reverse("v1:person-list")
response = client.get(url, {"verified": "false"}) response = client.get(url, {"verified": "false"})
results = response.json()["results"] results = response.json()["results"]
names = [(result["first_name"], result["last_name"]) for result in results] assert len(results) == 1
assert len(results) == 9 assert results[0]["first_name"] == "Bar"
assert ("Christopher", "Flores") not in names assert results[0]["last_name"] == "Bar"
@pytest.mark.django_db @pytest.mark.django_db
def test_add_role(client, person_foo): def test_add_role(
url = reverse("person_role-list", kwargs={"person_id": person_foo.id}) client, person_foo, role_visiting_professor, sponsor_guy, unit_human_resources
):
url = reverse("v1:person_role-list", kwargs={"person_id": person_foo.id})
roles_for_person = client.get(url).json()["results"] roles_for_person = client.get(url).json()["results"]
# Check that there are no roles for the person, and then add one # Check that there are no roles for the person, and then add one
assert len(roles_for_person) == 0 assert len(roles_for_person) == 0
role_data = { role_data = {
"role": "Visiting Professor", "role": "visiting_professor",
"start_date": "2021-06-10", "start_date": "2021-06-10",
"end_date": "2021-08-10", "end_date": "2021-08-10",
"registered_by": "1", "registered_by": "1",
...@@ -94,3 +139,284 @@ def test_add_role(client, person_foo): ...@@ -94,3 +139,284 @@ def test_add_role(client, person_foo):
# Check that the role shows up when listing roles for the person now # Check that the role shows up when listing roles for the person now
assert len(roles_for_person) == 1 assert len(roles_for_person) == 1
assert roles_for_person[0]["id"] == response_data["id"] assert roles_for_person[0]["id"] == response_data["id"]
@pytest.mark.django_db
def test_update_role(client, person_foo, role_data_guest):
url = reverse("v1:person_role-list", kwargs={"person_id": person_foo.id})
response = client.post(url, role_data_guest)
response_data = response.json()
assert response_data["start_date"] == "2021-06-10"
# Update the date and check that the change is registered
role_id = response.json()["id"]
updated_role = role_data_guest.copy()
updated_role["start_date"] = "2021-06-15"
url_detail = reverse(
"v1:person_role-detail", kwargs={"person_id": person_foo.id, "id": role_id}
)
client.patch(url_detail, updated_role)
updated_role_data = client.get(url)
updated_data = updated_role_data.json()["results"][0]
assert updated_data["id"] == role_id
assert updated_data["start_date"] == "2021-06-15"
@pytest.mark.django_db
def test_delete_role(client, person_foo, role_data_guest):
url = reverse("v1:person_role-list", kwargs={"person_id": person_foo.id})
role_id = client.post(url, role_data_guest).json()["id"]
roles_for_person = client.get(url).json()["results"]
assert len(roles_for_person) == 1
url_detail = reverse(
"v1:person_role-detail", kwargs={"person_id": person_foo.id, "id": role_id}
)
client.delete(url_detail)
assert len(client.get(url).json()["results"]) == 0
@pytest.mark.django_db
def test_identity_list(
client, person_foo, person_foo_verified, person_foo_not_verified
):
response = client.get(
reverse("v1:person-list"),
data={"first_name": person_foo.first_name, "last_name": person_foo.last_name},
)
person_id = response.json()["results"][0]["id"]
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_id})
)
data = response.json()["results"]
assert len(data) == 2
@pytest.mark.django_db
def test_identity_add(client, person_foo):
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
results = response.json()["results"]
assert len(results) == 0
data = {
"type": PersonIdentity.IdentityType.FEIDE_ID,
"source": "Test source",
"value": "12345",
}
client.post(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id}),
data=data,
)
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
results = response.json()["results"]
assert len(results) == 1
@pytest.mark.django_db
def test_identity_add_duplicate(client, person_foo, person_bar):
data = {
"type": PersonIdentity.IdentityType.FEIDE_ID,
"source": "Test source",
"value": "12345",
}
client.post(
reverse("v1:person_identity-list", kwargs={"person_id": person_bar.id}),
data=data,
)
with pytest.raises(ValidationError):
client.post(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id}),
data=data,
)
@pytest.mark.django_db
def test_identity_add_valid_duplicate(client, person_foo, person_bar):
data = {
"type": PersonIdentity.IdentityType.OTHER,
"source": "Test source",
"value": "12345",
}
client.post(
reverse("v1:person_identity-list", kwargs={"person_id": person_bar.id}),
data=data,
)
client.post(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id}),
data=data,
)
@pytest.mark.django_db
def test_identity_delete(client, person_foo):
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
results = response.json()["results"]
assert len(results) == 0
data = {
"type": PersonIdentity.IdentityType.FEIDE_ID,
"source": "Test source",
"value": "12345",
}
post_response = client.post(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id}),
data=data,
)
identity_id = post_response.json()["id"]
# Create two identities for the user
data = {
"type": PersonIdentity.IdentityType.PASSPORT,
"source": "Test",
"value": "1234413241235",
}
post_response2 = client.post(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id}),
data=data,
)
identity_id2 = post_response2.json()["id"]
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
results = response.json()["results"]
assert len(results) == 2
# Delete the first identity created
client.delete(
reverse(
"v1:person_identity-detail",
kwargs={"person_id": person_foo.id, "id": identity_id},
)
)
# Check that the other identity is still there
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
results = response.json()["results"]
assert len(results) == 1
assert results[0]["id"] == identity_id2
@pytest.mark.django_db
def test_identity_update(client, person_foo):
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
results = response.json()["results"]
assert len(results) == 0
data = {
"type": PersonIdentity.IdentityType.FEIDE_ID,
"source": "Test source",
"value": "12345",
}
client.post(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id}),
data=data,
)
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
results = response.json()["results"]
assert len(results) == 1
identity_id = results[0]["id"]
assert results[0]["type"] == data["type"]
assert results[0]["source"] == data["source"]
assert results[0]["value"] == data["value"]
data_updated = {
"type": PersonIdentity.IdentityType.PASSPORT,
"source": "Test",
"value": "10000",
}
patch_response = client.patch(
reverse(
"v1:person_identity-detail",
kwargs={"person_id": person_foo.id, "id": identity_id},
),
data=data_updated,
)
assert patch_response.status_code == status.HTTP_200_OK
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
results = response.json()["results"]
assert len(results) == 1
assert results[0]["type"] == data_updated["type"]
assert results[0]["source"] == data_updated["source"]
assert results[0]["value"] == data_updated["value"]
@pytest.mark.django_db
def test_remove_person(
client, person_foo, role_data_guest, person_foo_verified, person_foo_not_verified
):
url = reverse("v1:person_role-list", kwargs={"person_id": person_foo.id})
client.post(url, role_data_guest)
roles_for_person = client.get(url).json()["results"]
assert len(roles_for_person) == 1
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
assert len(response.json()["results"]) == 2
# Delete the person and check that the data has been removed
client.delete(reverse("v1:person-detail", kwargs={"id": person_foo.id}))
updated_role_data = client.get(url)
assert len(updated_role_data.json()["results"]) == 0
response = client.get(
reverse("v1:person_identity-list", kwargs={"person_id": person_foo.id})
)
assert len(response.json()["results"]) == 0
@pytest.mark.django_db
def test_add_duplicate_role_fails(
client, person_foo: Person, person_foo_role: PersonRole
):
url = reverse("v1:person_role-list", kwargs={"person_id": person_foo.id})
roles_for_person = client.get(url).json()["results"]
assert len(roles_for_person) == 1
role_data = {
"role": person_foo_role.role_id,
"start_date": person_foo_role.start_date,
"end_date": person_foo_role.end_date,
"registered_by": person_foo_role.registered_by,
"unit": person_foo_role.unit_id,
}
response = client.post(url, role_data)
# If the role cannot be create the return code is 400
assert response.status_code == status.HTTP_400_BAD_REQUEST
# Check that there is still only one role attached to the person
roles_for_person = client.get(url).json()["results"]
assert len(roles_for_person) == 1
import pytest
from rest_framework import status
from rest_framework.reverse import reverse
@pytest.mark.django_db
def test_add_sponsor(client):
data = {
"feide_id": "sponsor@example.org",
"first_name": "Test",
"last_name": "Sponsor",
}
post_response = client.post(reverse("v1:sponsor-list"), data=data)
assert post_response.status_code == status.HTTP_201_CREATED
response_data = post_response.json()
list_response = client.get(
reverse("v1:sponsor-detail", kwargs={"id": response_data["id"]})
)
list_response_data = list_response.json()
assert list_response_data["feide_id"] == data["feide_id"]
assert list_response_data["first_name"] == data["first_name"]
assert list_response_data["last_name"] == data["last_name"]
@pytest.mark.django_db
def test_sponsor_guest_list(client, sponsor_guy, person_foo_role):
url = reverse("v1:sponsor_guests-list", kwargs={"sponsor_id": sponsor_guy.id})
guests_for_sponsor = client.get(url).json()["results"]
assert len(guests_for_sponsor) == 1
assert guests_for_sponsor[0]["id"] == person_foo_role.person_id
@pytest.mark.django_db
def test_sponsor_empty_guest_list(client, sponsor_guy):
url = reverse("v1:sponsor_guests-list", kwargs={"sponsor_id": sponsor_guy.id})
guests_for_sponsor = client.get(url).json()["results"]
assert len(guests_for_sponsor) == 0
...@@ -4,27 +4,22 @@ from rest_framework.authtoken.models import Token ...@@ -4,27 +4,22 @@ from rest_framework.authtoken.models import Token
from rest_framework.test import APIClient from rest_framework.test import APIClient
from django.contrib.auth import get_user_model from django.contrib.auth import get_user_model
# faker spams the logs with localisation warnings
# see https://github.com/joke2k/faker/issues/753
import pytest import pytest
from greg.tests.populate_database import DatabasePopulation from greg.models import (
Person,
Sponsor,
PersonIdentity,
Role,
OrganizationalUnit,
PersonRole,
)
# faker spams the logs with localisation warnings
# see https://github.com/joke2k/faker/issues/753
logging.getLogger("faker").setLevel(logging.ERROR) logging.getLogger("faker").setLevel(logging.ERROR)
# The database is populated once when scope is session.
# If the scope is changed to function some additional
# logic is needed to make sure the old data is cleaned
# before the seeding is run again
@pytest.fixture(scope="session")
def setup_db_test_data(django_db_setup, django_db_blocker):
with django_db_blocker.unblock():
database_seeder = DatabasePopulation()
database_seeder.populate_database()
@pytest.fixture @pytest.fixture
def client() -> APIClient: def client() -> APIClient:
user, _ = get_user_model().objects.get_or_create(username="test") user, _ = get_user_model().objects.get_or_create(username="test")
...@@ -32,3 +27,82 @@ def client() -> APIClient: ...@@ -32,3 +27,82 @@ def client() -> APIClient:
client = APIClient() client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}") client.credentials(HTTP_AUTHORIZATION=f"Token {token.key}")
return client return client
@pytest.fixture
def person_foo() -> Person:
return Person.objects.create(
first_name="Foo",
last_name="Foo",
date_of_birth="2001-01-27",
email="test@example.org",
mobile_phone="123456788",
)
@pytest.fixture
def person_bar() -> Person:
return Person.objects.create(
first_name="Bar",
last_name="Bar",
date_of_birth="2000-07-01",
email="test2@example.org",
mobile_phone="123456789",
)
@pytest.fixture
def sponsor_guy() -> Sponsor:
return Sponsor.objects.create(
feide_id="guy@example.org", first_name="Sponsor", last_name="Guy"
)
@pytest.fixture
def person_foo_verified(person_foo, sponsor_guy) -> PersonIdentity:
return PersonIdentity.objects.create(
person=person_foo,
type=PersonIdentity.IdentityType.PASSPORT,
source="Test",
value="12345",
verified=PersonIdentity.Verified.MANUAL,
verified_by=sponsor_guy,
verified_when="2021-06-15",
)
@pytest.fixture
def person_foo_not_verified(person_foo) -> PersonIdentity:
return PersonIdentity.objects.create(
person=person_foo,
type=PersonIdentity.IdentityType.DRIVERS_LICENSE,
source="Test",
value="12345",
)
@pytest.fixture()
def role_test_guest() -> Role:
return Role.objects.create(type="Test Guest")
@pytest.fixture
def unit_foo() -> OrganizationalUnit:
return OrganizationalUnit.objects.create(orgreg_id="12345", name_en="foo_unit")
@pytest.fixture
def person_foo_role(
person_foo: Person,
role_test_guest: Role,
sponsor_guy: Sponsor,
unit_foo: OrganizationalUnit,
) -> PersonRole:
return PersonRole.objects.create(
person=person_foo,
role=role_test_guest,
start_date="2021-08-02",
end_date="2021-08-06",
registered_by=sponsor_guy,
unit=unit_foo,
)
import datetime
import pytest import pytest
from greg.models import ( from greg.models import (
Person, Person,
Consent, Consent,
PersonConsent,
) )
...@@ -17,9 +20,9 @@ def person() -> Person: ...@@ -17,9 +20,9 @@ def person() -> Person:
) )
@pytest.mark.django_db @pytest.fixture()
def test_add_consent_to_person(person): def consent() -> Consent:
consent = Consent.objects.create( return Consent.objects.create(
type="it_guidelines", type="it_guidelines",
consent_name_en="IT Guidelines", consent_name_en="IT Guidelines",
consent_name_nb="IT Regelverk", consent_name_nb="IT Regelverk",
...@@ -28,4 +31,25 @@ def test_add_consent_to_person(person): ...@@ -28,4 +31,25 @@ def test_add_consent_to_person(person):
consent_link_en="https://example.org/it_guidelines", consent_link_en="https://example.org/it_guidelines",
user_allowed_to_change=False, user_allowed_to_change=False,
) )
person.consents.add(consent, through_defaults={"consent_given_at": "2021-06-20"})
@pytest.mark.django_db
def test_add_consent_to_person(person: Person, consent: Consent):
consent_given_date = "2021-06-20"
person.consents.add(consent, through_defaults={"consent_given_at": consent_given_date}) # type: ignore
person_consent_links = PersonConsent.objects.filter(person_id=person.id)
assert len(person_consent_links) == 1
assert person_consent_links[0].person_id == person.id
assert person_consent_links[0].consent_id == consent.id
assert person_consent_links[0].consent_given_at == datetime.date(2021, 6, 20)
@pytest.mark.django_db
def test_add_not_acknowledged_consent_to_person(person: Person, consent: Consent):
person.consents.add(consent)
person_consent_links = PersonConsent.objects.filter(person_id=person.id)
assert len(person_consent_links) == 1
assert person_consent_links[0].person_id == person.id
assert person_consent_links[0].consent_id == consent.id
assert person_consent_links[0].consent_given_at is None
from datetime import (
datetime,
timedelta,
)
from functools import partial from functools import partial
import pytest import pytest
from django.db.models import ProtectedError
from greg.models import ( from greg.models import (
OrganizationalUnit, OrganizationalUnit,
Person, Person,
PersonIdentity,
PersonRole, PersonRole,
Role, Role,
Sponsor, Sponsor,
) )
person_role_with = partial( person_role_with = partial(
PersonRole.objects.create, PersonRole.objects.create,
start_date="2020-03-05", start_date="2020-03-05",
...@@ -33,7 +35,7 @@ def role_bar() -> Role: ...@@ -33,7 +35,7 @@ def role_bar() -> Role:
@pytest.fixture @pytest.fixture
def person(role_foo, role_bar) -> Person: def person(role_foo: Role, role_bar: Role) -> Person:
person = Person.objects.create( person = Person.objects.create(
first_name="Test", first_name="Test",
last_name="Tester", last_name="Tester",
...@@ -59,6 +61,51 @@ def person(role_foo, role_bar) -> Person: ...@@ -59,6 +61,51 @@ def person(role_foo, role_bar) -> Person:
return person return person
@pytest.fixture
def a_year_ago() -> datetime:
return datetime.now() - timedelta(days=365)
@pytest.fixture
def a_year_into_future() -> datetime:
return datetime.now() + timedelta(days=365)
@pytest.fixture
def feide_id(a_year_ago: datetime) -> PersonIdentity:
return PersonIdentity(
type=PersonIdentity.IdentityType.FEIDE_ID,
verified_when=a_year_ago,
)
@pytest.fixture
def passport(a_year_ago: datetime) -> PersonIdentity:
return PersonIdentity(
type=PersonIdentity.IdentityType.PASSPORT,
verified_when=a_year_ago,
)
@pytest.fixture
def unverified_passport() -> PersonIdentity:
return PersonIdentity(type=PersonIdentity.IdentityType.PASSPORT)
@pytest.fixture
def future_identity(a_year_into_future: datetime) -> PersonIdentity:
return PersonIdentity(
type=PersonIdentity.IdentityType.NATIONAL_ID_CARD,
verified_when=a_year_into_future,
)
@pytest.fixture
def feide_verified(person: Person, feide_id: PersonIdentity) -> Person:
person.identities.add(feide_id, bulk=False)
return person
@pytest.mark.django_db @pytest.mark.django_db
def test_add_multiple_roles_to_person(person, role_foo, role_bar): def test_add_multiple_roles_to_person(person, role_foo, role_bar):
person_roles = person.roles.all() person_roles = person.roles.all()
...@@ -68,9 +115,63 @@ def test_add_multiple_roles_to_person(person, role_foo, role_bar): ...@@ -68,9 +115,63 @@ def test_add_multiple_roles_to_person(person, role_foo, role_bar):
@pytest.mark.django_db @pytest.mark.django_db
def test_delete_person_with_roles(person): def test_identities(person: Person, feide_id: PersonIdentity, passport: PersonIdentity):
# it is not clear what cleanup needs to be done when removing a person, person.identities.add(feide_id, bulk=False)
# so for now it is prohibited to delete a person with role relationships assert list(person.identities.all()) == [feide_id]
# attached in other tables person.identities.add(passport, bulk=False)
with pytest.raises(ProtectedError): assert list(person.identities.all()) == [feide_id, passport]
person.delete()
@pytest.mark.django_db
def test_is_registered_incomplete(person):
assert person.registration_completed_date is None
assert not person.is_registered
@pytest.mark.django_db
def test_is_registered_completed_in_past(person, a_year_ago):
person.registration_completed_date = a_year_ago
assert person.is_registered
@pytest.mark.django_db
def test_is_registered_completed_in_future(person, a_year_into_future):
person.registration_completed_date = a_year_into_future
assert not person.is_registered
@pytest.mark.django_db
@pytest.mark.parametrize("identity_type", PersonIdentity.IdentityType)
def test_is_verified(identity_type, a_year_ago, person):
identity = PersonIdentity(type=identity_type, verified_when=a_year_ago)
person.identities.add(identity, bulk=False)
assert person.is_verified
@pytest.mark.django_db
def test_is_verified_multiple_identities(person, feide_id, passport):
person.identities.add(feide_id, passport, bulk=False)
assert person.is_verified
@pytest.mark.django_db
def is_verified_when_identity_is_unverified(person, unverified_passport):
person.identities.add(unverified_passport, bulk=False)
assert not person.is_verified
@pytest.mark.django_db
def is_verified_mixed_verified_and_unverified_identities(
person,
feide_id,
unverified_passport,
future_identity,
):
person.identities.add(feide_id, unverified_passport, future_identity, bulk=False)
assert person.is_verified
@pytest.mark.django_db
def is_verified_in_future(person, future_identity):
person.identities.add(future_identity, bulk=False)
assert not person.is_verified
from datetime import datetime, timedelta
import pytest
from django.db.models import Q
from django.utils import timezone
from greg.models import (
ScheduleTask,
Role,
Person,
OrganizationalUnit,
PersonRole,
Notification,
Sponsor,
)
from greg.schedule import ExpiringRolesNotification
@pytest.fixture
def role_task():
ScheduleTask.objects.create(
name="role_expiration", last_completed=timezone.now() - timedelta(days=1)
)
@pytest.fixture
def role_bar() -> Role:
return Role.objects.create(type="role_bar", name_en="Role Bar")
@pytest.fixture
def person(role_bar: Role) -> Person:
return Person.objects.create(
first_name="Test",
last_name="Tester",
date_of_birth="2000-01-27",
email="test@example.org",
mobile_phone="123456789",
)
@pytest.fixture
def organizational_unit() -> OrganizationalUnit:
return OrganizationalUnit.objects.create(orgreg_id="12345", name_en="Test unit")
@pytest.fixture
def sponsor() -> Sponsor:
return Sponsor.objects.create(feide_id="sponsor@example.org")
@pytest.fixture
def person_role(
person: Person,
role_bar: Role,
organizational_unit: OrganizationalUnit,
sponsor: Sponsor,
) -> PersonRole:
return PersonRole.objects.create(
person=person,
role=role_bar,
unit=organizational_unit,
start_date="2020-03-05",
end_date=datetime.today() + timedelta(days=30),
contact_person_unit="Contact Person",
available_in_search=True,
registered_by=sponsor,
)
@pytest.mark.django_db
def test_role_picked_up(role_task: ScheduleTask, person_role: PersonRole):
role_notification = ExpiringRolesNotification()
assert len(Notification.objects.filter(~Q(operation="add"))) == 0
role_notification.run()
notification = Notification.objects.get(operation="expire_reminder")
assert notification.identifier == person_role.id
role_notification.run()
notifications = Notification.objects.filter(operation="expire_reminder")
assert len(notifications) == 1
@pytest.mark.django_db
def test_no_notification_for_role_not_about_to_expire(
role_task: ScheduleTask, person_role: PersonRole
):
person_role.end_date = datetime.today() + timedelta(days=31)
person_role.save()
role_notification = ExpiringRolesNotification()
assert len(Notification.objects.filter(operation="expire_reminder")) == 0
# Role should not be picked up since it expires in 31 days
role_notification.run()
notifications = Notification.objects.filter(operation="expire_reminder")
assert len(notifications) == 0
import pytest
from greg.models import (
Person,
Notification,
Consent,
Role,
OrganizationalUnit,
Sponsor,
PersonConsent,
PersonIdentity,
)
@pytest.fixture
def person() -> Person:
return Person.objects.create(
first_name="Test",
last_name="Tester",
date_of_birth="2000-01-27",
email="test@example.org",
mobile_phone="123456789",
)
@pytest.fixture
def role_foo() -> Role:
return Role.objects.create(type="role_foo", name_en="Role Foo")
@pytest.fixture
def org_unit_bar() -> OrganizationalUnit:
return OrganizationalUnit.objects.create(orgreg_id="bar_unit")
@pytest.fixture
def sponsor() -> Sponsor:
return Sponsor.objects.create(feide_id="sponsor_id")
@pytest.fixture
def consent() -> Consent:
return Consent.objects.create(
type="it_guidelines",
consent_name_en="IT Guidelines",
consent_name_nb="IT Regelverk",
consent_description_en="IT Guidelines description",
consent_description_nb="IT Regelverk beskrivelse",
consent_link_en="https://example.org/it_guidelines",
user_allowed_to_change=False,
)
@pytest.fixture
def person_identity(person: Person) -> PersonIdentity:
return PersonIdentity.objects.create(
person=person,
type=PersonIdentity.IdentityType.PASSPORT,
source="Test",
value="12345678901",
)
@pytest.mark.django_db
def test_person_add_notification(person: Person):
notifications = Notification.objects.filter(object_type="Person")
assert len(notifications) == 1
assert notifications[0].operation == "add"
assert notifications[0].identifier == person.id
@pytest.mark.django_db
def test_person_update_notification(person: Person):
person.first_name = "New first name"
person.save()
notifications = Notification.objects.filter(object_type="Person")
assert len(notifications) == 2
assert notifications[1].operation == "update"
assert notifications[1].identifier == person.id
@pytest.mark.django_db
def test_person_delete_notification(person: Person):
person_id = person.id
person.delete()
notifications = Notification.objects.filter(object_type="Person")
assert len(notifications) == 2
assert notifications[1].operation == "delete"
assert notifications[1].identifier == person_id
@pytest.mark.django_db
def test_role_add_notification(
person: Person, role_foo: Role, org_unit_bar: OrganizationalUnit, sponsor: Sponsor
):
person.roles.add( # type: ignore
role_foo,
through_defaults={
"start_date": "2021-05-06",
"end_date": "2021-10-20",
"unit": org_unit_bar,
"registered_by": sponsor,
},
)
notifications = Notification.objects.filter(object_type="PersonRole")
assert len(notifications) == 1
assert notifications[0].operation == "add"
meta_data = notifications[0].meta
assert meta_data["person_id"] == person.id
assert meta_data["role_id"] == role_foo.id
@pytest.mark.django_db
def test_role_update_notification(
person: Person, role_foo: Role, org_unit_bar: OrganizationalUnit, sponsor: Sponsor
):
person.roles.add( # type: ignore
role_foo,
through_defaults={
"start_date": "2021-05-06",
"end_date": "2021-10-20",
"unit": org_unit_bar,
"registered_by": sponsor,
},
)
assert len(person.person_roles.all()) == 1
person_role = person.person_roles.all()[0]
person_role.end_date = "2021-10-21"
person_role.save()
notifications = Notification.objects.filter(object_type="PersonRole")
assert len(notifications) == 2
assert notifications[1].operation == "update"
meta_data = notifications[1].meta
assert meta_data["person_id"] == person.id
assert meta_data["role_id"] == role_foo.id
@pytest.mark.django_db
def test_role_delete_notification(
person: Person, role_foo: Role, org_unit_bar: OrganizationalUnit, sponsor: Sponsor
):
person.roles.add( # type: ignore
role_foo,
through_defaults={
"start_date": "2021-05-06",
"end_date": "2021-10-20",
"unit": org_unit_bar,
"registered_by": sponsor,
},
)
assert len(person.person_roles.all()) == 1
person_role = person.person_roles.all()[0]
person_role.delete()
notifications = Notification.objects.filter(object_type="PersonRole")
assert len(notifications) == 2
assert notifications[1].operation == "delete"
meta_data = notifications[1].meta
assert meta_data["person_id"] == person.id
assert meta_data["role_id"] == role_foo.id
@pytest.mark.django_db
def test_consent_add_notification(person: Person, consent: Consent):
person.consents.add(consent, through_defaults={"consent_given_at": "2021-06-20"}) # type: ignore
notifications = Notification.objects.filter(object_type="PersonConsent")
assert len(notifications) == 1
assert notifications[0].identifier == person.id
meta_data = notifications[0].meta
assert meta_data["person_id"] == person.id
assert meta_data["consent_id"] == consent.id
@pytest.mark.django_db
def test_consent_update_notification(person: Person, consent: Consent):
person.consents.add(consent, through_defaults={"consent_given_at": "2021-06-20"}) # type: ignore
person_consents = PersonConsent.objects.filter(person=person, consent=consent)
person_consents[0].consent_given_at = "2021-06-21"
person_consents[0].save()
notifications = Notification.objects.filter(object_type="PersonConsent")
assert len(notifications) == 2
assert notifications[0].identifier == person.id
meta_data = notifications[0].meta
assert meta_data["person_id"] == person.id
assert meta_data["consent_id"] == consent.id
@pytest.mark.django_db
def test_consent_delete_notification(person: Person, consent: Consent):
person.consents.add(consent, through_defaults={"consent_given_at": "2021-06-20"}) # type: ignore
person_consents = PersonConsent.objects.filter(person=person, consent=consent)
person_consents[0].delete()
notifications = Notification.objects.filter(object_type="PersonConsent")
assert len(notifications) == 2
assert notifications[1].identifier == person.id
assert notifications[1].operation == "delete"
meta_data = notifications[0].meta
assert meta_data["person_id"] == person.id
assert meta_data["consent_id"] == consent.id
@pytest.mark.django_db
def test_person_identity_add_notification(
person: Person, person_identity: PersonIdentity, sponsor: Sponsor
):
notifications = Notification.objects.filter(object_type="PersonIdentity")
assert len(notifications) == 1
assert notifications[0].identifier == person.id
assert notifications[0].operation == "add"
meta_data = notifications[0].meta
assert meta_data["person_id"] == person.id
assert meta_data["identity_id"] == person_identity.id
@pytest.mark.django_db
def test_person_identity_update_notification(
person: Person, person_identity: PersonIdentity, sponsor: Sponsor
):
person_identity.verified = PersonIdentity.Verified.MANUAL
person_identity.verified_by = sponsor
person_identity.verified_when = "2021-08-02"
person_identity.save()
notifications = Notification.objects.filter(object_type="PersonIdentity")
# One notification for adding person identity and one for updating it
assert len(notifications) == 2
assert notifications[1].operation == "update"
meta_data = notifications[1].meta
assert meta_data["person_id"] == person.id
assert meta_data["identity_id"] == person_identity.id
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment