NHSDigital
diff --git a/‎src/eligibility_signposting_api/common/date_util.py‎
Lines changed: 57 additions & 0 deletions b/‎src/eligibility_signposting_api/common/date_util.py‎
Lines changed: 57 additions & 0 deletions
diff --git a/‎src/eligibility_signposting_api/model/campaign_config.py‎
Lines changed: 38 additions & 52 deletions b/‎src/eligibility_signposting_api/model/campaign_config.py‎
Lines changed: 38 additions & 52 deletions
diff --git a/‎src/eligibility_signposting_api/services/processors/campaign_evaluator.py‎
Lines changed: 6 additions & 6 deletions b/‎src/eligibility_signposting_api/services/processors/campaign_evaluator.py‎
Lines changed: 6 additions & 6 deletions
diff --git a/‎src/rules_validation_api/app.py‎
Lines changed: 7 additions & 4 deletions b/‎src/rules_validation_api/app.py‎
Lines changed: 7 additions & 4 deletions
diff --git a/‎src/rules_validation_api/validators/campaign_config_validator.py‎
Lines changed: 3 additions & 3 deletions b/‎src/rules_validation_api/validators/campaign_config_validator.py‎
Lines changed: 3 additions & 3 deletions
@@ -0,0 +1,57 @@
+import re
+from collections.abc import Callable
+from datetime import date, datetime, time
+from zoneinfo import ZoneInfo
+
+UK_TIMEZONE = ZoneInfo("Europe/London")
+
+
+def now_uk() -> datetime:
+    return datetime.now(tz=UK_TIMEZONE)
+
+
+def date_with_uk_timezone(parsed_date: date) -> date:
+    return datetime.combine(parsed_date, time.min).replace(tzinfo=UK_TIMEZONE).date()
+
+
+def datetime_with_uk_timezone(parsed_date_time: datetime) -> datetime:
+    return parsed_date_time.replace(tzinfo=UK_TIMEZONE)
+
+
+def _parse_with_format[T](
+    value: str,
+    regex: str,
+    fmt: str,
+    error_info: tuple[str, str],
+    transform: Callable[[datetime], T],
+) -> T:
+    """Shared logic for regex validation and datetime parsing."""
+    label, expected_format = error_info
+
+    if not re.fullmatch(regex, value):
+        msg = f"Invalid format: {value}. Must be {expected_format}."
+        raise ValueError(msg)
+    try:
+        dt = datetime.strptime(value, fmt)  # noqa: DTZ007
+        return transform(dt)
+    except ValueError as err:
+        msg = f"Invalid {label} value: {value}."
+        raise ValueError(msg) from err
+
+
+def parse_date_yyyymmdd(v: str | date) -> date:
+    if isinstance(v, date):
+        return v
+    # Pass the last two strings as a single tuple inside parentheses
+    return _parse_with_format(str(v), r"\d{8}", "%Y%m%d", ("date", "YYYYMMDD"), lambda dt: dt.date())
+
+
+def parse_time_hhmmss(v: str | time | None) -> time | None:
+    if v is None:
+        return None
+    if isinstance(v, time):
+        return v
+    # Pass the last two strings as a single tuple inside parentheses
+    return _parse_with_format(
+        str(v).strip(), r"^\d{2}:\d{2}:\d{2}$", "%H:%M:%S", ("time", "HH:MM:SS"), lambda dt: dt.time()
+    )
@@ -1,10 +1,9 @@
 from __future__ import annotations
 
 import json
-import re
 import typing
 from collections import Counter
-from datetime import UTC, date, datetime, time
+from datetime import date, datetime, time
 from enum import StrEnum
 from functools import cached_property
 from operator import attrgetter
@@ -21,11 +20,19 @@
     model_validator,
 )
 
