tests: verify STUN diagnostics without publishing a foreign-port profile

tests/stream_netns/node.c

@@ -197,7 +197,9 @@ static void statistics(struct lab *lab, uint64_t id, int loads) {
         if (!stun_nat_get_socket_result(lab->inst, socket, &stun)) {
             printf(",\"stun\":{\"members\":%u,\"generation\":\"%" PRIu64 "\",\"evidence\":\"%" PRIu64 "\",\"test_local\":",
                    stun.members, stun.generation, stun.evidence_id);
-            json_string(sockaddr_storage_to_str(&stun.local).str); printf("}");
+            json_string(sockaddr_storage_to_str(&stun.local).str);
+            printf(",\"status\":%d,\"mapping\":\"%s\",\"filtering\":\"%s\"}", stun.status,
+                   stun_nat_mapping_name(stun.mapping), stun_nat_filtering_name(stun.filtering));
         }
         printf(",\"phase\":%u,\"reason\":\"%s\",\"group\":\"%" PRIu64 "\",\"observer\":\"%" PRIu64 "\",\"helper\":\"%" PRIu64 "\","
                "\"rounds\":%u,\"helper_reselections\":%u,\"cycling\":%u,\"history_complete\":%u,\"contacts\":[", status.phase,

tests/stream_netns/stun_scenario.py

@@ -1,4 +1,4 @@
-"""STUN на другом порту: реальные NAT и переключение на независимый ETCP-раунд."""
+"""STUN другого порта остаётся диагностикой; рабочий профиль назначает независимый ETCP-раунд."""
 
 import json
 from pathlib import Path
@@ -43,11 +43,17 @@ def check_stun(scenario):
             agents[owner_name] = Agent(network, owner, Path(scenario.args.binary).resolve(), scenario.args.seed,
                                        False, False, True, config)
             agents[owner_name].rpc("nat_auto", enabled=1)
-            data = scenario.wait(f"stun-fallback-{owner_name}", lambda statistics:
-                                 all(item["source"] == "STUN" for item in statistics[owner_name]["address_profiles"]), 130)
+            data = scenario.wait(f"stun-diagnostics-{owner_name}", lambda statistics:
+                                 all(item.get("stun", {}).get("status") in (1, 2)
+                                     for item in statistics[owner_name]["address_profiles"]), 130)
             udp, duplicate = profile(data[owner_name], "udp"), profile(data[owner_name], "duplicate")
-            if (udp["mapping"], udp["inbound"]) != (mapping, inbound):
-                raise RuntimeError(f"{owner_name}: STUN class mismatch: {udp}")
+            expected_mapping = {"DEPENDENT", "ADM", "APDM"} if mapping == "DEPENDENT" else {mapping}
+            expected_filtering = {"EIF"} if inbound == "OPEN" else {"ADF", "APDF"}
+            if udp["stun"]["mapping"] not in expected_mapping or udp["stun"]["filtering"] not in expected_filtering:
+                raise RuntimeError(f"{owner_name}: STUN diagnostic class mismatch: {udp}")
+            if any(item["source"] != "NONE" or item["mapping"] != "UNKNOWN" or item["inbound"] != "UNKNOWN"
+                   for item in (udp, duplicate)):
+                raise RuntimeError(f"{owner_name}: STUN diagnostics changed a working endpoint: {udp}, {duplicate}")
             if udp["stun"]["evidence"] != duplicate["stun"]["evidence"] or udp["stun"]["members"] != 2 or udp["external"]:
                 raise RuntimeError(f"{owner_name}: duplicated probe or foreign external port: {udp}, {duplicate}")
             for name in (observer_name,):
@@ -55,17 +61,19 @@ def check_stun(scenario):
                                       False, False, True)
             agents[owner_name].rpc("nat_link", snapshot=agents[observer_name].rpc("snapshot")["snapshot"], socket="udp")
             scenario.wait(f"stun-observer-link-{owner_name}", lambda statistics:
-                          agents[observer_name].identity in statistics[owner_name]["direct_up"], 20)
+                          agents[observer_name].identity in statistics[owner_name]["group_ready"] and
+                          agents[owner_name].identity in statistics[observer_name]["group_ready"], 20)
             agents[owner_name].rpc("nat_check", socket="udp", observer=agents[observer_name].identity)
             data = scenario.wait(f"stun-no-helper-{owner_name}", lambda statistics:
                                  profile(statistics[owner_name], "udp")["reason"] == "no_helper", 40)
-            if profile(data[owner_name], "udp")["source"] != "STUN":
-                raise RuntimeError(f"{owner_name}: missing fallback with single observer")
+            if profile(data[owner_name], "udp")["source"] != "NONE":
+                raise RuntimeError(f"{owner_name}: single observer assigned an unmeasured endpoint profile")
             agents[helper_name] = Agent(network, nodes[helper_name], Path(scenario.args.binary).resolve(), scenario.args.seed,
                                         False, False, True)
             agents[observer_name].rpc("nat_link", snapshot=agents[helper_name].rpc("snapshot")["snapshot"], socket="udp")
             scenario.wait(f"stun-helper-link-{owner_name}", lambda statistics:
-                          agents[helper_name].identity in statistics[observer_name]["direct_up"], 20)
+                          agents[helper_name].identity in statistics[observer_name]["group_ready"] and
+                          agents[observer_name].identity in statistics[helper_name]["group_ready"], 20)
             agents[owner_name].rpc("nat_check", socket="udp", observer=agents[observer_name].identity)
             data = scenario.wait(f"stun-etcp-promotion-{owner_name}", lambda statistics:
                                  profile(statistics[owner_name], "udp")["source"] == "ETCP", 40)