diff --git a/src/msgraph_core/tasks/large_file_upload.py b/src/msgraph_core/tasks/large_file_upload.py index 882b2ad6..048ba5f6 100644 --- a/src/msgraph_core/tasks/large_file_upload.py +++ b/src/msgraph_core/tasks/large_file_upload.py @@ -73,11 +73,18 @@ 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") + normalized_expiry = expiry[:-1] + '+00:00' if expiry.endswith('Z') else expiry + then = datetime.fromisoformat(normalized_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") 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