diff --git a/api/controllers/console/app/conversation.py b/api/controllers/console/app/conversation.py index d5fa70d678..57b6c314f3 100644 --- a/api/controllers/console/app/conversation.py +++ b/api/controllers/console/app/conversation.py @@ -1,7 +1,5 @@ -from datetime import datetime - -import pytz import sqlalchemy as sa +from flask import abort from flask_restx import Resource, marshal_with, reqparse from flask_restx.inputs import int_range from sqlalchemy import func, or_ @@ -19,7 +17,7 @@ from fields.conversation_fields import ( conversation_pagination_fields, conversation_with_summary_pagination_fields, ) -from libs.datetime_utils import naive_utc_now +from libs.datetime_utils import naive_utc_now, parse_time_range from libs.helper import DatetimeString from libs.login import current_account_with_tenant, login_required from models import Conversation, EndUser, Message, MessageAnnotation @@ -90,25 +88,17 @@ class CompletionConversationApi(Resource): account = current_user assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) + if start_datetime_utc: query = query.where(Conversation.created_at >= start_datetime_utc) - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=59) - - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) - + if end_datetime_utc: + end_datetime_utc = end_datetime_utc.replace(second=59) query = query.where(Conversation.created_at < end_datetime_utc) # FIXME, the type ignore in this file @@ -270,29 +260,21 @@ class ChatConversationApi(Resource): account = current_user assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) + if start_datetime_utc: match args["sort_by"]: case "updated_at" | "-updated_at": query = query.where(Conversation.updated_at >= start_datetime_utc) case "created_at" | "-created_at" | _: query = query.where(Conversation.created_at >= start_datetime_utc) - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=59) - - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) - + if end_datetime_utc: + end_datetime_utc = end_datetime_utc.replace(second=59) match args["sort_by"]: case "updated_at" | "-updated_at": query = query.where(Conversation.updated_at <= end_datetime_utc) diff --git a/api/controllers/console/app/statistic.py b/api/controllers/console/app/statistic.py index 0917a6e53c..8c8253b786 100644 --- a/api/controllers/console/app/statistic.py +++ b/api/controllers/console/app/statistic.py @@ -1,9 +1,7 @@ -from datetime import datetime from decimal import Decimal -import pytz import sqlalchemy as sa -from flask import jsonify +from flask import abort, jsonify from flask_restx import Resource, fields, reqparse from controllers.console import api, console_ns @@ -11,6 +9,7 @@ from controllers.console.app.wraps import get_app_model from controllers.console.wraps import account_initialization_required, setup_required from core.app.entities.app_invoke_entities import InvokeFrom from extensions.ext_database import db +from libs.datetime_utils import parse_time_range from libs.helper import DatetimeString from libs.login import current_account_with_tenant, login_required from models import AppMode, Message @@ -56,26 +55,16 @@ WHERE arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER} assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) + if start_datetime_utc: sql_query += " AND created_at >= :start" arg_dict["start"] = start_datetime_utc - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) - + if end_datetime_utc: sql_query += " AND created_at < :end" arg_dict["end"] = end_datetime_utc @@ -120,8 +109,11 @@ class DailyConversationStatistic(Resource): ) args = parser.parse_args() assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc + + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) stmt = ( sa.select( @@ -134,18 +126,10 @@ class DailyConversationStatistic(Resource): .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER) ) - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + if start_datetime_utc: stmt = stmt.where(Message.created_at >= start_datetime_utc) - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) + if end_datetime_utc: stmt = stmt.where(Message.created_at < end_datetime_utc) stmt = stmt.group_by("date").order_by("date") @@ -198,26 +182,17 @@ WHERE AND invoke_from != :invoke_from""" arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER} assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) + if start_datetime_utc: sql_query += " AND created_at >= :start" arg_dict["start"] = start_datetime_utc - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) - + if end_datetime_utc: sql_query += " AND created_at < :end" arg_dict["end"] = end_datetime_utc @@ -273,26 +248,17 @@ WHERE AND invoke_from != :invoke_from""" arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER} assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) + if start_datetime_utc: sql_query += " AND created_at >= :start" arg_dict["start"] = start_datetime_utc - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) - + if end_datetime_utc: sql_query += " AND created_at < :end" arg_dict["end"] = end_datetime_utc @@ -357,26 +323,17 @@ FROM AND m.invoke_from != :invoke_from""" arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER} assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) + if start_datetime_utc: sql_query += " AND c.created_at >= :start" arg_dict["start"] = start_datetime_utc - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) - + if end_datetime_utc: sql_query += " AND c.created_at < :end" arg_dict["end"] = end_datetime_utc @@ -446,26 +403,17 @@ WHERE AND m.invoke_from != :invoke_from""" arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER} assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) + if start_datetime_utc: sql_query += " AND m.created_at >= :start" arg_dict["start"] = start_datetime_utc - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) - + if end_datetime_utc: sql_query += " AND m.created_at < :end" arg_dict["end"] = end_datetime_utc @@ -525,26 +473,17 @@ WHERE AND invoke_from != :invoke_from""" arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER} assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) + if start_datetime_utc: sql_query += " AND created_at >= :start" arg_dict["start"] = start_datetime_utc - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) - + if end_datetime_utc: sql_query += " AND created_at < :end" arg_dict["end"] = end_datetime_utc @@ -602,26 +541,17 @@ WHERE AND invoke_from != :invoke_from""" arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER} assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - - start_datetime_timezone = timezone.localize(start_datetime) - start_datetime_utc = start_datetime_timezone.astimezone(utc_timezone) + try: + start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) + if start_datetime_utc: sql_query += " AND created_at >= :start" arg_dict["start"] = start_datetime_utc - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - - end_datetime_timezone = timezone.localize(end_datetime) - end_datetime_utc = end_datetime_timezone.astimezone(utc_timezone) - + if end_datetime_utc: sql_query += " AND created_at < :end" arg_dict["end"] = end_datetime_utc diff --git a/api/controllers/console/app/workflow_statistic.py b/api/controllers/console/app/workflow_statistic.py index c246b3ffd5..ef5205c1ee 100644 --- a/api/controllers/console/app/workflow_statistic.py +++ b/api/controllers/console/app/workflow_statistic.py @@ -1,7 +1,4 @@ -from datetime import datetime - -import pytz -from flask import jsonify +from flask import abort, jsonify from flask_restx import Resource, reqparse from sqlalchemy.orm import sessionmaker @@ -9,6 +6,7 @@ from controllers.console import api, console_ns from controllers.console.app.wraps import get_app_model from controllers.console.wraps import account_initialization_required, setup_required from extensions.ext_database import db +from libs.datetime_utils import parse_time_range from libs.helper import DatetimeString from libs.login import current_account_with_tenant, login_required from models.enums import WorkflowRunTriggeredFrom @@ -43,23 +41,11 @@ class WorkflowDailyRunsStatistic(Resource): args = parser.parse_args() assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - start_date = None - end_date = None - - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - start_datetime_timezone = timezone.localize(start_datetime) - start_date = start_datetime_timezone.astimezone(utc_timezone) - - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - end_datetime_timezone = timezone.localize(end_datetime) - end_date = end_datetime_timezone.astimezone(utc_timezone) + try: + start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) response_data = self._workflow_run_repo.get_daily_runs_statistics( tenant_id=app_model.tenant_id, @@ -100,23 +86,11 @@ class WorkflowDailyTerminalsStatistic(Resource): args = parser.parse_args() assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - start_date = None - end_date = None - - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - start_datetime_timezone = timezone.localize(start_datetime) - start_date = start_datetime_timezone.astimezone(utc_timezone) - - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - end_datetime_timezone = timezone.localize(end_datetime) - end_date = end_datetime_timezone.astimezone(utc_timezone) + try: + start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) response_data = self._workflow_run_repo.get_daily_terminals_statistics( tenant_id=app_model.tenant_id, @@ -157,23 +131,11 @@ class WorkflowDailyTokenCostStatistic(Resource): args = parser.parse_args() assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - start_date = None - end_date = None - - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - start_datetime_timezone = timezone.localize(start_datetime) - start_date = start_datetime_timezone.astimezone(utc_timezone) - - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - end_datetime_timezone = timezone.localize(end_datetime) - end_date = end_datetime_timezone.astimezone(utc_timezone) + try: + start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) response_data = self._workflow_run_repo.get_daily_token_cost_statistics( tenant_id=app_model.tenant_id, @@ -214,23 +176,11 @@ class WorkflowAverageAppInteractionStatistic(Resource): args = parser.parse_args() assert account.timezone is not None - timezone = pytz.timezone(account.timezone) - utc_timezone = pytz.utc - start_date = None - end_date = None - - if args["start"]: - start_datetime = datetime.strptime(args["start"], "%Y-%m-%d %H:%M") - start_datetime = start_datetime.replace(second=0) - start_datetime_timezone = timezone.localize(start_datetime) - start_date = start_datetime_timezone.astimezone(utc_timezone) - - if args["end"]: - end_datetime = datetime.strptime(args["end"], "%Y-%m-%d %H:%M") - end_datetime = end_datetime.replace(second=0) - end_datetime_timezone = timezone.localize(end_datetime) - end_date = end_datetime_timezone.astimezone(utc_timezone) + try: + start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone) + except ValueError as e: + abort(400, description=str(e)) response_data = self._workflow_run_repo.get_average_app_interaction_statistics( tenant_id=app_model.tenant_id, diff --git a/api/libs/datetime_utils.py b/api/libs/datetime_utils.py index e576a34629..88f45bd4de 100644 --- a/api/libs/datetime_utils.py +++ b/api/libs/datetime_utils.py @@ -2,6 +2,8 @@ import abc import datetime from typing import Protocol +import pytz + class _NowFunction(Protocol): @abc.abstractmethod @@ -20,3 +22,51 @@ def naive_utc_now() -> datetime.datetime: representing current UTC time. """ return _now_func(datetime.UTC).replace(tzinfo=None) + + +def parse_time_range( + start: str | None, end: str | None, tzname: str +) -> tuple[datetime.datetime | None, datetime.datetime | None]: + """ + Parse time range strings and convert to UTC datetime objects. + Handles DST ambiguity and non-existent times gracefully. + + Args: + start: Start time string (YYYY-MM-DD HH:MM) + end: End time string (YYYY-MM-DD HH:MM) + tzname: Timezone name + + Returns: + tuple: (start_datetime_utc, end_datetime_utc) + + Raises: + ValueError: When time range is invalid or start > end + """ + tz = pytz.timezone(tzname) + utc = pytz.utc + + def _parse(time_str: str | None, label: str) -> datetime.datetime | None: + if not time_str: + return None + + try: + dt = datetime.datetime.strptime(time_str, "%Y-%m-%d %H:%M").replace(second=0) + except ValueError as e: + raise ValueError(f"Invalid {label} time format: {e}") + + try: + return tz.localize(dt, is_dst=None).astimezone(utc) + except pytz.AmbiguousTimeError: + return tz.localize(dt, is_dst=False).astimezone(utc) + except pytz.NonExistentTimeError: + dt += datetime.timedelta(hours=1) + return tz.localize(dt, is_dst=None).astimezone(utc) + + start_dt = _parse(start, "start") + end_dt = _parse(end, "end") + + # Range validation + if start_dt and end_dt and start_dt > end_dt: + raise ValueError("start must be earlier than or equal to end") + + return start_dt, end_dt diff --git a/api/tests/unit_tests/libs/test_datetime_utils.py b/api/tests/unit_tests/libs/test_datetime_utils.py index e914ca4816..84f5b63fbf 100644 --- a/api/tests/unit_tests/libs/test_datetime_utils.py +++ b/api/tests/unit_tests/libs/test_datetime_utils.py @@ -1,8 +1,10 @@ import datetime +from unittest.mock import patch import pytest +import pytz -from libs.datetime_utils import naive_utc_now +from libs.datetime_utils import naive_utc_now, parse_time_range def test_naive_utc_now(monkeypatch: pytest.MonkeyPatch): @@ -20,3 +22,247 @@ def test_naive_utc_now(monkeypatch: pytest.MonkeyPatch): naive_time = naive_datetime.time() utc_time = tz_aware_utc_now.time() assert naive_time == utc_time + + +class TestParseTimeRange: + """Test cases for parse_time_range function.""" + + def test_parse_time_range_basic(self): + """Test basic time range parsing.""" + start, end = parse_time_range("2024-01-01 10:00", "2024-01-01 18:00", "UTC") + + assert start is not None + assert end is not None + assert start < end + assert start.tzinfo == pytz.UTC + assert end.tzinfo == pytz.UTC + + def test_parse_time_range_start_only(self): + """Test parsing with only start time.""" + start, end = parse_time_range("2024-01-01 10:00", None, "UTC") + + assert start is not None + assert end is None + assert start.tzinfo == pytz.UTC + + def test_parse_time_range_end_only(self): + """Test parsing with only end time.""" + start, end = parse_time_range(None, "2024-01-01 18:00", "UTC") + + assert start is None + assert end is not None + assert end.tzinfo == pytz.UTC + + def test_parse_time_range_both_none(self): + """Test parsing with both times None.""" + start, end = parse_time_range(None, None, "UTC") + + assert start is None + assert end is None + + def test_parse_time_range_different_timezones(self): + """Test parsing with different timezones.""" + # Test with US/Eastern timezone + start, end = parse_time_range("2024-01-01 10:00", "2024-01-01 18:00", "US/Eastern") + + assert start is not None + assert end is not None + assert start.tzinfo == pytz.UTC + assert end.tzinfo == pytz.UTC + # Verify the times are correctly converted to UTC + assert start.hour == 15 # 10 AM EST = 3 PM UTC (in January) + assert end.hour == 23 # 6 PM EST = 11 PM UTC (in January) + + def test_parse_time_range_invalid_start_format(self): + """Test parsing with invalid start time format.""" + with pytest.raises(ValueError, match="time data.*does not match format"): + parse_time_range("invalid-date", "2024-01-01 18:00", "UTC") + + def test_parse_time_range_invalid_end_format(self): + """Test parsing with invalid end time format.""" + with pytest.raises(ValueError, match="time data.*does not match format"): + parse_time_range("2024-01-01 10:00", "invalid-date", "UTC") + + def test_parse_time_range_invalid_timezone(self): + """Test parsing with invalid timezone.""" + with pytest.raises(pytz.exceptions.UnknownTimeZoneError): + parse_time_range("2024-01-01 10:00", "2024-01-01 18:00", "Invalid/Timezone") + + def test_parse_time_range_start_after_end(self): + """Test parsing with start time after end time.""" + with pytest.raises(ValueError, match="start must be earlier than or equal to end"): + parse_time_range("2024-01-01 18:00", "2024-01-01 10:00", "UTC") + + def test_parse_time_range_start_equals_end(self): + """Test parsing with start time equal to end time.""" + start, end = parse_time_range("2024-01-01 10:00", "2024-01-01 10:00", "UTC") + + assert start is not None + assert end is not None + assert start == end + + def test_parse_time_range_dst_ambiguous_time(self): + """Test parsing during DST ambiguous time (fall back).""" + # This test simulates DST fall back where 2:30 AM occurs twice + with patch("pytz.timezone") as mock_timezone: + # Mock timezone that raises AmbiguousTimeError + mock_tz = mock_timezone.return_value + + # Create a mock datetime object for the return value + mock_dt = datetime.datetime(2024, 1, 1, 10, 0, 0) + mock_utc_dt = mock_dt.replace(tzinfo=pytz.UTC) + + # Create a proper mock for the localized datetime + from unittest.mock import MagicMock + + mock_localized_dt = MagicMock() + mock_localized_dt.astimezone.return_value = mock_utc_dt + + # Set up side effects: first call raises exception, second call succeeds + mock_tz.localize.side_effect = [ + pytz.AmbiguousTimeError("Ambiguous time"), # First call for start + mock_localized_dt, # Second call for start (with is_dst=False) + pytz.AmbiguousTimeError("Ambiguous time"), # First call for end + mock_localized_dt, # Second call for end (with is_dst=False) + ] + + start, end = parse_time_range("2024-01-01 10:00", "2024-01-01 18:00", "US/Eastern") + + # Should use is_dst=False for ambiguous times + assert mock_tz.localize.call_count == 4 # 2 calls per time (first fails, second succeeds) + assert start is not None + assert end is not None + + def test_parse_time_range_dst_nonexistent_time(self): + """Test parsing during DST nonexistent time (spring forward).""" + with patch("pytz.timezone") as mock_timezone: + # Mock timezone that raises NonExistentTimeError + mock_tz = mock_timezone.return_value + + # Create a mock datetime object for the return value + mock_dt = datetime.datetime(2024, 1, 1, 10, 0, 0) + mock_utc_dt = mock_dt.replace(tzinfo=pytz.UTC) + + # Create a proper mock for the localized datetime + from unittest.mock import MagicMock + + mock_localized_dt = MagicMock() + mock_localized_dt.astimezone.return_value = mock_utc_dt + + # Set up side effects: first call raises exception, second call succeeds + mock_tz.localize.side_effect = [ + pytz.NonExistentTimeError("Non-existent time"), # First call for start + mock_localized_dt, # Second call for start (with adjusted time) + pytz.NonExistentTimeError("Non-existent time"), # First call for end + mock_localized_dt, # Second call for end (with adjusted time) + ] + + start, end = parse_time_range("2024-01-01 10:00", "2024-01-01 18:00", "US/Eastern") + + # Should adjust time forward by 1 hour for nonexistent times + assert mock_tz.localize.call_count == 4 # 2 calls per time (first fails, second succeeds) + assert start is not None + assert end is not None + + def test_parse_time_range_edge_cases(self): + """Test edge cases for time parsing.""" + # Test with midnight times + start, end = parse_time_range("2024-01-01 00:00", "2024-01-01 23:59", "UTC") + assert start is not None + assert end is not None + assert start.hour == 0 + assert start.minute == 0 + assert end.hour == 23 + assert end.minute == 59 + + def test_parse_time_range_different_dates(self): + """Test parsing with different dates.""" + start, end = parse_time_range("2024-01-01 10:00", "2024-01-02 10:00", "UTC") + assert start is not None + assert end is not None + assert start.date() != end.date() + assert (end - start).days == 1 + + def test_parse_time_range_seconds_handling(self): + """Test that seconds are properly set to 0.""" + start, end = parse_time_range("2024-01-01 10:30", "2024-01-01 18:45", "UTC") + assert start is not None + assert end is not None + assert start.second == 0 + assert end.second == 0 + + def test_parse_time_range_timezone_conversion_accuracy(self): + """Test accurate timezone conversion.""" + # Test with a known timezone conversion + start, end = parse_time_range("2024-01-01 12:00", "2024-01-01 12:00", "Asia/Tokyo") + + assert start is not None + assert end is not None + assert start.tzinfo == pytz.UTC + assert end.tzinfo == pytz.UTC + # Tokyo is UTC+9, so 12:00 JST = 03:00 UTC + assert start.hour == 3 + assert end.hour == 3 + + def test_parse_time_range_summer_time(self): + """Test parsing during summer time (DST).""" + # Test with US/Eastern during summer (EDT = UTC-4) + start, end = parse_time_range("2024-07-01 12:00", "2024-07-01 12:00", "US/Eastern") + + assert start is not None + assert end is not None + assert start.tzinfo == pytz.UTC + assert end.tzinfo == pytz.UTC + # 12:00 EDT = 16:00 UTC + assert start.hour == 16 + assert end.hour == 16 + + def test_parse_time_range_winter_time(self): + """Test parsing during winter time (standard time).""" + # Test with US/Eastern during winter (EST = UTC-5) + start, end = parse_time_range("2024-01-01 12:00", "2024-01-01 12:00", "US/Eastern") + + assert start is not None + assert end is not None + assert start.tzinfo == pytz.UTC + assert end.tzinfo == pytz.UTC + # 12:00 EST = 17:00 UTC + assert start.hour == 17 + assert end.hour == 17 + + def test_parse_time_range_empty_strings(self): + """Test parsing with empty strings.""" + # Empty strings are treated as None, so they should not raise errors + start, end = parse_time_range("", "2024-01-01 18:00", "UTC") + assert start is None + assert end is not None + + start, end = parse_time_range("2024-01-01 10:00", "", "UTC") + assert start is not None + assert end is None + + def test_parse_time_range_malformed_datetime(self): + """Test parsing with malformed datetime strings.""" + with pytest.raises(ValueError, match="time data.*does not match format"): + parse_time_range("2024-13-01 10:00", "2024-01-01 18:00", "UTC") + + with pytest.raises(ValueError, match="time data.*does not match format"): + parse_time_range("2024-01-01 10:00", "2024-01-32 18:00", "UTC") + + def test_parse_time_range_very_long_time_range(self): + """Test parsing with very long time range.""" + start, end = parse_time_range("2020-01-01 00:00", "2030-12-31 23:59", "UTC") + + assert start is not None + assert end is not None + assert start < end + assert (end - start).days > 3000 # More than 8 years + + def test_parse_time_range_negative_timezone(self): + """Test parsing with negative timezone offset.""" + start, end = parse_time_range("2024-01-01 12:00", "2024-01-01 12:00", "America/New_York") + + assert start is not None + assert end is not None + assert start.tzinfo == pytz.UTC + assert end.tzinfo == pytz.UTC