diff --git a/flow/record/fieldtypes/__init__.py b/flow/record/fieldtypes/__init__.py index 80cc89af..a26367af 100644 --- a/flow/record/fieldtypes/__init__.py +++ b/flow/record/fieldtypes/__init__.py @@ -308,6 +308,7 @@ def __new__(cls, *args, **kwargs): arg.second, arg.microsecond, tzinfo, + fold=arg.fold, ) else: obj = _dt.__new__(cls, *args, **kwargs) diff --git a/tests/fieldtypes/test_fieldtypes.py b/tests/fieldtypes/test_fieldtypes.py index 6c6eb099..08a273f7 100644 --- a/tests/fieldtypes/test_fieldtypes.py +++ b/tests/fieldtypes/test_fieldtypes.py @@ -13,6 +13,7 @@ import flow.record.fieldtypes from flow.record import RecordDescriptor, RecordReader, RecordWriter, fieldtypes from flow.record.fieldtypes import ( + HAS_ZONE_INFO, PY_312_OR_HIGHER, PY_313_OR_HIGHER, TYPE_POSIX, @@ -28,6 +29,9 @@ ) from flow.record.fieldtypes import datetime as dt +if HAS_ZONE_INFO: + from flow.record.fieldtypes import ZoneInfo + if TYPE_CHECKING: from collections.abc import Callable @@ -427,7 +431,7 @@ def test_datetime() -> None: ("2006-11-10T14:29:55.585192699999999-07:00", datetime(2006, 11, 10, 21, 29, 55, 585192, tzinfo=UTC)), ], ) -def test_datetime_formats(tmp_path: pathlib.Path, value: str, expected_dt: datetime) -> None: +def test_datetime_formats(tmp_path: pathlib.Path, value: str | datetime | float, expected_dt: datetime) -> None: TestRecord = RecordDescriptor( "test/datetime", [ @@ -448,6 +452,82 @@ def test_datetime_formats(tmp_path: pathlib.Path, value: str, expected_dt: datet assert record.dt == expected_dt +DATETIME_FOLD_PARAMS = [ + (datetime(2023, 1, 1, tzinfo=UTC, fold=1), datetime(2023, 1, 1, tzinfo=UTC)), +] +if HAS_ZONE_INFO: + DATETIME_FOLD_PARAMS.append( + ( + datetime(2025, 10, 26, 2, 0, 3, tzinfo=ZoneInfo("Europe/Amsterdam"), fold=1), + datetime(2025, 10, 26, 1, 0, 3, tzinfo=UTC), + ), + ) + + +@pytest.mark.skipif(not HAS_ZONE_INFO, reason="ZoneInfo is required for testing datetime fold parameter") +@pytest.mark.parametrize(("value", "expected_dt"), DATETIME_FOLD_PARAMS) +def test_datetime_formats_fold(tmp_path: pathlib.Path, value: datetime, expected_dt: datetime) -> None: + """test whether datetime accepts fold parameters and converts it correctly""" + TestRecord = RecordDescriptor( + "test/datetime", + [ + ("datetime", "dt"), + ], + ) + record = TestRecord(dt=value) + assert record.dt.fold == 1 + assert record.dt.astimezone(UTC) == expected_dt + + # test packing / serialization of datetime fields + path = tmp_path / "datetime.records" + with RecordWriter(path) as writer: + writer.write(record) + + # test unpacking / deserialization of datetime fields + with RecordReader(path) as reader: + record = next(iter(reader)) + # Need to convert it to UTC specifically as the timezones do not match + assert record.dt.astimezone(UTC) == expected_dt + + +@pytest.mark.skipif(not HAS_ZONE_INFO, reason="ZoneInfo is required for testing datetime fold parameter") +def test_datetime_fold_example() -> None: + """ + Test datetime fold parameter during daylight saving time changes in the Netherlands, which has a + timezone offset of +01:00 during wintertime and +02:00 during summertime. + """ + + TestRecord = RecordDescriptor( + "test/datetime", + [ + ("datetime", "dt"), + ], + ) + + # 2025-10-26 is the date of the end of daylight saving time in the Netherlands. + # At 3:00 AM the clock goes back to 2:00 AM, so 2:00 AM occurs twice. The first occurrence has fold=0 + record = TestRecord("2025-10-26T00:00:00+00:00") + nl_dt = record.dt.astimezone(ZoneInfo("Europe/Amsterdam")) + assert nl_dt.isoformat() == "2025-10-26T02:00:00+02:00" + assert nl_dt.hour == 2 + assert nl_dt.fold == 0 + + # test that both datetimes are considered equal when converted to UTC + record2 = TestRecord(nl_dt) + assert record.dt.astimezone(UTC) == record2.dt.astimezone(UTC) + + # wintertime, the clock goes back from 3:00 AM to 2:00 AM, so 2:00 AM occurs twice. The second occurrence has fold=1 + record = TestRecord("2025-10-26T01:00:00+00:00") + nl_dt = record.dt.astimezone(ZoneInfo("Europe/Amsterdam")) + assert nl_dt.isoformat() == "2025-10-26T02:00:00+01:00" + assert nl_dt.hour == 2 + assert nl_dt.fold == 1 + + # test that both datetimes are considered equal when converted to UTC + record2 = TestRecord(nl_dt) + assert record.dt.astimezone(UTC) == record2.dt.astimezone(UTC) + + def test_digest() -> None: TestRecord = RecordDescriptor( "test/digest",