+from eligibility_signposting_api.common.date_util import (
+    UK_TIMEZONE,
+    datetime_with_uk_timezone,
+    now_uk,
+    parse_date_yyyymmdd,
+    parse_time_hhmmss,
+)
 from eligibility_signposting_api.config.constants import ALLOWED_CONDITIONS, RULE_STOP_DEFAULT
 
 if typing.TYPE_CHECKING:  # pragma: no cover
     from pydantic import SerializationInfo
 
+
 CampaignName = NewType("CampaignName", str)
 CampaignVersion = NewType("CampaignVersion", int)
 CampaignID = NewType("CampaignID", str)
@@ -51,38 +58,6 @@
 RuleText = NewType("RuleText", str)
 
 
-class DateUtil:
-    @staticmethod
-    def parse_date_yyyymmdd(v: str | date) -> date:
-        if isinstance(v, date):
-            return v
-        v_str = str(v)
-        if not re.fullmatch(r"\d{8}", v_str):
-            msg = f"Invalid format: {v_str}. Must be YYYYMMDD."
-            raise ValueError(msg)
-        try:
-            return datetime.strptime(v_str, "%Y%m%d").date()  # noqa: DTZ007
-        except ValueError as err:
-            msg = f"Invalid date value: {v_str}."
-            raise ValueError(msg) from err
-
-    @staticmethod
-    def parse_time_hhmmss(v: str | time | None) -> time | None:
-        if not v:
-            return None
-        if isinstance(v, time):
-            return v
-        v_str = str(v).strip()
-        if re.fullmatch(r"^\d{2}:\d{2}:\d{2}$", v_str):
-            try:
-                return datetime.strptime(v_str, "%H:%M:%S").time()  # noqa: DTZ007
-            except ValueError as err:
-                msg = f"Invalid time value: {v_str}."
-                raise ValueError(msg) from err
-        msg = f"Invalid format: {v_str}. Must be HH:MM:SS."
-        raise ValueError(msg)
-
-
 class RuleType(StrEnum):
     filter = "F"
     suppression = "S"
@@ -294,8 +269,12 @@ class Iteration(BaseModel):
     id: IterationID = Field(..., alias="ID")
     version: IterationVersion = Field(..., alias="Version")
     name: IterationName = Field(..., alias="Name")
-    iteration_date: IterationDate = Field(..., alias="IterationDate")
-    iteration_time: IterationTime | None = Field(default=None, alias="IterationTime")
+    iteration_date: IterationDate = Field(
+        ..., alias="IterationDate", description="Iteration start date in Europe/London time Zone"
+    )
+    iteration_time: IterationTime | None = Field(
+        default=None, alias="IterationTime", description="Iteration start time in Europe/London time Zone"
+    )
     iteration_number: int | None = Field(None, alias="IterationNumber")
     approval_minimum: int | None = Field(None, alias="ApprovalMinimum")
     approval_maximum: int | None = Field(None, alias="ApprovalMaximum")
@@ -319,13 +298,14 @@ def _link_parent_to_iteration_rules(self) -> typing.Self:
 
     @field_validator("iteration_date", mode="before")
     @classmethod
-    def parse_dates(cls, v: str | date) -> date:
-        return DateUtil.parse_date_yyyymmdd(v)
+    def parse_dates_as_uk_local(cls, v: str | date) -> date:
+        parsed_date = parse_date_yyyymmdd(v)
+        return datetime.combine(parsed_date, time.min).replace(tzinfo=UK_TIMEZONE).date()
 
     @field_validator("iteration_time", mode="before")
     @classmethod
-    def parse_times(cls, v: str | time) -> time | None:
-        return DateUtil.parse_time_hhmmss(v)
+    def parse_times_as_uk_local(cls, v: str | time) -> time | None:
+        return parse_time_hhmmss(v)
 
     @field_serializer("iteration_date", when_used="always")
     @staticmethod
@@ -344,6 +324,9 @@ def set_parent(self, parent: CampaignConfig) -> None:
 
     @cached_property
     def iteration_datetime(self) -> datetime:
