Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pcapkit/protocols/internet/mh.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
5 changes: 2 additions & 3 deletions pcapkit/protocols/schema/internet/mh.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down
66 changes: 66 additions & 0 deletions tests/protocols/internet/test_mh_mesg_id_fraction_unit.py
Original file line number Diff line number Diff line change
@@ -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()
Loading