diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 0f7356fde..516fa8328 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -8100,7 +8100,7 @@ def _make_opt_mesg_id(self, type: 'Enum_Option', option: 'Optional[Data_MesgIDOp int_ts = interval.timestamp() ts_sec = math.floor(int_ts) - ts_frc = math.ceil(((int_ts - ts_sec) * 1_000_000)) * 2**32 + ts_frc = min(round((int_ts - ts_sec) * 2**32), 0xFFFF_FFFF) # 1/2**32 seconds per unit timestamp = NTPTimestamp(seconds=ts_sec + 2_208_988_800, # 70 years fraction=ts_frc) diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index da71ebb49..6545dccb0 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -3,7 +3,6 @@ """header schema for mobility header""" import datetime -import math from typing import TYPE_CHECKING from pcapkit.const.mh.access_type import AccessType as Enum_AccessType @@ -782,9 +781,9 @@ def post_process(self, packet: 'dict[str, Any]') -> 'MesgIDOption': # convert timestamp to datetime # c.f., http://tickelton.gitlab.io/articles/ntp-timestamps/ ts_sec = self.seconds - 2_208_988_800 # 70 years - ts_usec = math.floor(self.fraction / 2**32) + ts_frac = self.fraction / 2**32 # fraction of a second - self.timestamp = datetime.datetime.fromtimestamp(ts_sec + ts_usec, tz=datetime.timezone.utc) + self.timestamp = datetime.datetime.fromtimestamp(ts_sec + ts_frac, tz=datetime.timezone.utc) return self diff --git a/tests/protocols/internet/test_mh_mesg_id_fraction_unit.py b/tests/protocols/internet/test_mh_mesg_id_fraction_unit.py new file mode 100644 index 000000000..125f76fd1 --- /dev/null +++ b/tests/protocols/internet/test_mh_mesg_id_fraction_unit.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import datetime +import unittest +from unittest import mock + +from tests._support import reimport_once_per_class + +EPOCH = 1_700_000_000 + + +class MHMesgIDFractionUnitTests(unittest.TestCase): + """The Message ID option carries the NTP fraction as a 32-bit binary fraction.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def roundtrip(self, interval: 'datetime.datetime') -> 'tuple[int, datetime.datetime]': + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + from pcapkit.protocols.schema.internet.mh import MesgIDOption + + proto = object.__new__(MH) + made = proto._make_opt_mesg_id(Option.MESG_ID_OPTION_TYPE, interval=interval) + read = MesgIDOption.unpack(made.pack()) + self.assertEqual(read.fraction, made.fraction) + return made.fraction, read.timestamp + + def test_fraction_roundtrip(self) -> None: + for usec, fraction in ((0, 0), (500_000, 2**31), (999_999, None)): + with self.subTest(usec=usec): + interval = datetime.datetime.fromtimestamp(EPOCH, tz=datetime.timezone.utc).replace(microsecond=usec) + packed, timestamp = self.roundtrip(interval) + if fraction is not None: + self.assertEqual(packed, fraction) + self.assertAlmostEqual(packed / 2**32, usec / 1_000_000, delta=1e-6) + self.assertEqual(timestamp, interval) + + def test_exact_integer_second(self) -> None: + interval = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) + packed, timestamp = self.roundtrip(interval) + self.assertEqual(packed, 0) + self.assertEqual(timestamp, interval) + + def test_fraction_clamped_to_field_width(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + + proto = object.__new__(MH) + interval = mock.Mock(timestamp=lambda: 1 - 2**-40) # rounds up to 2**32 + made = proto._make_opt_mesg_id(Option.MESG_ID_OPTION_TYPE, interval=interval) + self.assertEqual(made.fraction, 0xFFFF_FFFF) + self.assertEqual(made.seconds, 2_208_988_800) + + def test_read_fraction_is_seconds(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.protocols.schema.internet.mh import MesgIDOption + + raw = MesgIDOption(type=Option.MESG_ID_OPTION_TYPE, length=8, + seconds=EPOCH + 2_208_988_800, fraction=2**30).pack() + self.assertEqual(MesgIDOption.unpack(raw).timestamp, + datetime.datetime.fromtimestamp(EPOCH + 0.25, tz=datetime.timezone.utc)) + + +if __name__ == '__main__': + unittest.main()