+        """iteration_datetime is the datetime of the iteration,
+        including the iteration_time if set, otherwise the parent's iteration_time.
+        the return type is datetime in Europe/London time zone."""
         if self.iteration_time:
             iteration_time = self.iteration_time
         elif self._parent:
@@ -352,7 +335,7 @@ def iteration_datetime(self) -> datetime:
             msg = f"No iteration_time and no parent linked for iteration {self.id}"
             raise ValueError(msg)
 
-        return datetime.combine(self.iteration_date, iteration_time).replace(tzinfo=UTC)
+        return datetime_with_uk_timezone(datetime.combine(self.iteration_date, iteration_time))
 
     def __str__(self) -> str:
         return json.dumps(self.model_dump(by_alias=True), indent=2)
@@ -369,10 +352,14 @@ class CampaignConfig(BaseModel):
     reviewer: list[str] | None = Field(None, alias="Reviewer")
     iteration_frequency: Literal["X", "D", "W", "M", "Q", "A"] = Field(..., alias="IterationFrequency")
     iteration_type: Literal["A", "M", "S", "O"] = Field(..., alias="IterationType")
-    iteration_time: IterationTime = Field(default=IterationTime(time(0, 0, 0)), alias="IterationTime")
+    iteration_time: IterationTime = Field(
+        default=IterationTime(time(0, 0, 0)),
+        alias="IterationTime",
+        description="Default Iteration start time in Europe/London time Zone",
+    )
     default_comms_routing: str | None = Field(None, alias="DefaultCommsRouting")
-    start_date: StartDate = Field(..., alias="StartDate")
-    end_date: EndDate = Field(..., alias="EndDate")
+    start_date: StartDate = Field(..., alias="StartDate", description="Campaign start date in Europe/London time Zone")
+    end_date: EndDate = Field(..., alias="EndDate", description="Campaign end date in Europe/London time Zone")
     approval_minimum: int | None = Field(None, alias="ApprovalMinimum")
     approval_maximum: int | None = Field(None, alias="ApprovalMaximum")
     iterations: list[Iteration] = Field(..., min_length=1, alias="Iterations")
@@ -388,13 +375,14 @@ def _link_parent_to_iterations(self) -> typing.Self:
 
     @field_validator("start_date", "end_date", mode="before")
     @classmethod
-    def parse_dates(cls, v: str | date) -> date:
-        return DateUtil.parse_date_yyyymmdd(v)
+    def parse_dates_as_uk_local(cls, v: str | date) -> date:
+        parsed_date = parse_date_yyyymmdd(v)
+        return datetime.combine(parsed_date, time.min).replace(tzinfo=UK_TIMEZONE).date()
 
     @field_validator("iteration_time", mode="before")
     @classmethod
-    def parse_times(cls, v: str | time) -> time | None:
-        return DateUtil.parse_time_hhmmss(v)
+    def parse_times_as_uk_local(cls, v: str | time) -> time | None:
+        return parse_time_hhmmss(v)
 
     @field_serializer("start_date", "end_date", when_used="always")
     @staticmethod
@@ -435,14 +423,12 @@ def check_no_overlapping_iterations(self) -> typing.Self:
 
     @cached_property
     def campaign_live(self) -> bool:
-        today = datetime.now(tz=UTC).date()
-        return self.start_date <= today <= self.end_date
+        return self.start_date <= now_uk().date() <= self.end_date
 
     @cached_property
     def current_iteration(self) -> Iteration:
-        now = datetime.now(tz=UTC)
         iterations_by_date_descending = sorted(self.iterations, key=attrgetter("iteration_datetime"), reverse=True)
-        return next(i for i in iterations_by_date_descending if i.iteration_datetime <= now)
+        return next(i for i in iterations_by_date_descending if i.iteration_datetime <= now_uk())
 
     def __str__(self) -> str:
         return json.dumps(self.model_dump(by_alias=True), indent=2)
 
