MT#64955 test-b2b: skip retransmitted B2B INVITEs from previous dialogs

recvB2BINVITE now tracks the Call-ID of every completed dialog in a
class-level set (_stale_b2b_cids) and skips any INVITE whose Call-ID
appears in that set.  tearDown populates the set from _current_b2b_cid,
which recvB2BINVITE sets on each successful return.

This closes the narrow race window left by the setUp non-blocking drain:
a retransmission that arrives between setUp and the recvB2BINVITE call
can no longer poison b2b_cid for the current test, regardless of timing.

Change-Id: I206ff35e0c04f80d4bcd12eaa4a7dd56acb08152

t/sems_tester.py

@@ -9,8 +9,13 @@ import unittest
 
 
 class TestCase(unittest.TestCase):
+    _stale_b2b_dialogs: typing.ClassVar[typing.Set] = set()
+    _current_b2b_dialogs: typing.ClassVar[typing.List] = []
+
     @classmethod
     def setUpClass(cls):
+        cls._stale_b2b_dialogs = set()
+        cls._current_b2b_dialogs = []
         tmpdir = tempfile.TemporaryDirectory()
         tmpsock = tmpdir.name + "/notify.sock"
 
@@ -132,12 +137,31 @@ class TestCase(unittest.TestCase):
         """Send SIP message to a specific address from a bound socket."""
         sock.sendto(msg.replace("\n", "\r\n").encode("utf-8"), addr)
 
+    def tearDown(self):
+        type(self)._stale_b2b_dialogs.update(type(self)._current_b2b_dialogs)
+        type(self)._current_b2b_dialogs = []
+
+    @staticmethod
+    def _b2b_dialog_key(msg: str) -> tuple:
+        cid      = re.search(r"Call-ID:\s*(\S+)", msg)
+        from_tag = re.search(r"From:[^\r\n]*tag=([A-Za-z0-9\-_\.]+)", msg)
+        to_tag   = re.search(r"To:[^\r\n]*tag=([A-Za-z0-9\-_\.]+)", msg)
+        return (
+            cid.group(1)      if cid      else "",
+            from_tag.group(1) if from_tag else "",
+            to_tag.group(1)   if to_tag   else "",
+        )
+
     def recvB2BINVITE(self, sock: socket.socket) -> typing.Tuple[str, tuple]:
-        """Receive a B2B INVITE from the shared UAS socket, skipping stale non-INVITE messages
-        (e.g. BYE/ACK retransmissions from the previous test that slipped past tearDown)."""
+        """Receive a B2B INVITE, skipping retransmissions from previous dialogs
+        identified by (Call-ID, From-tag, To-tag) per RFC 3261 §12."""
         for _ in range(10):
             msg, addr = self.recvFromSIP(sock)
             if msg.startswith("INVITE"):
+                key = self._b2b_dialog_key(msg)
+                if key in type(self)._stale_b2b_dialogs:
+                    continue
+                type(self)._current_b2b_dialogs.append(key)
                 return msg, addr
         raise AssertionError("No B2B INVITE received after 10 attempts")
 

t/test-b2b.py

@@ -290,6 +290,7 @@ def _to_tag(msg):
     return m.group(1) if m else ""
 
 
+
 def _sdp_media_port(msg):
     """Return the first m=audio port from a SIP message body."""
     m = _re.search(r"m=audio (\d+)", msg)
@@ -340,6 +341,7 @@ class TestB2B(sems_tester.TestCase):
         """Drain leftover messages from the shared UAS socket between tests.
         Uses 1.5*T1 timeout to catch SIP retransmissions that SEMS may still
         be generating on a loaded CI host when the next test starts."""
+        super().tearDown()
         self._uas_sock.settimeout(1.5)
         try:
             while True:
@@ -1585,13 +1587,7 @@ class TestB2B(sems_tester.TestCase):
         # --- Setup call ---
         self.sendSIP(_make_invite(branch, call_id, src_port=src_port), leg_a)
         self.recvSIP(leg_a)  # 100 Trying
-        b2b_invite = sems_addr = None
-        for _ in range(10):
-            msg, addr = self.recvFromSIP(self._uas_sock)
-            if msg.startswith("INVITE"):
-                b2b_invite, sems_addr = msg, addr
-                break
-        self.assertIsNotNone(b2b_invite, "SEMS did not forward INVITE to leg B")
+        b2b_invite, sems_addr = self.recvB2BINVITE(self._uas_sock)
         b2b_cid = _hdr(b2b_invite, "Call-ID")
         self.sendToSIP(_make_provisional(180, "Ringing", b2b_invite),
                        sems_addr, self._uas_sock)