#!/usr/bin/env python3 # SPDX-License-Identifier: MIT # Copyright (c) 2026 Olympian LED """Tests for current_probe_kit.py: every test end to end against the simulated SCPI scope on 127.0.0.1 (its firmware's habits as parameters), the restore, the refusals, and the published files kept free of addresses, serials and bench names. Run: python3 -I test_current_probe_kit.py""" import argparse import io import json import re import shutil import sys import tempfile import types import unittest from contextlib import redirect_stdout from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import current_probe_kit as kit # noqa: E402 def args(test, **kw): """The namespace main() builds for a run, without the command line.""" p = argparse.ArgumentParser() kit.add_run_args(p, test) a = p.parse_args([]) a.test, a.port, a.quiet = test, kit.SCPI_PORT, True for k, v in kw.items(): setattr(a, k, v) return a class KitCase(unittest.TestCase): def setUp(self): self.tmp = Path(tempfile.mkdtemp(prefix="cpkit-test-")) def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) def run_kit(self, test, sim=None, **kw): sim = sim or kit.SimScope(world=kit.SimWorld(sens=kit.RANGES[kw.get("range", "5A")]["sens"])) a = args(test, out_dir=str(self.tmp / test), state=str(self.tmp / "state.json"), **kw) try: code = kit.run_test(a, clock=kit.SimClock(), sim=sim) finally: sim.close() return code, a._last, sim def assertRestored(self, doc, sim): self.assertTrue(doc["restore"]["complete"], doc["restore"]["still_different"]) self.assertTrue(sim.restored(), "the simulated scope's settings are not what it started with") def assertClean(self, sim): sent = "\n".join(sim.lines) for bad in (r"^:?MEAS", r"^:?DISP(lay)?:DATA", r"^\*RST", r"^:?SAVE:IMAG", r"^[:*]?CAL", r"^:?AUT"): self.assertIsNone(re.search(bad, sent, re.I | re.M), "the kit sent %s" % bad) class TestAmps(KitCase): def test_guide_firmware_passes_and_restores(self): code, doc, sim = self.run_kit("amps", range="5A", firmware="stock") self.assertEqual(code, 0, doc.get("error")) self.assertTrue(doc["verdict"]) self.assertEqual(len([p for p in doc["points"] if "pass" in p]), 5) self.assertEqual(doc["results"]["channel"]["units"], "AMP") self.assertRestored(doc, sim) self.assertClean(sim) self.assertIn("unit_date_u", doc["stamp_start"]) self.assertTrue((self.tmp / "amps" / "amps-5A.json").is_file()) self.assertTrue((self.tmp / "amps" / "amps-5A.csv").is_file()) self.assertNotIn(sim.serial, json.dumps(doc)) # the trigger level and the cursor in amperes self.assertAlmostEqual(kit.parse_num(doc["results"]["trigger_level"]["read"]), 1.0) self.assertIn("cursor", doc["results"]) def test_every_command_logged(self): code, doc, sim = self.run_kit("amps", range="5A") sent = [e["send"] for e in doc["trace"] if e["dev"] == "scope"] self.assertEqual(sent, sim.lines) self.assertTrue(any(e["dev"] in ("load", "psu") for e in doc["trace"])) def test_30a_range_guide(self): code, doc, sim = self.run_kit("amps", range="30A") self.assertEqual(code, 0, doc.get("error")) self.assertTrue(doc["verdict"]) self.assertAlmostEqual(doc["results"]["channel"]["ratio_sent"], 0.1) def test_multiplier_firmware_is_a_finding(self): sim = kit.SimScope(world=kit.SimWorld(sens=0.1), convention="x") code, doc, _ = self.run_kit("amps", sim=sim, range="30A") self.assertEqual(code, 0, doc.get("error")) self.assertFalse(doc["verdict"]) ids = [f["id"] for f in doc["findings"]] self.assertIn("not-amps", ids) self.assertIn("ratio-convention", ids) self.assertAlmostEqual(doc["results"]["factor"]["median"], 0.01, places=3) self.assertAlmostEqual(doc["results"]["other_convention"]["factor"], 1.0, places=2) self.assertRestored(doc, sim) def test_wrong_probe_range_is_named(self): # the probe left on its 5 A range (1 V/A) while the run says 30 A sim = kit.SimScope(world=kit.SimWorld(sens=1.0)) code, doc, _ = self.run_kit("amps", sim=sim, range="30A") self.assertIn("other range", doc["results"]["factor"]["means"]) def test_units_not_taken_is_a_finding_not_a_failure(self): sim = kit.SimScope(units_ok=False) code, doc, _ = self.run_kit("amps", sim=sim, range="5A") self.assertEqual(code, 0, doc.get("error")) self.assertIn("units-not-taken", [f["id"] for f in doc["findings"]]) self.assertFalse(doc["verdict"]) self.assertRestored(doc, sim) def test_custom_ratio_by_turns(self): # three turns through the jaws: 3 V/A, off the guide's list sim = kit.SimScope(world=kit.SimWorld(sens=1.0, turns=3)) code, doc, _ = self.run_kit("amps", sim=sim, range="5A", turns=3, top=1.0) self.assertEqual(code, 0, doc.get("error")) self.assertTrue(doc["verdict"], doc["findings"]) self.assertTrue(any("custom ratio" in n for n in doc["notes"])) def test_custom_ratio_refused_is_a_finding(self): sim = kit.SimScope(world=kit.SimWorld(sens=1.0, turns=3), custom_ok=False) code, doc, _ = self.run_kit("amps", sim=sim, range="5A", turns=3, top=1.0) self.assertIn("custom-ratio-not-taken", [f["id"] for f in doc["findings"]]) def test_dho_has_no_impedance_command(self): sim = kit.SimScope(model="DHO924S") code, doc, _ = self.run_kit("amps", sim=sim, range="5A") self.assertEqual(code, 0, doc.get("error")) self.assertFalse(any("IMPedance" in l for l in sim.lines)) self.assertRestored(doc, sim) class TestOthers(KitCase): def test_power(self): code, doc, sim = self.run_kit("power", range="5A") self.assertEqual(code, 0, doc.get("error")) self.assertTrue(doc["verdict"]) self.assertTrue(all(abs(p["math_error_pct"]) < 2 for p in doc["points"])) self.assertRestored(doc, sim) self.assertClean(sim) def test_step_fires_every_single(self): code, doc, sim = self.run_kit("step", range="5A") self.assertEqual(code, 0, doc.get("error")) st = doc["results"]["step"] self.assertEqual(st["fired"], st["singles"]) self.assertTrue(st["pass"], st) self.assertAlmostEqual(st["lo"], 0.5, delta=0.01) self.assertAlmostEqual(st["hi"], 1.5, delta=0.01) self.assertRestored(doc, sim) def test_step_trigger_that_never_fires(self): # a level outside the step: NORMal sweep never stops; every single is reported as not fired old = dict(kit.STEP) kit.STEP.update(level=3.0, singles=3) try: code, doc, _ = self.run_kit("step", range="5A") finally: kit.STEP.clear() kit.STEP.update(old) self.assertEqual(doc["results"]["step"]["fired"], 0) self.assertFalse(doc["verdict"]) def test_drift(self): sim = kit.SimScope(world=kit.SimWorld(sens=1.0)) code, doc, _ = self.run_kit("drift", sim=sim, range="5A", minutes=5.0, interval=60.0, source="none") self.assertEqual(code, 0, doc.get("error")) self.assertEqual(doc["results"]["drift"]["samples"], 6) self.assertLess(abs(doc["results"]["drift"]["drift"]), 0.001) self.assertRestored(doc, sim) def test_persist_setup(self): code, doc, sim = self.run_kit("persist-setup", range="30A", source="none") self.assertEqual(code, 0, doc.get("error")) self.assertTrue(doc["verdict"]) self.assertIn(kit.SETUP_FILE, sim.files) self.assertRestored(doc, sim) def test_persist_setup_unsupported_is_a_finding(self): sim = kit.SimScope(file_ok=False) code, doc, _ = self.run_kit("persist-setup", sim=sim, range="30A", source="none") self.assertFalse(doc["verdict"]) self.assertIn("scpi-errors", [f["id"] for f in doc["findings"]]) def _power_cycle(self, keeps): sim = kit.SimScope(keeps=keeps) a = args("persist-arm", out_dir=str(self.tmp / "arm"), state=str(self.tmp / "state.json"), range="30A", source="none") self.assertEqual(kit.run_test(a, clock=kit.SimClock(), sim=sim), 0, a._last.get("error")) self.assertTrue((self.tmp / "state.json").is_file()) sim.power_cycle() b = args("persist-check", out_dir=str(self.tmp / "check"), state=str(self.tmp / "state.json"), range="30A", source="none") code = kit.run_test(b, clock=kit.SimClock(), sim=sim) sim.close() return code, b._last, sim def test_power_cycle_kept(self): code, doc, sim = self._power_cycle(True) self.assertEqual(code, 0, doc.get("error")) self.assertTrue(doc["verdict"]) self.assertRestored(doc, sim) def test_power_cycle_lost_is_a_finding(self): code, doc, sim = self._power_cycle(False) self.assertEqual(code, 0, doc.get("error")) self.assertFalse(doc["verdict"]) self.assertIn("power-cycle-units", [f["id"] for f in doc["findings"]]) self.assertRestored(doc, sim) class TestRefusals(unittest.TestCase): def test_never_sent(self): s = kit.Scpi("127.0.0.1", 1) for cmd in (":MEASure:ITEM? VAVG,CHANnel1", ":MEAS:ITEM VAVG", ":MEASure:DELete", ":DISPlay:DATA?", ":DISP:DATA? PNG", "*RST", ":SAVE:IMAGe C:/a.png", ":SAVE:SETup C:/x.stp", ":LOAD:SETup C:/x.stp", ":CALibrate:STARt", ":SYSTem:SETup?", ":AUToscale", ":SYSTem:REBoot"): with self.assertRaises(kit.KitError, msg=cmd): s.check(cmd) for ok in (":SAVE:OVERlap?", ":SAVE:OVERlap ON", ":CHANnel1:UNITs AMPere", ":WAVeform:DATA?"): s.check(ok) s.allow.add(":SAVE:SETup C:/x.stp") s.check(":SAVE:SETup C:/x.stp") def test_ladder(self): self.assertEqual([kit.ladder_up(x) for x in (0.033, 0.1, 0.967, 4.0, 0.0034)], [0.05, 0.1, 1.0, 5.0, 0.005]) def test_factor_meanings(self): self.assertEqual(kit.nearest_factor(1.0, 0.1)[0], "reads amps") self.assertIn("multiplies", kit.nearest_factor(0.0101, 0.1)[0]) self.assertIn("other range", kit.nearest_factor(9.8, 0.1)[0]) self.assertIn("other range", kit.nearest_factor(0.1, 1.0)[0]) class TestDryRun(unittest.TestCase): def test_dry_run_touches_nothing(self): tmp = tempfile.mkdtemp(prefix="cpkit-dry-test-") try: buf = io.StringIO() with redirect_stdout(buf): code = kit.main(["amps", "--host", "192.0.2.1", "--range", "5A", "--dry-run", "--out-dir", tmp]) self.assertEqual(code, 0) self.assertIn("DRY RUN", buf.getvalue()) self.assertIn("plan:", buf.getvalue()) finally: shutil.rmtree(tmp, ignore_errors=True) class TestHygiene(unittest.TestCase): PUBLIC = ("current_probe_kit.py", "test_current_probe_kit.py", "README.md", "METHOD.md", "LICENSE") # the published files name no bench unit, no internal tool, path or repository # (written in pieces, so that this file does not match itself) INTERNAL = re.compile("|".join(("led" "-fpga", "led" "_fpga", "Nova" "Star", "nova" "_devices", "nova" "-devtools", r"devices\.local", "load" "-1", "psu" "-1", "scope" "-1", r"mho" r"-\w", "bench" "_plan", "crates" "/", r"tools/local\.json", "Kisz" "ka")), re.I) def test_public_files_are_clean(self): for name in self.PUBLIC: text = (HERE / name).read_text(encoding="utf-8") hits = sorted(set(m.group(0) for m in self.INTERNAL.finditer(text))) self.assertEqual(hits, [], name) def test_no_addresses_or_serials(self): for f in HERE.iterdir(): if f.suffix in (".py", ".md", ".json"): text = f.read_text(encoding="utf-8") ips = [m for m in re.findall(r"(?