improve error reporting and update test cases for validation errors

Author: zhenrongliew

src/bloqade/analysis/validation/nocloning/analysis.py

@@ -7,11 +7,9 @@
 
 from bloqade.analysis.address import (
     Address,
-    AddressReg,
-    AddressQubit,
     AddressAnalysis,
 )
-from bloqade.analysis.address.lattice import QubitLike
+from bloqade.analysis.address.lattice import AddressReg, AddressQubit
 
 from .lattice import QubitValidation
 
@@ -28,7 +26,7 @@ class NoCloningValidation(Forward[QubitValidation]):
     _address_frame: ForwardFrame[Address] = field(init=False)
     _type_frame: ForwardFrame = field(init=False)
     method: ir.Method
-    violations: int = field(default=0, init=False)
+    _validation_errors: list[str] = field(default_factory=list, init=False)
 
     def __init__(self, mtd: ir.Method):
         """
@@ -41,7 +39,7 @@ def __init__(self, mtd: ir.Method):
 
     def initialize(self):
         super().initialize()
-
+        self._validation_errors = []
         address_analysis = AddressAnalysis(self.dialects)
         address_analysis.initialize()
         self._address_frame, _ = address_analysis.run_analysis(self.method)
@@ -88,7 +86,8 @@ def eval_stmt_fallback(
             return tuple(QubitValidation.top() for _ in stmt.results)
 
         has_qubit_args = any(
-            isinstance(address_frame.get(arg), QubitLike) for arg in stmt.args
+            isinstance(address_frame.get(arg), (AddressQubit, AddressReg))
+            for arg in stmt.args
         )
 
         if not has_qubit_args:
@@ -106,7 +105,10 @@ def eval_stmt_fallback(
 
         for qubit_addr in used_addrs:
             if qubit_addr in seen:
-                violations.append(f"Qubit[{qubit_addr}] at {stmt_info}")
+                violations.append(f"Qubit[{qubit_addr}] on {stmt_info}")
+                self._validation_errors.append(
+                    f"Qubit[{qubit_addr}] on {stmt_info} in {stmt.source}"
+                )
             seen.add(qubit_addr)
 
         if not violations:
@@ -120,3 +122,7 @@ def run_method(
     ) -> tuple[ForwardFrame[QubitValidation], QubitValidation]:
         self_mt = self.method_self(method)
         return self.run_callable(method.code, (self_mt,) + args)
+
+    def get_validation_errors(self) -> str:
+        """Retrieve collected validation error messages."""
+        return "\n".join(self._validation_errors)

test/analysis/validation/test_no_cloning.py

@@ -12,7 +12,7 @@
 
 
 @pytest.mark.parametrize("control_gate", [squin.cx, squin.cy, squin.cz])
-def test_control_gate_fail(control_gate: ir.Method[[Qubit, Qubit], Any]):
+def test_fail(control_gate: ir.Method[[Qubit, Qubit], Any]):
     @squin.kernel
     def bad_control():
         q = squin.qalloc(1)
@@ -28,7 +28,7 @@ def bad_control():
 
 
 @pytest.mark.parametrize("control_gate", [squin.cx, squin.cy, squin.cz])
-def test_control_gate_conditionals_fail(control_gate: ir.Method[[Qubit, Qubit], Any]):
+def test_conditionals_fail(control_gate: ir.Method[[Qubit, Qubit], Any]):
     @squin.kernel
     def bad_control(cond: bool):
         q = squin.qalloc(10)
@@ -44,12 +44,11 @@ def bad_control(cond: bool):
     print()
     bad_control.print(analysis=frame.entries)
     validation_errors = collect_validation_errors(frame, QubitValidation)
-    # print("Violations:", validation_errors)
     assert len(validation_errors) == 2
 
 
 @pytest.mark.parametrize("control_gate", [squin.cx, squin.cy, squin.cz])
-def test_control_gate_parallel_fail(control_gate: ir.Method[[Qubit, Qubit], Any]):
+def test_pass(control_gate: ir.Method[[Qubit, Qubit], Any]):
     @squin.kernel
     def bad_control():
         q = squin.qalloc(2)
@@ -64,7 +63,7 @@ def bad_control():
     assert len(validation_errors) == 0
 
 
-def test_control_gate_parallel_pass():
+def test_fail_2():
     @squin.kernel
     def good_kernel():
         q = squin.qalloc(2)
@@ -74,7 +73,22 @@ def good_kernel():
     validation = NoCloningValidation(good_kernel)
     validation.initialize()
     frame, _ = validation.run_analysis(good_kernel)
-    print()
-    good_kernel.print(analysis=frame.entries)
     validation_errors = collect_validation_errors(frame, QubitValidation)
     assert len(validation_errors) == 1
+    print(validation.get_validation_errors())
+
+
+def test_parallel_fail():
+    @squin.kernel
+    def bad_kernel():
+        q = squin.qalloc(5)
+        squin.broadcast.cx(IList([q[0], q[1], q[2]]), IList([q[1], q[2], q[3]]))
+
+    validation = NoCloningValidation(bad_kernel)
+    validation.initialize()
+    frame, _ = validation.run_analysis(bad_kernel)
+    print()
+    bad_kernel.print(analysis=frame.entries)
+    validation_errors = collect_validation_errors(frame, QubitValidation)
+    assert len(validation_errors) == 2
+    print(validation.get_validation_errors())

test/analysis/validation/util.py

@@ -1,17 +1,17 @@
-from typing import TypeVar
+from typing import List
 
 from kirin.analysis import ForwardFrame
 
 from bloqade.analysis.validation.nocloning.lattice import QubitValidation
 
-T = TypeVar("T", bound=QubitValidation)
-
 
 def collect_validation_errors(
-    frame: ForwardFrame[QubitValidation], typ: type[T]
-) -> list[T]:
-    return [
-        validation_errors
-        for validation_errors in frame.entries.values()
-        if isinstance(validation_errors, typ) and len(validation_errors.violations) > 0
-    ]
+    frame: ForwardFrame[QubitValidation], typ: type[QubitValidation]
+) -> List[str]:
+    """Collect individual violation strings from all QubitValidation entries of type `typ`."""
+    violations: List[str] = []
+    for validation_val in frame.entries.values():
+        if isinstance(validation_val, typ):
+            for v in getattr(validation_val, "violations", ()):
+                violations.append(v)
+    return violations