1  """Pytest harness for keccak_bench.
  2  
  3  Validates:
  4    - static test/vectors.json against hashlib (catches drift in the file)
  5    - `--validate-hex` path against hashlib (catches miner-side bugs)
  6    - a short hot-path run (real beat records)
  7  
  8  Run with:  pytest -xvs test/test_bench.py
  9  
 10  Or from the repo root:  make test
 11  """
 12  from __future__ import annotations
 13  
 14  import hashlib
 15  import json
 16  import pathlib
 17  import subprocess
 18  import sys
 19  
 20  HERE = pathlib.Path(__file__).resolve().parent
 21  REPO = HERE.parent
 22  VECTORS_PATH = HERE / "vectors.json"
 23  MINER_LTO = REPO / "src" / "keccak_bench_lto"
 24  MINER_PLAIN = REPO / "src" / "keccak_bench"
 25  
 26  
 27  def _load_vectors() -> dict:
 28      return json.loads(VECTORS_PATH.read_text())
 29  
 30  
 31  def test_vectors_file_self_consistent() -> None:
 32      data = _load_vectors()
 33      prefix = data["prefix"].encode()
 34      for v in data["vectors"]:
 35          x = bytes.fromhex(v["x_hex"])
 36          expected = hashlib.sha3_256(prefix + x).hexdigest()
 37          assert v["expected_hash_hex"] == expected, v
 38  
 39  
 40  def _run_validate_hex(miner: pathlib.Path, x_hex: str) -> str:
 41      out = subprocess.check_output([str(miner), "--validate-hex", x_hex],
 42                                    text=True, timeout=10)
 43      rec = json.loads(out.strip().splitlines()[-1])
 44      assert rec["kind"] == "validate"
 45      return rec["hash_hex"]
 46  
 47  
 48  def _test_vectors_for(miner: pathlib.Path) -> None:
 49      if not miner.exists():
 50          import pytest
 51          pytest.skip(f"miner missing: {miner} — run `make build`")
 52      data = _load_vectors()
 53      for v in data["vectors"]:
 54          x_hex = v["x_hex"]
 55          if not x_hex:
 56              # --validate-hex with empty string is handled as 0-length input.
 57              pass
 58          got = _run_validate_hex(miner, x_hex)
 59          assert got == v["expected_hash_hex"], f"{x_hex}: got={got} expected={v['expected_hash_hex']}"
 60  
 61  
 62  def test_vectors_via_miner_lto() -> None:
 63      _test_vectors_for(MINER_LTO)
 64  
 65  
 66  def test_vectors_via_miner_plain() -> None:
 67      _test_vectors_for(MINER_PLAIN)
 68  
 69  
 70  def test_hotpath_short() -> None:
 71      """Spawn the miner for a few seconds, verify emitted 'best' records
 72      re-hash with hashlib. Short run so the test suite stays fast."""
 73      if not MINER_LTO.exists():
 74          import pytest
 75          pytest.skip(f"miner missing: {MINER_LTO}")
 76      data = _load_vectors()
 77      prefix = data["prefix"].encode()
 78      p = subprocess.Popen(
 79          [str(MINER_LTO), "--workers", "1", "--salt-hex", "0102030405", "--source", "pytest"],
 80          stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1,
 81      )
 82      assert p.stdout is not None
 83      import time
 84      t0 = time.time()
 85      best_records: list[dict] = []
 86      canary_fail = False
 87      try:
 88          for line in p.stdout:
 89              if time.time() - t0 > 3.0:
 90                  break
 91              line = line.strip()
 92              if not line.startswith("{"):
 93                  continue
 94              try:
 95                  r = json.loads(line)
 96              except Exception:
 97                  continue
 98              if r.get("kind") == "canary_fail":
 99                  canary_fail = True
100                  break
101              if r.get("kind") == "best":
102                  best_records.append(r)
103      finally:
104          try:
105              p.terminate()
106              p.wait(timeout=5)
107          except Exception:
108              p.kill()
109      assert not canary_fail, "miner emitted canary_fail"
110      assert len(best_records) >= 5, f"expected >= 5 best records, got {len(best_records)}"
111      for r in best_records:
112          x = bytes.fromhex(r["input_hex"])
113          expected = hashlib.sha3_256(prefix + x).hexdigest()
114          assert r["hash_hex"] == expected, r
115  
116  
117  if __name__ == "__main__":
118      # Allow running as a script too: `python test/test_bench.py`
119      for fn in [test_vectors_file_self_consistent,
120                 test_vectors_via_miner_lto,
121                 test_vectors_via_miner_plain,
122                 test_hotpath_short]:
123          try:
124              fn()
125              print(f"OK  {fn.__name__}")
126          except Exception as e:  # noqa: BLE001
127              print(f"FAIL {fn.__name__}: {e}")
128              sys.exit(1)
129      print("all tests passed")