#!/usr/bin/env python3 """ PTP BMC behavior test for netX DUT. What this script does: 1) Sends Announce messages from two emulated grandmasters via UDP multicast. 2) Queries the DUT with a PTP management GET(PARENT_DATA_SET) message. 3) Prints the currently selected grandmaster from DUT responses. Test case specification (BMC-focused): This script can execute a multi-phase BMCA verification plan. Each phase defines two emulated masters and an expected winner. The DUT winner is read via PARENT_DATA_SET and checked against the expected grandmaster identity. Master identities used in all phases: - good source/gm clockIdentity: 08:61:95:ff:fe:e5:b2:4b - bad source/gm clockIdentity: 6c:14:f7:16:05:0a:00:00 - sourcePortNumber: 1 - stepsRemoved: 1 - timeSource: good=0x20, bad=0xA0 Phase table: +-------+----------------------------------+---------------------------------------+----------------------------------------+-------------+-------------+ | Phase | Purpose | Good dataset | Bad dataset | Announce on | Expected GM | +-------+----------------------------------+---------------------------------------+----------------------------------------+-------------+-------------+ | A | Baseline | p1=50 c=6 a=0x30 v=0x1000 p2=50 | p1=248 c=248 a=0xFE v=0xFFFF p2=248 | good,bad | good | | B | Priority1 decision | p1=248 c=6 a=0x30 v=0x1000 p2=100 | p1=50 c=6 a=0x30 v=0x1000 p2=100 | good,bad | bad | | C | Priority2 decision | p1=100 c=6 a=0x30 v=0x1000 p2=248 | p1=100 c=6 a=0x30 v=0x1000 p2=50 | good,bad | bad | | D | ClockClass decision | p1=100 c=248 a=0x30 v=0x1000 p2=100 | p1=100 c=6 a=0x30 v=0x1000 p2=100 | good,bad | bad | | E | ClockAccuracy decision | p1=100 c=6 a=0xFE v=0x1000 p2=100 | p1=100 c=6 a=0x30 v=0x1000 p2=100 | good,bad | bad | | F | OffsetScaledLogVariance decision | p1=100 c=6 a=0x30 v=0xFFFF p2=100 | p1=100 c=6 a=0x30 v=0x1000 p2=100 | good,bad | bad | | G | Failover | same as A | same as A | bad only | bad | | H | Recovery | same as A | same as A | good,bad | good | +-------+----------------------------------+---------------------------------------+----------------------------------------+-------------+-------------+ Legend: - p1=priority1, c=clockClass, a=clockAccuracy, v=offsetScaledLogVariance, p2=priority2 Additional diagnostics: - Optional UDP preflight probe for DUT ports 319/320 - Optional management destination modes (multicast, unicast, both) - Optional management request variants for targetPortIdentity handling - Optional syslog markers for phase boundaries (useful in Wireshark) Only required input: DUT IP address. Example: python Tools/ptp_bmc_behavior_test.py --dut-ip 192.168.2.242 """ from __future__ import annotations import argparse import ipaddress import socket import struct import threading import time from dataclasses import dataclass, replace from typing import Optional, Dict, Any, List, Set, Tuple PTP_MCAST_IP = "224.0.1.129" PTP_GENERAL_PORT = 320 PTP_EVENT_PORT = 319 PTP_MSGTYPE_ANNOUNCE = 0x0B PTP_MSGTYPE_MANAGEMENT = 0x0D PTP_VERSION = 2 PTP_CONTROLFIELD_ALL_OTHERS = 0x05 PTP_LOGMESSAGEINTERVAL_DEFAULT = 0x7F PTP_MM_ACTION_GET = 0x00 PTP_MM_ACTION_RESPONSE = 0x02 PTP_MM_ID_PARENT_DATA_SET = 0x2002 PTP_TLV_TYPE_MANAGEMENT = 0x0001 PTP_TLV_TYPE_MANAGEMENT_ERROR_STATUS = 0x0002 SYSLOG_DEFAULT_PORT = 514 SYSLOG_DEFAULT_FACILITY_SEVERITY = 134 # local0.info SYSLOG_DEFAULT_BROADCAST_IP = "255.255.255.255" @dataclass class EmulatedGrandmaster: source_clock_identity: bytes source_port_number: int grandmaster_identity: bytes priority1: int clock_class: int clock_accuracy: int offset_scaled_log_variance: int priority2: int steps_removed: int time_source: int @dataclass class ParentDatasetResponse: src_ip: str source_port_identity: str parent_port_identity: str grandmaster_identity: str priority1: int clock_class: int clock_accuracy: int offset_scaled_log_variance: int priority2: int sequence_id: int @dataclass class BMCTestPhase: name: str duration_s: float expected_label: str good_master: EmulatedGrandmaster bad_master: EmulatedGrandmaster good_enabled: bool = True bad_enabled: bool = True @dataclass class BMCTestPhaseResult: name: str expected_gm: str observed_gm: str samples: int passed: bool @dataclass class SyslogConfig: enabled: bool dst_ip: str dst_port: int app_name: str def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description="PTP BMC behavior test for a netX DUT") p.add_argument("--dut-ip", required=True, help="DUT IPv4 address") p.add_argument("--src-ip", help="Optional local IPv4 address of the interface to use") p.add_argument("--domain", type=int, default=0, help="PTP domain number (default: 0)") p.add_argument("--announce-duration", type=float, default=20.0, help="Seconds to send announce traffic") p.add_argument("--announce-interval", type=float, default=1.0, help="Announce interval in seconds") p.add_argument("--query-interval", type=float, default=2.0, help="Management query interval in seconds") p.add_argument("--mgmt-timeout", type=float, default=1.0, help="Timeout waiting for a management response") p.add_argument( "--test-plan", choices=["single", "phases"], default="phases", help="BMC verification mode: single static check or multi-phase switching checks", ) p.add_argument( "--phase-duration", type=float, default=12.0, help="Duration in seconds for each BMC test phase when --test-plan phases", ) p.add_argument( "--port-scan", action="store_true", help="Run a UDP preflight scan for DUT ports 319 and 320 before test start", ) p.add_argument( "--port-scan-timeout", type=float, default=0.6, help="Timeout in seconds for each UDP port probe", ) p.add_argument( "--mgmt-dst", choices=["multicast", "unicast", "both"], default="both", help="How to send management GET requests (default: both)", ) p.add_argument( "--diag-mgmt-variants", action="store_true", help="Send multiple management GET variants and print raw DUT management TLV details", ) p.add_argument( "--diag-announce-timeout", type=float, default=3.0, help="Seconds to observe DUT announce for targetPortIdentity variant", ) p.add_argument( "--syslog", action="store_true", help="Send syslog markers for test and phase transitions", ) p.add_argument( "--syslog-ip", help="Syslog destination IPv4 address (default: 255.255.255.255 broadcast)", ) p.add_argument( "--syslog-port", type=int, default=SYSLOG_DEFAULT_PORT, help="Syslog destination UDP port (default: 514)", ) p.add_argument( "--syslog-app", default="ptp-bmc-test", help="App name included in syslog marker messages", ) return p.parse_args() def validate_ip(ip: str) -> None: ipaddress.IPv4Address(ip) def local_ip_for_target(target_ip: str) -> str: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect((target_ip, 1)) return s.getsockname()[0] finally: s.close() def hex_to_clock_identity(v: str) -> bytes: cleaned = v.replace(":", "").replace("-", "").replace("0x", "") b = bytes.fromhex(cleaned) if len(b) != 8: raise ValueError("ClockIdentity must be exactly 8 bytes") return b def fmt_clock_identity(b: bytes) -> str: return ":".join(f"{x:02x}" for x in b) def clone_master(master: EmulatedGrandmaster, **changes: Any) -> EmulatedGrandmaster: return replace(master, **changes) def build_bmc_test_phases(base_good: EmulatedGrandmaster, base_bad: EmulatedGrandmaster, duration_s: float) -> List[BMCTestPhase]: # Phase A: Baseline comparison where the good master should win. phase_a_good = clone_master(base_good, priority1=50, priority2=50, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0x1000) phase_a_bad = clone_master(base_bad, priority1=248, priority2=248, clock_class=248, clock_accuracy=0xFE, offset_scaled_log_variance=0xFFFF) # Phase B: Priority1 swap only (all other quality fields equal). phase_b_good = clone_master(base_good, priority1=248, priority2=100, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0x1000) phase_b_bad = clone_master(base_bad, priority1=50, priority2=100, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0x1000) # Phase C: Priority2 swap only (all other quality fields equal). phase_c_good = clone_master(base_good, priority1=100, priority2=248, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0x1000) phase_c_bad = clone_master(base_bad, priority1=100, priority2=50, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0x1000) # Phase D: ClockClass swap only. phase_d_good = clone_master(base_good, priority1=100, priority2=100, clock_class=248, clock_accuracy=0x30, offset_scaled_log_variance=0x1000) phase_d_bad = clone_master(base_bad, priority1=100, priority2=100, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0x1000) # Phase E: ClockAccuracy swap only. phase_e_good = clone_master(base_good, priority1=100, priority2=100, clock_class=6, clock_accuracy=0xFE, offset_scaled_log_variance=0x1000) phase_e_bad = clone_master(base_bad, priority1=100, priority2=100, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0x1000) # Phase F: Variance swap only. phase_f_good = clone_master(base_good, priority1=100, priority2=100, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0xFFFF) phase_f_bad = clone_master(base_bad, priority1=100, priority2=100, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0x1000) # Phase G: Failover by disabling the currently better master. phase_g_good = clone_master(phase_a_good) phase_g_bad = clone_master(phase_a_bad) # Phase H: Recovery after re-enabling better master. phase_h_good = clone_master(phase_a_good) phase_h_bad = clone_master(phase_a_bad) return [ BMCTestPhase("A_baseline_good_wins", duration_s, "good", phase_a_good, phase_a_bad, True, True), BMCTestPhase("B_priority1_swap_bad_wins", duration_s, "bad", phase_b_good, phase_b_bad, True, True), BMCTestPhase("C_priority2_swap_bad_wins", duration_s, "bad", phase_c_good, phase_c_bad, True, True), BMCTestPhase("D_clockClass_swap_bad_wins", duration_s, "bad", phase_d_good, phase_d_bad, True, True), BMCTestPhase("E_clockAccuracy_swap_bad_wins", duration_s, "bad", phase_e_good, phase_e_bad, True, True), BMCTestPhase("F_variance_swap_bad_wins", duration_s, "bad", phase_f_good, phase_f_bad, True, True), BMCTestPhase("G_failover_good_disabled_bad_wins", duration_s, "bad", phase_g_good, phase_g_bad, False, True), BMCTestPhase("H_recovery_good_reenabled_wins", duration_s, "good", phase_h_good, phase_h_bad, True, True), ] def build_ptp_header( message_type: int, message_length: int, domain: int, source_clock_identity: bytes, source_port_number: int, sequence_id: int, control_field: int, log_message_interval: int, ) -> bytes: # IEEE 1588v2 header: 34 bytes, all multi-byte fields in network byte order. transport_specific = 0 version_ptp = PTP_VERSION & 0x0F first_byte = ((transport_specific & 0x0F) << 4) | (message_type & 0x0F) second_byte = version_ptp return b"".join( [ struct.pack("!B", first_byte), struct.pack("!B", second_byte), struct.pack("!H", message_length), struct.pack("!B", domain & 0xFF), struct.pack("!B", 0), struct.pack("!H", 0), struct.pack("!Q", 0), struct.pack("!I", 0), source_clock_identity, struct.pack("!H", source_port_number & 0xFFFF), struct.pack("!H", sequence_id & 0xFFFF), struct.pack("!B", control_field & 0xFF), struct.pack("!b", log_message_interval), ] ) def build_announce_payload(gm: EmulatedGrandmaster) -> bytes: # PTP_ANNOUNCED_FRAME_T payload fields (30 bytes total): # originTimestamp(10), currentUtcOffset(2), reserved(1), priority1(1), class(1), # accuracy(1), variance(2), priority2(1), grandmasterIdentity(8), stepsRemoved(2), timeSource(1) origin_timestamp = struct.pack("!HII", 0, 0, 0) current_utc_offset = struct.pack("!H", 37) reserved = struct.pack("!B", 0) payload = b"".join( [ origin_timestamp, current_utc_offset, reserved, struct.pack("!B", gm.priority1 & 0xFF), struct.pack("!B", gm.clock_class & 0xFF), struct.pack("!B", gm.clock_accuracy & 0xFF), struct.pack("!H", gm.offset_scaled_log_variance & 0xFFFF), struct.pack("!B", gm.priority2 & 0xFF), gm.grandmaster_identity, struct.pack("!H", gm.steps_removed & 0xFFFF), struct.pack("!B", gm.time_source & 0xFF), ] ) if len(payload) != 30: raise RuntimeError(f"Unexpected Announce payload size: {len(payload)}") return payload def build_announce_frame(gm: EmulatedGrandmaster, domain: int, sequence_id: int) -> bytes: payload = build_announce_payload(gm) msg_len = 34 + len(payload) header = build_ptp_header( message_type=PTP_MSGTYPE_ANNOUNCE, message_length=msg_len, domain=domain, source_clock_identity=gm.source_clock_identity, source_port_number=gm.source_port_number, sequence_id=sequence_id, control_field=PTP_CONTROLFIELD_ALL_OTHERS, log_message_interval=1, ) return header + payload def build_parent_dataset_get_request( domain: int, requester_clock_identity: bytes, requester_port_number: int, sequence_id: int, ) -> bytes: # Management common (14 bytes): targetPortIdentity(10), startingBoundaryHops(1), boundaryHops(1), action(1), reserved(1) # TLV (6 bytes): tlvType(2), lengthField(2), managementId(2) target_port_identity = (b"\xff" * 8) + struct.pack("!H", 0xFFFF) common = b"".join( [ target_port_identity, struct.pack("!B", 0), struct.pack("!B", 0), struct.pack("!B", PTP_MM_ACTION_GET), struct.pack("!B", 0), ] ) tlv = b"".join( [ struct.pack("!H", PTP_TLV_TYPE_MANAGEMENT), struct.pack("!H", 2), struct.pack("!H", PTP_MM_ID_PARENT_DATA_SET), ] ) msg_len = 34 + len(common) + len(tlv) header = build_ptp_header( message_type=PTP_MSGTYPE_MANAGEMENT, message_length=msg_len, domain=domain, source_clock_identity=requester_clock_identity, source_port_number=requester_port_number, sequence_id=sequence_id, control_field=0, log_message_interval=PTP_LOGMESSAGEINTERVAL_DEFAULT, ) return header + common + tlv def parse_ptp_header(frame: bytes) -> Optional[dict]: if len(frame) < 34: return None first_byte, second_byte = struct.unpack_from("!BB", frame, 0) msg_type = first_byte & 0x0F version = second_byte & 0x0F msg_len = struct.unpack_from("!H", frame, 2)[0] domain = frame[4] source_clock_identity = frame[20:28] source_port_number = struct.unpack_from("!H", frame, 28)[0] sequence_id = struct.unpack_from("!H", frame, 30)[0] return { "msg_type": msg_type, "version": version, "msg_len": msg_len, "domain": domain, "source_clock_identity": source_clock_identity, "source_port_number": source_port_number, "sequence_id": sequence_id, } def parse_management_meta(frame: bytes) -> Optional[Dict[str, Any]]: if len(frame) < 54: return None hdr = parse_ptp_header(frame) if hdr is None or hdr["msg_type"] != PTP_MSGTYPE_MANAGEMENT: return None action = frame[46] tlv_type, length_field = struct.unpack_from("!HH", frame, 48) info: Dict[str, Any] = { "seq": hdr["sequence_id"], "version": hdr["version"], "msg_len": hdr["msg_len"], "action": action, "tlv_type": tlv_type, "length_field": length_field, } if tlv_type == PTP_TLV_TYPE_MANAGEMENT and len(frame) >= 54: mgmt_id = struct.unpack_from("!H", frame, 52)[0] info["management_id"] = mgmt_id elif tlv_type == PTP_TLV_TYPE_MANAGEMENT_ERROR_STATUS and len(frame) >= 60: err_id, mgmt_id = struct.unpack_from("!HH", frame, 52) info["management_error_id"] = err_id info["management_id"] = mgmt_id return info def observe_dut_port_identity(io_sock: socket.socket, dut_ip: str, timeout_s: float) -> Optional[Tuple[bytes, int]]: deadline = time.time() + timeout_s while time.time() < deadline: io_sock.settimeout(max(0.05, deadline - time.time())) try: data, addr = io_sock.recvfrom(2048) except socket.timeout: return None if addr[0] != dut_ip: continue hdr = parse_ptp_header(data) if hdr is None: continue if hdr["msg_type"] == PTP_MSGTYPE_ANNOUNCE and hdr["version"] == PTP_VERSION: return hdr["source_clock_identity"], hdr["source_port_number"] return None def udp_probe_port(src_ip: str, dst_ip: str, dst_port: int, timeout_s: float) -> str: # UDP probing is best-effort: timeout usually means open|filtered. s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) s.settimeout(timeout_s) try: s.bind((src_ip, 0)) s.connect((dst_ip, dst_port)) s.send(b"PTP_PROBE") try: _ = s.recv(256) return "open" except ConnectionRefusedError: return "closed" except socket.timeout: return "open|filtered" except ConnectionRefusedError: return "closed" except OSError: return "filtered|error" finally: s.close() def run_udp_port_preflight(src_ip: str, dut_ip: str, timeout_s: float) -> None: print(f"Running UDP preflight scan on DUT {dut_ip} from {src_ip}...") for port in (PTP_EVENT_PORT, PTP_GENERAL_PORT): state = udp_probe_port(src_ip, dut_ip, port, timeout_s) print(f"UDP/{port}: {state}") print("Note: UDP timeout typically means open|filtered and is not a definitive closed result.") def open_syslog_socket(local_ip: str) -> socket.socket: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) s.bind((local_ip, 0)) s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) return s def send_syslog_marker( sock: Optional[socket.socket], cfg: SyslogConfig, marker: str, kv: Optional[Dict[str, Any]] = None, ) -> None: if not cfg.enabled or sock is None: return ts = time.strftime("%b %d %H:%M:%S", time.localtime()) host = socket.gethostname() details = "" if kv: details = " " + " ".join(f"{k}={v}" for k, v in kv.items()) payload = f"<{SYSLOG_DEFAULT_FACILITY_SEVERITY}>{ts} {host} {cfg.app_name}: {marker}{details}" try: sock.sendto(payload.encode("ascii", errors="replace"), (cfg.dst_ip, cfg.dst_port)) except OSError: # Do not fail the test due to telemetry marker transport issues. pass def parse_parent_dataset_response(src_ip: str, frame: bytes, expected_seq: int) -> Optional[ParentDatasetResponse]: hdr = parse_ptp_header(frame) if hdr is None: return None if hdr["msg_type"] != PTP_MSGTYPE_MANAGEMENT or hdr["version"] != PTP_VERSION: return None if hdr["msg_len"] > len(frame): return None if hdr["sequence_id"] != expected_seq: return None if len(frame) < 54: return None action = frame[46] if action != PTP_MM_ACTION_RESPONSE: return None tlv_type, length_field, management_id = struct.unpack_from("!HHH", frame, 48) if tlv_type == PTP_TLV_TYPE_MANAGEMENT_ERROR_STATUS: return None if tlv_type != PTP_TLV_TYPE_MANAGEMENT: return None if management_id != PTP_MM_ID_PARENT_DATA_SET: return None data_offset = 54 if len(frame) < data_offset + 32: return None parent_clock_id = frame[data_offset : data_offset + 8] parent_port = struct.unpack_from("!H", frame, data_offset + 8)[0] priority1 = frame[data_offset + 18] clock_class = frame[data_offset + 19] clock_accuracy = frame[data_offset + 20] variance = struct.unpack_from("!H", frame, data_offset + 21)[0] priority2 = frame[data_offset + 23] grandmaster_id = frame[data_offset + 24 : data_offset + 32] source_pi = f"{fmt_clock_identity(hdr['source_clock_identity'])}:{hdr['source_port_number']}" parent_pi = f"{fmt_clock_identity(parent_clock_id)}:{parent_port}" return ParentDatasetResponse( src_ip=src_ip, source_port_identity=source_pi, parent_port_identity=parent_pi, grandmaster_identity=fmt_clock_identity(grandmaster_id), priority1=priority1, clock_class=clock_class, clock_accuracy=clock_accuracy, offset_scaled_log_variance=variance, priority2=priority2, sequence_id=hdr["sequence_id"], ) class AnnounceSender(threading.Thread): def __init__(self, io_sock: socket.socket, domain: int, interval_s: float, stop_evt: threading.Event): super().__init__(daemon=True) self.io_sock = io_sock self.domain = domain self.interval_s = interval_s self.stop_evt = stop_evt self.good = EmulatedGrandmaster( source_clock_identity=hex_to_clock_identity("08:61:95:ff:fe:e5:b2:4b"), source_port_number=1, grandmaster_identity=hex_to_clock_identity("08:61:95:ff:fe:e5:b2:4b"), priority1=50, clock_class=6, clock_accuracy=0x30, offset_scaled_log_variance=0x1000, priority2=50, steps_removed=1, time_source=0x20, ) self.bad = EmulatedGrandmaster( source_clock_identity=hex_to_clock_identity("6c:14:f7:16:05:0a:00:00"), source_port_number=1, grandmaster_identity=hex_to_clock_identity("6c:14:f7:16:05:0a:00:00"), priority1=248, clock_class=248, clock_accuracy=0xFE, offset_scaled_log_variance=0xFFFF, priority2=248, steps_removed=1, time_source=0xA0, ) self.seq_good = 1 self.seq_bad = 1 self.good_enabled = True self.bad_enabled = True self.lock = threading.Lock() def configure( self, good_master: EmulatedGrandmaster, bad_master: EmulatedGrandmaster, good_enabled: bool, bad_enabled: bool, ) -> None: with self.lock: self.good = good_master self.bad = bad_master self.good_enabled = good_enabled self.bad_enabled = bad_enabled def run(self) -> None: while not self.stop_evt.is_set(): with self.lock: good = self.good bad = self.bad good_enabled = self.good_enabled bad_enabled = self.bad_enabled if good_enabled: frame_good = build_announce_frame(good, self.domain, self.seq_good) self.io_sock.sendto(frame_good, (PTP_MCAST_IP, PTP_GENERAL_PORT)) self.seq_good = (self.seq_good + 1) & 0xFFFF if bad_enabled: frame_bad = build_announce_frame(bad, self.domain, self.seq_bad) self.io_sock.sendto(frame_bad, (PTP_MCAST_IP, PTP_GENERAL_PORT)) self.seq_bad = (self.seq_bad + 1) & 0xFFFF self.stop_evt.wait(self.interval_s) def open_ptp_general_socket(local_ip: str, ttl: int = 1) -> socket.socket: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # Use one socket for RX and TX on UDP/320 to match common PTP management behavior. s.bind((local_ip, PTP_GENERAL_PORT)) s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl) s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(local_ip)) mreq = socket.inet_aton(PTP_MCAST_IP) + socket.inet_aton(local_ip) s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) return s def query_parent_dataset( io_sock: socket.socket, dut_ip: str, domain: int, requester_clock_identity: bytes, requester_port_number: int, sequence_id: int, timeout_s: float, mgmt_dst_mode: str, diag_variants: bool, dut_port_identity: Optional[Tuple[bytes, int]], ) -> Tuple[Optional[ParentDatasetResponse], int]: requests: List[Tuple[str, int, bytes]] = [] if diag_variants: seq = sequence_id req_wild = build_parent_dataset_get_request( domain=domain, requester_clock_identity=requester_clock_identity, requester_port_number=requester_port_number, sequence_id=seq, ) requests.append(("wildcard_all", seq, req_wild)) seq = (seq + 1) & 0xFFFF req_zero = build_parent_dataset_get_request( domain=domain, requester_clock_identity=requester_clock_identity, requester_port_number=requester_port_number, sequence_id=seq, ) req_zero = req_zero[:34] + (b"\x00" * 10) + req_zero[44:] requests.append(("target_zero", seq, req_zero)) if dut_port_identity is not None: seq = (seq + 1) & 0xFFFF req_dut = build_parent_dataset_get_request( domain=domain, requester_clock_identity=requester_clock_identity, requester_port_number=requester_port_number, sequence_id=seq, ) clock_id, port_no = dut_port_identity req_dut = req_dut[:34] + clock_id + struct.pack("!H", port_no) + req_dut[44:] requests.append(("target_dut_port_identity", seq, req_dut)) else: req = build_parent_dataset_get_request( domain=domain, requester_clock_identity=requester_clock_identity, requester_port_number=requester_port_number, sequence_id=sequence_id, ) requests.append(("default", sequence_id, req)) sent_seqs: Set[int] = set() for name, seq, req in requests: if mgmt_dst_mode in ("multicast", "both"): io_sock.sendto(req, (PTP_MCAST_IP, PTP_GENERAL_PORT)) if mgmt_dst_mode in ("unicast", "both"): io_sock.sendto(req, (dut_ip, PTP_GENERAL_PORT)) sent_seqs.add(seq) if diag_variants: print(f"sent mgmt variant={name} seq={seq:05d}") deadline = time.time() + timeout_s while time.time() < deadline: remaining = max(0.01, deadline - time.time()) io_sock.settimeout(remaining) try: data, addr = io_sock.recvfrom(2048) except socket.timeout: return None, len(requests) src_ip = addr[0] if src_ip != dut_ip: continue meta = parse_management_meta(data) if meta is not None and diag_variants: if meta["tlv_type"] == PTP_TLV_TYPE_MANAGEMENT_ERROR_STATUS: print( f"DUT mgmt error: seq={meta['seq']:05d} action={meta['action']} " f"errId=0x{meta.get('management_error_id', 0):04x} mgmtId=0x{meta.get('management_id', 0):04x}" ) else: print( f"DUT mgmt frame: seq={meta['seq']:05d} action={meta['action']} " f"tlv=0x{meta['tlv_type']:04x} mgmtId=0x{meta.get('management_id', 0):04x}" ) hdr = parse_ptp_header(data) if hdr is None or hdr["sequence_id"] not in sent_seqs: continue parsed = parse_parent_dataset_response(src_ip, data, hdr["sequence_id"]) if parsed is not None: return parsed, len(requests) if diag_variants and hdr["msg_type"] == PTP_MSGTYPE_MANAGEMENT: print( f"received mgmt frame from DUT but no ParentDataSet match " f"(version={hdr['version']}, msgLen={hdr['msg_len']}, seq={hdr['sequence_id']})" ) return None, len(requests) def main() -> None: args = parse_args() validate_ip(args.dut_ip) if args.src_ip: validate_ip(args.src_ip) if args.syslog_ip: validate_ip(args.syslog_ip) local_ip = args.src_ip if args.src_ip else local_ip_for_target(args.dut_ip) syslog_cfg = SyslogConfig( enabled=args.syslog, dst_ip=args.syslog_ip if args.syslog_ip else SYSLOG_DEFAULT_BROADCAST_IP, dst_port=args.syslog_port, app_name=args.syslog_app, ) print(f"Using local interface IP: {local_ip}") print(f"DUT IP: {args.dut_ip}") if syslog_cfg.enabled: print(f"Syslog markers enabled: dst={syslog_cfg.dst_ip}:{syslog_cfg.dst_port} app={syslog_cfg.app_name}") if args.port_scan: run_udp_port_preflight(local_ip, args.dut_ip, args.port_scan_timeout) io_sock = open_ptp_general_socket(local_ip) syslog_sock = open_syslog_socket(local_ip) if syslog_cfg.enabled else None print(f"PTP socket bound to: {io_sock.getsockname()[0]}:{io_sock.getsockname()[1]}") print(f"Joined multicast group {PTP_MCAST_IP} on interface {local_ip}") send_syslog_marker( syslog_sock, syslog_cfg, "TEST_START", { "dut": args.dut_ip, "domain": args.domain, "plan": args.test_plan, "mgmtdst": args.mgmt_dst, "src": local_ip, }, ) stop_evt = threading.Event() sender = AnnounceSender(io_sock=io_sock, domain=args.domain, interval_s=args.announce_interval, stop_evt=stop_evt) sender.start() # Fixed requester identity used for management queries from the PC. requester_clock_identity = hex_to_clock_identity("02:00:00:00:00:00:00:01") requester_port_number = 1 print("Sending Announce traffic from two emulated grandmasters...") print( f"Querying DUT ParentDataSet (domain={args.domain}, mgmt-dst={args.mgmt_dst}) " "to read selected Grandmaster..." ) dut_port_identity = None if args.diag_mgmt_variants: print("Diagnostic mode enabled: probing additional management request variants") dut_port_identity = observe_dut_port_identity(io_sock, args.dut_ip, args.diag_announce_timeout) if dut_port_identity is None: print("Could not learn DUT sourcePortIdentity from Announce within timeout") else: print( "Learned DUT sourcePortIdentity from Announce: " f"{fmt_clock_identity(dut_port_identity[0])}:{dut_port_identity[1]}" ) end_time = time.time() + args.announce_duration seq = 1 last = None try: if args.test_plan == "single": send_syslog_marker(syslog_sock, syslog_cfg, "PHASE_START", {"name": "single", "expect": "good"}) while time.time() < end_time: rsp, used_seq_count = query_parent_dataset( io_sock=io_sock, dut_ip=args.dut_ip, domain=args.domain, requester_clock_identity=requester_clock_identity, requester_port_number=requester_port_number, sequence_id=seq, timeout_s=args.mgmt_timeout, mgmt_dst_mode=args.mgmt_dst, diag_variants=args.diag_mgmt_variants, dut_port_identity=dut_port_identity, ) if rsp is None: print(f"seq={seq:05d} no ParentDataSet response from DUT") else: changed = "" if last is None or last.grandmaster_identity != rsp.grandmaster_identity: changed = " <-- GM changed" print( f"seq={rsp.sequence_id:05d} dut={rsp.src_ip} " f"gm={rsp.grandmaster_identity} p1={rsp.priority1} p2={rsp.priority2} " f"class={rsp.clock_class} acc=0x{rsp.clock_accuracy:02x} " f"var=0x{rsp.offset_scaled_log_variance:04x}{changed}" ) last = rsp seq = (seq + used_seq_count) & 0xFFFF time.sleep(args.query_interval) send_syslog_marker(syslog_sock, syslog_cfg, "PHASE_END", {"name": "single"}) else: phases = build_bmc_test_phases(sender.good, sender.bad, args.phase_duration) phase_results: List[BMCTestPhaseResult] = [] for phase in phases: sender.configure(phase.good_master, phase.bad_master, phase.good_enabled, phase.bad_enabled) send_syslog_marker( syslog_sock, syslog_cfg, "PHASE_START", { "name": phase.name, "expect": phase.expected_label, "good_en": int(phase.good_enabled), "bad_en": int(phase.bad_enabled), }, ) print(f"\n=== Phase: {phase.name} ===") print( f"expect={phase.expected_label} duration={phase.duration_s:.1f}s " f"good_enabled={phase.good_enabled} bad_enabled={phase.bad_enabled}" ) phase_end = time.time() + phase.duration_s observed_last_gm = "none" sample_count = 0 while time.time() < phase_end: rsp, used_seq_count = query_parent_dataset( io_sock=io_sock, dut_ip=args.dut_ip, domain=args.domain, requester_clock_identity=requester_clock_identity, requester_port_number=requester_port_number, sequence_id=seq, timeout_s=args.mgmt_timeout, mgmt_dst_mode=args.mgmt_dst, diag_variants=args.diag_mgmt_variants, dut_port_identity=dut_port_identity, ) if rsp is None: print(f"phase={phase.name} seq={seq:05d} no ParentDataSet response") else: observed_last_gm = rsp.grandmaster_identity sample_count += 1 print( f"phase={phase.name} seq={rsp.sequence_id:05d} " f"gm={rsp.grandmaster_identity} p1={rsp.priority1} p2={rsp.priority2} " f"class={rsp.clock_class} acc=0x{rsp.clock_accuracy:02x} " f"var=0x{rsp.offset_scaled_log_variance:04x}" ) last = rsp seq = (seq + used_seq_count) & 0xFFFF time.sleep(args.query_interval) expected_gm = fmt_clock_identity(phase.good_master.grandmaster_identity) if phase.expected_label == "bad": expected_gm = fmt_clock_identity(phase.bad_master.grandmaster_identity) passed = (observed_last_gm == expected_gm) phase_results.append( BMCTestPhaseResult( name=phase.name, expected_gm=expected_gm, observed_gm=observed_last_gm, samples=sample_count, passed=passed, ) ) print( f"phase={phase.name} expected={expected_gm} observed={observed_last_gm} " f"samples={sample_count} result={'PASS' if passed else 'FAIL'}" ) send_syslog_marker( syslog_sock, syslog_cfg, "PHASE_END", { "name": phase.name, "result": "PASS" if passed else "FAIL", "expected": expected_gm, "observed": observed_last_gm, "samples": sample_count, }, ) print("\n=== BMC Phase Summary ===") for result in phase_results: print( f"{result.name}: {'PASS' if result.passed else 'FAIL'} " f"expected={result.expected_gm} observed={result.observed_gm} samples={result.samples}" ) total = len(phase_results) passed_total = sum(1 for r in phase_results if r.passed) print(f"BMC phase result: {passed_total}/{total} phases passed") send_syslog_marker(syslog_sock, syslog_cfg, "TEST_PHASE_SUMMARY", {"passed": passed_total, "total": total}) finally: stop_evt.set() sender.join(timeout=2.0) send_syslog_marker(syslog_sock, syslog_cfg, "TEST_END") if syslog_sock is not None: syslog_sock.close() io_sock.close() print("\nTest finished.") if last is None: print("No valid ParentDataSet response received from DUT.") return expected_good_gm = fmt_clock_identity(sender.good.grandmaster_identity) if last.grandmaster_identity == expected_good_gm: print("Result: DUT selected the expected (better) Grandmaster.") else: print("Result: DUT selected a different Grandmaster than expected.") print(f"Expected better GM: {expected_good_gm}") print(f"Observed GM: {last.grandmaster_identity}") if __name__ == "__main__": main()