Eli 716/convert time values from utc to gmt bst

src/eligibility_signposting_api/common/date_util.py

@@ -0,0 +1,55 @@
+import re
+from collections.abc import Callable
+from datetime import date, datetime, time
+from zoneinfo import ZoneInfo
+
+
+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 convert_from_uk_to_utc(value: datetime | date) -> datetime:
+    if isinstance(value, date) and not isinstance(value, datetime):
+        value = datetime.combine(value, time.min)
+
+    uk = ZoneInfo("Europe/London")
+    utc = ZoneInfo("UTC")
+
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=uk)
+    return value.astimezone(utc)
+
+
+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()
+    )

src/eligibility_signposting_api/model/campaign_config.py

@@ -21,6 +21,7 @@
     model_validator,
 )
 
+from eligibility_signposting_api.common.date_util import convert_from_uk_to_utc, 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
@@ -294,8 +295,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")
@@ -320,12 +325,12 @@ 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)
+        return parse_date_yyyymmdd(v)
 
     @field_validator("iteration_time", mode="before")
     @classmethod
     def parse_times(cls, v: str | time) -> time | None:
-        return DateUtil.parse_time_hhmmss(v)
+        return parse_time_hhmmss(v)
 
     @field_serializer("iteration_date", when_used="always")
     @staticmethod
@@ -343,7 +348,7 @@ def set_parent(self, parent: CampaignConfig) -> None:
         self._parent = parent
 
     @cached_property
-    def iteration_datetime(self) -> datetime:
+    def iteration_datetime_utc(self) -> datetime:
         if self.iteration_time:
             iteration_time = self.iteration_time
         elif self._parent:
@@ -352,7 +357,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 convert_from_uk_to_utc(datetime.combine(self.iteration_date, iteration_time))
 
     def __str__(self) -> str:
         return json.dumps(self.model_dump(by_alias=True), indent=2)
@@ -369,32 +374,43 @@ 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")
 
     model_config = {"populate_by_name": True, "arbitrary_types_allowed": True, "extra": "ignore"}
 
-    @model_validator(mode="after")
-    def _link_parent_to_iterations(self) -> typing.Self:
+    def __init__(self, **data: dict[str, typing.Any]) -> None:
+        super().__init__(**data)
+        # Ensure each rule knows its parent iteration
         for iteration in self.iterations:
             iteration.set_parent(self)
 
-        return self
+    @cached_property
+    def start_date_utc(self) -> datetime:
+        return convert_from_uk_to_utc(datetime.combine(self.start_date, time.min))
+
+    @cached_property
+    def end_date_utc(self) -> datetime:
+        return convert_from_uk_to_utc(datetime.combine(self.end_date, time.min))
 
     @field_validator("start_date", "end_date", mode="before")
     @classmethod
     def parse_dates(cls, v: str | date) -> date:
-        return DateUtil.parse_date_yyyymmdd(v)
+        return parse_date_yyyymmdd(v)
 
     @field_validator("iteration_time", mode="before")
     @classmethod
     def parse_times(cls, v: str | time) -> time | None:
-        return DateUtil.parse_time_hhmmss(v)
+        return parse_time_hhmmss(v)
 
     @field_serializer("start_date", "end_date", when_used="always")
     @staticmethod
@@ -436,13 +452,14 @@ 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
+        today_midnight = datetime.combine(today, time.min, tzinfo=UTC)
+        return self.start_date_utc <= today_midnight <= self.end_date_utc
 
     @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)
+        iterations_by_date_descending = sorted(self.iterations, key=attrgetter("iteration_datetime_utc"), reverse=True)
+        return next(i for i in iterations_by_date_descending if i.iteration_datetime_utc <= now)
 
     def __str__(self) -> str:
         return json.dumps(self.model_dump(by_alias=True), indent=2)

src/eligibility_signposting_api/services/processors/campaign_evaluator.py

@@ -31,7 +31,7 @@ def get_campaign_with_latest_iteration(self, active_campaigns: list[CampaignConf
 
         for cc in active_campaigns:
             try:
-                valid_items.append((cc.current_iteration.iteration_datetime, cc))
+                valid_items.append((cc.current_iteration.iteration_datetime_utc, cc))
             except StopIteration:
                 logger.info(
                     "Skipping campaign ID %s as no active iteration was found.",
@@ -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
 

src/rules_validation_api/app.py

@@ -93,10 +93,10 @@ def display_current_iteration(result: RulesValidation) -> None:
                 sys.stdout.write(
                     f"{YELLOW}Current active Iteration Number: {RESET}{GREEN}{current.iteration_number}{RESET}\n"
                 )
-                tz = current.iteration_datetime.tzinfo
+                tz = current.iteration_datetime_utc.tzinfo
                 sys.stdout.write(
                     f"{YELLOW}Current active Iteration's date&time: "
-                    f"{RESET}{GREEN}{current.iteration_datetime} ({tz}){RESET}\n"
+                    f"{RESET}{GREEN}{current.iteration_datetime_utc} ({tz}){RESET}\n"
                 )
         except StopIteration:
             sys.stdout.write(f"{YELLOW}No active iteration could be determined{RESET}\n")
@@ -120,10 +120,10 @@ def display_current_iteration(result: RulesValidation) -> None:
                 sys.stdout.write(
                     f"{YELLOW}Next active Iteration Number: {RESET}{GREEN}{next_iteration.iteration_number}{RESET}\n"
                 )
-                tz = next_iteration.iteration_datetime.tzinfo
+                tz = next_iteration.iteration_datetime_utc.tzinfo
                 sys.stdout.write(
                     f"{YELLOW}Next active Iteration's date&time: "
-                    f"{RESET}{GREEN}{next_iteration.iteration_datetime} ({tz}){RESET}\n"
+                    f"{RESET}{GREEN}{next_iteration.iteration_datetime_utc} ({tz}){RESET}\n"
                 )
         except StopIteration:
             sys.stdout.write(f"{YELLOW}No next active iteration could be determined{RESET}\n")

src/rules_validation_api/validators/campaign_config_validator.py

@@ -46,15 +46,19 @@ 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"))
+        for iteration in self.iterations:
+            if iteration._parent is None:  # noqa : SLF001
+                iteration.set_parent(self)
+
+        iterations_by_date = sorted(self.iterations, key=attrgetter("iteration_datetime_utc"))
 
         for idx, iteration in enumerate(iterations_by_date):
-            if iteration.iteration_date < self.start_date:
+            if iteration.iteration_datetime_utc < self.start_date_utc:
                 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_utc > self.end_date_utc:
                 errors.append(
                     f"CampaignConfig.Iterations.{idx}.IterationDate : "
                     f"Starts after campaign end date {self.end_date} [type=invalid]"