From 4e04a12e6d982bc497b10ae3231117b555d2a7de Mon Sep 17 00:00:00 2001 From: mango <287602771+mango-tart@users.noreply.github.com> Date: Mon, 25 May 2026 16:53:06 +1000 Subject: [PATCH 1/2] fix(uploader): TypeError between offset-native and offset-aware datetimes --- src/msgraph_core/tasks/large_file_upload.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/msgraph_core/tasks/large_file_upload.py b/src/msgraph_core/tasks/large_file_upload.py index 882b2ad6..f0f170d0 100644 --- a/src/msgraph_core/tasks/large_file_upload.py +++ b/src/msgraph_core/tasks/large_file_upload.py @@ -73,11 +73,17 @@ def upload_session_expired(self, upload_session: Optional[Parsable] = None) -> b if expiry is None: raise ValueError("Expiry is None") if isinstance(expiry, str): - then = datetime.strptime(expiry, "%Y-%m-%dT%H:%M:%S") + then = datetime.fromisoformat(expiry) elif isinstance(expiry, datetime): then = expiry else: raise ValueError("Expiry is not a string or datetime") + + if then.tzinfo is not None and then.tzinfo.utcoffset(then) is not None: + then = then.astimezone(timezone.utc) + else: + then = then.replace(tzinfo=timezone.utc) + interval = now - then if not isinstance(interval, timedelta): raise ValueError("Interval is not a timedelta") From 77e529935212ba1eb177a868c811d1e68a52a05f Mon Sep 17 00:00:00 2001 From: mango <287602771+mango-tart@users.noreply.github.com> Date: Thu, 1 Oct 2026 16:01:12 +1000 Subject: [PATCH 2/2] chore(test): large_file_upload expiry timestamps --- src/msgraph_core/tasks/large_file_upload.py | 3 +- tests/tasks/test_large_file_upload.py | 64 +++++++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) create mode 100644 tests/tasks/test_large_file_upload.py diff --git a/src/msgraph_core/tasks/large_file_upload.py b/src/msgraph_core/tasks/large_file_upload.py index f0f170d0..048ba5f6 100644 --- a/src/msgraph_core/tasks/large_file_upload.py +++ b/src/msgraph_core/tasks/large_file_upload.py @@ -73,7 +73,8 @@ def upload_session_expired(self, upload_session: Optional[Parsable] = None) -> b if expiry is None: raise ValueError("Expiry is None") if isinstance(expiry, str): - then = datetime.fromisoformat(expiry) + normalized_expiry = expiry[:-1] + '+00:00' if expiry.endswith('Z') else expiry + then = datetime.fromisoformat(normalized_expiry) elif isinstance(expiry, datetime): then = expiry else: diff --git a/tests/tasks/test_large_file_upload.py b/tests/tasks/test_large_file_upload.py new file mode 100644 index 00000000..c2b187e6 --- /dev/null +++ b/tests/tasks/test_large_file_upload.py @@ -0,0 +1,64 @@ +from datetime import datetime, timedelta, timezone +from io import BytesIO +from unittest.mock import Mock + +import pytest + +from msgraph_core.models import LargeFileUploadSession +from msgraph_core.tasks.large_file_upload import LargeFileUploadTask + + +def create_upload_task(expiration_date_time): + upload_session = LargeFileUploadSession( + expiration_date_time=expiration_date_time, + next_expected_ranges=['0-'], + upload_url='https://example.org/upload', + ) + return LargeFileUploadTask(upload_session, Mock(), BytesIO(b'test payload')) + + +def test_upload_session_expired_with_naive_datetime(): + expiration_date_time = (datetime.now(timezone.utc) - timedelta(minutes=5)).replace(tzinfo=None) + task = create_upload_task(expiration_date_time) + + assert task.upload_session_expired() is True + + +def test_upload_session_not_expired_with_naive_datetime(): + expiration_date_time = (datetime.now(timezone.utc) + timedelta(minutes=5)).replace(tzinfo=None) + task = create_upload_task(expiration_date_time) + + assert task.upload_session_expired() is False + + +def test_upload_session_expired_with_offset_aware_datetime(): + expiration_date_time = datetime.now( + timezone(timedelta(hours=-5)) + ) - timedelta(minutes=5) + task = create_upload_task(expiration_date_time) + + assert task.upload_session_expired() is True + + +def test_upload_session_not_expired_with_offset_aware_datetime(): + expiration_date_time = datetime.now( + timezone(timedelta(hours=-5)) + ) + timedelta(minutes=5) + task = create_upload_task(expiration_date_time) + + assert task.upload_session_expired() is False + + +@pytest.mark.parametrize( + ('expiration_date_time', 'expected'), + [ + ((datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat().replace('+00:00', 'Z'), True), + ((datetime.now(timezone.utc) + timedelta(minutes=5)).astimezone( + timezone(timedelta(hours=2)) + ).isoformat(), False), + ], +) +def test_upload_session_expired_with_iso8601_string(expiration_date_time, expected): + task = create_upload_task(expiration_date_time) + + assert task.upload_session_expired() is expected