CGM_Public/pretix_original
2
0
Fork 1
mirror of https://github.com/pretix/pretix.git synced 2026-05-03 14:54:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

API: Add endpoints for scheduled exports (#3659)

Browse Source 
* API: Add endpoints for scheduled exports

* ADd note to docs
This commit is contained in:
Raphael Michel
2023-10-27 17:15:53 +02:00
committed by GitHub
parent 26cbc24a10
commit 3b64e6046c
10 changed files with 1389 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
src/pretix/api/serializers/exporters.py
View File

@@ -20,11 +20,14 @@
# <https://www.gnu.org/licenses/>.
#
from django import forms
from django.conf import settings
from django.http import QueryDict
from pytz import common_timezones
from rest_framework import serializers
from rest_framework.exceptions import ValidationError
from pretix.base.exporter import OrganizerLevelExportMixin
from pretix.base.models import ScheduledEventExport, ScheduledOrganizerExport
from pretix.base.timeframes import DateFrameField, SerializerDateFrameField
@@ -197,3 +200,92 @@ class JobRunSerializer(serializers.Serializer):
raise ValidationError(self.errors)
return not bool(self._errors)
class ScheduledExportSerializer(serializers.ModelSerializer):
schedule_next_run = serializers.DateTimeField(read_only=True)
export_identifier = serializers.ChoiceField(choices=[])
locale = serializers.ChoiceField(choices=settings.LANGUAGES, default='en')
owner = serializers.SlugRelatedField(slug_field='email', read_only=True)
error_counter = serializers.IntegerField(read_only=True)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.fields['export_identifier'].choices = [(e, e) for e in self.context['exporters']]
def validate(self, attrs):
if attrs.get("export_form_data"):
identifier = attrs.get('export_identifier', self.instance.export_identifier if self.instance else None)
exporter = self.context['exporters'].get(identifier)
if exporter:
try:
JobRunSerializer(exporter=exporter).to_internal_value(attrs["export_form_data"])
except ValidationError as e:
raise ValidationError({"export_form_data": e.detail})
else:
raise ValidationError({"export_identifier": ["Unknown exporter."]})
return attrs
def validate_mail_additional_recipients(self, value):
d = value.replace(' ', '')
if len(d.split(',')) > 25:
raise ValidationError('Please enter less than 25 recipients.')
return d
def validate_mail_additional_recipients_cc(self, value):
d = value.replace(' ', '')
if len(d.split(',')) > 25:
raise ValidationError('Please enter less than 25 recipients.')
return d
def validate_mail_additional_recipients_bcc(self, value):
d = value.replace(' ', '')
if len(d.split(',')) > 25:
raise ValidationError('Please enter less than 25 recipients.')
return d
class ScheduledEventExportSerializer(ScheduledExportSerializer):
class Meta:
model = ScheduledEventExport
fields = [
'id',
'owner',
'export_identifier',
'export_form_data',
'locale',
'mail_additional_recipients',
'mail_additional_recipients_cc',
'mail_additional_recipients_bcc',
'mail_subject',
'mail_template',
'schedule_rrule',
'schedule_rrule_time',
'schedule_next_run',
'error_counter',
]
class ScheduledOrganizerExportSerializer(ScheduledExportSerializer):
timezone = serializers.ChoiceField(default=settings.TIME_ZONE, choices=[(a, a) for a in common_timezones])
class Meta:
model = ScheduledOrganizerExport
fields = [
'id',
'owner',
'export_identifier',
'export_form_data',
'locale',
'mail_additional_recipients',
'mail_additional_recipients_cc',
'mail_additional_recipients_bcc',
'mail_subject',
'mail_template',
'schedule_rrule',
'schedule_rrule_time',
'schedule_next_run',
'timezone',
'error_counter',
]

2
src/pretix/api/urls.py
View File

@@ -63,6 +63,7 @@ orga_router.register(r'teams', organizer.TeamViewSet)
orga_router.register(r'devices', organizer.DeviceViewSet)
orga_router.register(r'orders', order.OrganizerOrderViewSet)
orga_router.register(r'invoices', order.InvoiceViewSet)
orga_router.register(r'scheduled_exports', exporters.ScheduledOrganizerExportViewSet)
orga_router.register(r'exporters', exporters.OrganizerExportersViewSet, basename='exporters')
team_router = routers.DefaultRouter()
@@ -88,6 +89,7 @@ event_router.register(r'taxrules', event.TaxRuleViewSet)
event_router.register(r'waitinglistentries', waitinglist.WaitingListViewSet)
event_router.register(r'checkinlists', checkin.CheckinListViewSet)
event_router.register(r'cartpositions', cart.CartPositionViewSet)
event_router.register(r'scheduled_exports', exporters.ScheduledEventExportViewSet)
event_router.register(r'exporters', exporters.EventExportersViewSet, basename='exporters')
event_router.register(r'shredders', shredders.EventShreddersViewSet, basename='shredders')
event_router.register(r'item_meta_properties', event.ItemMetaPropertiesViewSet)

159
src/pretix/api/views/exporters.py
View File

@@ -29,14 +29,20 @@ from django.utils.functional import cached_property
from django.utils.timezone import now
from rest_framework import status, viewsets
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied
from rest_framework.response import Response
from rest_framework.reverse import reverse
from pretix.api.pagination import TotalOrderingFilter
from pretix.api.serializers.exporters import (
ExporterSerializer, JobRunSerializer,
ExporterSerializer, JobRunSerializer, ScheduledEventExportSerializer,
ScheduledOrganizerExportSerializer,
)
from pretix.base.exporter import OrganizerLevelExportMixin
from pretix.base.models import CachedFile, Device, Event, TeamAPIToken
from pretix.base.models import (
CachedFile, Device, Event, ScheduledEventExport, ScheduledOrganizerExport,
TeamAPIToken,
)
from pretix.base.services.export import export, multiexport
from pretix.base.signals import (
register_data_exporters, register_multievent_data_exporters,
@@ -199,3 +205,152 @@ class OrganizerExportersViewSet(ExportersMixin, viewsets.ViewSet):
'provider': instance.identifier,
'form_data': data
})
class ScheduledExportersViewSet(viewsets.ModelViewSet):
filter_backends = (TotalOrderingFilter,)
ordering = ('id',)
ordering_fields = ('id', 'export_identifier', 'schedule_next_run')
class ScheduledEventExportViewSet(ScheduledExportersViewSet):
serializer_class = ScheduledEventExportSerializer
queryset = ScheduledEventExport.objects.none()
permission = 'can_view_orders'
def get_queryset(self):
perm_holder = self.request.auth if isinstance(self.request.auth, (TeamAPIToken, Device)) else self.request.user
if not perm_holder.has_event_permission(self.request.organizer, self.request.event, 'can_change_event_settings',
request=self.request):
if self.request.user.is_authenticated:
qs = self.request.event.scheduled_exports.filter(owner=self.request.user)
else:
raise PermissionDenied('Scheduled exports require either permission to change event settings or '
'user-specific API access.')
else:
qs = self.request.event.scheduled_exports
return qs.select_related("owner")
def perform_create(self, serializer):
if not self.request.user.is_authenticated:
raise PermissionDenied('Creation of exports requires user-specific API access.')
serializer.save(event=self.request.event, owner=self.request.user)
serializer.instance.compute_next_run()
serializer.instance.save(update_fields=["schedule_next_run"])
self.request.event.log_action(
'pretix.event.export.schedule.added',
user=self.request.user,
auth=self.request.auth,
data=self.request.data
)
def get_serializer_context(self):
ctx = super().get_serializer_context()
ctx['event'] = self.request.event
ctx['exporters'] = self.exporters
return ctx
@cached_property
def exporters(self):
responses = register_data_exporters.send(self.request.event)
exporters = [response(self.request.event, self.request.organizer) for r, response in responses if response]
return {e.identifier: e for e in exporters}
def perform_update(self, serializer):
serializer.save(event=self.request.event)
serializer.instance.compute_next_run()
serializer.instance.error_counter = 0
serializer.instance.error_last_message = None
serializer.instance.save(update_fields=["schedule_next_run", "error_counter", "error_last_message"])
self.request.event.log_action(
'pretix.event.export.schedule.changed',
user=self.request.user,
auth=self.request.auth,
data=self.request.data
)
def perform_destroy(self, instance):
self.request.event.log_action(
'pretix.event.export.schedule.deleted',
user=self.request.user,
auth=self.request.auth,
)
super().perform_destroy(instance)
class ScheduledOrganizerExportViewSet(ScheduledExportersViewSet):
serializer_class = ScheduledOrganizerExportSerializer
queryset = ScheduledOrganizerExport.objects.none()
permission = None
def get_queryset(self):
perm_holder = self.request.auth if isinstance(self.request.auth, (TeamAPIToken, Device)) else self.request.user
if not perm_holder.has_organizer_permission(self.request.organizer, 'can_change_organizer_settings',
request=self.request):
if self.request.user.is_authenticated:
qs = self.request.organizer.scheduled_exports.filter(owner=self.request.user)
else:
raise PermissionDenied('Scheduled exports require either permission to change organizer settings or '
'user-specific API access.')
else:
qs = self.request.organizer.scheduled_exports
return qs.select_related("owner")
def perform_create(self, serializer):
if not self.request.user.is_authenticated:
raise PermissionDenied('Creation of exports requires user-specific API access.')
serializer.save(organizer=self.request.organizer, owner=self.request.user)
serializer.instance.compute_next_run()
serializer.instance.save(update_fields=["schedule_next_run"])
self.request.organizer.log_action(
'pretix.organizer.export.schedule.added',
user=self.request.user,
auth=self.request.auth,
data=self.request.data
)
def get_serializer_context(self):
ctx = super().get_serializer_context()
ctx['organizer'] = self.request.organizer
ctx['exporters'] = self.exporters
return ctx
@cached_property
def events(self):
if isinstance(self.request.auth, (TeamAPIToken, Device)):
return self.request.auth.get_events_with_permission('can_view_orders')
elif self.request.user.is_authenticated:
return self.request.user.get_events_with_permission('can_view_orders', self.request).filter(
organizer=self.request.organizer
)
@cached_property
def exporters(self):
responses = register_multievent_data_exporters.send(self.request.organizer)
exporters = [
response(Event.objects.none() if issubclass(response, OrganizerLevelExportMixin) else self.events,
self.request.organizer)
for r, response in responses if response
]
return {e.identifier: e for e in exporters}
def perform_update(self, serializer):
serializer.save(organizer=self.request.organizer)
serializer.instance.compute_next_run()
serializer.instance.error_counter = 0
serializer.instance.error_last_message = None
serializer.instance.save(update_fields=["schedule_next_run", "error_counter", "error_last_message"])
self.request.organizer.log_action(
'pretix.organizer.export.schedule.changed',
user=self.request.user,
auth=self.request.auth,
data=self.request.data
)
def perform_destroy(self, instance):
self.request.organizer.log_action(
'pretix.organizer.export.schedule.deleted',
user=self.request.user,
auth=self.request.auth,
)
super().perform_destroy(instance)

