Allow specifying multiple offset days to create additional events

README.md

@@ -8,7 +8,7 @@ Individual calendars can be protected by basic authentication if required (value
 
 Calendars are served based on their [name](#configuration), at `/{name}.ics`.
 
-Calendars which allow custom offsets (`allow_custom_offset = true`) can add `?offset_days=3` to customize the offset. Events can only be offset ±10 years. When not configured, this parameter is ignored.
+Additional events cen be added at offsets from the original date using the `offset_days` configuration. Events can only be offset ±10 years.
 
 ### Static
 

calmerge/calendars.py

@@ -36,16 +36,38 @@ async def fetch_merged_calendar(calendar_config: CalendarConfig) -> icalendar.Ca
     return merged_calendar
 
 
-def offset_calendar(calendar: icalendar.Calendar, offset_days: int) -> None:
+def shift_event_by_offset(event: icalendar.cal.Component, offset: timedelta) -> None:
     """
-    Mutate a calendar and move events by a given offset
+    Mutate a calendar event and shift its dates to a given offset
     """
-    offset = timedelta(days=offset_days)
+    if "DTSTART" in event:
+        event["DTSTART"].dt += offset
+    if "DTEND" in event:
+        event["DTEND"].dt += offset
+    if "DTSTAMP" in event:
+        event["DTSTAMP"].dt += offset
+
+
+def create_offset_calendar_events(
+    calendar: icalendar.Calendar, duplicate_days: list[int]
+) -> None:
+    """
+    Mutate a calendar and add additional events at given offsets
+    """
+    new_components = []
 
     for component in calendar.walk():
-        if "DTSTART" in component:
-            component["DTSTART"].dt += offset
-        if "DTEND" in component:
-            component["DTEND"].dt += offset
-        if "DTSTAMP" in component:
-            component["DTSTAMP"].dt += offset
+        for days in duplicate_days:
+            day_component = component.copy()
+
+            shift_event_by_offset(day_component, timedelta(days=days))
+
+            if "SUMMARY" in day_component:
+                day_component["SUMMARY"] += (
+                    f" ({days} days {'after' if days > 0 else 'before'})"
+                )
+
+            new_components.append(day_component)
+
+    for component in new_components:
+        calendar.add_component(component)

calmerge/config.py

@@ -36,7 +36,7 @@ def validate_header(self, auth_header: str) -> bool:
 class CalendarConfig(BaseModel):
     name: str
     urls: list[HttpUrl]
-    offset_days: int = Field(default=0, le=MAX_OFFSET, ge=-MAX_OFFSET)
+    offset_days: list[int] = Field(default_factory=list)
     auth: AuthConfig | None = None
 
     @field_validator("urls")
@@ -51,6 +51,28 @@ def check_urls_unique(cls, urls: list[HttpUrl]) -> list[HttpUrl]:
     def expand_url_vars(cls, urls: list[str]) -> list[str]:
         return [expandvars(url) for url in urls]
 
+    @field_validator("offset_days")
+    @classmethod
+    def validate_offset_days(cls, offset_days: list[int]) -> list[int]:
+        if len(set(offset_days)) != len(offset_days):
+            raise PydanticCustomError(
+                "unique_offset_days", "Offset days must be unique"
+            )
+
+        if any(day == 0 for day in offset_days):
+            raise PydanticCustomError(
+                "zero_offset_days",
+                "Offset days must not be zero",
+            )
+
+        if not all(-MAX_OFFSET <= day <= MAX_OFFSET for day in offset_days):
+            raise PydanticCustomError(
+                "offset_days_range",
+                f"Offset days must be between -{MAX_OFFSET} and {MAX_OFFSET}",
+            )
+
+        return offset_days
+
 
 class Config(BaseModel):
     calendars: list[CalendarConfig] = Field(alias="calendar", default_factory=list)

calmerge/static.py

@@ -1,14 +1,14 @@
 import asyncio
 from pathlib import Path
 
-from .calendars import fetch_merged_calendar, offset_calendar
+from .calendars import create_offset_calendar_events, fetch_merged_calendar
 from .config import CalendarConfig
 
 
 def write_calendar(calendar_config: CalendarConfig, output_file: Path) -> None:
     merged_calendar = asyncio.run(fetch_merged_calendar(calendar_config))
 
     if offset_days := calendar_config.offset_days:
-        offset_calendar(merged_calendar, offset_days)
+        create_offset_calendar_events(merged_calendar, offset_days)
 
     output_file.write_bytes(merged_calendar.to_ical())

calmerge/views.py

@@ -1,8 +1,6 @@
 from aiohttp import web
 
-from .calendars import fetch_merged_calendar, offset_calendar
-from .config import MAX_OFFSET
-from .utils import try_parse_int
+from .calendars import create_offset_calendar_events, fetch_merged_calendar
 
 
 async def healthcheck(request: web.Request) -> web.Response:
@@ -24,19 +22,7 @@ async def calendar(request: web.Request) -> web.Response:
 
     calendar = await fetch_merged_calendar(calendar_config)
 
-    offset_days = calendar_config.offset_days
-
-    if custom_offset_days := request.query.get("offset_days", ""):
-        offset_days = try_parse_int(custom_offset_days)
-
-        if offset_days is None:
-            raise web.HTTPBadRequest(reason="offset_days is invalid")
-        elif abs(offset_days) > MAX_OFFSET:
-            raise web.HTTPBadRequest(
-                reason=f"offset_days is too large (must be between -{MAX_OFFSET} and {MAX_OFFSET})"
-            )
-
-    if offset_days is not None:
-        offset_calendar(calendar, offset_days)
+    if (offset_days := calendar_config.offset_days):
+        create_offset_calendar_events(calendar, offset_days)
 
     return web.Response(body=calendar.to_ical())

tests/calendars.toml

@@ -9,7 +9,7 @@ name = "python-offset"
 urls = [
     "https://endoflife.date/calendar/python.ics",
 ]
-offset_days = 365
+offset_days = [365]
 
 [[calendar]]
 name = "python-authed"

tests/test_calendar_view.py

@@ -1,12 +1,9 @@
 from datetime import timedelta
 
 import icalendar
-import pytest
 from aiohttp import BasicAuth
 from aiohttp.test_utils import TestClient
 
-from calmerge.config import MAX_OFFSET
-
 
 async def test_retrieves_calendars(client: TestClient) -> None:
     response = await client.get("/python.ics")
@@ -54,16 +51,18 @@ async def test_offset_calendar_matches(client: TestClient) -> None:
     assert not offset_calendar.is_broken
     assert not original_calendar.is_broken
 
-    assert len(offset_calendar.walk("VEVENT")) > 1
-
-    assert len(offset_calendar.walk("VEVENT")) == len(original_calendar.walk("VEVENT"))
+    assert (
+        len(offset_calendar.walk("VEVENT")) == len(original_calendar.walk("VEVENT")) * 2
+    )
 
     original_events_by_summary = {
         event["SUMMARY"]: event for event in original_calendar.walk("VEVENT")
     }
 
     for offset_event in offset_calendar.walk("VEVENT"):
-        original_event = original_events_by_summary[offset_event["SUMMARY"]]
+        original_event = original_events_by_summary[
+            offset_event["SUMMARY"].removesuffix(" (365 days after)")
+        ]
 
         assert offset_event["dtstart"].dt == (
             original_event["dtstart"].dt + timedelta(days=365)
@@ -78,60 +77,3 @@ async def test_offset_calendar_matches(client: TestClient) -> None:
         )
 
         assert offset_event["description"] == original_event["description"]
-
-
-@pytest.mark.parametrize("offset", [100, -100, MAX_OFFSET, -MAX_OFFSET])
-async def test_custom_offset(client: TestClient, offset: int) -> None:
-    offset_response = await client.get(
-        "/python-offset.ics",
-        params={"offset_days": offset},
-    )
-    offset_calendar = icalendar.Calendar.from_ical(await offset_response.text())
-
-    original_response = await client.get("/python.ics")
-    original_calendar = icalendar.Calendar.from_ical(await original_response.text())
-
-    assert not offset_calendar.is_broken
-    assert not original_calendar.is_broken
-
-    original_events_by_summary = {
-        event["SUMMARY"]: event for event in original_calendar.walk("VEVENT")
-    }
-
-    delta = timedelta(days=offset)
-
-    for offset_event in offset_calendar.walk("VEVENT"):
-        original_event = original_events_by_summary[offset_event["SUMMARY"]]
-
-        assert offset_event["dtstart"].dt == (original_event["dtstart"].dt + delta)
-
-        assert offset_event["dtend"].dt == (original_event["dtend"].dt + delta)
-
-        assert offset_event["dtstamp"].dt == (original_event["dtstamp"].dt + delta)
-
-        assert offset_event["description"] == original_event["description"]
-
-
-@pytest.mark.parametrize("offset", [MAX_OFFSET + 1, -MAX_OFFSET - 1])
-async def test_out_of_bounds_custom_offset(client: TestClient, offset: int) -> None:
-    response = await client.get(
-        "/python-offset.ics",
-        params={"offset_days": offset},
-    )
-
-    assert response.status == 400
-    assert (
-        await response.text()
-        == f"400: offset_days is too large (must be between -{MAX_OFFSET} and {MAX_OFFSET})"
-    )
-
-
-@pytest.mark.parametrize("offset", ["invalid", "\0"])
-async def test_invalid_offset(client: TestClient, offset: str) -> None:
-    response = await client.get(
-        "/python-offset.ics",
-        params={"offset_days": offset},
-    )
-
-    assert response.status == 400
-    assert await response.text() == "400: offset_days is invalid"

tests/test_config.py

@@ -7,7 +7,7 @@
 from pydantic_core import Url
 from tomllib import TOMLDecodeError
 
-from calmerge.config import AuthConfig, CalendarConfig, Config
+from calmerge.config import MAX_OFFSET, AuthConfig, CalendarConfig, Config
 
 
 def test_non_unique_urls() -> None:
@@ -17,6 +17,31 @@ def test_non_unique_urls() -> None:
     assert e.value.errors()[0]["msg"] == "URLs must be unique"
 
 
+def test_non_unique_offset_days() -> None:
+    with pytest.raises(ValidationError) as e:
+        CalendarConfig(
+            name="test",
+            urls=["https://example.com"],  # type: ignore [list-item]
+            offset_days=[1, 2, 3, 2, 1],
+        )
+
+    assert e.value.errors()[0]["msg"] == "Offset days must be unique"
+
+
+def test_invalid_offset_days() -> None:
+    with pytest.raises(ValidationError) as e:
+        CalendarConfig(
+            name="test",
+            urls=["https://example.com"],  # type: ignore [list-item]
+            offset_days=[MAX_OFFSET + 1],
+        )
+
+    assert (
+        e.value.errors()[0]["msg"]
+        == f"Offset days must be between -{MAX_OFFSET} and {MAX_OFFSET}"
+    )
+
+
 def test_urls_expand_env_var(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("FOO", "BAR")