api: work around ambiguous time error on DST change (#37238)

src/authentic2/api_views.py

@@ -14,9 +14,11 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from functools import partial
 import logging
 import smtplib
 
+from pytz.exceptions import AmbiguousTimeError
 import django
 from django.db import models
 from django.contrib.auth import get_user_model
@@ -24,6 +26,7 @@ from django.contrib.auth.hashers import identify_hasher
 from django.core.exceptions import MultipleObjectsReturned
 from django.utils.translation import ugettext as _
 from django.utils.encoding import force_text
+from django.utils.dateparse import parse_datetime
 from django.views.decorators.vary import vary_on_headers
 from django.views.decorators.cache import cache_control
 from django.shortcuts import get_object_or_404
@@ -46,6 +49,9 @@ from rest_framework.filters import BaseFilterBackend
 from rest_framework.settings import api_settings
 
 from django_filters.rest_framework import FilterSet
+from django_filters.filters import IsoDateTimeFilter
+from django_filters.fields import IsoDateTimeField
+from django_filters.utils import handle_timezone
 
 from .passwords import get_password_checker
 from .custom_user.models import User
@@ -559,6 +565,42 @@ class RoleSerializer(serializers.ModelSerializer):
         extra_kwargs = {'uuid': {'read_only': True}}
 
 
+# override to handle ambiguous naive DateTime on DST change
+class IsoDateTimeField(IsoDateTimeField):
+    def __init__(self, *args, **kwargs):
+        self.bound = kwargs.pop('bound')
+        assert self.bound in ['upper', 'lesser']
+        super(IsoDateTimeField, self).__init__(*args, **kwargs)
+
+    def strptime(self, value, format):
+        try:
+            return super(IsoDateTimeField, self).strptime(value, format)
+        except AmbiguousTimeError:
+            parsed = parse_datetime(value)
+            possible = sorted([
+                handle_timezone(parsed, is_dst=True),
+                handle_timezone(parsed, is_dst=False),
+            ])
+            if self.bound == 'lesser':
+                return possible[0]
+            elif self.bound == 'upper':
+                return possible[1]
+
+
+class IsoDateTimeFilter(IsoDateTimeFilter):
+    @property
+    def field_class(self):
+        if self.lookup_expr.startswith('gt'):
+            return partial(IsoDateTimeField, bound='lesser')
+        elif self.lookup_expr.startswith('lt'):
+            return partial(IsoDateTimeField, bound='upper')
+        else:
+            raise NotImplementedError
+
+    def filter(self, qs, value):
+        return super(IsoDateTimeFilter, self).filter(qs, value)
+
+
 class UsersFilter(FilterSet):
     class Meta:
         model = get_user_model()
@@ -599,6 +641,11 @@ class UsersFilter(FilterSet):
                 'exact',
             ],
         }
+        filter_overrides = {
+            models.DateTimeField: {
+                'filter_class': IsoDateTimeFilter,
+            }
+        }
 
 
 class ChangeEmailSerializer(serializers.Serializer):

tests/test_api.py

@@ -29,6 +29,8 @@ from django.contrib.auth import get_user_model
 from django.contrib.contenttypes.models import ContentType
 from django.core import mail
 from django.core.urlresolvers import reverse
+from django.utils.timezone import now
+from django.utils.http import urlencode
 
 from django_rbac.models import SEARCH_OP
 from django_rbac.utils import get_role_model, get_ou_model
@@ -1683,3 +1685,18 @@ def test_filter_users_by_service(app, admin, simple_user, role_random, service):
 
     resp = app.get('/api/users/?service-slug=service&service-ou=default')
     assert len(resp.json['results']) == 1
+
+
+def test_filter_users_by_last_modification(app, admin, simple_user, freezer):
+    app.authorization = ('Basic', (admin.username, admin.username))
+
+    freezer.move_to('2019-10-27T02:00:00Z')
+
+    admin.save()
+    simple_user.save()
+
+    # AmbiguousTimeError
+    resp = app.get('/api/users/', params={'modified__gt': '2019-10-27T02:58:07'})
+    assert len(resp.json['results']) == 2
+    resp = app.get('/api/users/', params={'modified__lt': '2019-10-27T02:58:07'})
+    assert len(resp.json['results']) == 0