Changeset 11060 in ntrip for trunk/BNC/tests/run_tests.py

Timestamp:

Oct 1, 2026, 5:35:37 PM (8 days ago)

Author:

stuerze

Message:

another test is added

File:

• trunk/BNC/tests/run_tests.py (modified) (4 diffs)

trunk/BNC/tests/run_tests.py

@@ -24 +24 @@
                  debug_trace  true for BNC_DEBUG_SSR stderr traces: reported as
                             SKIP instead of FAIL if the build has no trace output
+  fakecaster   optional local Ntrip caster for an upload during the run:
+               {"port", "out" (raw file, relative to the work dir), "sta",
+                "time" (epoch stamped on the raw file), "timeout"}
+  post         optional list of further BNC runs after the main run, each
+               {"dir" (subdirectory of the work dir), "config", "raw"
+                (relative to the work dir), "args", "exit_code", "timeout"};
+               outputs may refer to their files as "<dir>/Output/..."
 
 BNC runs in tests/work/<name>/, which contains an "Input" symlink to
@@ -146 +153 @@
 
 
+def run_bnc(bnc, workdir, spec, rawBase):
+    """Run BNC once in workdir (created with Input link and Output dir); spec
+    holds config, optional raw (relative to tests/, or to rawBase if given),
+    args and timeout. Returns the exit code or None on timeout."""
+    os.makedirs(os.path.join(workdir, "Output"), exist_ok=True)
+    if not os.path.exists(os.path.join(workdir, "Input")):
+        os.symlink(os.path.join(ROOT, "Example_Configs", "Input"), os.path.join(workdir, "Input"))
+    cmd = [bnc, "--nw", "--platform", "offscreen", "--conf", resolve(ROOT, spec["config"])]
+    if spec.get("raw"):
+        cmd += ["--file", resolve(rawBase or TESTS, spec["raw"])]
+    cmd += spec.get("args", [])
+    env = dict(os.environ, QT_QPA_PLATFORM="offscreen", LC_ALL="C", TZ="UTC")
+    with open(os.path.join(workdir, "stdout.txt"), "wb") as out, \
+         open(os.path.join(workdir, "stderr.txt"), "wb") as err:
+        try:
+            code = subprocess.run(cmd, cwd=workdir, stdout=out, stderr=err, env=env,
+                                  timeout=spec.get("timeout", 600)).returncode
+        except subprocess.TimeoutExpired:
+            code = None
+    with open(os.path.join(workdir, "command.txt"), "w") as f:
+        f.write(" ".join(cmd) + "\n")
+    return code
+
+
 def run_case(name, bnc, update):
     res = Result(name)
@@ -153 +184 @@
 
     shutil.rmtree(work, ignore_errors=True)
-    os.makedirs(os.path.join(work, "Output"))
-    os.symlink(os.path.join(ROOT, "Example_Configs", "Input"), os.path.join(work, "Input"))
-
-    cmd = [bnc, "--nw", "--platform", "offscreen",
-           "--conf", resolve(ROOT, case["config"])]
-    if case.get("raw"):
-        cmd += ["--file", resolve(TESTS, case["raw"])]
-    cmd += case.get("args", [])
-
-    env = dict(os.environ, QT_QPA_PLATFORM="offscreen", LC_ALL="C", TZ="UTC")
     t0 = time.time()
-    with open(os.path.join(work, "stdout.txt"), "wb") as out, \
-         open(os.path.join(work, "stderr.txt"), "wb") as err:
+
+    # Optional local Ntrip caster receiving an upload (see tools/fakecaster.py)
+    fc = case.get("fakecaster")
+    fcProc = None
+    if fc:
+        os.makedirs(os.path.join(work, "Output"))
+        fcProc = subprocess.Popen(
+            [sys.executable, os.path.join(TESTS, "tools", "fakecaster.py"),
+             "--port", str(fc["port"]), "--out", os.path.join(work, fc["out"]),
+             "--sta", fc.get("sta", "CLOCK"), "--time", fc["time"],
+             "--timeout", str(fc.get("timeout", 10))],
+            stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        fcProc.stdout.readline()  # "listening"
+
+    code = run_bnc(bnc, work, case, "")
+    if fcProc:
         try:
-            proc = subprocess.run(cmd, cwd=work, stdout=out, stderr=err, env=env,
-                                  timeout=case.get("timeout", 600))
-            code = proc.returncode
+            fcProc.communicate(timeout=fc.get("timeout", 10) + 60)
         except subprocess.TimeoutExpired:
-            code = None
-    res.seconds = time.time() - t0
-    with open(os.path.join(work, "command.txt"), "w") as f:
-        f.write(" ".join(cmd) + "\n")
+            fcProc.kill()
+            res.fail("fake caster did not finish")
 
     if code is None:
@@ -181 +212 @@
     if code != case.get("exit_code", 0):
         res.fail("exit code %d, expected %d" % (code, case.get("exit_code", 0)))
+
+    # Optional further BNC runs, e.g. decoding the uploaded stream; each runs
+    # in its own subdirectory of the work directory, raw relative to work/
+    for post in case.get("post", []):
+        pcode = run_bnc(bnc, os.path.join(work, post["dir"]), post, work)
+        if pcode is None or pcode != post.get("exit_code", 0):
+            res.fail("post step %s: exit code %s" % (post["dir"], pcode))
+    res.seconds = time.time() - t0
 
     # Collect (reference name, actual file, spec) triples