@@ -42,19 +42,19 @@ def get_campaign_with_latest_iteration(self, active_campaigns: list[CampaignConf
             latest_campaign = None
         else:
             max_date_time = max(item[0] for item in valid_items)
-            cc_with_max_iteration_date: list[CampaignConfig] = [
+            cc_with_max_iteration_date_time: list[CampaignConfig] = [
                 item[1] for item in valid_items if item[0] == max_date_time
             ]
-            if len(cc_with_max_iteration_date) > 1:
+            if len(cc_with_max_iteration_date_time) > 1:
                 err_msg = (
-                    f"Ambiguous result: '{len(cc_with_max_iteration_date)}' active iterations "
-                    f"for target {cc_with_max_iteration_date[0].target} "
+                    f"Ambiguous result: '{len(cc_with_max_iteration_date_time)}' active iterations "
+                    f"for target {cc_with_max_iteration_date_time[0].target} "
                     f"found for datetime '{max_date_time}' "
-                    f"across campaign(s) {[cc.id for cc in cc_with_max_iteration_date]}"
+                    f"across campaign(s) {[cc.id for cc in cc_with_max_iteration_date_time]}"
                 )
                 raise ValueError(err_msg)
 
-            latest_campaign = cc_with_max_iteration_date[0]
+            latest_campaign = cc_with_max_iteration_date_time[0]
 
         return latest_campaign
 
 
@@ -3,9 +3,10 @@
 import logging
 import sys
 from collections import defaultdict
-from datetime import UTC, datetime
+from datetime import datetime
 from operator import attrgetter
 from pathlib import Path
+from zoneinfo import ZoneInfo
 
 from pydantic import ValidationError
 
@@ -24,6 +25,8 @@
 RED = "\033[91m"
 BLUE = "\033[34m"
 
+UK_TIMEZONE = ZoneInfo("Europe/London")
+
 
 def refine_error(e: ValidationError) -> str:
     """Return a very short, single-line error message."""
@@ -79,10 +82,10 @@ def display_current_iteration(result: RulesValidation) -> None:
     config = result.campaign_config
     iterations = config.iterations
     is_campaign_live = config.campaign_live
-    today = datetime.now(tz=UTC).date()
+    today_uk = datetime.now(tz=UK_TIMEZONE).date()
 
     no_of_iterations = len(iterations)
-    is_campaign_expired = config.end_date < today
+    is_campaign_expired = config.end_date < today_uk
 
     # ---- Current Iteration ----
     if is_campaign_live:
@@ -114,7 +117,7 @@ def display_current_iteration(result: RulesValidation) -> None:
         sorted_iterations = sorted(iterations, key=attrgetter("iteration_date"))
 
         try:
-            next_iteration = next((i for i in sorted_iterations if i.iteration_date > today), None)
+            next_iteration = next((i for i in sorted_iterations if i.iteration_date > today_uk), None)
 
             if next_iteration:
                 sys.stdout.write(
 
@@ -46,15 +46,15 @@ def validate_iterations_have_unique_number(self) -> typing.Self:
     @model_validator(mode="after")
     def validate_campaign_has_iteration_within_schedule(self) -> typing.Self:
         errors: list[str] = []
-        iterations_by_date = sorted(self.iterations, key=attrgetter("iteration_date"))
+        iterations_by_date = sorted(self.iterations, key=attrgetter("iteration_datetime"))
 
         for idx, iteration in enumerate(iterations_by_date):
-            if iteration.iteration_date < self.start_date:
+            if iteration.iteration_datetime.date() < self.start_date:
                 errors.append(
                     f"CampaignConfig.Iterations.{idx}.IterationDate : "
                     f"Starts before campaign start date {self.start_date} [type=invalid]"
                 )
-            if iteration.iteration_date > self.end_date:
+            if iteration.iteration_datetime.date() > self.end_date:
                 errors.append(
                     f"CampaignConfig.Iterations.{idx}.IterationDate : "
                     f"Starts after campaign end date {self.end_date} [type=invalid]"