2
src/pretix/base/models/exports.py
View File

@@ -79,7 +79,7 @@ class AbstractScheduledExport(LoggedModel):
)
schedule_rrule = models.TextField(
null=True, blank=True, validators=[RRuleValidator()]
null=True, blank=True, validators=[RRuleValidator(enforce_simple=True)]
)
schedule_rrule_time = models.TimeField(
verbose_name=_("Requested start time"),

47
src/pretix/base/validators.py
View File

@@ -19,7 +19,9 @@
# You should have received a copy of the GNU Affero General Public License along with this program. If not, see
# <https://www.gnu.org/licenses/>.
#
from dateutil.rrule import rrulestr
import calendar
from dateutil.rrule import DAILY, MONTHLY, WEEKLY, YEARLY, rrule, rrulestr
from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.validators import validate_email
@@ -40,7 +42,6 @@ from django.utils.translation import gettext_lazy as _
class BanlistValidator:
banlist = []
def __call__(self, value):
@@ -55,7 +56,6 @@ class BanlistValidator:
@deconstructible
class EventSlugBanlistValidator(BanlistValidator):
banlist = [
'download',
'healthcheck',
@@ -77,7 +77,6 @@ class EventSlugBanlistValidator(BanlistValidator):
@deconstructible
class OrganizerSlugBanlistValidator(BanlistValidator):
banlist = [
'download',
'healthcheck',
@@ -98,7 +97,6 @@ class OrganizerSlugBanlistValidator(BanlistValidator):
@deconstructible
class EmailBanlistValidator(BanlistValidator):
banlist = [
settings.PRETIX_EMAIL_NONE_VALUE,
]
@@ -112,8 +110,45 @@ def multimail_validate(val):
class RRuleValidator:
def __init__(self, enforce_simple=False):
self.enforce_simple = enforce_simple
def __call__(self, value):
try:
rrulestr(value)
parsed = rrulestr(value)
except Exception:
raise ValidationError("Not a valid rrule.")
if self.enforce_simple:
# Validate that only things are used that we can represent in our UI for later editing
if not isinstance(parsed, rrule):
raise ValidationError("Only a single RRULE is allowed, no combination of rules.")
if parsed._freq not in (YEARLY, MONTHLY, WEEKLY, DAILY):
raise ValidationError("Unsupported FREQ value")
if parsed._wkst != calendar.firstweekday():
raise ValidationError("Unsupported WKST value")
if parsed._bysetpos:
if len(parsed._bysetpos) > 1:
raise ValidationError("Only one BYSETPOS value allowed")
if parsed._freq == YEARLY and parsed._bysetpos not in (1, 2, 3, -1):
raise ValidationError("BYSETPOS value not allowed, should be 1, 2, 3 or -1")
elif parsed._freq == MONTHLY and parsed._bysetpos not in (1, 2, 3, -1):
raise ValidationError("BYSETPOS value not allowed, should be 1, 2, 3 or -1")
elif parsed._freq not in (YEARLY, MONTHLY):
raise ValidationError("BYSETPOS not allowed for this FREQ")
if parsed._bymonthday:
raise ValidationError("BYMONTHDAY not supported")
if parsed._byyearday:
raise ValidationError("BYYEARDAY not supported")
if parsed._byeaster:
raise ValidationError("BYEASTER not supported")
if parsed._byweekno:
raise ValidationError("BYWEEKNO not supported")
if len(parsed._byhour) > 1 or set(parsed._byhour) != {parsed._dtstart.hour}:
raise ValidationError("BYHOUR not supported")
if len(parsed._byminute) > 1 or set(parsed._byminute) != {parsed._dtstart.minute}:
raise ValidationError("BYMINUTE not supported")
if len(parsed._bysecond) > 1 or set(parsed._bysecond) != {parsed._dtstart.second}:
raise ValidationError("BYSECOND not supported")

534
src/tests/api/test_exporters.py
View File

@@ -34,10 +34,13 @@
import copy
import uuid
import zoneinfo
from datetime import time
import pytest
from django.utils.timezone import now
from pretix.base.models import CachedFile
from pretix.base.models import CachedFile, User
SAMPLE_EXPORTER_CONFIG = {
"identifier": "orderlist",
@@ -277,3 +280,532 @@ def test_org_level_export(token_client, organizer, team, event):
'_format': 'xlsx',
}, format='json')
assert resp.status_code == 404
@pytest.fixture
def event_scheduled_export(event, user):
e = event.scheduled_exports.create(
owner=user,
export_identifier="orderlist",
export_form_data={
"_format": "xlsx",
"date_range": "year_this"
},
locale="en",
mail_additional_recipients="foo@example.org",
mail_subject="Current order list",
mail_template="Here is the current order list",
schedule_rrule="DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
schedule_rrule_time=time(4, 0, 0),
)
e.compute_next_run()
e.save()
return e
TEST_SCHEDULED_EXPORT_RES = {
"owner": "dummy@dummy.dummy",
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "foo@example.org",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
"error_counter": 0,
}
@pytest.mark.django_db
def test_event_scheduled_export_list_token(token_client, organizer, event, user, team, event_scheduled_export):
res = dict(TEST_SCHEDULED_EXPORT_RES)
res["id"] = event_scheduled_export.pk
res["schedule_next_run"] = event_scheduled_export.schedule_next_run.astimezone(zoneinfo.ZoneInfo("UTC")). \
isoformat().replace("+00:00", "Z")
# Token can see it because it has change permission
resp = token_client.get('/api/v1/organizers/{}/events/{}/scheduled_exports/'.format(organizer.slug, event.slug))
assert resp.status_code == 200
assert [res] == resp.data['results']
team.can_change_event_settings = False
team.save()
# Token can no longer sees it an gets error message
resp = token_client.get('/api/v1/organizers/{}/events/{}/scheduled_exports/'.format(organizer.slug, event.slug))
assert resp.status_code == 403
@pytest.mark.django_db
def test_event_scheduled_export_list_user(user_client, organizer, event, user, team, event_scheduled_export):
user2 = User.objects.create_user('dummy2@dummy.dummy', 'dummy')
team.members.add(user2)
res = dict(TEST_SCHEDULED_EXPORT_RES)
res["id"] = event_scheduled_export.pk
res["schedule_next_run"] = event_scheduled_export.schedule_next_run.astimezone(zoneinfo.ZoneInfo("UTC")).\
isoformat().replace("+00:00", "Z")
# User can see it because its their own
resp = user_client.get('/api/v1/organizers/{}/events/{}/scheduled_exports/'.format(organizer.slug, event.slug))
assert [res] == resp.data['results']
team.can_change_event_settings = False
team.save()
# Owner still can
resp = user_client.get('/api/v1/organizers/{}/events/{}/scheduled_exports/'.format(organizer.slug, event.slug))
assert [res] == resp.data['results']
# Other user can't see it and gets empty list
user_client.force_authenticate(user=user2)
resp = user_client.get('/api/v1/organizers/{}/events/{}/scheduled_exports/'.format(organizer.slug, event.slug))
assert resp.status_code == 200
assert [] == resp.data['results']
@pytest.mark.django_db
def test_event_scheduled_export_detail(token_client, organizer, event, user, event_scheduled_export):
res = dict(TEST_SCHEDULED_EXPORT_RES)
res["id"] = event_scheduled_export.pk
res["schedule_next_run"] = event_scheduled_export.schedule_next_run.astimezone(zoneinfo.ZoneInfo("UTC")).\
isoformat().replace("+00:00", "Z")
resp = token_client.get(
'/api/v1/organizers/{}/events/{}/scheduled_exports/{}/'.format(
organizer.slug, event.slug, event_scheduled_export.pk
)
)
assert resp.status_code == 200
assert res == resp.data
@pytest.mark.django_db
def test_event_scheduled_export_create(user_client, organizer, event, user):
resp = user_client.post(
'/api/v1/organizers/{}/events/{}/scheduled_exports/'.format(organizer.slug, event.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "foo@example.org",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 201
created = event.scheduled_exports.get(id=resp.data["id"])
assert created.export_form_data == {"_format": "xlsx", "date_range": "year_this"}
assert created.owner == user
assert created.schedule_next_run > now()
@pytest.mark.django_db
def test_event_scheduled_export_create_requires_user(token_client, organizer, event, user):
resp = token_client.post(
'/api/v1/organizers/{}/events/{}/scheduled_exports/'.format(organizer.slug, event.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "foo@example.org",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 403
@pytest.mark.django_db
def test_event_scheduled_export_delete_token(token_client, organizer, event, user, event_scheduled_export):
resp = token_client.delete(
'/api/v1/organizers/{}/events/{}/scheduled_exports/{}/'.format(
organizer.slug, event.slug, event_scheduled_export.pk,
),
)
assert resp.status_code == 204
assert not event.scheduled_exports.exists()
@pytest.mark.django_db
def test_event_scheduled_export_update_token(token_client, organizer, event, user, event_scheduled_export):
resp = token_client.patch(
'/api/v1/organizers/{}/events/{}/scheduled_exports/{}/'.format(
organizer.slug, event.slug, event_scheduled_export.pk,
),
data={
"export_form_data": {"_format": "xlsx", "date_range": "month_this"},
},
format='json'
)
assert resp.status_code == 200
created = event.scheduled_exports.get(id=resp.data["id"])
assert created.export_form_data == {"_format": "xlsx", "date_range": "month_this"}
@pytest.fixture
def org_scheduled_export(organizer, user):
e = organizer.scheduled_exports.create(
owner=user,
export_identifier="orderlist",
export_form_data={
"_format": "xlsx",
"date_range": "year_this"
},
locale="en",
mail_additional_recipients="foo@example.org",
mail_subject="Current order list",
mail_template="Here is the current order list",
schedule_rrule="DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
schedule_rrule_time=time(4, 0, 0),
)
e.compute_next_run()
e.save()
return e
@pytest.mark.django_db
def test_org_scheduled_export_list_token(token_client, organizer, user, team, org_scheduled_export):
res = dict(TEST_SCHEDULED_EXPORT_RES)
res["id"] = org_scheduled_export.pk
res["schedule_next_run"] = org_scheduled_export.schedule_next_run.astimezone(zoneinfo.ZoneInfo("UTC")). \
isoformat().replace("+00:00", "Z")
res["timezone"] = "UTC"
# Token can see it because it has change permission
resp = token_client.get('/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug))
assert resp.status_code == 200
assert [res] == resp.data['results']
team.can_change_organizer_settings = False
team.save()
# Token can no longer sees it an gets error message
resp = token_client.get('/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug))
assert resp.status_code == 403
@pytest.mark.django_db
def test_org_scheduled_export_list_user(user_client, organizer, user, team, org_scheduled_export):
user2 = User.objects.create_user('dummy2@dummy.dummy', 'dummy')
team.members.add(user2)
res = dict(TEST_SCHEDULED_EXPORT_RES)
res["id"] = org_scheduled_export.pk
res["schedule_next_run"] = org_scheduled_export.schedule_next_run.astimezone(zoneinfo.ZoneInfo("UTC")). \
isoformat().replace("+00:00", "Z")
res["timezone"] = "UTC"
# User can see it because its their own
resp = user_client.get('/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug))
assert [res] == resp.data['results']
team.can_change_organizer_settings = False
team.save()
# Owner still can
resp = user_client.get('/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug))
assert [res] == resp.data['results']
# Other user can't see it and gets empty list
user_client.force_authenticate(user=user2)
resp = user_client.get('/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug))
assert resp.status_code == 200
assert [] == resp.data['results']
@pytest.mark.django_db
def test_org_scheduled_export_detail(token_client, organizer, user, org_scheduled_export):
res = dict(TEST_SCHEDULED_EXPORT_RES)
res["id"] = org_scheduled_export.pk
res["schedule_next_run"] = org_scheduled_export.schedule_next_run.astimezone(zoneinfo.ZoneInfo("UTC")). \
isoformat().replace("+00:00", "Z")
res["timezone"] = "UTC"
resp = token_client.get(
'/api/v1/organizers/{}/scheduled_exports/{}/'.format(
organizer.slug, org_scheduled_export.pk
)
)
assert resp.status_code == 200
assert res == resp.data
@pytest.mark.django_db
def test_org_scheduled_export_create(user_client, organizer, user):
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "foo@example.org",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 201
created = organizer.scheduled_exports.get(id=resp.data["id"])
assert created.export_form_data == {"_format": "xlsx", "date_range": "year_this", "event_date_range": "/"}
assert created.owner == user
assert created.schedule_next_run > now()
@pytest.mark.django_db
def test_org_scheduled_export_create_requires_user(token_client, organizer, user):
resp = token_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "foo@example.org",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 403
@pytest.mark.django_db
def test_org_scheduled_export_delete_token(token_client, organizer, user, org_scheduled_export):
resp = token_client.delete(
'/api/v1/organizers/{}/scheduled_exports/{}/'.format(
organizer.slug, org_scheduled_export.pk,
),
)
assert resp.status_code == 204
assert not organizer.scheduled_exports.exists()
@pytest.mark.django_db
def test_org_scheduled_export_update_token(token_client, organizer, user, org_scheduled_export):
resp = token_client.patch(
'/api/v1/organizers/{}/scheduled_exports/{}/'.format(
organizer.slug, org_scheduled_export.pk,
),
data={
"export_form_data": {"_format": "xlsx", "date_range": "month_this"},
"timezone": "America/New_York"
},
format='json'
)
assert resp.status_code == 200
created = organizer.scheduled_exports.get(id=resp.data["id"])
assert created.export_form_data == {"_format": "xlsx", "date_range": "month_this", "event_date_range": "/"}
assert created.timezone == "America/New_York"
@pytest.mark.django_db
def test_org_scheduled_export_validate_identifier(user_client, organizer, user):
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "unknownorg",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "foo@example.org",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 400
assert resp.data == {"export_identifier": ["\"unknownorg\" is not a valid choice."]}
@pytest.mark.django_db
def test_org_scheduled_export_validate_form_data(user_client, organizer, user):
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "UNKNOWN"},
"locale": "en",
"mail_additional_recipients": "foo@example.org",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 400
assert resp.data == {"export_form_data": {"date_range": ["Invalid date frame"]}}
@pytest.mark.django_db
def test_org_scheduled_export_validate_locale(user_client, organizer, user):
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "BLÖDSINN",
"mail_additional_recipients": "",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 400
assert resp.data == {"locale": ["\"BLÖDSINN\" is not a valid choice."]}
@pytest.mark.django_db
def test_org_scheduled_export_validate_timezone(user_client, organizer, user):
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "de",
"mail_additional_recipients": "",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
"timezone": "Invalid"
},
format='json',
)
assert resp.status_code == 400
assert resp.data == {"timezone": ["\"Invalid\" is not a valid choice."]}
@pytest.mark.django_db
def test_org_scheduled_export_validate_additional_recipients(user_client, organizer, user):
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "aaaaaa",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 400
assert resp.data == {"mail_additional_recipients": ["Enter a valid email address."]}
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,"
"a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,"
"a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com,a@b.com",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 400
assert resp.data == {"mail_additional_recipients": ["Please enter less than 25 recipients."]}
@pytest.mark.django_db
def test_org_scheduled_export_validate_rrule(user_client, organizer, user):
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "invalid content",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 400
assert resp.data == {"schedule_rrule": ["Not a valid rrule."]}
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=WEEKLY;BYDAY=TU,WE,TH\nEXRULE:FREQ=WEEKLY;COUNT=4;INTERVAL=2;BYDAY=TU,TH",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 400
assert resp.data == {"schedule_rrule": ["Only a single RRULE is allowed, no combination of rules."]}
resp = user_client.post(
'/api/v1/organizers/{}/scheduled_exports/'.format(organizer.slug),
data={
"export_identifier": "orderlist",
"export_form_data": {"_format": "xlsx", "date_range": "year_this"},
"locale": "en",
"mail_additional_recipients": "",
"mail_additional_recipients_cc": "",
"mail_additional_recipients_bcc": "",
"mail_subject": "Current order list",
"mail_template": "Here is the current order list",
"schedule_rrule": "DTSTART:20230118T000000\nRRULE:FREQ=YEARLY;BYEASTER=0",
"schedule_rrule_time": "04:00:00",
},
format='json',
)
assert resp.status_code == 400
assert resp.data == {"schedule_rrule": ["BYEASTER not supported"]}