From 342ef3cf158cb39d1182bcce4b49da6da7b1f202 Mon Sep 17 00:00:00 2001
From: anurag <anurag@amd.com>
Date: Thu, 7 May 2026 15:42:20 -0600
Subject: [PATCH 1/2] add test flow

Signed-off-by: anurag <anurag@amd.com>
---
 .github/workflows/test.yml |  54 ++++++++
 ext/run.py                 | 170 ++++++++++++++++++++++++
 ext/run_tests.py           | 261 +++++++++++++++++++++++++++++++++++++
 3 files changed, 485 insertions(+)
 create mode 100644 .github/workflows/test.yml
 create mode 100644 ext/run.py
 create mode 100644 ext/run_tests.py

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
new file mode 100644
index 0000000..c151539
--- /dev/null
+++ b/.github/workflows/test.yml
@@ -0,0 +1,54 @@
+# This workflow will install Python dependencies, run tests and lint with a single version of Python
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: MLDebugger Test
+
+on:
+  pull_request:
+    branches: [ "main" ]
+  push:
+    branches: [ "main" ]
+  workflow_dispatch:
+
+permissions:
+  contents: write
+
+#env:
+#  # Differentiate manual testing from automated testing
+#  GIT_MODE: 1
+#  # pip install to conda env
+#  PYTHONNOUSERSITE: 1
+
+jobs:
+  build:
+    runs-on: [ vdi ]
+    steps:
+    - name: Set correct Git path
+      run: echo "/usr/bin" >> $GITHUB_PATH
+    - uses: actions/checkout@v4
+      with:
+        clean: true
+        lfs: true
+
+    - name: Pull LFS files
+      run: git lfs pull
+
+    - name: Lint with flake8, pylint
+      run: |
+        # stop the build if there are Python syntax errors or undefined names
+        flake8 . --exclude=.venv --count --select=E9,F63,F7,F82 --show-source --statistics --indent-size 2
+        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
+        # flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics --indent-size 2
+        find src/ -type f -name "*.py"  | xargs pylint --ignore=.venv --indent-string='  ' --exit-zero --max-line-length=120
+
+    - name: Update submodules
+      run: |
+        git submodule update --init --recursive
+        # Pull LFS files inside any submodules that use LFS
+        git submodule foreach --recursive 'git lfs pull || true'
+
+    - name: Run debugger with test backend
+      run: |
+        cd ext
+        git clone https://gitenterprise.xilinx.com/XDP/MLDebugTest
+        python run_tests.py
diff --git a/ext/run.py b/ext/run.py
new file mode 100644
index 0000000..ec52ab6
--- /dev/null
+++ b/ext/run.py
@@ -0,0 +1,170 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (C) 2024-2025 Advanced Micro Devices, Inc. All rights reserved.
+
+"""
+Runs onnx test and mldebugger
+"""
+
+import argparse
+import numpy as np
+import onnxruntime
+import os
+import shutil
+import subprocess
+import sys
+import threading
+
+from pathlib import Path
+from subprocess import PIPE
+
+os.environ["XLNX_ENABLE_CACHE"] = "0"
+os.environ["DEBUG_VAIML_PARTITION"] = "2"
+os.environ["DEBUG_LOG_LEVEL"] = "info"
+
+def setup_halt():
+  content = (
+    "[Debug]\n"
+    "aie_halt=true\n"
+    "[AIE_halt_settings]\n"
+    "control_code=aieHalt1x4x4.elf\n"
+    "[Runtime]\n"
+    "cert_timeout_ms = 0x99999999\n"
+  )
+  with open("xrt.ini", "w", encoding="utf-8") as fd:
+    fd.write(content)
+  shutil.copy2("MLDebug/ext/initial_halt_elfs/telluride/aieHalt1x4x4.elf", ".")
+
+class ProcessWithNotification:
+  def __init__(self, args):
+    self.continue_event = threading.Event()
+    self.halfway_event = threading.Event()
+    self.thread = None
+    self.args = args
+    self.cache_key = None
+
+  def _process_function(self):
+    model_file = Path(self.args.input_model)
+    self.cache_key = model_file.stem
+    provider_options_dict = {
+      "config_file": "vitisai_config.json",
+      "cache_dir": os.getcwd(),
+      "cache_key": self.cache_key,
+    }
+    session = onnxruntime.InferenceSession(
+      model_file,
+      # providers=["CPUExecutionProvider"],
+      providers=["VitisAIExecutionProvider"],
+      provider_options=[provider_options_dict],
+    )
+    input_name = {
+      inp.name: np.load(f"ifm_{count}.npy")
+      for count, inp in enumerate(session.get_inputs())
+    }
+    outputs = [x.name for x in session.get_outputs()]
+    ofm = None
+    try:
+      ofm = session.run(outputs, input_name)
+    except:
+      # print(f"ERROR: onnx_session failed run the inference with exception: {e}")
+      print("Reached MLDebugger Halt with hang")
+      self.halfway_event.set()
+      print("Waiting for signal to continue...\n")
+      self.continue_event.wait()
+      return
+    print("Reached MLDebugger Halt without hang")
+    self.halfway_event.set()
+    print("Waiting for signal to continue...\n")
+    self.continue_event.wait()
+    count = 0
+    for i, j in zip(outputs, ofm):
+      np.save(f"ofm_{count}.npy", j)
+      print(f"INFO: Saved OFM {i} size = {j.shape}")
+      count = count + 1
+    del session
+    for i in range(count):
+      ofm = np.load(f"ofm_{i}.npy")
+      ofm_ref = np.load(f"ofm_{i}_ref.npy")
+      max_diff = np.max(np.abs(ofm - ofm_ref))
+      print(f"Max absolute difference for ofm_{i}.npy: {max_diff}")
+
+  def start(self):
+    """Start the processing in a separate thread."""
+    self.thread = threading.Thread(target=self._process_function)
+    self.thread.start()
+    return self
+
+  def wait_for_halfway(self, timeout=None):
+    """Wait for the thread to reach the halfway point."""
+    return self.halfway_event.wait(timeout)
+
+  def continue_processing(self):
+    """Signal the thread to continue processing."""
+    self.continue_event.set()
+
+  def join(self, timeout=None):
+    """Wait for the thread to complete."""
+    if self.thread:
+      self.thread.join(timeout)
+
+def check_stdout(stdout):
+  """
+  Check if the output of the command contains valid results.
+  """
+  check_strings = [
+    "Finished Execution",
+    "Reached Final iteration",
+    "now exiting InteractiveConsole...",
+  ]
+  return any(check_str in stdout for check_str in check_strings)
+
+def run_mldebugger(processor):
+  os.chdir("MLDebug")
+  cmdseq = None
+  CMD = "python mldebug.py -v ../ -f l2_ifm_dump"
+  user_input = b""
+  if cmdseq:
+    # Simulate user input for interactive mode
+    user_input = cmdseq.encode("utf-8")
+  with subprocess.Popen(CMD, shell=True, stdout=PIPE, stderr=PIPE, stdin=PIPE) as process:
+    stdout, stderr = process.communicate(input=user_input)
+  stdout_txt = stdout.decode(encoding="utf-8")
+  stderr_txt = stderr.decode(encoding="utf-8")
+  with open("../test.log", "w", encoding="utf-8") as fd:
+    fd.write(stdout_txt)
+    fd.write(stderr_txt)
+  if process.returncode != 0 or not check_stdout(stdout_txt + stderr_txt):
+    print(stdout_txt, stderr_txt)
+    print(f"FAILED: {CMD} RET_CODE: {process.returncode}")
+    return 1
+  return 0
+
+def main():
+  parser = argparse.ArgumentParser(
+    description="onnxruntime compile and run using python API"
+  )
+  parser.add_argument(
+    "--input_model",
+    "-i",
+    help="Path to the model (.onnx for ONNX) for onnxruntime compile",
+  )
+  args = parser.parse_args()
+  setup_halt()
+  status = 0
+
+  processor = ProcessWithNotification(args)
+  processor.start()
+  print("Main thread: Wait for halfway")
+  processor.wait_for_halfway()
+  print("Main thread: Received halt notification! Running MLDebugger")
+  status = run_mldebugger(processor)
+  print("Finished MLDebugger")
+  processor.continue_processing()
+  processor.join()
+  if status != 0:
+    print("FAILED: MLDebugger")
+  else:
+    print("SUCCESS")
+  sys.exit(status)
+
+if __name__ == "__main__":
+  main()
diff --git a/ext/run_tests.py b/ext/run_tests.py
new file mode 100644
index 0000000..2249ad3
--- /dev/null
+++ b/ext/run_tests.py
@@ -0,0 +1,261 @@
+#!/usr/bin/env python3
+
+"""
+Test script for mldebugger
+"""
+
+from pathlib import Path
+
+import subprocess
+import time
+import os
+import sys
+import filecmp
+
+# Change to parent directory
+os.chdir("..")
+TEST_DIR="ext/test_outputs"
+Path(TEST_DIR).mkdir(parents=True, exist_ok=True)
+
+if os.getenv("GIT_MODE"):
+  CMD = "mldebug -x test "
+else:
+  CMD = "python mldebug.py -x test "
+
+BATCH_TESTS = {
+  "NLP": f"{CMD} -a ext/tests/nlp -b ext/tests/nlp/buffer_info.json -f skip_dump --verbose",
+  "STAMPED_DESIGN1": f"{CMD} -a ext/tests/stamped2 -b ext/tests/stamped2/buffer_info.json -f multistamp skip_dump",
+  "STAMPED_DESIGN2": f"{CMD} -a ext/tests/stamped -b ext/tests/stamped/buffer_info.json -f multistamp skip_dump -o 2x4x4",
+  "PEANO_BATCH": f"{CMD} -a ext/tests/peano -b ext/tests/peano/buffer_info.json -f l2_ifm_dump --peano",
+  "PEANO_L2_DUMP": f"{CMD} -a ext/tests/peano -b ext/tests/peano/buffer_info.json -f l2_ifm_dump --peano -e 15",
+  "WTS_ITER_FLAGS": f"{CMD} -a ext/tests/wts_iter -b ext/tests/wts_iter/buffer_info.json"
+  " -e 2 -f layer_status text_dump l1_ofm_dump",
+  "VAIML": f"{CMD} -v ext/tests/vaiml -f skip_dump",
+ # "X2": f"{CMD} -a ext/tests/x2 -b ext/tests/x2/buffer_info.json -f skip_dump",
+}
+
+INTERACTIVE_TESTS = {
+  "VAIML_INTERACTIVE": [f"{CMD} -v ext/tests/vaiml -i", "i\na\nv\nd\ns\nn\nc\nq", ""],
+  "VAIML_STANDALONE": [
+    f"{CMD} -v ext/tests/vaiml -s",
+    "info()\ngoto_pc(10)\nstatus()\nrmem(0,0,0,4)\nrreg(0,0,0)\npreg(0,0,0)\n"
+    "wreg(0,0,0,0)\nrlcp(0,2)\ncontrol_instr()\nfuncs()\nunhalt()\npc_brkpt(10, 0)\n"
+    "step()\npc(1)\n",
+    "",
+  ]
+  #"X2_INTERACTIVE": [f"{CMD}  -a ext/tests/x2 -b ext/tests/x2/buffer_info.json -i", "i\na\nv\nd\ns\nn\nc\nq", ""],
+}
+
+# Status tests that compare generated output with golden files
+STATUS_TESTS = {
+  "STATUS_STX": [f"{CMD} -d stx -s", f"status(advanced=True, filename='{TEST_DIR}/stx_st.log')\nexit()\n",
+                 "stx", f"{TEST_DIR}/stx_st.log"],
+  "STATUS_TELLURIDE": [f"{CMD} -d telluride -s", f"status(advanced=True, filename='{TEST_DIR}/tel_st.log')\nexit()\n",
+                       "telluride", f"{TEST_DIR}/tel_st.log"],
+}
+
+COREDUMP_TESTS = {
+
+  "COREDUMP_TELLURIDE": [f"{CMD} -d telluride -c ext/tests/coredump/telluride.bin -s",
+                         "status(advanced=True, filename='.cd_telluride')\nexit()\n", "golden_tel", ".cd_telluride"]
+}
+
+
+def check_stdout(stdout, extra_check_str):
+  """
+  Check if the output of the command contains valid results.
+  """
+  check_strings = [
+    "Finished Execution",
+    "Exiting debugger at Layer",
+    "Reached Final iteration",
+    "now exiting InteractiveConsole...",
+  ]
+  if extra_check_str and extra_check_str not in stdout:
+    return False
+  return any(check_str in stdout for check_str in check_strings)
+
+
+def run_test(name, command, cmdseq=None, extra_check_str=""):
+  """
+  Run a command in batch mode and log the output.
+  """
+  user_input = b""
+  if cmdseq:
+    # Simulate user input for interactive mode
+    user_input = cmdseq.encode("utf-8")
+  with subprocess.Popen(
+    command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE
+  ) as process:
+    stdout, stderr = process.communicate(input=user_input)
+  stdout_txt = stdout.decode(encoding="utf-8")
+  stderr_txt = stderr.decode(encoding="utf-8")
+  with open(f"{TEST_DIR}/{name}.log", "w", encoding="utf-8") as fd:
+    fd.write(stdout_txt)
+    fd.write(stderr_txt)
+  if process.returncode != 0 or not check_stdout(stdout_txt + stderr_txt, extra_check_str):
+    print(stdout_txt)
+    print(stderr_txt)
+    print(f"{name}: FAIL")
+    print(f"Command: {command}")
+    print(f"RET_CODE: {process.returncode}")
+    return False
+  print(f"{name}: PASS")
+  return True
+
+
+def run_status_test(name, command, cmdseq, device_name, generated_file, coredump=False):
+  """
+  Run a status test and compare the generated file with the golden file.
+  """
+  user_input = cmdseq.encode("utf-8")
+  with subprocess.Popen(
+    command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE
+  ) as process:
+    stdout, stderr = process.communicate(input=user_input)
+  stdout_txt = stdout.decode(encoding="utf-8")
+  stderr_txt = stderr.decode(encoding="utf-8")
+
+  # Log the output
+  with open(f"{TEST_DIR}/{name}.log", "w", encoding="utf-8") as fd:
+    fd.write(stdout_txt)
+    fd.write(stderr_txt)
+
+  # Check if the command executed successfully
+  if process.returncode != 0:
+    print(stdout_txt)
+    print(stderr_txt)
+    print(f"{name}: FAIL - Command failed with return code {process.returncode}")
+    return False
+
+  if coredump:
+    golden_file = "ext/tests/coredump/golden_tel"
+  else:
+    golden_file = f"ext/tests/status_golden/{device_name}"
+
+  if not os.path.exists(generated_file):
+    print(f"{name}: FAIL - Generated file '{generated_file}' not found")
+    return False
+
+  if not os.path.exists(golden_file):
+    print(f"{name}: FAIL - Golden file '{golden_file}' not found")
+    return False
+
+  # Compare files
+  if filecmp.cmp(generated_file, golden_file, shallow=False):
+    print(f"{name}: PASS")
+    return True
+  else:
+    print(f"{name}: FAIL - Generated file differs from golden file")
+    # Show diff for debugging
+    diff_result = subprocess.run(
+      ["diff", "-u", golden_file, generated_file],
+      capture_output=True,
+      text=True,
+      check=True
+    )
+    print(diff_result.stdout)
+    return False
+
+
+def run_guidance_tests():
+  """
+  Run guidance unit tests
+  """
+  print("\n" + "="*80)
+  print("Running Guidance Tests")
+  print("="*80)
+
+  # Run guidance unit tests
+  test_result = subprocess.run(
+    ["python", "tests/guidance/test_guidance.py"],
+    cwd="ext",
+    capture_output=True,
+    text=True,
+    check=True
+  )
+
+  print(test_result.stdout)
+  if test_result.stderr:
+    print(test_result.stderr)
+
+  if test_result.returncode != 0:
+    print("Guidance Tests: FAIL")
+    return False
+
+  print("Guidance Tests: PASS")
+
+  # Run integration tests
+  print("\n" + "="*80)
+  print("Running Guidance Integration Tests")
+  print("="*80)
+
+  test_result = subprocess.run(
+    ["python", "tests/guidance/test_integration.py"],
+    cwd="ext",
+    capture_output=True,
+    text=True,
+    check=True
+  )
+
+  print(test_result.stdout)
+  if test_result.stderr:
+    print(test_result.stderr)
+
+  if test_result.returncode != 0:
+    print("Guidance Integration Tests: FAIL")
+    return False
+
+  print("Guidance Integration Tests: PASS")
+  return True
+
+
+def test():
+  """
+  Toplevel
+  """
+  # Run guidance tests first
+  guidance_pass = run_guidance_tests()
+
+  print("Begin status tests")
+  status_pass = True
+  for test_name, [cmdline, cmdseq, device_name, gen_file] in STATUS_TESTS.items():
+    if not run_status_test(test_name, cmdline, cmdseq, device_name, gen_file):
+      status_pass = False
+
+  print("Begin coredump tests")
+  cd_pass = True
+  for test_name, [cmdline, cmdseq, device_name, gen_file] in COREDUMP_TESTS.items():
+    if not run_status_test(test_name, cmdline, cmdseq, device_name, gen_file, coredump=True):
+      cd_pass = False
+
+  print("Begin batch tests")
+  batch_pass = True
+  for test_name, cmdline in BATCH_TESTS.items():
+    if not run_test(test_name, cmdline, False):
+      batch_pass = False
+
+  print("Begin interactive tests")
+  interactive_pass = True
+  for test_name, [cmdline, cmdseq, extra_check_str] in INTERACTIVE_TESTS.items():
+    if not run_test(test_name, cmdline, cmdseq, extra_check_str):
+      interactive_pass = False
+
+  # Return 0 only if all tests pass
+  return 0 if (guidance_pass and batch_pass and interactive_pass and status_pass and cd_pass) else 1
+
+
+def main():
+  """
+  Main function to run the tests.
+  """
+  start_time = time.time()
+  status = test()
+  end_time = time.time()
+  elapsed_time = int(end_time - start_time)
+  print(f"\nElapsed time: {elapsed_time} seconds")
+  sys.exit(status)
+
+
+if __name__ == "__main__":
+  main()

From f0f1ab9a76ada41504e656de3ae24249e85c95d0 Mon Sep 17 00:00:00 2001
From: anurag <anurag@amd.com>
Date: Thu, 7 May 2026 15:44:58 -0600
Subject: [PATCH 2/2] fix path

Signed-off-by: anurag <anurag@amd.com>
---
 .github/workflows/test.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index c151539..36b26b3 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -50,5 +50,5 @@ jobs:
     - name: Run debugger with test backend
       run: |
         cd ext
-        git clone https://gitenterprise.xilinx.com/XDP/MLDebugTest
+        git clone https://gitenterprise.xilinx.com/XDP/MLDebugTest tests
         python run_tests.py