Wire live stream scenarios into isolated bench launcher

tests/Makefile.am

@@ -2,7 +2,7 @@
 
 EXTRA_DIST = fixtures/reality_openssl_hello.hex data/silero_speech.pcm data/silero_speech.txt test_net.h \
     run_tests.py test_run_tests.py test_build_script.py stream_netns/topology.py stream_netns/network.py \
-    stream_netns/probe.py stream_netns/run.py
+    stream_netns/probe.py stream_netns/run.py stream_netns/scenario.py
 
 # Порядок по замерам Linux: быстрые проверки раньше длительных; состав зависит от прежних configure-условий.
 
@@ -190,6 +190,10 @@ if OS_LINUX
 EXTRA_PROGRAMS += stream_netns/stream_lab_node
 stream_netns_stream_lab_node_SOURCES = stream_netns/node.c
 stream_netns_stream_lab_node_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS)
+
+.PHONY: check-stream-netns
+check-stream-netns: stream_netns/stream_lab_node$(EXEEXT)
+	python3 "$(srcdir)/stream_netns/run.py" --binary "$(abs_builddir)/stream_netns/stream_lab_node$(EXEEXT)" $(STREAM_NETNS_ARGS)
 endif
 
 # Разрывы двух узлов и race трёх SQLite-баз; не входит в all/check/check.sh.

tests/stream_netns/run.py

@@ -13,6 +13,7 @@ import time
 
 from network import Network
 from probe import Probes
+from scenario import Scenario
 from topology import manifest
 
 
@@ -44,6 +45,7 @@ def inside(args):
     network = Network(directory, args.seed, args.host_netns)
     result = {"pass": False}
     probes = Probes(network)
+    scenario = None
     try:
         print("Creating 25 isolated network namespaces (16 nodes, 3 routers, 6 NAT gateways)", flush=True)
         network.build()
@@ -54,12 +56,21 @@ def inside(args):
         (directory / "calibration.json").write_text(json.dumps(calibration, indent=2) + "\n")
         probes.close()
         print("Network calibration passed", flush=True)
-        result["pass"] = True
+        if args.network_only:
+            result["pass"] = True
+        else:
+            scenario = Scenario(network, args)
+            result = scenario.run()
     except BaseException as error:
         result["error"] = str(error)
         print(f"FAIL: {type(error).__name__}: {error}", file=sys.stderr, flush=True)
     finally:
         probes.close()
+        if scenario:
+            failures = scenario.close()
+            if failures:
+                result["pass"] = False
+                result["agent_exit_errors"] = failures
         try:
             network.snapshot("final")
         except Exception as error:
@@ -74,9 +85,18 @@ def main():
     parser.add_argument("--output", help="New directory for logs and results")
     parser.add_argument("--seed", type=int, default=20261011)
     parser.add_argument("--network-only", action="store_true", help="Only topology, WAN and NAT calibration")
+    parser.add_argument("--binary", default=str(Path(__file__).parent / "stream_lab_node"))
+    parser.add_argument("--publisher", choices=[f"n{i:02d}" for i in range(1, 17)], default="n01")
+    parser.add_argument("--rate", type=int, default=1000000, help="Application bitrate in bits/s, before FEC")
+    parser.add_argument("--duration", type=int, default=120, help="Baseline duration in real seconds")
+    parser.add_argument("--faults", action="store_true", help="Degrade WAN, recover and check unsubscribe")
+    parser.add_argument("--fault-duration", type=int, default=30)
+    parser.add_argument("--tcp", action="store_true", help="Also expose TCP sockets, permitting ETCP multilink")
     parser.add_argument("--inside", action="store_true", help=argparse.SUPPRESS)
     parser.add_argument("--host-netns", help=argparse.SUPPRESS)
     args = parser.parse_args()
+    if not args.network_only and (not Path(args.binary).is_file() or args.rate <= 0 or args.duration < 1 or args.fault_duration < 1):
+        parser.error("Build stream_lab_node first; rate and durations must be positive")
     if args.inside:
         return inside(args)
     for program in ("ip", "tc", "nft", "ethtool", "sysctl", "unshare", "nsenter"):
@@ -91,9 +111,15 @@ def main():
     command = ["unshare", "--user", "--map-root-user", "--keep-caps", "--mount", "--net", "--pid", "--fork",
                "--mount-proc", "--propagation", "private", "--kill-child", "--", sys.executable,
                str(Path(__file__).resolve()), "--inside", "--host-netns", original_namespace,
-               "--output", str(directory), "--seed", str(args.seed)]
+               "--output", str(directory), "--seed", str(args.seed), "--binary", str(Path(args.binary).resolve()),
+               "--publisher", args.publisher, "--rate", str(args.rate), "--duration", str(args.duration),
+               "--fault-duration", str(args.fault_duration)]
     if args.network_only:
         command.append("--network-only")
+    if args.faults:
+        command.append("--faults")
+    if args.tcp:
+        command.append("--tcp")
     print(f"Artifacts: {directory}", flush=True)
     child = subprocess.Popen(command)
     try: