!112 !94: tests commons modules that contains most of the logic + zoneinfo (#113)

!112

Co-authored-by: bkfox <thomas bkfox net>
Reviewed-on: rc/aircox#113

aircox/admin/filters.py

@@ -13,7 +13,7 @@ class DateFieldFilter(filters.FieldListFilter):
     input_type = "date"
 
     def __init__(self, field, request, params, model, model_admin, field_path):
-        self.field_generic = "%s__" % field_path
+        self.field_generic = f"{field_path}__"
         self.date_params = {
             k: v for k, v in params.items() if k.startswith(self.field_generic)
         }

aircox/admin/mixins.py

@@ -1,41 +0,0 @@
-class UnrelatedInlineMixin:
-    """Inline class that can be included in an admin change view whose model is
-    not directly related to inline's model."""
-
-    view_model = None
-    parent_model = None
-    parent_fk = ""
-
-    def __init__(self, parent_model, admin_site):
-        self.view_model = parent_model
-        super().__init__(self.parent_model, admin_site)
-
-    def get_parent(self, view_obj):
-        """Get formset's instance from `obj` of AdminSite's change form."""
-        field = self.parent_model._meta.get_field(self.parent_fk).remote_field
-        return getattr(view_obj, field.name, None)
-
-    def save_parent(self, parent, view_obj):
-        """Save formset's instance."""
-        setattr(parent, self.parent_fk, view_obj)
-        parent.save()
-        return parent
-
-    def get_formset(self, request, obj):
-        ParentFormSet = super().get_formset(request, obj)
-        inline = self
-
-        class FormSet(ParentFormSet):
-            view_obj = None
-
-            def __init__(self, *args, instance=None, **kwargs):
-                self.view_obj = instance
-                instance = inline.get_parent(instance)
-                self.instance = instance
-                super().__init__(*args, instance=instance, **kwargs)
-
-            def save(self):
-                inline.save_parent(self.instance, self.view_obj)
-                return super().save()
-
-        return FormSet

aircox/controllers/log_archiver.py

@@ -80,7 +80,7 @@ class LogArchiver:
     def load_file(self, path):
         with gzip.open(path, "rb") as archive:
             data = archive.read()
-            logs = yaml.load(data)
+            logs = yaml.safe_load(data)
 
             # we need to preload diffusions, sounds and tracks
             rels = {

aircox/controllers/sound_stats.py

@@ -84,7 +84,6 @@ class SoundStats:
         self.stats = [SoxStats(self.path)]
         position = 0
         length = self.stats[0].get("length")
-        print(self.stats, "-----")
         if not self.sample_length:
             return
 

aircox/middleware.py

@@ -1,4 +1,5 @@
-import pytz
+from zoneinfo import ZoneInfo
+
 from django.db.models import Q
 from django.utils import timezone as tz
 
@@ -11,38 +12,36 @@ __all__ = ("AircoxMiddleware",)
 class AircoxMiddleware(object):
     """Middleware used to get default info for the given website.
 
-    Theses
+    It provide following request attributes:
+    - ``station``: current Station
+
     This middleware must be set after the middleware
         'django.contrib.auth.middleware.AuthenticationMiddleware',
     """
 
+    timezone_session_key = "aircox.timezone"
+
     def __init__(self, get_response):
         self.get_response = get_response
 
     def get_station(self, request):
         """Return station for the provided request."""
-        expr = Q(default=True) | Q(hosts__contains=request.get_host())
+        host = request.get_host()
-        # case = Case(When(hosts__contains=request.get_host(), then=Value(0)),
+        expr = Q(default=True) | Q(hosts=host) | Q(hosts__contains=host + "\n")
-        #            When(default=True, then=Value(32)))
         return Station.objects.filter(expr).order_by("default").first()
-        #              .annotate(resolve_priority=case) \
-        # .order_by('resolve_priority').first()
 
     def init_timezone(self, request):
         # note: later we can use http://freegeoip.net/ on user side if
         # required
         timezone = None
         try:
-            timezone = request.session.get("aircox.timezone")
+            timezone = request.session.get(self.timezone_session_key)
             if timezone:
-                timezone = pytz.timezone(timezone)
+                timezone = ZoneInfo(timezone)
+                tz.activate(timezone)
         except Exception:
             pass
 
-        if not timezone:
-            timezone = tz.get_current_timezone()
-            tz.activate(timezone)
-
     def __call__(self, request):
         self.init_timezone(request)
         request.station = self.get_station(request)

aircox/models/diffusion.py

@@ -39,8 +39,10 @@ class DiffusionQuerySet(RerunQuerySet):
     def date(self, date=None, order=True):
         """Diffusions occuring date."""
         date = date or datetime.date.today()
-        start = tz.datetime.combine(date, datetime.time())
+        start = tz.make_aware(tz.datetime.combine(date, datetime.time()))
-        end = tz.datetime.combine(date, datetime.time(23, 59, 59, 999))
+        end = tz.make_aware(
+            tz.datetime.combine(date, datetime.time(23, 59, 59, 999))
+        )
         # start = tz.get_current_timezone().localize(start)
         # end = tz.get_current_timezone().localize(end)
         qs = self.filter(start__range=(start, end))

aircox/models/schedule.py

@@ -1,6 +1,6 @@
 import calendar
+import zoneinfo
 
-import pytz
 from django.db import models
 from django.utils import timezone as tz
 from django.utils.functional import cached_property
@@ -49,9 +49,9 @@ class Schedule(Rerun):
     )
     timezone = models.CharField(
         _("timezone"),
-        default=lambda: tz.get_current_timezone().zone,
+        default=lambda: tz.get_current_timezone().key,
         max_length=100,
-        choices=[(x, x) for x in pytz.all_timezones],
+        choices=[(x, x) for x in zoneinfo.available_timezones()],
         help_text=_("timezone used for the date"),
     )
     duration = models.TimeField(
@@ -82,9 +82,7 @@ class Schedule(Rerun):
     @cached_property
     def tz(self):
         """Pytz timezone of the schedule."""
-        import pytz
+        return zoneinfo.ZoneInfo(self.timezone)
-
-        return pytz.timezone(self.timezone)
 
     @cached_property
     def start(self):
@@ -110,7 +108,7 @@ class Schedule(Rerun):
         """Return a datetime set to schedule's time for the provided date,
         handling timezone (based on schedule's timezone)."""
         date = tz.datetime.combine(date, self.time)
-        return self.tz.normalize(self.tz.localize(date))
+        return date.replace(tzinfo=self.tz)
 
     def dates_of_month(self, date):
         """Return normalized diffusion dates of provided date's month."""

aircox/models/signals.py

@@ -53,8 +53,10 @@ def page_post_save(sender, instance, created, *args, **kwargs):
 def program_post_save(sender, instance, created, *args, **kwargs):
     """Clean-up later diffusions when a program becomes inactive."""
     if not instance.active:
-        Diffusion.object.program(instance).after(tz.now()).delete()
+        Diffusion.objects.program(instance).after(tz.now()).delete()
-        Episode.object.parent(instance).filter(diffusion__isnull=True).delete()
+        Episode.objects.parent(instance).filter(
+            diffusion__isnull=True
+        ).delete()
 
     cover = getattr(instance, "__initial_cover", None)
     if cover is None and instance.cover is not None:

aircox/models/station.py

@@ -60,7 +60,7 @@ class Station(models.Model):
         max_length=512,
         null=True,
         blank=True,
-        help_text=_("specify one url per line"),
+        help_text=_("specify one domain per line, without 'http://' prefix"),
     )
     audio_streams = models.TextField(
         _("audio streams"),

aircox/test.py

@@ -121,7 +121,7 @@ class SpoofMixin:
         if funcs is not None:
             self.funcs = funcs
 
-    def get_trace(self, name, args=False, kw=False):
+    def get_trace(self, name="__call__", args=False, kw=False):
         """Get a function call parameters.
 
         :param str name: function name
@@ -137,7 +137,7 @@ class SpoofMixin:
             raise ValueError(f"{name} called multiple times.")
         return self._get_trace(trace, args=args, kw=kw)
 
-    def get_traces(self, name, args=False, kw=False):
+    def get_traces(self, name="__call__", args=False, kw=False):
         """Get a tuple of all call parameters.
 
         Parameters are the same as `get()`.

aircox/tests/admin/test_filters.py

@@ -0,0 +1,62 @@
+from datetime import date, timedelta
+
+from django.contrib.admin import filters as d_filters
+from django.utils.translation import gettext_lazy as _
+import pytest
+
+from ..conftest import req_factory
+from aircox import models
+from aircox.admin import filters
+
+
+class FakeFilter(d_filters.FieldListFilter):
+    def __init__(self, field, request, params, model, model_admin, field_path):
+        self.field = field
+        self.request = request
+        self.params = params
+        self.model = model
+        self.model_admin = model_admin
+        self.field_path = field_path
+
+
+class DateFieldFilter(filters.DateFieldFilter, FakeFilter):
+    pass
+
+
+today = date.today()
+tomorrow = date.today() + timedelta(days=1)
+
+
+@pytest.fixture
+def req():
+    return req_factory.get("/test", {"pub_date__gte": today})
+
+
+@pytest.fixture
+def date_filter(req):
+    return DateFieldFilter(
+        models.Page._meta.get_field("pub_date"),
+        req,
+        {"pub_date__lte": tomorrow, "other_param": 13},
+        models.Page,
+        None,
+        "pub_date",
+    )
+
+
+class TestDateFieldFilter:
+    def test___init__(self, date_filter):
+        assert date_filter.date_params == {"pub_date__lte": tomorrow}
+
+        date_filter.links = [
+            (str(link[0]), *list(link[1:])) for link in date_filter.links
+        ]
+        assert date_filter.links == [
+            (str(_("None")), "pub_date__isnull", None, "1"),
+            (str(_("Exact")), "pub_date__date", date_filter.input_type),
+            (str(_("Since")), "pub_date__gte", date_filter.input_type),
+            (str(_("Until")), "pub_date__lte", date_filter.input_type),
+        ]
+        assert date_filter.query_attrs == {
+            "pub_date__gte": today.strftime("%Y-%m-%d")
+        }

aircox/tests/conftest.py

@@ -2,6 +2,10 @@ from datetime import time, timedelta
 import itertools
 import logging
 
+from django.conf import settings
+from django.contrib.auth.models import User
+from django.test import RequestFactory
+
 import pytest
 from model_bakery import baker
 
@@ -9,6 +13,21 @@ from aircox import models
 from aircox.test import Interface
 
 
+req_factory = RequestFactory()
+"""Request Factory used among different tests."""
+
+
+settings.ALLOWED_HOSTS = list(settings.ALLOWED_HOSTS) + [
+    "sub.server.org",
+    "server.org",
+]
+
+
+@pytest.fixture
+def staff_user():
+    return baker.make(User, is_active=True, is_staff=True)
+
+
 @pytest.fixture
 def logger():
     logger = Interface(
@@ -18,8 +37,24 @@ def logger():
 
 
 @pytest.fixture
-def stations():
+def station():
-    return baker.make(models.Station, _quantity=2)
+    return baker.make(
+        models.Station,
+        hosts="server.org",
+        audio_streams="stream 1\nstream 2",
+        default=True,
+    )
+
+
+@pytest.fixture
+def sub_station():
+    """Non-default station."""
+    return baker.make(models.Station, hosts="sub.server.org", default=False)
+
+
+@pytest.fixture
+def stations(station, sub_station):
+    return [station, sub_station]
 
 
 @pytest.fixture
@@ -28,7 +63,11 @@ def programs(stations):
         itertools.chain(
             *(
                 baker.make(
-                    models.Program, station=station, cover=None, _quantity=2
+                    models.Program,
+                    station=station,
+                    cover=None,
+                    status=models.Program.STATUS_PUBLISHED,
+                    _quantity=2,
                 )
                 for station in stations
             )

aircox/tests/models/test_schedule.py

@@ -24,7 +24,7 @@ class TestSchedule:
     @pytest.mark.django_db
     def test_tz(self, schedules):
         for schedule in schedules:
-            assert schedule.timezone == schedule.tz.zone
+            assert schedule.timezone == schedule.tz.key
 
     @pytest.mark.django_db
     def test_start(self, schedules):
@@ -45,7 +45,7 @@ class TestSchedule:
     def test_normalize(self, schedules):
         for schedule in schedules:
             dt = datetime.combine(schedule.date, schedule.time)
-            assert schedule.normalize(dt).tzinfo.zone == schedule.timezone
+            assert schedule.normalize(dt).tzinfo.key == schedule.timezone
 
     @pytest.mark.django_db
     def test_dates_of_month_ponctual(self):
@@ -117,7 +117,7 @@ class TestSchedule:
         assert dt.month == at.month
         assert dt.weekday() == schedule.date.weekday()
         assert dt.time() == schedule.time
-        assert dt.tzinfo.zone == schedule.timezone
+        assert dt.tzinfo.key == schedule.timezone
 
     @pytest.mark.django_db
     def test_diffusions_of_month(self, sched_initials):

aircox/tests/test_admin_site.py

@@ -0,0 +1,45 @@
+from django.urls import path, reverse
+from django.utils.translation import gettext_lazy as _
+
+import pytest
+
+from aircox import admin_site, urls as _urls
+from .conftest import req_factory
+
+
+# Just for code quality: urls module is required because we need some
+# url resolvers to be registered in order to run tests.
+_urls
+
+
+@pytest.fixture
+def site():
+    return admin_site.AdminSite()
+
+
+class TestAdminSite:
+    @pytest.mark.django_db
+    def test_each_context(self, site, staff_user):
+        req = req_factory.get("admin/test")
+        req.user = staff_user
+        context = site.each_context(req)
+        assert "programs" in context
+        assert "diffusions" in context
+        assert "comments" in context
+
+    def test_get_urls(self, site):
+        extra_url = path("test/path", lambda *_, **kw: _)
+        site.extra_urls.append(extra_url)
+        urls = site.get_urls()
+        assert extra_url in urls
+
+    def test_get_tools(self, site):
+        tools = site.get_tools()
+        tools = dict(tools)
+        assert tools == {
+            _("Statistics"): reverse("admin:tools-stats"),
+        }
+
+    def test_route_view(self, site):
+        # TODO
+        pass

aircox/tests/test_converters.py

@@ -0,0 +1,54 @@
+from datetime import date
+from django.utils.safestring import SafeString
+
+import pytest
+
+from aircox import converters
+
+
+@pytest.fixture
+def page_path_conv():
+    return converters.PagePathConverter()
+
+
+@pytest.fixture
+def week_conv():
+    return converters.WeekConverter()
+
+
+@pytest.fixture
+def date_conv():
+    return converters.DateConverter()
+
+
+class TestPagePathConverter:
+    def test_to_python(self, page_path_conv):
+        val = "path_value"
+        result = page_path_conv.to_python(val)
+        assert result == "/" + val + "/"
+
+    def test_to_url(self, page_path_conv):
+        val = "/val"
+        result = page_path_conv.to_url(val)
+        assert isinstance(result, SafeString)
+        assert result == val[1:] + "/"
+
+
+class TestWeekConverter:
+    def test_to_python(self, week_conv):
+        val = "2023/02"
+        assert week_conv.to_python(val) == date(2023, 1, 9)
+
+    def test_to_url(self, week_conv):
+        val = date(2023, 1, 10)
+        assert week_conv.to_url(val) == "2023/02"
+
+
+class TestDateConverter:
+    def test_to_python(self, date_conv):
+        val = "2023/02/05"
+        assert date_conv.to_python(val) == date(2023, 2, 5)
+
+    def test_to_url(self, date_conv):
+        val = date(2023, 5, 10)
+        assert date_conv.to_url(val) == "2023/05/10"

aircox/utils.py

@@ -4,7 +4,6 @@ import django.utils.timezone as tz
 __all__ = (
     "Redirect",
     "redirect",
-    "date_range",
     "cast_date",
     "date_or_default",
     "to_timedelta",
@@ -29,31 +28,17 @@ def redirect(url):
 
 
 def str_to_date(value, sep="/"):
-    """Return a date from the provided `value` string, formated as "yyyy/mm/dd"
+    """Return a date from the provided `value` string, formated as
-    (or "dd/mm/yyyy" if `reverse` is True).
+    "yyyy/mm/dd".
 
-    Raises ValueError for incorrect value format.
+    :raises: ValueError for incorrect value format.
     """
     value = value.split(sep)[:3]
     if len(value) < 3:
-        return ValueError("incorrect date format")
+        raise ValueError("incorrect date format")
     return datetime.date(int(value[0]), int(value[1]), int(value[2]))
 
 
-def date_range(date, delta=None, **delta_kwargs):
-    """Return a range of provided date such as `[date-delta, date+delta]`.
-
-    :param date: the reference date
-    :param delta: timedelta
-    :param **delta_kwargs: timedelta init arguments
-
-    Return a datetime range for a given day, as:
-    ```(date, 0:0:0:0; date, 23:59:59:999)```.
-    """
-    delta = tz.timedelta(**delta_kwargs) if delta is None else delta
-    return [date - delta, date + delta]
-
-
 def cast_date(date, into=datetime.date):
     """Cast a given date into the provided class' instance.
 

aircox/views/base.py

@@ -48,9 +48,7 @@ class BaseView(TemplateResponseMixin, ContextMixin):
                 kwargs["sidebar_list_url"] = self.get_sidebar_url()
 
         if "audio_streams" not in kwargs:
-            streams = self.station.audio_streams
+            kwargs["audio_streams"] = self.station.streams
-            streams = streams and streams.split("\n")
-            kwargs["audio_streams"] = streams
 
         if "model" not in kwargs:
             model = (
@@ -63,7 +61,7 @@ class BaseView(TemplateResponseMixin, ContextMixin):
         return super().get_context_data(**kwargs)
 
 
-# FIXME: rename to sth like [Base]?StationAPIView
+# FIXME: rename to sth like [Base]?StationAPIView/Mixin
 class BaseAPIView:
     @property
     def station(self):

aircox/views/mixins.py

@@ -91,6 +91,8 @@ class AttachedToMixin:
         return super().get_page()
 
 
+# FIXME: django-filter provides filter mixin, but I don't remember why this is
+# used.
 class FiltersMixin:
     """Mixin integrating Django filters' filter set."""
 

aircox_streamer/management/commands/streamer.py

@@ -6,11 +6,11 @@ to:
 - cancels Diffusions that have an archive but could not have been played;
 - run Liquidsoap
 """
+from datetime import timezone
 import time
 
 from argparse import RawTextHelpFormatter
 
-import pytz
 from django.core.management.base import BaseCommand
 from django.utils import timezone as tz
 
@@ -19,7 +19,7 @@ from aircox_streamer.controllers import Monitor, Streamer
 
 
 # force using UTC
-tz.activate(pytz.UTC)
+tz.activate(timezone.UTC)
 
 
 class Command(BaseCommand):

instance/settings/base.py

@@ -1,7 +1,7 @@
 import os
 import sys
+from zoneinfo import ZoneInfo
 
-import pytz
 from django.utils import timezone
 
 sys.path.insert(1, os.path.dirname(os.path.realpath(__file__)))
@@ -65,7 +65,7 @@ USE_I18N = True
 USE_L10N = True
 USE_TZ = True
 
-timezone.activate(pytz.timezone(TIME_ZONE))
+timezone.activate(ZoneInfo(TIME_ZONE))
 
 try:
     import locale

instance/settings/sample.py

@@ -10,6 +10,7 @@ For Django settings see:
     https://docs.djangoproject.com/en/3.1/topics/settings/
     https://docs.djangoproject.com/en/3.1/ref/settings/
 """
+from zoneinfo import ZoneInfo
 from .prod import *
 
 # FOR dev: from .dev import *
@@ -20,7 +21,7 @@ LANGUAGE_CODE = "fr-BE"
 LC_LOCALE = "fr_BE.UTF-8"
 TIME_ZONE = "Europe/Brussels"
 
-timezone.activate(pytz.timezone(TIME_ZONE))
+timezone.activate(ZoneInfo(TIME_ZONE))
 
 # Secret key: you MUST put a consistent secret key. You can generate one
 # at https://djecrety.ir/

notes.md

@@ -76,6 +76,6 @@ cms:
 
 # For the next version:
 ## Refactorisation
-Move:
+- move into `aircox_streamer`: `Log`, `Port`
-- into `aircox_streamer`: `Log`, `Port`
+- move into `aircox_cms`: `Page`, `NavItem`, `Category`, `StaticPage`, etc.
-- into `aircox_cms`: `Page`, `NavItem`, `Category`, `StaticPage`, etc.
+- use TextChoice and IntegerChoices in models fields enums