Merge pull request #1110 from JarryShaw/fix/1093-mh-mesg-id-fraction

Author: JarryShaw

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)

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
 

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()