Script 'mail_helper' called by obssrc
Hello community,
here is the log from the commit of package python-django-eremaea2 for
openSUSE:Factory checked in at 2025-09-23 20:47:10
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Comparing /work/SRC/openSUSE:Factory/python-django-eremaea2 (Old)
and /work/SRC/openSUSE:Factory/.python-django-eremaea2.new.27445 (New)
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-django-eremaea2"
Tue Sep 23 20:47:10 2025 rev:10 rq:1306762 version:2.1.1
Changes:
--------
---
/work/SRC/openSUSE:Factory/python-django-eremaea2/python-django-eremaea2.changes
2024-10-04 17:11:20.970855590 +0200
+++
/work/SRC/openSUSE:Factory/.python-django-eremaea2.new.27445/python-django-eremaea2.changes
2025-09-23 20:48:17.806379765 +0200
@@ -1,0 +2,6 @@
+Tue Sep 23 08:48:30 UTC 2025 - Matwey Kornilov <[email protected]>
+
+- Version 2.1.1
+ - Rework API structure
+
+-------------------------------------------------------------------
Old:
----
django_eremaea2-2.0.22.tar.gz
New:
----
django_eremaea2-2.1.1.tar.gz
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Other differences:
------------------
++++++ python-django-eremaea2.spec ++++++
--- /var/tmp/diff_new_pack.jE7xui/_old 2025-09-23 20:48:18.494408850 +0200
+++ /var/tmp/diff_new_pack.jE7xui/_new 2025-09-23 20:48:18.494408850 +0200
@@ -1,7 +1,7 @@
#
# spec file for package python-django-eremaea2
#
-# Copyright (c) 2024 SUSE LLC
+# Copyright (c) 2025 SUSE LLC
#
# All modifications and additions to the file contributed by third parties
# remain the property of their copyright owners, unless otherwise agreed
@@ -20,7 +20,7 @@
%define skip_python36 1
%{?sle15_python_module_pythons}
Name: python-django-eremaea2
-Version: 2.0.22
+Version: 2.1.1
Release: 0
Summary: A simple Django application to store and show webcam snapshots
License: BSD-2-Clause
@@ -30,7 +30,9 @@
BuildArch: noarch
BuildRequires: %{python_module click}
BuildRequires: %{python_module devel}
+BuildRequires: %{python_module django-filter}
BuildRequires: %{python_module djangorestframework >= 3.7.0}
+BuildRequires: %{python_module drf-spectacular}
BuildRequires: %{python_module pip}
BuildRequires: %{python_module pytest-django}
# https://github.com/matwey/django-eremaea2/issues/15
@@ -44,12 +46,14 @@
BuildRequires: python-rpm-macros
Requires: eremaea = %{version}
Requires: python-click
+Requires: python-django-filter
Requires: python-djangorestframework >= 3.7.0
+Requires: python-drf-spectacular
Requires: python-python-magic
Requires: python-requests
Requires: python-requests-toolbelt
Requires(post): update-alternatives
-Requires(postun):update-alternatives
+Requires(postun): update-alternatives
%python_subpackages
%description
++++++ django_eremaea2-2.0.22.tar.gz -> django_eremaea2-2.1.1.tar.gz ++++++
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/PKG-INFO
new/django_eremaea2-2.1.1/PKG-INFO
--- old/django_eremaea2-2.0.22/PKG-INFO 2024-10-03 15:53:29.807284600 +0200
+++ new/django_eremaea2-2.1.1/PKG-INFO 2025-09-23 10:41:26.318097000 +0200
@@ -1,6 +1,6 @@
-Metadata-Version: 2.1
+Metadata-Version: 2.4
Name: django-eremaea2
-Version: 2.0.22
+Version: 2.1.1
Summary: A simple Django application to store and show webcam snapshots
Author-email: "Matwey V. Kornilov" <[email protected]>
License: BSD-2-Clause
@@ -20,7 +20,9 @@
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: click
+Requires-Dist: django-filter
Requires-Dist: djangorestframework
+Requires-Dist: drf_spectacular
Requires-Dist: python-magic
Requires-Dist: requests
Requires-Dist: requests_toolbelt
@@ -30,6 +32,7 @@
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-django; extra == "dev"
Requires-Dist: requests-mock; extra == "dev"
+Dynamic: license-file
# django-eremaea2
@@ -73,8 +76,7 @@
To perform actual cleanup a POST request to the endpoint
```http://example.com/eremaea/retention_policies/{name}/purge``` is required.
The ```Collection``` model has two parameters: a lookup field ```name``` and
```default_retention_policy```.
The ```Snapshot``` model has the following field: associated ```collection```
and ```retention_policy```, ```file``` object, and auto-now ```date```.
-New images are uploaded by POST request to the endpoint
```http://example.com/eremaea/snapshots/?collection=collection_name&retention_policy=retention_policy_name```.
-The latest query parameter is optional one.
+New images are uploaded by POST request to the endpoint
```http://example.com/eremaea/snapshots/{collection}/``` with optional
```retention_policy={retention_policy_name}``` query string parameter.
The other ways to configure the application are Django fixtures, Django admin
interface, [django-rest-framework] web browsable interface, and REST API itself.
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/README.md
new/django_eremaea2-2.1.1/README.md
--- old/django_eremaea2-2.0.22/README.md 2024-10-03 15:53:20.000000000
+0200
+++ new/django_eremaea2-2.1.1/README.md 2025-09-23 10:41:21.000000000 +0200
@@ -40,8 +40,7 @@
To perform actual cleanup a POST request to the endpoint
```http://example.com/eremaea/retention_policies/{name}/purge``` is required.
The ```Collection``` model has two parameters: a lookup field ```name``` and
```default_retention_policy```.
The ```Snapshot``` model has the following field: associated ```collection```
and ```retention_policy```, ```file``` object, and auto-now ```date```.
-New images are uploaded by POST request to the endpoint
```http://example.com/eremaea/snapshots/?collection=collection_name&retention_policy=retention_policy_name```.
-The latest query parameter is optional one.
+New images are uploaded by POST request to the endpoint
```http://example.com/eremaea/snapshots/{collection}/``` with optional
```retention_policy={retention_policy_name}``` query string parameter.
The other ways to configure the application are Django fixtures, Django admin
interface, [django-rest-framework] web browsable interface, and REST API itself.
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/pyproject.toml
new/django_eremaea2-2.1.1/pyproject.toml
--- old/django_eremaea2-2.0.22/pyproject.toml 2024-10-03 15:53:20.000000000
+0200
+++ new/django_eremaea2-2.1.1/pyproject.toml 2025-09-23 10:41:21.000000000
+0200
@@ -4,7 +4,7 @@
[project]
name = "django-eremaea2"
-version = "2.0.22"
+version = "2.1.1"
authors = [
{name = "Matwey V. Kornilov", email = "[email protected]"},
]
@@ -26,7 +26,9 @@
requires-python = ">= 3.7"
dependencies = [
"click",
+ "django-filter",
"djangorestframework",
+ "drf_spectacular",
"python-magic",
"requests",
"requests_toolbelt",
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/django_eremaea2-2.0.22/src/django_eremaea2.egg-info/PKG-INFO
new/django_eremaea2-2.1.1/src/django_eremaea2.egg-info/PKG-INFO
--- old/django_eremaea2-2.0.22/src/django_eremaea2.egg-info/PKG-INFO
2024-10-03 15:53:29.000000000 +0200
+++ new/django_eremaea2-2.1.1/src/django_eremaea2.egg-info/PKG-INFO
2025-09-23 10:41:26.000000000 +0200
@@ -1,6 +1,6 @@
-Metadata-Version: 2.1
+Metadata-Version: 2.4
Name: django-eremaea2
-Version: 2.0.22
+Version: 2.1.1
Summary: A simple Django application to store and show webcam snapshots
Author-email: "Matwey V. Kornilov" <[email protected]>
License: BSD-2-Clause
@@ -20,7 +20,9 @@
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: click
+Requires-Dist: django-filter
Requires-Dist: djangorestframework
+Requires-Dist: drf_spectacular
Requires-Dist: python-magic
Requires-Dist: requests
Requires-Dist: requests_toolbelt
@@ -30,6 +32,7 @@
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-django; extra == "dev"
Requires-Dist: requests-mock; extra == "dev"
+Dynamic: license-file
# django-eremaea2
@@ -73,8 +76,7 @@
To perform actual cleanup a POST request to the endpoint
```http://example.com/eremaea/retention_policies/{name}/purge``` is required.
The ```Collection``` model has two parameters: a lookup field ```name``` and
```default_retention_policy```.
The ```Snapshot``` model has the following field: associated ```collection```
and ```retention_policy```, ```file``` object, and auto-now ```date```.
-New images are uploaded by POST request to the endpoint
```http://example.com/eremaea/snapshots/?collection=collection_name&retention_policy=retention_policy_name```.
-The latest query parameter is optional one.
+New images are uploaded by POST request to the endpoint
```http://example.com/eremaea/snapshots/{collection}/``` with optional
```retention_policy={retention_policy_name}``` query string parameter.
The other ways to configure the application are Django fixtures, Django admin
interface, [django-rest-framework] web browsable interface, and REST API itself.
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/django_eremaea2-2.0.22/src/django_eremaea2.egg-info/SOURCES.txt
new/django_eremaea2-2.1.1/src/django_eremaea2.egg-info/SOURCES.txt
--- old/django_eremaea2-2.0.22/src/django_eremaea2.egg-info/SOURCES.txt
2024-10-03 15:53:29.000000000 +0200
+++ new/django_eremaea2-2.1.1/src/django_eremaea2.egg-info/SOURCES.txt
2025-09-23 10:41:26.000000000 +0200
@@ -23,6 +23,8 @@
src/eremaea/ctl/commandline.py
src/eremaea/ctl/file.py
src/eremaea/migrations/0001_squashed_0002_drop_index_together.py
+src/eremaea/migrations/0003_initial_retention_policies.py
+src/eremaea/migrations/0004_use_slug_for_names.py
src/eremaea/migrations/__init__.py
tests/__init__.py
tests/settings.py
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/django_eremaea2-2.0.22/src/django_eremaea2.egg-info/requires.txt
new/django_eremaea2-2.1.1/src/django_eremaea2.egg-info/requires.txt
--- old/django_eremaea2-2.0.22/src/django_eremaea2.egg-info/requires.txt
2024-10-03 15:53:29.000000000 +0200
+++ new/django_eremaea2-2.1.1/src/django_eremaea2.egg-info/requires.txt
2025-09-23 10:41:26.000000000 +0200
@@ -1,5 +1,7 @@
click
+django-filter
djangorestframework
+drf_spectacular
python-magic
requests
requests_toolbelt
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/src/eremaea/ctl/client.py
new/django_eremaea2-2.1.1/src/eremaea/ctl/client.py
--- old/django_eremaea2-2.0.22/src/eremaea/ctl/client.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/src/eremaea/ctl/client.py 2025-09-23
10:41:21.000000000 +0200
@@ -9,12 +9,12 @@
self._session.headers.update({'Authorization': 'Token
{}'.format(token)})
def upload(self, file, collection, retention_policy = None):
- url = self.api + '/snapshots/'
+ url = self.api + '/snapshots/{}/'.format(collection)
headers = {
'Content-Disposition': 'attachment;
filename=\"{}\"'.format(file.name),
'Content-Type': file.mimetype
}
- params = {'collection': collection}
+ params = {}
if retention_policy:
params['retention_policy'] = retention_policy
r = self._session.post(url, params=params, headers=headers,
data=file.content)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/django_eremaea2-2.0.22/src/eremaea/migrations/0003_initial_retention_policies.py
new/django_eremaea2-2.1.1/src/eremaea/migrations/0003_initial_retention_policies.py
---
old/django_eremaea2-2.0.22/src/eremaea/migrations/0003_initial_retention_policies.py
1970-01-01 01:00:00.000000000 +0100
+++
new/django_eremaea2-2.1.1/src/eremaea/migrations/0003_initial_retention_policies.py
2025-09-23 10:41:21.000000000 +0200
@@ -0,0 +1,28 @@
+# Generated by Django 4.2.4 on 2023-08-11 08:25
+
+from datetime import timedelta
+from django.db import migrations
+
+
+def populate_retention_policies(apps, schema_editor):
+ INITIAL_RETENTION_POLICIES = [
+ ("annual", 365),
+ ("monthly", 30),
+ ("weekly", 7),
+ ("daily", 1),
+ ]
+
+ RetentionPolicy = apps.get_model("eremaea", "RetentionPolicy")
+
+ for (name, duration) in INITIAL_RETENTION_POLICIES:
+ RetentionPolicy.objects.get_or_create(name = name, duration =
timedelta(days = duration))
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('eremaea', '0002_drop_index_together'),
+ ]
+
+ operations = [
+ migrations.RunPython(populate_retention_policies),
+ ]
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/django_eremaea2-2.0.22/src/eremaea/migrations/0004_use_slug_for_names.py
new/django_eremaea2-2.1.1/src/eremaea/migrations/0004_use_slug_for_names.py
---
old/django_eremaea2-2.0.22/src/eremaea/migrations/0004_use_slug_for_names.py
1970-01-01 01:00:00.000000000 +0100
+++ new/django_eremaea2-2.1.1/src/eremaea/migrations/0004_use_slug_for_names.py
2025-09-23 10:41:21.000000000 +0200
@@ -0,0 +1,23 @@
+# Generated by Django 5.2.3 on 2025-08-12 09:14
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('eremaea', '0003_initial_retention_policies'),
+ ]
+
+ operations = [
+ migrations.AlterField(
+ model_name='collection',
+ name='name',
+ field=models.SlugField(max_length=256, unique=True),
+ ),
+ migrations.AlterField(
+ model_name='retentionpolicy',
+ name='name',
+ field=models.SlugField(max_length=256, unique=True),
+ ),
+ ]
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/src/eremaea/models.py
new/django_eremaea2-2.1.1/src/eremaea/models.py
--- old/django_eremaea2-2.0.22/src/eremaea/models.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/src/eremaea/models.py 2025-09-23
10:41:21.000000000 +0200
@@ -69,39 +69,17 @@
if storage.exists(path):
storage.delete(path)
- def get_next(self):
- try:
- return self.get_next_by_date(collection =
self.collection)
- except Snapshot.DoesNotExist:
- pass
- def get_previous(self):
- try:
- return self.get_previous_by_date(collection =
self.collection)
- except Snapshot.DoesNotExist:
- pass
-
class Meta:
ordering = ['-date']
get_latest_by = 'date'
class Collection(models.Model):
- name = models.CharField(max_length=256, blank=False, unique=True,
db_index=True)
+ name = models.SlugField(max_length=256, unique=True, db_index=True)
default_retention_policy = models.ForeignKey('RetentionPolicy',
on_delete=models.PROTECT)
- def get_latest(self):
- try:
- return Snapshot.objects.filter(collection =
self).latest()
- except Snapshot.DoesNotExist:
- pass
- def get_earliest(self):
- try:
- return Snapshot.objects.filter(collection =
self).earliest()
- except Snapshot.DoesNotExist:
- pass
-
class RetentionPolicy(models.Model):
- name = models.CharField(max_length=256, blank=False, null=False,
unique=True, db_index=True)
- duration = models.DurationField(blank=False, null=False)
+ name = models.SlugField(max_length=256, unique=True, db_index=True)
+ duration = models.DurationField()
class Meta:
db_table = 'retention_policy'
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/src/eremaea/serializers.py
new/django_eremaea2-2.1.1/src/eremaea/serializers.py
--- old/django_eremaea2-2.0.22/src/eremaea/serializers.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/src/eremaea/serializers.py 2025-09-23
10:41:21.000000000 +0200
@@ -1,29 +1,69 @@
from eremaea import models
+from django.shortcuts import get_object_or_404
from rest_framework import serializers
+from rest_framework.reverse import reverse
+
+
+class SnapshotHyperlinkRelated(serializers.HyperlinkedRelatedField):
+ view_name = 'snapshot-detail'
+
+ def get_url(self, obj, view_name, request, format):
+ url_kwargs = {
+ 'collection': obj.collection.name,
+ 'pk': obj.pk,
+ }
+
+ return reverse(view_name, kwargs=url_kwargs, request=request,
format=format)
+
+ def get_object(self, view_name, view_args, view_kwargs):
+ return self.get_queryset().get(collection__name =
view_kwargs['collection'], pk = view_kwargs['pk'])
+
+class SnapshotHyperlinkIdentity(SnapshotHyperlinkRelated):
+ def __init__(self, **kwargs):
+ kwargs['read_only'] = True
+ kwargs['source'] = '*'
+ super(SnapshotHyperlinkIdentity, self).__init__(**kwargs)
+
+class CurrentCollectionDefault:
+ requires_context = True
+
+ def __call__(self, serializer_field):
+ view = serializer_field.context['view']
+ collection_name = view.kwargs['collection']
+ collection = get_object_or_404(models.Collection, name =
collection_name)
+ return collection
class RetentionPolicySerializer(serializers.HyperlinkedModelSerializer):
url = serializers.HyperlinkedIdentityField(lookup_field='name',
view_name='retention_policy-detail')
class Meta:
model = models.RetentionPolicy
- fields = ('url', 'name', 'duration')
+ fields = '__all__'
+
+class CreateSnapshotSerializer(serializers.HyperlinkedModelSerializer):
+ url = SnapshotHyperlinkIdentity()
+ collection =
serializers.SlugRelatedField(queryset=models.Collection.objects.all(),
slug_field='name', allow_null=False,
default=serializers.CreateOnlyDefault(CurrentCollectionDefault()))
+ retention_policy =
serializers.SlugRelatedField(queryset=models.RetentionPolicy.objects.all(),
slug_field='name', allow_null=True)
+
+ class Meta:
+ model = models.Snapshot
+ fields = '__all__'
+
+class SnapshotSerializer(CreateSnapshotSerializer):
+ class Meta(CreateSnapshotSerializer.Meta):
+ read_only_fields = ['file']
-class SnapshotSerializer(serializers.HyperlinkedModelSerializer):
- collection =
serializers.HyperlinkedRelatedField(view_name='collection-detail',
read_only=True, lookup_field='name')
- retention_policy =
serializers.HyperlinkedRelatedField(view_name='retention_policy-detail',
queryset=models.RetentionPolicy.objects.all(), lookup_field='name')
- next = serializers.HyperlinkedRelatedField(view_name='snapshot-detail',
read_only=True, source="get_next")
- prev = serializers.HyperlinkedRelatedField(view_name='snapshot-detail',
read_only=True, source="get_previous")
+class ListSnapshotSerializer(SnapshotSerializer):
+ collection = None
class Meta:
model = models.Snapshot
- fields = ('url', 'collection', 'file', 'date',
'retention_policy', 'next', 'prev')
+ exclude = ['collection',]
class CollectionSerializer(serializers.HyperlinkedModelSerializer):
url = serializers.HyperlinkedIdentityField(lookup_field='name',
view_name='collection-detail')
- default_retention_policy =
serializers.HyperlinkedRelatedField(view_name='retention_policy-detail',
queryset=models.RetentionPolicy.objects.all(), lookup_field='name')
- begin =
serializers.HyperlinkedRelatedField(view_name='snapshot-detail',
read_only=True, source="get_earliest")
- end = serializers.HyperlinkedRelatedField(view_name='snapshot-detail',
read_only=True, source="get_latest")
+ default_retention_policy =
serializers.SlugRelatedField(queryset=models.RetentionPolicy.objects.all(),
slug_field='name')
class Meta:
model = models.Collection
- fields = ('url', 'name', 'default_retention_policy', 'begin',
'end')
+ fields = '__all__'
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/src/eremaea/urls.py
new/django_eremaea2-2.1.1/src/eremaea/urls.py
--- old/django_eremaea2-2.0.22/src/eremaea/urls.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/src/eremaea/urls.py 2025-09-23
10:41:21.000000000 +0200
@@ -4,7 +4,7 @@
router = DefaultRouter()
router.register(r'collections', views.CollectionViewSet)
-router.register(r'snapshots', views.SnapshotViewSet)
+router.register(r'snapshots/(?P<collection>[-\w]+)', views.SnapshotViewSet)
router.register(r'retention_policies', views.RetentionPolicyViewSet,
basename='retention_policy')
if django.VERSION[0] > 1:
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/src/eremaea/views.py
new/django_eremaea2-2.1.1/src/eremaea/views.py
--- old/django_eremaea2-2.0.22/src/eremaea/views.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/src/eremaea/views.py 2025-09-23
10:41:21.000000000 +0200
@@ -1,11 +1,129 @@
-from django.shortcuts import get_object_or_404
+import datetime
+from django.conf import settings
from django.db.models.deletion import ProtectedError
-from django.utils.cache import patch_cache_control
+from django.shortcuts import get_object_or_404
+from django.utils import timezone
+from django.utils.cache import patch_cache_control, patch_response_headers
+from django.utils.http import http_date
+from django_filters import rest_framework as filters
from eremaea import models, serializers
from rest_framework import status, viewsets
-from rest_framework.decorators import action
+from rest_framework.decorators import action, parser_classes
+from rest_framework.pagination import Cursor, CursorPagination, _positive_int
from rest_framework.parsers import FileUploadParser
from rest_framework.response import Response
+from rest_framework.settings import api_settings
+from rest_framework.utils.urls import replace_query_param
+from drf_spectacular.utils import extend_schema
+from drf_spectacular.types import OpenApiTypes
+
+
+class CollectionFilter(filters.FilterSet):
+ default_retention_policy =
filters.ModelChoiceFilter(queryset=models.RetentionPolicy.objects.all(),
to_field_name="name")
+
+ class Meta:
+ model = models.Collection
+ fields = ['default_retention_policy']
+
+class SnapshotPagination(CursorPagination):
+ ordering = '-date'
+ page_size_query_param = 'page_size'
+ cursor_separator = '.'
+ # Cursor text is always in UTC for consistency. However, it must be
+ # converted to Django format accoring to actual settings.USE_TZ
+ time_origin = datetime.datetime(2000, 1, 1,
tzinfo=datetime.timezone.utc)
+
+ @staticmethod
+ def _datetime_to_django(datetime):
+ if not settings.USE_TZ and timezone.is_aware(datetime):
+ datetime = timezone.make_naive(datetime,
timezone.get_default_timezone())
+ return datetime
+
+ @staticmethod
+ def _datetime_from_django(datetime):
+ if timezone.is_naive(datetime):
+ datetime = timezone.make_aware(datetime,
timezone.get_default_timezone())
+ return datetime
+
+ def decode_cursor(self, request):
+ encoded = request.query_params.get(self.cursor_query_param)
+ if encoded is None:
+ return None
+
+ try:
+ position, offset, reverse =
encoded.split(self.cursor_separator)
+
+ if not position:
+ position = None
+ else:
+ position =
self._datetime_to_django(self.time_origin + datetime.datetime.resolution *
_positive_int(position))
+ offset = _positive_int(offset,
cutoff=self.offset_cutoff)
+ reverse = bool(int(reverse))
+ except (TypeError, ValueError):
+ raise NotFound(self.invalid_cursor_message)
+
+ return Cursor(offset=offset, reverse=reverse, position=position)
+
+ def encode_cursor(self, cursor):
+ if cursor.position is not None:
+ position =
str(int((self._datetime_from_django(cursor.position) - self.time_origin) /
datetime.datetime.resolution))
+ else:
+ position = ''
+ offset = str(cursor.offset)
+ reverse = str(int(cursor.reverse))
+ encoded = self.cursor_separator.join([position, offset,
reverse])
+
+ return replace_query_param(self.base_url,
self.cursor_query_param, encoded)
+
+ def _get_position_from_instance(self, instance, ordering):
+ field_name = ordering[0].lstrip('-')
+ if isinstance(instance, dict):
+ attr = instance[field_name]
+ else:
+ attr = getattr(instance, field_name)
+
+ assert isinstance(attr, datetime.datetime), (
+ 'Invalid ordering value type. Expected
datetime.datetime type, but got {type}'.format(type=type(attr).__name__)
+ )
+
+ return attr
+
+ def get_paginated_response(self, data):
+ response = super(SnapshotPagination,
self).get_paginated_response(data)
+
+ date_now = timezone.now()
+ date = self.page[0].date.timestamp() if self.page else
date_now.timestamp()
+ response['Date'] = http_date(date)
+
+ reverse = self.cursor.reverse if self.cursor else False
+ current_position = self.cursor.position if self.cursor else None
+ is_expires = (current_position is not None and current_position
< date_now and not reverse) or (reverse and self.has_previous)
+
+ if self.page and is_expires:
+ last_instance = self.page[-1]
+ retention_policy = last_instance.retention_policy
+ response['Expires'] =
http_date(last_instance.date.timestamp() +
retention_policy.duration.total_seconds())
+ else:
+ patch_cache_control(response, no_cache=True, max_age=0)
+
+ return response
+
+class
SnapshotContentNegotiation(api_settings.DEFAULT_CONTENT_NEGOTIATION_CLASS):
+ def select_parser(self, request, parsers):
+ viewset = request.parser_context['view']
+ if viewset.action == 'create':
+ return FileUploadParser()
+
+ return super(SnapshotContentNegotiation,
self).select_parser(request, parsers)
+
+class SnapshotFilter(filters.FilterSet):
+ retention_policy =
filters.ModelChoiceFilter(queryset=models.RetentionPolicy.objects.all(),
to_field_name="name")
+ date = filters.IsoDateTimeFromToRangeFilter()
+
+ class Meta:
+ model = models.Snapshot
+ fields = ['retention_policy', 'date']
+
class RetentionPolicyViewSet(viewsets.ModelViewSet):
queryset = models.RetentionPolicy.objects.all()
@@ -17,54 +135,97 @@
return super(RetentionPolicyViewSet,
self).destroy(request, name)
except ProtectedError as e:
return Response(status=status.HTTP_400_BAD_REQUEST)
+
+ @extend_schema(request=None)
@action(methods=['post'], detail=True)
def purge(self, request, name):
- retention_policy = get_object_or_404(models.RetentionPolicy,
name = name)
+ retention_policy = self.get_object()
retention_policy.purge()
return Response(status=status.HTTP_201_CREATED)
class CollectionViewSet(viewsets.ModelViewSet):
- queryset =
models.Collection.objects.select_related("default_retention_policy")
+ queryset =
models.Collection.objects.select_related('default_retention_policy')
serializer_class = serializers.CollectionSerializer
lookup_field = 'name'
-
- def retrieve(self, request, name=None):
- response = super(CollectionViewSet, self).retrieve(request,
name)
- link = []
- if 'begin' in response.data:
- link.append("{0};
rel=begin".format(response.data['begin']))
- if 'end' in response.data:
- link.append("{0}; rel=end".format(response.data['end']))
- if link:
- response['Link'] = ", ".join(link)
- patch_cache_control(response, max_age=0, must_revalidate=True)
- return response
+ filter_backends = [filters.DjangoFilterBackend]
+ filterset_class = CollectionFilter
class SnapshotViewSet(viewsets.ModelViewSet):
- queryset = models.Snapshot.objects.select_related("collection",
"retention_policy")
+ queryset = models.Snapshot.objects.select_related('collection',
'retention_policy')
serializer_class = serializers.SnapshotSerializer
- parser_classes = (FileUploadParser,)
+ create_serializer_class = serializers.CreateSnapshotSerializer
+ list_serializer_class = serializers.ListSnapshotSerializer
+ pagination_class = SnapshotPagination
+ content_negotiation_class = SnapshotContentNegotiation
+ filter_backends = [filters.DjangoFilterBackend]
+ filterset_class = SnapshotFilter
+
+ def get_queryset(self):
+ collection_name = self.kwargs['collection']
+ queryset = super(SnapshotViewSet, self).get_queryset()
+
+ # For list action we need to distinguish
+ # empty response from non-existing collection
+ if self.action == 'list':
+ collection = get_object_or_404(models.Collection, name
= collection_name)
+ return queryset.filter(collection = collection)
+
+ return queryset.filter(collection__name = collection_name)
+
+ def _get_create_serializer(self, request):
+ file = getattr(request, 'data', {}).get('file')
+ retention_policy = getattr(request, 'query_params',
{}).get('retention_policy')
+
+ data = {
+ 'file': file,
+ 'retention_policy': retention_policy,
+ }
+
+ return super(SnapshotViewSet, self).get_serializer(data = data)
+
+ def get_serializer(self, *args, **kwargs):
+ if self.action == 'create':
+ return self._get_create_serializer(self.request)
+
+ return super(SnapshotViewSet, self).get_serializer(*args,
**kwargs)
+
+ def get_serializer_class(self):
+ if self.action == 'create':
+ return self.create_serializer_class
+ elif self.action == 'list':
+ return self.list_serializer_class
+
+ return super(SnapshotViewSet, self).get_serializer_class()
+
+ @extend_schema(request={'image/*': OpenApiTypes.BINARY})
+ def create(self, request, collection=None):
+ return super(SnapshotViewSet, self).create(request, collection)
+
+ def retrieve(self, request, pk=None, collection=None):
+ response = super(SnapshotViewSet, self).retrieve(request, pk =
pk, collection = collection)
+ instance = response.data.serializer.instance
+ retention_policy = instance.retention_policy
+
+ response['Link'] = '{0};
rel=alternate'.format(response.data['file'])
+ response['Date'] = http_date(instance.date.timestamp())
+ patch_response_headers(response,
cache_timeout=retention_policy.duration.total_seconds())
+
+ return response
+
+ def list(self, request, collection=None):
+ response = super(SnapshotViewSet, self).list(request,
collection = collection)
+
+ # Paginated response handled in
SnapshotPagination.get_paginated_response()
+ if not hasattr(response.data, 'serializer'):
+ return response
+
+ query_set = response.data.serializer.instance
+ instance = query_set.first()
+
+ date_now = timezone.now()
+ date = instance.date.timestamp() if instance is not None else
date_now.timestamp()
+ response['Date'] = http_date(date)
+
+ patch_cache_control(response, no_cache=True, max_age=0)
- def create(self, request):
- if 'collection' not in request.query_params:
- return Response(status=status.HTTP_400_BAD_REQUEST,
data={'detail':'collection is not specified'})
- if 'file' not in request.data:
- return Response(status=status.HTTP_400_BAD_REQUEST,
data={'detail':'file is not supplied'})
- kwargs = {}
- kwargs['file'] = request.data['file']
- kwargs['collection'] = models.Collection.objects.get(name =
request.query_params['collection'])
- if 'retention_policy' in request.query_params:
- kwargs['retention_policy'] =
models.RetentionPolicy.objects.get(name =
request.query_params['retention_policy'])
- models.Snapshot.objects.create(**kwargs)
- return Response(status=status.HTTP_201_CREATED)
- def retrieve(self, request, pk=None):
- response = super(SnapshotViewSet, self).retrieve(request, pk)
- link = []
- if response.data['next'] is not None:
- link.append("{0};
rel=next".format(response.data['next']))
- if response.data['prev'] is not None:
- link.append("{0};
rel=prev".format(response.data['prev']))
- link.append("{0}; rel=alternate".format(response.data['file']))
- if link:
- response['Link'] = ", ".join(link)
return response
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/tests/ctl/test_client.py
new/django_eremaea2-2.1.1/tests/ctl/test_client.py
--- old/django_eremaea2-2.0.22/tests/ctl/test_client.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/tests/ctl/test_client.py 2025-09-23
10:41:21.000000000 +0200
@@ -1,4 +1,5 @@
-from datetime import datetime, timedelta
+from datetime import timedelta
+from django.utils import timezone
from django.core.files.base import ContentFile
from django.contrib.auth.models import User
from django.test import LiveServerTestCase
@@ -36,7 +37,7 @@
file.name = "file.jpg"
retention_policy =
models.RetentionPolicy.objects.create(name="hourly",
duration=timedelta(hours=1))
collection = models.Collection.objects.create(name="mycol",
default_retention_policy=retention_policy)
- models.Snapshot.objects.create(collection = collection, date =
datetime.now() - timedelta(minutes=90), file=file)
+ models.Snapshot.objects.create(collection = collection, date =
timezone.now() - timedelta(minutes=90), file=file)
self.assertTrue(self.client.purge("hourly"))
snapshots = models.Snapshot.objects.all()
self.assertEqual(len(snapshots), 0)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/tests/settings.py
new/django_eremaea2-2.1.1/tests/settings.py
--- old/django_eremaea2-2.0.22/tests/settings.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/tests/settings.py 2025-09-23 10:41:21.000000000
+0200
@@ -9,9 +9,15 @@
'django.contrib.auth',
'rest_framework',
'rest_framework.authtoken',
+ 'drf_spectacular',
+ 'django_filters',
'eremaea',
]
+MIDDLEWARE = [
+ 'django.middleware.http.ConditionalGetMiddleware',
+]
+
TEMPLATES = [{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'APP_DIRS': True,
@@ -26,6 +32,7 @@
'DEFAULT_PARSER_CLASSES': (
'rest_framework.parsers.JSONParser',
),
+ 'DEFAULT_SCHEMA_CLASS': 'drf_spectacular.openapi.AutoSchema',
}
DATABASES = {
'default': {
@@ -48,5 +55,5 @@
MIDDLEWARE_CLASSES = []
ROOT_URLCONF = 'tests.urls'
TIME_ZONE = 'UTC'
-
-DEFAULT_AUTO_FIELD='django.db.models.AutoField'
+USE_TZ = True
+DEFAULT_AUTO_FIELD = 'django.db.models.AutoField'
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/tests/test_collection.py
new/django_eremaea2-2.1.1/tests/test_collection.py
--- old/django_eremaea2-2.0.22/tests/test_collection.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/tests/test_collection.py 2025-09-23
10:41:21.000000000 +0200
@@ -2,9 +2,11 @@
from django.urls import reverse
except ImportError:
from django.core.urlresolvers import reverse
+from django.core.files.base import ContentFile
from django.test import TestCase
from rest_framework import status
from rest_framework.test import APIClient
+from rest_framework.utils.urls import replace_query_param
from eremaea import models
from datetime import timedelta
from urllib.parse import urlparse
@@ -12,51 +14,79 @@
class CollectionTest(TestCase):
def setUp(self):
self.client = APIClient()
- self.retention =
models.RetentionPolicy.objects.create(name="daily", duration=timedelta(days=1))
- def tearDown(self):
- pass
-
+
def assertEqualUrl(self, x, y):
path_x = urlparse(x).path
path_y = urlparse(y).path
return self.assertEqual(path_x, path_y)
- def test_collection_create1(self):
+
+ def test_collection_create(self):
+ url = reverse('collection-list')
+ retention = 'daily'
+
+ response = self.client.post(url, {'name': 'collection',
'default_retention_policy': retention}, format='json')
+ self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+ self.assertIn('Location', response)
+ self.assertIn(reverse('collection-detail', args =
('collection',)), response['Location'])
+ collection = models.Collection.objects.get(name='collection')
+ self.assertEqual(collection.default_retention_policy,
models.RetentionPolicy.objects.get(name='daily'))
+
+ def test_collection_create_duplicate(self):
url = reverse('collection-list')
- retention_url = reverse('retention_policy-detail',
args=('daily',))
- response = self.client.post(url, {'name': 'name1',
'default_retention_policy': retention_url}, format='json')
+ retention = 'daily'
+
+ response = self.client.post(url, {'name': 'collection',
'default_retention_policy': retention}, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
- response = self.client.post(url, {'name': 'name1',
'default_retention_policy': retention_url}, format='json')
+
+ response = self.client.post(url, {'name': 'collection',
'default_retention_policy': retention}, format='json')
+
self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
- group1 = models.Collection.objects.get(name="name1")
- self.assertEqual(group1.default_retention_policy,
self.retention)
- url = reverse('collection-detail', args=['name1'])
- response = self.client.get(url, format='json')
- self.assertEqual(response.status_code, status.HTTP_200_OK)
- def test_collection_get1(self):
- collection =
models.Collection.objects.create(name="collection",
default_retention_policy=self.retention)
- snapshots = []
- for i in range(0,3):
-
snapshots.append(models.Snapshot.objects.create(collection = collection))
- url = reverse('collection-detail', args=('collection',))
- response = self.client.get(url)
- self.assertEqualUrl(response.data['begin'],
reverse("snapshot-detail", args=(snapshots[0].id,)))
- self.assertEqualUrl(response.data['end'],
reverse("snapshot-detail", args=(snapshots[2].id,)))
- def test_collection_head1(self):
- collection =
models.Collection.objects.create(name="collection",
default_retention_policy=self.retention)
- snapshots = []
- for i in range(0,3):
-
snapshots.append(models.Snapshot.objects.create(collection = collection))
+
+ def test_collection_update(self):
+ retention_policy =
models.RetentionPolicy.objects.get(name='daily')
+ retention_alternative = 'weekly'
+ collection =
models.Collection.objects.create(name='collection',
default_retention_policy=retention_policy)
+
url = reverse('collection-detail', args=('collection',))
- response = self.client.head(url)
+ response = self.client.put(url, {
+ 'name': 'alternative',
+ 'default_retention_policy': retention_alternative
+ }, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
- self.assertEqualUrl(response.data['begin'],
reverse("snapshot-detail", args=(snapshots[0].id,)))
- self.assertEqualUrl(response.data['end'],
reverse("snapshot-detail", args=(snapshots[2].id,)))
- def test_collection_delete1(self):
- group1 = models.Collection.objects.create(name="delete1",
default_retention_policy=self.retention)
- url = reverse('collection-detail', args=['delete1'])
+
+ collection.refresh_from_db()
+ self.assertEqual(collection.name, 'alternative')
+ self.assertEqual(collection.default_retention_policy,
models.RetentionPolicy.objects.get(name='weekly'))
+
+ def test_collection_delete(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+
+ retention_policy =
models.RetentionPolicy.objects.get(name='daily')
+ collection =
models.Collection.objects.create(name='collection',
default_retention_policy=retention_policy)
+ snapshot = models.Snapshot.objects.create(collection =
collection, file = file)
+ storage = snapshot.file.storage
+ filepath = snapshot.file.name
+
+ url = reverse('collection-detail', args=(collection.name,))
response = self.client.delete(url)
self.assertEqual(response.status_code,
status.HTTP_204_NO_CONTENT)
+ self.assertRaises(models.Collection.DoesNotExist,
models.Collection.objects.get, name='collection')
+ self.assertRaises(models.Snapshot.DoesNotExist,
models.Snapshot.objects.get, pk=snapshot.id)
+ # Known issue:
+ # self.assertFalse(storage.exists(filepath))
+
response = self.client.delete(url)
self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
- self.assertRaises(models.Collection.DoesNotExist,
models.Collection.objects.get, name="delete1")
+ def test_collection_filter_by_retention_policy(self):
+ daily = models.RetentionPolicy.objects.get(name='daily')
+ weekly = models.RetentionPolicy.objects.get(name='weekly')
+ collection =
models.Collection.objects.create(name='collection',
default_retention_policy=daily)
+ alternative =
models.Collection.objects.create(name='alternative',
default_retention_policy=weekly)
+
+ url = reverse('collection-list')
+ url = replace_query_param(url, 'default_retention_policy',
'daily')
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(len(response.data), 1)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/tests/test_retention.py
new/django_eremaea2-2.1.1/tests/test_retention.py
--- old/django_eremaea2-2.0.22/tests/test_retention.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/tests/test_retention.py 2025-09-23
10:41:21.000000000 +0200
@@ -7,66 +7,124 @@
from rest_framework import status
from rest_framework.test import APIClient
from eremaea import models
-from datetime import datetime,timedelta
+from datetime import timedelta
+from django.utils import timezone
class RetentionPolicyTest(TestCase):
def setUp(self):
self.client = APIClient()
- def test_retention_create1(self):
- url = reverse('retention_policy-list')
- response = self.client.post(url, {'name': 'daily', 'duration':
'01 00'}, format='json')
- self.assertEqual(response.status_code, status.HTTP_201_CREATED)
- response = self.client.post(url, {'name': 'daily', 'duration':
'02 00'}, format='json')
- self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
- group1 = models.RetentionPolicy.objects.get(name="daily")
- self.assertEqual(group1.duration, timedelta(days=1))
+ def test_retention_get_daily(self):
url = reverse('retention_policy-detail', args=['daily'])
response = self.client.get(url, format='json')
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['duration'], '1 00:00:00')
+
+ def test_retention_get_weekly(self):
+ url = reverse('retention_policy-detail', args=['weekly'])
+ response = self.client.get(url, format='json')
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(response.data['duration'], '7 00:00:00')
+
+ def test_retention_get_monthly(self):
+ url = reverse('retention_policy-detail', args=['monthly'])
+ response = self.client.get(url, format='json')
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(response.data['duration'], '30 00:00:00')
+
+ def test_retention_get_annual(self):
+ url = reverse('retention_policy-detail', args=['annual'])
+ response = self.client.get(url, format='json')
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(response.data['duration'], '365 00:00:00')
+
+ def test_retention_create_biweekly(self):
+ url = reverse('retention_policy-list')
+
+ response = self.client.post(url, {
+ 'name': 'biweekly',
+ 'duration': '14 00'
+ }, format='json')
+ self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+ self.assertIn('Location', response)
+ self.assertIn(reverse('retention_policy-detail', args =
('biweekly',)), response['Location'])
+
+ biweekly = models.RetentionPolicy.objects.get(name='biweekly')
+ self.assertEqual(biweekly.duration, timedelta(days=14))
+
+ def test_retention_create_duplicate(self):
+ url = reverse('retention_policy-list')
+
+ response = self.client.post(url, {
+ 'name': 'weekly',
+ 'duration': '7 00'
+ }, format='json')
+ self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
+
+ def test_retention_update_weekly(self):
+ weekly = models.RetentionPolicy.objects.get(name='weekly')
+ url = reverse('retention_policy-detail', args=['weekly'])
+
+ response = self.client.put(url, {
+ 'name': 'biweekly',
+ 'duration': '14 00'
+ }, format='json')
self.assertEqual(response.status_code, status.HTTP_200_OK)
- def test_retention_delete1(self):
- group1 = models.RetentionPolicy.objects.create(id=123,
name="test", duration=timedelta(days=1))
- url = reverse('retention_policy-detail', args=['test'])
+
+ weekly.refresh_from_db()
+ self.assertEqual(weekly.name, 'biweekly')
+ self.assertEqual(weekly.duration, timedelta(days=14))
+
+ def test_retention_delete_biweekly(self):
+ biweekly =
models.RetentionPolicy.objects.create(name='biweekly',
duration=timedelta(days=14))
+
+ url = reverse('retention_policy-detail', args=['biweekly'])
response = self.client.delete(url)
self.assertEqual(response.status_code,
status.HTTP_204_NO_CONTENT)
+
response = self.client.delete(url)
self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
- self.assertRaises(models.RetentionPolicy.DoesNotExist,
models.RetentionPolicy.objects.get, name="test")
- def test_retention_delete2(self):
+ self.assertRaises(models.RetentionPolicy.DoesNotExist,
models.RetentionPolicy.objects.get, name='biweekly')
+
+ def test_retention_delete_protected_by_collection(self):
# Retention deletion is protected by collection
- retention = models.RetentionPolicy.objects.create(name='daily',
duration=timedelta(days=1))
- collection = models.Collection.objects.create(name='test',
default_retention_policy=retention)
+ retention, created =
models.RetentionPolicy.objects.get_or_create(name='daily',
duration=timedelta(days=1))
+ collection =
models.Collection.objects.create(name='collection',
default_retention_policy=retention)
+
url = reverse('retention_policy-detail', args=['daily'])
response = self.client.delete(url)
self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
- def test_retention_delete3(self):
+
+ def test_retention_delete_protected_by_snapshot(self):
# Retention deletion is protected by snapshot
- retention_daily =
models.RetentionPolicy.objects.create(name='daily', duration=timedelta(days=1))
- retention_hourly =
models.RetentionPolicy.objects.create(name='hourly',
duration=timedelta(hours=1))
- collection = models.Collection.objects.create(name='test',
default_retention_policy=retention_daily)
+ retention_daily, created =
models.RetentionPolicy.objects.get_or_create(name='daily',
duration=timedelta(days=1))
+ retention_hourly, created =
models.RetentionPolicy.objects.get_or_create(name='hourly',
duration=timedelta(hours=1))
+ collection =
models.Collection.objects.create(name='collection',
default_retention_policy=retention_daily)
snapshot = models.Snapshot.objects.create(collection =
collection, retention_policy=retention_hourly)
+
url = reverse('retention_policy-detail', args=['hourly'])
response = self.client.delete(url)
self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
- def test_retention_purge1(self):
- file = ContentFile(b"123")
- file.name = "file.jpg"
- retention_daily =
models.RetentionPolicy.objects.create(name='daily', duration=timedelta(days=1))
- retention_hourly =
models.RetentionPolicy.objects.create(name='hourly',
duration=timedelta(hours=1))
- collection = models.Collection.objects.create(name='test',
default_retention_policy=retention_hourly)
- dates = [datetime.now(), datetime.now() -
timedelta(minutes=30), datetime.now() - timedelta(minutes=90)]
+
+ def test_retention_purge(self):
+ retention_daily, created =
models.RetentionPolicy.objects.get_or_create(name='daily',
duration=timedelta(days=1))
+ retention_hourly, created =
models.RetentionPolicy.objects.get_or_create(name='hourly',
duration=timedelta(hours=1))
+ collection =
models.Collection.objects.create(name='collection',
default_retention_policy=retention_hourly)
+
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ dates = [timezone.now(), timezone.now() -
timedelta(minutes=30), timezone.now() - timedelta(minutes=90)]
snapshots = [models.Snapshot.objects.create(collection =
collection, date = x, file = file) for x in dates]
snapshots.append(models.Snapshot.objects.create(collection =
collection, date = dates[-1], retention_policy = retention_daily, file = file))
storage2, filepath2 = snapshots[2].file.storage,
snapshots[2].file.name
+
url = reverse('retention_policy-purge', args=['hourly'])
response = self.client.post(url)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
- snapshots2 = models.Snapshot.objects.all()
- self.assertEqual(len(snapshots2), 3)
- self.assertEqual(snapshots2[0], snapshots[0])
- self.assertEqual(snapshots2[1], snapshots[1])
- self.assertEqual(snapshots2[2], snapshots[3])
+
+ snapshots_expected = snapshots[:2] + snapshots[3:]
+ snapshots_actual = list(models.Snapshot.objects.all())
+ self.assertListEqual(snapshots_actual, snapshots_expected)
self.assertFalse(storage2.exists(filepath2))
def test_retention_purge2(self):
url = reverse('retention_policy-purge', args=['not_exists'])
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/django_eremaea2-2.0.22/tests/test_snapshot.py
new/django_eremaea2-2.1.1/tests/test_snapshot.py
--- old/django_eremaea2-2.0.22/tests/test_snapshot.py 2024-10-03
15:53:20.000000000 +0200
+++ new/django_eremaea2-2.1.1/tests/test_snapshot.py 2025-09-23
10:41:21.000000000 +0200
@@ -1,154 +1,428 @@
+from django.conf import settings
from django.core.files.base import ContentFile
-from django.test import TestCase
+from django.test import TestCase, override_settings
+from django.utils import timezone
from rest_framework import status
from rest_framework.reverse import reverse
from rest_framework.test import APIClient
+from rest_framework.utils.urls import replace_query_param
from mimetypes import guess_all_extensions
from os.path import splitext
from eremaea import models
-from datetime import timedelta
+from datetime import datetime, timedelta
from urllib.parse import urlparse
-class SnapshotTest(TestCase):
+
+class SnapshotTestBase(TestCase):
+ __test__ = False
+
def setUp(self):
self.client = APIClient()
- self.retention =
models.RetentionPolicy.objects.create(name="daily", duration=timedelta(days=1))
- self.collection =
models.Collection.objects.create(name="mycol",
default_retention_policy=self.retention)
+ self.retention =
models.RetentionPolicy.objects.get(name='daily')
+ self.collection =
models.Collection.objects.create(name='collection', default_retention_policy =
self.retention)
+
def tearDown(self):
self.collection.delete()
- self.retention.delete()
def assertEqualUrl(self, x, y):
path_x = urlparse(x).path
path_y = urlparse(y).path
return self.assertEqual(path_x, path_y)
- def test_snapshot_create1(self):
- content = b'123'
- url = reverse('snapshot-list')
+
+ @staticmethod
+ def make_datetime(*args, **kwargs):
+ obj = datetime(*args, **kwargs)
+
+ if settings.USE_TZ:
+ default_timezone = timezone.get_default_timezone()
+ obj = timezone.make_aware(obj, default_timezone)
+
+ return obj
+
+ def test_snapshot_create_in_not_existing_collection(self):
+ content = b'test'
+
+ url = reverse('snapshot-list', kwargs = {'collection':
'not_exists'})
response = self.client.post(url, content,
content_type='image/jpeg', HTTP_CONTENT_DISPOSITION='attachment;
filename=upload.jpg')
- self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
- def test_snapshot_create2(self):
- content = b'123'
- url = reverse('snapshot-list') + '?collection=mycol'
+ self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
+
+ def test_snapshot_create(self):
+ content = b'test'
+
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
response = self.client.post(url, content,
content_type='image/jpeg', HTTP_CONTENT_DISPOSITION='attachment;
filename=upload.jpg')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
snapshot = models.Snapshot.objects.all()[0]
self.assertEqual(snapshot.retention_policy, self.retention)
self.assertEqual(snapshot.file.read(), content)
- def test_snapshot_create3(self):
- content = b'123'
- retention_hourly =
models.RetentionPolicy.objects.create(name="hourly",
duration=timedelta(hours=1))
- url = reverse('snapshot-list') +
'?collection=mycol&retention_policy=hourly'
+ self.assertIn('Location', response)
+ self.assertIn(reverse('snapshot-detail', kwargs = {
+ 'collection': self.collection.name,
+ 'pk': snapshot.id
+ }), response['Location'])
+
+ def test_snapshot_create_with_retention_policy(self):
+ content = b'test'
+ retention_policy =
models.RetentionPolicy.objects.get(name='weekly')
+
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
+ url = replace_query_param(url, 'retention_policy',
retention_policy.name)
response = self.client.post(url, content,
content_type='image/jpeg', HTTP_CONTENT_DISPOSITION='attachment;
filename=upload.jpg')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
snapshot = models.Snapshot.objects.all()[0]
- self.assertEqual(snapshot.retention_policy, retention_hourly)
+ self.assertEqual(snapshot.retention_policy, retention_policy)
self.assertEqual(snapshot.file.read(), content)
- def test_snapshot_create4(self):
+
+ def test_snapshot_create_with_not_existing_retention_policy(self):
+ content = b'test'
+
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
+ url = replace_query_param(url, 'retention_policy', 'not_exists')
+ response = self.client.post(url, content,
content_type='image/jpeg', HTTP_CONTENT_DISPOSITION='attachment;
filename=upload.jpg')
+ self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
+
+ def test_snapshot_create_with_empty_payload(self):
content = b''
- url = reverse('snapshot-list') + '?collection=mycol'
+
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
response = self.client.post(url, content,
content_type='image/jpeg', HTTP_CONTENT_DISPOSITION='attachment;
filename=upload.jpg')
self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
- def test_snapshot_create5(self):
- url = reverse('snapshot-list') + '?collection=mycol'
+
+ def test_snapshot_create_guess_by_content_type(self):
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
+
response = self.client.post(url, {}, content_type='image/jpeg',
HTTP_CONTENT_DISPOSITION='attachment; filename=upload.png')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
snapshot = models.Snapshot.objects.all()[0]
self.assertIn(splitext(snapshot.file.name)[1],
guess_all_extensions('image/jpeg'))
- def test_snapshot_create6(self):
- url = reverse('snapshot-list') + '?collection=mycol'
+
+ def test_snapshot_create_guess_by_content_type_ext(self):
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
+
response = self.client.post(url, {}, content_type='image/jpeg',
HTTP_CONTENT_DISPOSITION='attachment; filename=upload.jpg')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
snapshot = models.Snapshot.objects.all()[0]
- self.assertEqual(splitext(snapshot.file.name)[1], ".jpg")
- def test_snapshot_create7(self):
- url = reverse('snapshot-list') + '?collection=mycol'
+ self.assertEqual(splitext(snapshot.file.name)[1], '.jpg')
+
+ def test_snapshot_create_guess_by_filename(self):
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
+
response = self.client.post(url, {},
HTTP_CONTENT_DISPOSITION='attachment; filename=upload.png')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
snapshot = models.Snapshot.objects.all()[0]
- self.assertEqual(splitext(snapshot.file.name)[1], ".png")
- def test_snapshot_create8(self):
+ self.assertEqual(splitext(snapshot.file.name)[1], '.png')
+
+ def test_snapshot_create_guess_by_content(self):
content =
b'\x89\x50\x4e\x47\x0d\x0a\x1a\x0a\x00\x00\x00\x0d\x49\x48\x44\x52\x00\x00\x00\x01\x00\x00\x00\x01\x08\x04\x00\x00\x00\xb5\x1c\x0c\x02\x00\x00\x00\x0b\x49\x44\x41\x54\x78\x9c\x63\x62\x60\x00\x00\x00\x09\x00\x03\x19\x11\xd9\xe4\x00\x00\x00\x00\x49\x45\x4e\x44\xae\x42\x60\x82'
- url = reverse('snapshot-list') + '?collection=mycol'
+
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
response = self.client.post(url, content,
content_type='application/x-null', HTTP_CONTENT_DISPOSITION='attachment;
filename=upload')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+
snapshot = models.Snapshot.objects.all()[0]
- self.assertEqual(splitext(snapshot.file.name)[1], ".png")
- def test_snapshot_get1(self):
- file = ContentFile(b"123")
- file.name = "file.jpg"
+ self.assertEqual(splitext(snapshot.file.name)[1], '.png')
+
+ def test_snapshot_get(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
- url = reverse('snapshot-detail', args=[snapshot.id])
- response = self.client.get(url)
- self.assertIsNone(response.data['next'])
- self.assertIsNone(response.data['prev'])
- def test_snapshot_get2(self):
- file = ContentFile(b"123")
- file.name = "file.jpg"
- snapshot1 = models.Snapshot.objects.create(collection =
self.collection, file = file)
- snapshot2 = models.Snapshot.objects.create(collection =
self.collection, file = file)
- snapshot3 = models.Snapshot.objects.create(collection =
self.collection, file = file)
- url = reverse('snapshot-detail', args=[snapshot1.id])
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': self.collection.name,
+ 'pk': snapshot.id})
response = self.client.get(url)
- self.assertEqualUrl(response.data['next'],
reverse("snapshot-detail", args=(snapshot2.id,)))
- self.assertIsNone(response.data['prev'])
- url = reverse('snapshot-detail', args=[snapshot2.id])
- response = self.client.get(url)
- self.assertEqualUrl(response.data['next'],
reverse("snapshot-detail", args=(snapshot3.id,)))
- self.assertEqualUrl(response.data['prev'],
reverse("snapshot-detail", args=(snapshot1.id,)))
- url = reverse('snapshot-detail', args=[snapshot3.id])
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(response['Link'], '{};
rel=alternate'.format(response.data['file']))
+ self.assertIn('Date', response)
+ self.assertIn('Expires', response)
+
+ def test_snapshot_get_from_not_existing_collection(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': 'not_exists',
+ 'pk': snapshot.id})
response = self.client.get(url)
- self.assertIsNone(response.data['next'])
- self.assertEqualUrl(response.data['prev'],
reverse("snapshot-detail", args=(snapshot2.id,)))
- def test_snapshot_get3(self):
- file = ContentFile(b"123")
- file.name = "file.jpg"
- retention_hourly =
models.RetentionPolicy.objects.create(name="hourly",
duration=timedelta(hours=1))
- collection1 =
models.Collection.objects.create(name="collection1",
default_retention_policy=retention_hourly)
- collection2 =
models.Collection.objects.create(name="collection2",
default_retention_policy=retention_hourly)
- snapshots1 = []
- snapshots2 = []
- for i in range(0,3):
-
snapshots1.append(models.Snapshot.objects.create(collection = collection1, file
= file))
-
snapshots2.append(models.Snapshot.objects.create(collection = collection2, file
= file))
- # first collection
- self.assertEqual(snapshots1[0].get_next(), snapshots1[1])
- self.assertEqual(snapshots1[1].get_next(), snapshots1[2])
- self.assertEqual(snapshots1[1].get_previous(), snapshots1[0])
- self.assertEqual(snapshots1[2].get_previous(), snapshots1[1])
- self.assertEqual(collection1.get_earliest(), snapshots1[0])
- self.assertEqual(collection1.get_latest(), snapshots1[2])
- # second collection
- self.assertEqual(snapshots2[0].get_next(), snapshots2[1])
- self.assertEqual(snapshots2[1].get_next(), snapshots2[2])
- self.assertEqual(snapshots2[1].get_previous(), snapshots2[0])
- self.assertEqual(snapshots2[2].get_previous(), snapshots2[1])
- self.assertEqual(collection2.get_earliest(), snapshots2[0])
- self.assertEqual(collection2.get_latest(), snapshots2[2])
- def test_snapshot_get4(self):
- file = ContentFile(b"123")
- file.name = "file.jpg"
+ self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
+
+ def test_snapshot_get_from_wrong_collection(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
- url = reverse('snapshot-detail', args=[snapshot.id])
+ collection =
models.Collection.objects.create(name='alternative', default_retention_policy =
self.retention)
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': collection.name,
+ 'pk': snapshot.id})
response = self.client.get(url)
+ self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
+
+ def test_snapshot_head(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': self.collection.name,
+ 'pk': snapshot.id})
+ response = self.client.head(url)
link_hdr = response['Link']
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(link_hdr, '{};
rel=alternate'.format(response.data['file']))
- def test_snapshot_head1(self):
- file = ContentFile(b"123")
- file.name = "file.jpg"
+
+ def test_shapshot_update_retention_policy(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
- url = reverse('snapshot-detail', args=[snapshot.id])
- response = self.client.head(url)
+
+ retention = 'weekly'
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': self.collection.name,
+ 'pk': snapshot.id})
+ response = self.client.patch(url, {
+ 'retention_policy': retention
+ }, format='json')
+
self.assertEqual(response.status_code, status.HTTP_200_OK)
- self.assertIsNone(response.data['next'])
- self.assertIsNone(response.data['prev'])
- def test_snapshot_delete1(self):
- file = ContentFile(b"123")
- file.name = "file.jpg"
+ snapshot.refresh_from_db()
+ self.assertEqual(snapshot.retention_policy,
models.RetentionPolicy.objects.get(name='weekly'))
+
+ def test_shapshot_update_retention_policy_to_non_existing(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': self.collection.name,
+ 'pk': snapshot.id})
+ response = self.client.patch(url, {
+ 'retention_policy': 'not_exists'
+ }, format='json')
+
+ self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
+
+ def test_shapshot_update_collection(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
+
+ collection =
models.Collection.objects.create(name='alternative', default_retention_policy =
self.retention)
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': self.collection.name,
+ 'pk': snapshot.id})
+ response = self.client.patch(url, {
+ 'collection': collection.name
+ }, format='json')
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ snapshot.refresh_from_db()
+ self.assertEqual(snapshot.collection, collection)
+
+ def test_shapshot_update_collection_to_non_existing(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': self.collection.name,
+ 'pk': snapshot.id})
+ response = self.client.patch(url, {
+ 'collection': 'not_exists'
+ }, format='json')
+
+ self.assertEqual(response.status_code,
status.HTTP_400_BAD_REQUEST)
+
+ def test_shapshot_update_file(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
+ storage = snapshot.file.storage
+ filepath = snapshot.file.name
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': self.collection.name,
+ 'pk': snapshot.id})
+ response = self.client.patch(url, {
+ 'file': 'not_exists'
+ }, format='json')
+
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertTrue(storage.exists(filepath))
+
+ def test_snapshot_delete(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
storage = snapshot.file.storage
filepath = snapshot.file.name
- url = reverse('snapshot-detail', args=[snapshot.id])
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': self.collection.name,
+ 'pk': snapshot.id})
response = self.client.delete(url)
self.assertEqual(response.status_code,
status.HTTP_204_NO_CONTENT)
self.assertFalse(storage.exists(filepath))
+
+ def test_snapshot_delete_from_wrong_collection(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ snapshot = models.Snapshot.objects.create(collection =
self.collection, file = file)
+ storage = snapshot.file.storage
+ filepath = snapshot.file.name
+ collection =
models.Collection.objects.create(name='alternative', default_retention_policy =
self.retention)
+
+ url = reverse('snapshot-detail', kwargs = {
+ 'collection': collection.name,
+ 'pk': snapshot.id})
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
+
+ snapshot.refresh_from_db()
+ self.assertIsNotNone(snapshot.id)
+
+ def test_snapshot_list(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ snapshot1 = models.Snapshot.objects.create(collection =
self.collection, file = file)
+ snapshot2 = models.Snapshot.objects.create(collection =
self.collection, file = file)
+ snapshot3 = models.Snapshot.objects.create(collection =
self.collection, file = file)
+ url = reverse('snapshot-list', kwargs = {
+ 'collection': self.collection.name
+ })
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(len(response.data), 3)
+ self.assertIn('Date', response)
+ self.assertNotIn('Expires', response)
+ self.assertIn('no-cache', response["Cache-Control"])
+
+ def test_snapshot_list_from_not_existing_collection(self):
+ url = reverse('snapshot-list', kwargs = {'collection':
'not_exists'})
+ response = self.client.get(url)
+ self.assertEqual(response.status_code,
status.HTTP_404_NOT_FOUND)
+
+ def test_snapshot_list_from_empty_collection(self):
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+ def test_snapshot_filter_by_retention_policy(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ daily = models.RetentionPolicy.objects.get(name='daily')
+ weekly = models.RetentionPolicy.objects.get(name='weekly')
+ snapshot1 = models.Snapshot.objects.create(collection =
self.collection, file = file, retention_policy = daily)
+ snapshot2 = models.Snapshot.objects.create(collection =
self.collection, file = file, retention_policy = weekly)
+
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
+ url = replace_query_param(url, 'retention_policy', 'daily')
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(len(response.data), 1)
+ self.assertIn('Date', response)
+ self.assertNotIn('Expires', response)
+ self.assertIn('no-cache', response["Cache-Control"])
+
+ def test_snapshot_filter_by_date_range(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ snapshot1 = models.Snapshot.objects.create(collection =
self.collection,
+ file = file, date = self.make_datetime(2001, 1, 1))
+ snapshot2 = models.Snapshot.objects.create(collection =
self.collection,
+ file = file, date = self.make_datetime(2001, 1, 3))
+ snapshot3 = models.Snapshot.objects.create(collection =
self.collection,
+ file = file, date = self.make_datetime(2001, 1, 5))
+
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
+ url = replace_query_param(url, 'date_after',
'2001-01-02T00:00:00')
+ url = replace_query_param(url, 'date_before',
'2001-01-03T00:00:00')
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertEqual(len(response.data), 1)
+ self.assertIn('Date', response)
+ self.assertNotIn('Expires', response)
+ self.assertIn('no-cache', response["Cache-Control"])
+
+ def test_snapshot_list_pagination(self):
+ file = ContentFile(b'123')
+ file.name = 'file.jpg'
+ date1 = self.make_datetime(2001, 1, 1)
+ date2 = self.make_datetime(2001, 1, 2)
+ snapshot1 = models.Snapshot.objects.create(collection =
self.collection, file = file, date = date1)
+ snapshot2 = models.Snapshot.objects.create(collection =
self.collection, file = file, date = date1)
+ snapshot2 = models.Snapshot.objects.create(collection =
self.collection, file = file, date = date2)
+
+ url = reverse('snapshot-list', kwargs = {'collection':
self.collection.name})
+ url = replace_query_param(url, 'page_size', 1)
+
+ # First item
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertIn('next', response.data)
+ self.assertIsNone(response.data['previous'])
+ self.assertEqual(len(response.data['results']), 1)
+ self.assertIn('Date', response)
+ self.assertNotIn('Expires', response)
+ self.assertIn('no-cache', response["Cache-Control"])
+ url = response.data['next']
+
+ # Second item
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertIn('next', response.data)
+ self.assertIn('previous', response.data)
+ self.assertEqual(len(response.data['results']), 1)
+ self.assertIn('Date', response)
+ self.assertIn('Expires', response)
+ self.assertNotIn('Cache-Control', response)
+ url = response.data['next']
+
+ # Third item
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertIsNone(response.data['next'])
+ self.assertIn('previous', response.data)
+ self.assertEqual(len(response.data['results']), 1)
+ self.assertIn('Date', response)
+ self.assertIn('Expires', response)
+ self.assertNotIn('Cache-Control', response)
+ url = response.data['previous']
+
+ # Second item in reverse
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertIn('next', response.data)
+ self.assertIn('previous', response.data)
+ self.assertEqual(len(response.data['results']), 1)
+ self.assertIn('Date', response)
+ self.assertIn('Expires', response)
+ self.assertNotIn('Cache-Control', response)
+ url = response.data['previous']
+
+ # First item in reverse
+ response = self.client.get(url)
+ self.assertEqual(response.status_code, status.HTTP_200_OK)
+ self.assertIn('next', response.data)
+ self.assertIsNone(response.data['previous'])
+ self.assertEqual(len(response.data['results']), 1)
+ self.assertIn('Date', response)
+ self.assertNotIn('Expires', response)
+ self.assertIn('no-cache', response['Cache-Control'])
+ url = response.data['previous']
+
+
+@override_settings(USE_TZ=False)
+class SnapshotTestNoTZ(SnapshotTestBase):
+ __test__ = True
+
+@override_settings(USE_TZ=True)
+class SnapshotTestWithTZ(SnapshotTestBase):
+ __test__ = True
