| import numpy as np
|
| import pytest
|
| from dense_evolution import DenseSVSimulator, GATES, PARAMETRIC_GATES, NoiseModel, NoiseSpec, QuantumTranspiler, QASMParser, Chunk
|
| from dense_evolution import QASMCircuit
|
| from dense_evolution import healing
|
|
|
| import inspect
|
| import jax
|
| import jax.numpy as jnp
|
|
|
|
|
| def patched_measure_for_tests(self, qubit_idx: int) -> int:
|
| """
|
| Misura un singolo qubit e collassa lo stato quantistico.
|
| """
|
| import numpy as np
|
|
|
| if not 0 <= qubit_idx < self.n:
|
| raise ValueError(f"Qubit {qubit_idx} out of bounds")
|
|
|
| xp = self.xp
|
|
|
| phys_q = self.n - 1 - qubit_idx
|
| stride = 1 << phys_q
|
|
|
| if xp is jnp:
|
|
|
| probs = self.xp.abs(self.sv)**2
|
| sv_shape = [2] * self.n
|
| sv_nd = probs.reshape(sv_shape)
|
|
|
| moved_probs = jnp.moveaxis(sv_nd, qubit_idx, 0)
|
| prob_0 = float(jnp.sum(moved_probs[0]))
|
| prob_1 = float(jnp.sum(moved_probs[1]))
|
| else:
|
|
|
| sv_reshaped = self.sv.reshape(-1, 2, stride)
|
| prob_0 = float(xp.sum(xp.abs(sv_reshaped[:, 0, :])**2))
|
| prob_1 = float(xp.sum(xp.abs(sv_reshaped[:, 1, :])**2))
|
|
|
| total = prob_0 + prob_1
|
| if total > 1e-12:
|
| prob_0 /= total
|
| prob_1 /= total
|
|
|
|
|
| result = int(np.random.choice([0, 1], p=[prob_0, prob_1]))
|
|
|
| if xp is jnp:
|
| sv_shape = [2] * self.n
|
| sv_nd = self.sv.reshape(sv_shape)
|
| moved_sv = jnp.moveaxis(sv_nd, qubit_idx, 0)
|
|
|
| moved_sv = moved_sv.at[1 - result].set(0.0)
|
| self.sv = jnp.moveaxis(moved_sv, 0, qubit_idx).ravel()
|
| else:
|
| sv_reshaped = self.sv.reshape(-1, 2, stride)
|
|
|
| sv_reshaped[:, 1 if result == 0 else 0, :] = 0.0
|
| self.sv = sv_reshaped.ravel()
|
|
|
| self.normalize()
|
| return result
|
|
|
|
|
| DenseSVSimulator.measure = patched_measure_for_tests
|
|
|
|
|
|
|
|
|
|
|
| @pytest.fixture
|
| def sim2():
|
| """Fresh 2-qubit simulator (NumPy CPU, float64)"""
|
| return DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
|
|
| @pytest.fixture
|
| def sim3():
|
| """Fresh 3-qubit simulator (NumPy CPU, float64)"""
|
| return DenseSVSimulator(n_qubits=3, use_gpu=False, use_float32=False)
|
|
|
| @pytest.fixture
|
| def sim4():
|
| """Fresh 4-qubit simulator (NumPy CPU, float64)"""
|
| return DenseSVSimulator(n_qubits=4, use_gpu=False, use_float32=False)
|
|
|
|
|
|
|
|
|
|
|
| def norm(sim):
|
| return float(np.linalg.norm(sim.get_statevector()))
|
|
|
| def probs(sim):
|
| return sim.get_probabilities()
|
|
|
|
|
|
|
|
|
|
|
| class TestInitialization:
|
|
|
| def test_initial_state_is_zero(self, sim2):
|
| sv = sim2.get_statevector()
|
| expected = np.zeros(4, dtype=complex)
|
| expected[0] = 1.0
|
| np.testing.assert_allclose(sv, expected, atol=1e-12)
|
|
|
| def test_initial_norm_is_one(self, sim2):
|
| assert abs(norm(sim2) - 1.0) < 1e-12
|
|
|
| def test_initial_probabilities(self, sim2):
|
| p = probs(sim2)
|
| assert abs(p[0] - 1.0) < 1e-12
|
| assert np.all(p[1:] < 1e-12)
|
|
|
| def test_custom_initial_state(self, sim2):
|
| sv_in = np.array([1, 0, 0, 1], dtype=complex) / np.sqrt(2)
|
| sim2.set_initial_state(sv_in)
|
| sv_out = sim2.get_statevector()
|
| np.testing.assert_allclose(np.abs(sv_out), np.abs(sv_in), atol=1e-12)
|
|
|
| def test_invalid_state_raises(self, sim2):
|
| with pytest.raises(ValueError):
|
| sim2.set_initial_state(np.array([1, 0, 0], dtype=complex))
|
|
|
| def test_zero_norm_state_raises(self, sim2):
|
| with pytest.raises(ValueError):
|
| sim2.set_initial_state(np.zeros(4, dtype=complex))
|
|
|
| def test_set_initial_state_explicit_none_resets_to_zero(self, sim2):
|
| sim2.set_initial_state(np.array([1, 0, 0, 1], dtype=complex) / np.sqrt(2))
|
| sim2.set_initial_state(None)
|
| sv = sim2.get_statevector()
|
| expected = np.zeros(4, dtype=complex)
|
| expected[0] = 1.0
|
| np.testing.assert_allclose(sv, expected, atol=1e-12)
|
|
|
| def test_set_state_is_an_alias_for_set_initial_state(self, sim2):
|
| sv_in = np.array([0, 1, 0, 0], dtype=complex)
|
| sim2.set_state(sv_in)
|
| np.testing.assert_allclose(np.abs(sim2.get_statevector()), np.abs(sv_in), atol=1e-12)
|
|
|
| def test_n_qubits_out_of_range_raises(self):
|
| with pytest.raises(ValueError):
|
| DenseSVSimulator(n_qubits=0)
|
| with pytest.raises(ValueError):
|
| DenseSVSimulator(n_qubits=35)
|
|
|
|
|
|
|
|
|
|
|
| class TestSingleQubitGates:
|
|
|
| def test_x_gate_flips_qubit(self, sim2):
|
| """X|0⟩ = |1⟩"""
|
| sim2.apply_gate_1q(GATES['x'], 0)
|
| p = probs(sim2)
|
|
|
| assert p[2] > 0.99
|
|
|
| def test_x_gate_double_application_identity(self, sim2):
|
| """XX = I"""
|
| sim2.apply_gate_1q(GATES['x'], 0)
|
| sim2.apply_gate_1q(GATES['x'], 0)
|
| p = probs(sim2)
|
| assert p[0] > 0.99
|
|
|
| def test_h_gate_creates_superposition(self, sim2):
|
| """H|0⟩ = (|0⟩+|1⟩)/√2 on qubit 0"""
|
| sim2.apply_gate_1q(GATES['h'], 0)
|
| p = probs(sim2)
|
| assert abs(p[0] - 0.5) < 1e-10
|
| assert abs(p[2] - 0.5) < 1e-10
|
|
|
| def test_h_gate_is_self_inverse(self, sim2):
|
| """HH = I"""
|
| sim2.apply_gate_1q(GATES['h'], 0)
|
| sim2.apply_gate_1q(GATES['h'], 0)
|
| p = probs(sim2)
|
| assert p[0] > 0.99
|
|
|
| def test_z_gate_on_zero_state_no_change(self, sim2):
|
| """Z|0⟩ = |0⟩ (phase change invisible in probabilities)"""
|
| sim2.apply_gate_1q(GATES['z'], 0)
|
| p = probs(sim2)
|
| assert p[0] > 0.99
|
|
|
| def test_z_gate_on_superposition_flips_phase(self, sim2):
|
| """Z applied after H: |+⟩ → |-⟩, then H gives |1⟩"""
|
| sim2.apply_gate_1q(GATES['h'], 0)
|
| sim2.apply_gate_1q(GATES['z'], 0)
|
| sim2.apply_gate_1q(GATES['h'], 0)
|
| p = probs(sim2)
|
|
|
| assert (p[2] + p[3]) > 0.99
|
|
|
| def test_norm_preserved_after_1q_gate(self, sim2):
|
| for g in ['h', 'x', 'y', 'z', 's', 't']:
|
| sim2.apply_gate_1q(GATES[g], 0)
|
| assert abs(norm(sim2) - 1.0) < 1e-12
|
|
|
| def test_out_of_bounds_qubit_raises(self, sim2):
|
| with pytest.raises((ValueError, IndexError)):
|
| sim2.apply_gate_1q(GATES['x'], 5)
|
|
|
|
|
| class TestQubitRangeValidationBypassesJIT:
|
| """run_circuit_jit_beast_mode / run_parametric_batch_jit build their own
|
| compiled_ops and never call apply_gate_1q/apply_gate_2q (which already
|
| validate) — an out-of-range qubit index there used to silently corrupt
|
| the entire statevector to zero instead of raising, because the fast
|
| JAX path encodes qubit indices as bit-shift amounts inside
|
| jax.lax.scan/switch with no bounds check. Verified before the fix:
|
| a single gate on an out-of-range qubit on an otherwise normalized
|
| state left get_probabilities().sum() == 0.0, no exception."""
|
|
|
| def test_beast_mode_1q_gate_out_of_range_raises(self, sim4):
|
| with pytest.raises(ValueError):
|
| sim4.run_circuit_jit_beast_mode([['x', 5, -1]])
|
|
|
| def test_beast_mode_2q_gate_out_of_range_raises(self, sim4):
|
| with pytest.raises(ValueError):
|
| sim4.run_circuit_jit_beast_mode([['cx', 0, 5]])
|
|
|
| def test_beast_mode_valid_circuit_unaffected(self, sim4):
|
|
|
| sim4.run_circuit_jit_beast_mode([['h', 0, -1], ['cx', 0, 1]])
|
| p = probs(sim4)
|
| assert abs(p.sum() - 1.0) < 1e-9
|
|
|
| def test_parametric_batch_qubit_out_of_range_raises(self, sim4):
|
| with pytest.raises(ValueError):
|
| sim4.run_parametric_batch_jit([['rx', 5]], np.zeros((1, 1)))
|
|
|
|
|
| class TestBeastModeGateDispatchGaps:
|
| """run_circuit_jit_beast_mode used to silently DROP cy/cp/crz/u1/p/sx —
|
| they weren't in GATE_IDS, so `if name not in GATE_IDS: continue` skipped
|
| them with no error (verified: h(0);h(1);crz(0,1,1.2) produced the exact
|
| same output as h(0);h(1) alone — the crz vanished). Fixed by adding the
|
| missing GATE_IDS entries and (for cy/crz/sx, which had no kernel at all)
|
| new branches in _apply_gate_fast_step. crz specifically needed its own
|
| kernel, not reuse of cp's: CP phases |11> only, CRZ phases the target
|
| conditioned on its own bit value — mathematically different gates."""
|
|
|
| def test_previously_dropped_gates_are_not_no_ops(self):
|
|
|
|
|
| cases = [
|
| ("cy", [('h', 0), ('cy', 0, 1)]),
|
| ("cp", [('h', 0), ('h', 1), ('cp', 0, 1, 0.7)]),
|
| ("crz", [('h', 0), ('h', 1), ('crz', 0, 1, 1.2)]),
|
| ("u1", [('h', 0), ('u1', 0, 0.9)]),
|
| ("p", [('h', 0), ('p', 0, 0.5)]),
|
| ("sx", [('sx', 0)]),
|
| ]
|
| for name, circuit in cases:
|
| sim_with = DenseSVSimulator(n_qubits=2)
|
| sim_with.run_circuit_jit_beast_mode(circuit)
|
| without = [c for c in circuit if c[0] != name]
|
| sim_without = DenseSVSimulator(n_qubits=2)
|
| sim_without.run_circuit_jit_beast_mode(without)
|
| assert not np.allclose(
|
| np.asarray(sim_with.get_statevector()),
|
| np.asarray(sim_without.get_statevector()), atol=1e-9,
|
| ), f"'{name}' still has no effect in beast mode"
|
|
|
| @pytest.mark.parametrize("name,circuit", [
|
| ("cy", [('h', 0), ('cy', 0, 1)]),
|
| ("cp", [('h', 0), ('h', 1), ('cp', 0, 1, 0.7)]),
|
| ("crz", [('h', 0), ('h', 1), ('crz', 0, 1, 1.2)]),
|
| ("u1", [('h', 0), ('u1', 0, 0.9)]),
|
| ("p", [('h', 0), ('p', 0, 0.5)]),
|
| ("sx_q0", [('sx', 0)]),
|
| ("sx_q1", [('sx', 1)]),
|
| ("cp_and_crz_and_cy_and_rx", [('rx', 0, 0.3), ('cx', 0, 1), ('cy', 1, 0), ('crz', 0, 1, 0.8), ('p', 1, 0.4)]),
|
| ])
|
| def test_matches_run_circuit(self, name, circuit):
|
|
|
|
|
|
|
|
|
| n = 2
|
| ref = DenseSVSimulator(n_qubits=n)
|
| ref.run_circuit(circuit)
|
| fast = DenseSVSimulator(n_qubits=n)
|
| fast.run_circuit_jit_beast_mode(circuit)
|
| np.testing.assert_allclose(
|
| np.asarray(ref.get_statevector()), np.asarray(fast.get_statevector()), atol=1e-9,
|
| )
|
|
|
| def test_crz_is_not_cp(self):
|
|
|
|
|
|
|
|
|
|
|
| sim_cp = DenseSVSimulator(n_qubits=2)
|
| sim_cp.run_circuit_jit_beast_mode([('x', 1), ('cp', 1, 0, 1.5)])
|
| sim_crz = DenseSVSimulator(n_qubits=2)
|
| sim_crz.run_circuit_jit_beast_mode([('x', 1), ('crz', 1, 0, 1.5)])
|
| assert not np.allclose(
|
| np.asarray(sim_cp.get_statevector()), np.asarray(sim_crz.get_statevector()), atol=1e-9,
|
| )
|
|
|
| def test_sx_squared_is_x(self):
|
|
|
| sim = DenseSVSimulator(n_qubits=1)
|
| sim.run_circuit_jit_beast_mode([('sx', 0), ('sx', 0)])
|
| p = probs(sim)
|
| assert p[1] > 0.999
|
|
|
| def test_previously_working_gates_unaffected(self, sim2):
|
|
|
|
|
| sim2.run_circuit_jit_beast_mode([('h', 0), ('cx', 0, 1), ('rz', 1, 0.6)])
|
| p = probs(sim2)
|
| assert abs(p.sum() - 1.0) < 1e-9
|
|
|
|
|
| class TestUnknownGateRaises:
|
| """Issue #4: a typo'd/unrecognized gate name used to be silently
|
| dropped from the circuit in all three execution paths instead of
|
| raising -- verified: h(0);ch(0,1);x(2) executed as if 'ch' wasn't
|
| there, no exception, no warning."""
|
|
|
| def test_run_circuit_raises_on_unknown_gate(self):
|
| sim = DenseSVSimulator(n_qubits=2)
|
| with pytest.raises(ValueError, match="unknown gate"):
|
| sim.run_circuit([('h', 0), ('ch', 0, 1)])
|
|
|
| def test_beast_mode_raises_on_unknown_gate(self):
|
| sim = DenseSVSimulator(n_qubits=2)
|
| with pytest.raises(ValueError, match="unknown gate"):
|
| sim.run_circuit_jit_beast_mode([('h', 0), ('ch', 0, 1)])
|
|
|
| def test_parametric_batch_raises_on_unknown_gate(self):
|
| sim = DenseSVSimulator(n_qubits=2)
|
| with pytest.raises(ValueError, match="unknown gate"):
|
| sim.run_parametric_batch_jit(
|
| [('h', 0), ('ch', 0, 1)], np.zeros((3, 0))
|
| )
|
|
|
| def test_known_gates_still_run_unaffected(self, sim2):
|
|
|
|
|
| sim2.run_circuit([('h', 0), ('cx', 0, 1), ('rz', 1, 0.6)])
|
| p = probs(sim2)
|
| assert abs(p.sum() - 1.0) < 1e-9
|
|
|
|
|
| class TestParametricBatchColumnMismatchRaises:
|
| """Issue #6: run_parametric_batch_jit assigns one parameter_batch
|
| column per parametric gate, in gate-appearance order -- including
|
| literal-float rotation gates, which silently ignored their literal
|
| and consumed a column anyway. A column-count mismatch used to be
|
| clipped silently by JAX's default out-of-bounds indexing instead of
|
| raising -- verified (pre-fix) with a statevector delta of 0.66
|
| against the intended circuit."""
|
|
|
| def test_too_few_columns_raises(self):
|
|
|
| sim = DenseSVSimulator(n_qubits=2)
|
| circuit = [('rx', 0, None), ('ry', 1, None)]
|
| with pytest.raises(ValueError, match="parameter_batch"):
|
| sim.run_parametric_batch_jit(circuit, np.zeros((5, 1)))
|
|
|
| def test_too_many_columns_raises(self):
|
| sim = DenseSVSimulator(n_qubits=2)
|
| circuit = [('rx', 0, None)]
|
| with pytest.raises(ValueError, match="parameter_batch"):
|
| sim.run_parametric_batch_jit(circuit, np.zeros((5, 2)))
|
|
|
| def test_literal_float_rotation_still_consumes_a_column(self):
|
|
|
|
|
| sim = DenseSVSimulator(n_qubits=2)
|
| circuit = [('rx', 0, 0.5), ('ry', 1, None)]
|
| with pytest.raises(ValueError, match="parameter_batch"):
|
| sim.run_parametric_batch_jit(circuit, np.zeros((5, 1)))
|
|
|
| def test_matching_column_count_runs_correctly(self):
|
| sim = DenseSVSimulator(n_qubits=2)
|
| circuit = [('rx', 0, None), ('ry', 1, None)]
|
| out = sim.run_parametric_batch_jit(circuit, np.zeros((3, 2)))
|
| assert out.shape == (3, 4)
|
|
|
|
|
| class TestBeastModeQubitOrdering:
|
| """run_circuit_jit_beast_mode used raw qubit index as bit position
|
| (LSB-first: qubit 0 = least significant bit) inside _apply_gate_fast_step
|
| (do_1q/do_2q), while the rest of the simulator — run_circuit(),
|
| apply_gate_1q(), apply_gate_2q(), measure() — uses the documented
|
| MSB-first convention (qubit 0 = most significant bit, phys = n-1-qubit,
|
| see simulator.py's class docstring and _qubit_stride_pairs). Pre-existing,
|
| not introduced by the cy/cp/crz/u1/p/sx dispatch fix above — found while
|
| verifying that fix, masked until then because every circuit tested this
|
| session against beast_mode happened to be symmetric under qubit reversal
|
| (Bell states, GHZ states, uniform superpositions). Fixed by computing
|
| physical bit positions (n_qubits-1-qubit) in do_1q/do_2q instead of using
|
| the raw qubit index directly."""
|
|
|
| def test_x_on_qubit_0_matches_msb_first_convention(self):
|
|
|
|
|
|
|
| sim = DenseSVSimulator(n_qubits=3)
|
| sim.run_circuit_jit_beast_mode([('x', 0)])
|
| p = probs(sim)
|
| assert p[4] > 0.999
|
| assert p[1] < 1e-9
|
|
|
| @pytest.mark.parametrize("circuit", [
|
| [('x', 0), ('cx', 0, 2)],
|
| [('x', 0), ('cy', 0, 2)],
|
| [('x', 0), ('crz', 0, 2, 1.1)],
|
| [('x', 1), ('cp', 1, 2, 0.8)],
|
| [('rx', 0, 0.4), ('ry', 1, 0.9), ('rz', 2, 1.3), ('cx', 0, 2), ('cx', 1, 2)],
|
| [('h', 0), ('rx', 1, 0.3), ('cx', 0, 1), ('cy', 1, 2), ('crz', 0, 2, 0.9),
|
| ('t', 0), ('sdg', 1), ('sx', 2), ('cp', 2, 0, 0.5)],
|
| ])
|
| def test_asymmetric_circuits_match_run_circuit(self, circuit):
|
| n = 3
|
| ref = DenseSVSimulator(n_qubits=n)
|
| ref.run_circuit(circuit)
|
| fast = DenseSVSimulator(n_qubits=n)
|
| fast.run_circuit_jit_beast_mode(circuit)
|
| np.testing.assert_allclose(
|
| np.asarray(ref.get_statevector()), np.asarray(fast.get_statevector()), atol=1e-9,
|
| )
|
|
|
| def test_run_parametric_batch_jit_matches_run_circuit(self):
|
|
|
| sim = DenseSVSimulator(n_qubits=3)
|
| batch = sim.run_parametric_batch_jit([('rx', 0, None), ('cx', 0, 2)], np.array([[0.5]]))
|
| ref = DenseSVSimulator(n_qubits=3)
|
| ref.run_circuit([('rx', 0, 0.5), ('cx', 0, 2)])
|
| np.testing.assert_allclose(np.asarray(batch[0]), ref.get_statevector(), atol=1e-9)
|
|
|
|
|
| class TestBeastModeFloat32:
|
| """use_float32=True used to crash unconditionally in run_circuit_jit_beast_mode
|
| (the JIT fast path) — not just for circuits with 2-qubit gates, even a
|
| circuit with only 1-qubit gates hit it, because jax.lax.cond traces
|
| every branch of _apply_gate_fast_step's dispatch (do_1q AND do_2q)
|
| regardless of which gates are actually present. Root cause: inside
|
| do_2q, apply_cp built its exp_pos constant hardcoded to complex128,
|
| while the identity branch of that same lax.cond (`lambda s: s`)
|
| preserved sv's real dtype (complex64 under use_float32=True) —
|
| 'cond branches must have equal output types but they differ'. Fixed by
|
| deriving every constant in _apply_gate_fast_step from sv.dtype instead
|
| of a hardcoded complex128."""
|
|
|
| def test_1q_only_circuit_runs_under_float32(self):
|
| sim = DenseSVSimulator(n_qubits=3, use_float32=True)
|
| sim.run_circuit_jit_beast_mode([['h', 0, -1], ['x', 1, -1]])
|
| assert sim.sv.dtype == np.complex64
|
| assert abs(float(np.sum(np.abs(np.asarray(sim.sv)) ** 2)) - 1.0) < 1e-6
|
|
|
| def test_2q_gates_run_under_float32(self):
|
| sim = DenseSVSimulator(n_qubits=4, use_float32=True)
|
| sim.run_circuit_jit_beast_mode(
|
| [['h', 0, -1], ['cx', 0, 1, 0], ['cz', 1, 2, 0], ['cp', 2, 3, 0.7]]
|
| )
|
| assert sim.sv.dtype == np.complex64
|
| assert abs(float(np.sum(np.abs(np.asarray(sim.sv)) ** 2)) - 1.0) < 1e-6
|
|
|
| def test_float32_matches_float64_within_precision(self):
|
| circuit = [
|
| ['h', 0, -1], ['h', 1, -1], ['rx', 2, 0.5], ['ry', 3, 1.1],
|
| ['cx', 0, 1, 0], ['cz', 1, 2, 0], ['cp', 2, 3, 0.7], ['crz', 0, 3, 1.3],
|
| ]
|
| sim32 = DenseSVSimulator(n_qubits=4, use_float32=True)
|
| sim32.run_circuit_jit_beast_mode(circuit)
|
| sim64 = DenseSVSimulator(n_qubits=4, use_float32=False)
|
| sim64.run_circuit_jit_beast_mode(circuit)
|
| np.testing.assert_allclose(probs(sim32), probs(sim64), atol=1e-6)
|
|
|
|
|
| class TestBeastModeDonateArgnums:
|
| """run_circuit_jit_beast_mode's self.sv = ... call used to allocate a
|
| fresh statevector buffer on every call instead of letting XLA reuse the
|
| memory of the one it's replacing — zero donate_argnums anywhere in the
|
| codebase, confirmed via audit. Only THIS call site is safe to donate:
|
| self.sv is always rebound immediately after, and no code path anywhere
|
| (including run_circuit_with_chunking's repeated calls, or separate
|
| DenseSVSimulator instances) keeps a stale reference to the old buffer
|
| across the call. run_parametric_batch_jit (vmap-broadcasts its init_sv
|
| closure across the whole batch) and circuit_to_energy_fn's energy_fn
|
| (the VQE loop reuses the same stato_zero every epoch) are NOT safe to
|
| donate — verified by tracing every call site of the shared
|
| _compile_and_run_circuit_jit before touching anything — so they keep
|
| using the plain, non-donating wrapper, untouched by this change."""
|
|
|
| def test_result_unchanged_by_donation(self):
|
| circuit = [['h', 0, -1], ['cx', 0, 1, 0], ['rz', 1, 0.6]]
|
| sim = DenseSVSimulator(n_qubits=3, use_float32=False)
|
| sim.run_circuit_jit_beast_mode(circuit)
|
| expected = probs(sim)
|
|
|
|
|
| sim2 = DenseSVSimulator(n_qubits=3, use_float32=False)
|
| sim2.run_circuit_jit_beast_mode(circuit)
|
| np.testing.assert_allclose(probs(sim2), expected, atol=1e-12)
|
| assert abs(expected.sum() - 1.0) < 1e-9
|
|
|
| def test_donation_actually_reuses_the_buffer(self):
|
|
|
|
|
|
|
|
|
|
|
| import jax
|
| sim = DenseSVSimulator(n_qubits=3, use_float32=False)
|
| old_sv = sim.sv
|
| sim.run_circuit_jit_beast_mode([['h', 0, -1]])
|
| with pytest.raises(RuntimeError, match="deleted"):
|
| jax.block_until_ready(old_sv)
|
|
|
| def test_chunked_repeated_calls_stay_correct(self):
|
|
|
|
|
|
|
|
|
| sim = DenseSVSimulator(n_qubits=4, use_float32=False)
|
| circuit = [('h', i % 4) for i in range(50)] + [('cx', i % 3, (i % 3) + 1) for i in range(50)]
|
| sim.run_circuit_with_chunking(circuit, chunk_size=7)
|
| assert abs(float(np.sum(probs(sim))) - 1.0) < 1e-9
|
|
|
| def test_memory_rss_donated_vs_non_donated(self, capsys):
|
|
|
|
|
|
|
|
|
|
|
| import gc
|
| import psutil
|
| from dense_evolution.compiler import (
|
| _compile_and_run_circuit_jit, _compile_and_run_circuit_jit_donated,
|
| )
|
|
|
| n_qubits = 22
|
| n_gates = 300
|
| sim_setup = DenseSVSimulator(n_qubits=n_qubits, use_float32=False)
|
|
|
| ops_jnp = jnp.array(
|
| [[1.0, float(i % n_qubits), 0.0, 0.0] for i in range(n_gates)],
|
| dtype=jnp.float64,
|
| )
|
|
|
| proc = psutil.Process()
|
|
|
| def run(fn):
|
| sv = sim_setup.sv
|
| gc.collect()
|
| before = proc.memory_info().rss
|
| out = fn(sv, ops_jnp)
|
| jnp.asarray(out).block_until_ready()
|
| gc.collect()
|
| after = proc.memory_info().rss
|
| return (after - before) / 1e6
|
|
|
|
|
| sim_setup.sv = jnp.zeros(2 ** n_qubits, dtype=jnp.complex128).at[0].set(1.0)
|
| delta_plain = run(_compile_and_run_circuit_jit)
|
| sim_setup.sv = jnp.zeros(2 ** n_qubits, dtype=jnp.complex128).at[0].set(1.0)
|
| delta_donated = run(_compile_and_run_circuit_jit_donated)
|
|
|
| with capsys.disabled():
|
| print(f"\n[donate_argnums RSS] n_qubits={n_qubits} "
|
| f"plain=+{delta_plain:.1f}MB donated=+{delta_donated:.1f}MB")
|
|
|
|
|
|
|
|
|
|
|
|
|
| class TestTwoQubitGates:
|
|
|
| def test_cx_on_zero_state_no_change(self, sim2):
|
| """CNOT with ctrl=0 in |0⟩: no flip"""
|
| sim2.apply_cx(0, 1)
|
| p = probs(sim2)
|
| assert p[0] > 0.99
|
|
|
| def test_cx_flips_target_when_control_is_one(self, sim2):
|
| """CNOT with ctrl=1: |10⟩ → |11⟩"""
|
| sim2.apply_gate_1q(GATES['x'], 0)
|
| sim2.apply_cx(0, 1)
|
| p = probs(sim2)
|
|
|
| assert p[3] > 0.99
|
|
|
| def test_cx_double_application_identity(self, sim2):
|
| sim2.apply_gate_1q(GATES['x'], 0)
|
| sim2.apply_cx(0, 1)
|
| sim2.apply_cx(0, 1)
|
| p = probs(sim2)
|
| assert p[2] > 0.99
|
|
|
| def test_cz_no_change_on_zero_state(self, sim2):
|
| sim2.apply_cz(0, 1)
|
| p = probs(sim2)
|
| assert p[0] > 0.99
|
|
|
| def test_norm_preserved_after_2q_gate(self, sim2):
|
| sim2.apply_gate_1q(GATES['h'], 0)
|
| sim2.apply_cx(0, 1)
|
| assert abs(norm(sim2) - 1.0) < 1e-12
|
|
|
| def test_invalid_qubit_indices_raise(self, sim2):
|
| with pytest.raises(ValueError):
|
| sim2.apply_cx(0, 0)
|
| with pytest.raises(ValueError):
|
| sim2.apply_cx(0, 5)
|
|
|
| def test_apply_gate_2q_direct_validation(self, sim2):
|
|
|
|
|
|
|
| with pytest.raises(ValueError):
|
| sim2.apply_gate_2q(np.eye(4), 0, 0)
|
| with pytest.raises(ValueError):
|
| sim2.apply_gate_2q(np.eye(4), 0, 5)
|
|
|
| def test_apply_cz_invalid_qubit_indices_raise(self, sim2):
|
| with pytest.raises(ValueError):
|
| sim2.apply_cz(1, 1)
|
| with pytest.raises(ValueError):
|
| sim2.apply_cz(0, 5)
|
|
|
|
|
|
|
|
|
|
|
| class TestGHZState:
|
|
|
| def test_ghz_3qubit_probabilities(self, sim3):
|
| """H-CX-CX: generates |000⟩+|111⟩ / √2"""
|
| circuit = [('h', 0), ('cx', 0, 1), ('cx', 1, 2)]
|
| sim3.run_circuit(circuit)
|
| p = probs(sim3)
|
| assert abs(p[0] - 0.5) < 1e-10
|
| assert abs(p[7] - 0.5) < 1e-10
|
|
|
| for i in [1, 2, 3, 4, 5, 6]:
|
| assert p[i] < 1e-10
|
|
|
| def test_ghz_norm(self, sim3):
|
| circuit = [('h', 0), ('cx', 0, 1), ('cx', 1, 2)]
|
| sim3.run_circuit(circuit)
|
| assert abs(norm(sim3) - 1.0) < 1e-12
|
|
|
| def test_ghz_statevector_shape(self, sim3):
|
| circuit = [('h', 0), ('cx', 0, 1), ('cx', 1, 2)]
|
| sim3.run_circuit(circuit)
|
| sv = sim3.get_statevector()
|
| assert sv.shape == (8,)
|
| assert sv.dtype == np.complex128
|
|
|
|
|
|
|
|
|
|
|
| class TestBellState:
|
|
|
| def test_bell_phi_plus(self, sim2):
|
| """H + CNOT creates |Φ+⟩ = (|00⟩+|11⟩)/√2"""
|
| sim2.apply_gate_1q(GATES['h'], 0)
|
| sim2.apply_cx(0, 1)
|
| p = probs(sim2)
|
| assert abs(p[0] - 0.5) < 1e-10
|
| assert abs(p[3] - 0.5) < 1e-10
|
| assert p[1] < 1e-10
|
| assert p[2] < 1e-10
|
|
|
| def test_bell_entanglement_norm(self, sim2):
|
| sim2.apply_gate_1q(GATES['h'], 0)
|
| sim2.apply_cx(0, 1)
|
| assert abs(norm(sim2) - 1.0) < 1e-12
|
|
|
|
|
|
|
|
|
|
|
| class TestParametricGates:
|
|
|
| def test_rx_pi_equals_x(self, sim2):
|
| """Rx(π)|0⟩ ≈ X|0⟩ up to global phase"""
|
| sim2.apply_rx(0, np.pi)
|
| p = probs(sim2)
|
| assert p[2] > 0.99
|
|
|
| def test_rz_no_change_in_probabilities(self, sim2):
|
| """Rz only changes phase, not populations"""
|
| p_before = probs(sim2).copy()
|
| sim2.apply_rz(0, np.pi / 3)
|
| p_after = probs(sim2)
|
| np.testing.assert_allclose(p_before, p_after, atol=1e-12)
|
|
|
| def test_ry_half_pi_superposition(self, sim2):
|
| """Ry(π/2)|0⟩ gives equal superposition"""
|
| sim2.apply_ry(0, np.pi / 2)
|
| p = probs(sim2)
|
| assert abs(p[0] - 0.5) < 1e-10
|
| assert abs(p[2] - 0.5) < 1e-10
|
|
|
| def test_norm_preserved_after_parametric(self, sim2):
|
| for theta in [0.1, np.pi / 4, np.pi / 2, np.pi]:
|
| sim2_local = DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
| sim2_local.apply_rx(0, theta)
|
| assert abs(norm(sim2_local) - 1.0) < 1e-12
|
|
|
| def test_run_circuit_u3_three_parameter_gate(self, sim2):
|
|
|
|
|
|
|
| sim2.run_circuit([('u3', 0, np.pi, 0.3, 0.7)], transpile=True)
|
| assert abs(norm(sim2) - 1.0) < 1e-12
|
|
|
| def test_run_parametric_batch_jit_cp_gate(self):
|
|
|
|
|
|
|
| sim = DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
| sim.apply_gate_1q(GATES['h'], 0)
|
| sim.apply_gate_1q(GATES['h'], 1)
|
| out = sim.run_parametric_batch_jit([('cp', 0, 1, None)], np.array([[0.5]]))
|
| out_np = np.asarray(out)
|
| assert out_np.shape == (1, 4)
|
| np.testing.assert_allclose(np.sum(np.abs(out_np) ** 2, axis=1), 1.0, atol=1e-6)
|
|
|
|
|
|
|
|
|
|
|
| class TestMeasurement:
|
|
|
| def test_measure_zero_state_returns_zero(self):
|
| sim = DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
| result = sim.measure(0)
|
| assert result == 0
|
|
|
| def test_measure_one_state_returns_one(self):
|
| sim = DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
| sim.apply_gate_1q(GATES['x'], 0)
|
| result = sim.measure(0)
|
| assert result == 1
|
|
|
| def test_measure_collapses_state_norm(self):
|
| sim = DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
| sim.apply_gate_1q(GATES['h'], 0)
|
| sim.measure(0)
|
| assert abs(norm(sim) - 1.0) < 1e-12
|
|
|
| def test_measure_returns_binary_value(self):
|
| sim = DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
| sim.apply_gate_1q(GATES['h'], 0)
|
| results = set()
|
| for _ in range(30):
|
| s = DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
| s.apply_gate_1q(GATES['h'], 0)
|
| results.add(s.measure(0))
|
| assert results == {0, 1}
|
|
|
| def test_measure_out_of_bounds_raises(self):
|
| sim = DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
| with pytest.raises(ValueError):
|
| sim.measure(5)
|
|
|
|
|
|
|
|
|
|
|
| class TestNoiseModel:
|
|
|
| def test_ideal_model_no_change(self):
|
| sv = np.array([1.0, 0.0], dtype=complex)
|
| sv_out = NoiseModel.apply_to_sv(sv, n=1, model='ideal', p=0.1)
|
| np.testing.assert_allclose(sv_out, sv, atol=1e-12)
|
|
|
| def test_depolarizing_preserves_norm(self):
|
| sv = np.array([1.0, 0.0], dtype=complex)
|
| rng = np.random.default_rng(42)
|
| sv_out = NoiseModel.apply_to_sv(sv, n=1, model='depolarizing', p=0.2, rng=rng)
|
| assert abs(np.linalg.norm(sv_out) - 1.0) < 1e-10
|
|
|
| def test_bitflip_preserves_norm(self):
|
| sv = np.array([1.0, 0.0], dtype=complex)
|
| rng = np.random.default_rng(7)
|
| sv_out = NoiseModel.apply_to_sv(sv, n=1, model='bitflip', p=0.3, rng=rng)
|
| assert abs(np.linalg.norm(sv_out) - 1.0) < 1e-10
|
|
|
| def test_phaseflip_preserves_norm(self):
|
| sv = np.array([1.0, 0.0], dtype=complex)
|
| rng = np.random.default_rng(99)
|
| sv_out = NoiseModel.apply_to_sv(sv, n=1, model='phaseflip', p=0.5, rng=rng)
|
| assert abs(np.linalg.norm(sv_out) - 1.0) < 1e-10
|
|
|
| def test_amplitude_damping_preserves_norm(self):
|
| sv = np.array([0.0, 1.0], dtype=complex)
|
| rng = np.random.default_rng(42)
|
| sv_out = NoiseModel.apply_to_sv(sv, n=1, model='amplitude_damping', p=0.2, rng=rng)
|
| assert abs(np.linalg.norm(sv_out) - 1.0) < 1e-10
|
|
|
| def test_zero_probability_no_change(self):
|
| sv = np.array([1.0, 0.0], dtype=complex)
|
| rng = np.random.default_rng(0)
|
| sv_out = NoiseModel.apply_to_sv(sv, n=1, model='depolarizing', p=0.0, rng=rng)
|
| np.testing.assert_allclose(sv_out, sv, atol=1e-12)
|
|
|
| def test_kraus_description_returns_dict(self):
|
| for model in NoiseModel.MODELS:
|
| desc = NoiseModel.kraus_description(model)
|
| assert isinstance(desc, dict)
|
| assert 'kraus' in desc
|
|
|
| def test_depolarizing_matches_analytic_prediction(self):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| rng = np.random.default_rng(42)
|
| sv1 = np.array([0.0, 1.0], dtype=complex)
|
| n_shots = 30000
|
| p = 0.3
|
| counts = np.zeros(2)
|
| for _ in range(n_shots):
|
| sv = NoiseModel.apply_to_sv(sv1.copy(), n=1, model='depolarizing', p=p, rng=rng)
|
| probs_ = np.abs(sv) ** 2
|
| probs_ /= probs_.sum()
|
| counts[rng.choice(2, p=probs_)] += 1
|
| freq = counts / n_shots
|
| expected = np.array([2 * p / 3, 1 - 2 * p / 3])
|
| np.testing.assert_allclose(freq, expected, atol=0.02)
|
|
|
| def test_depolarizing_pauli_choice_is_uniform_given_fire(self):
|
|
|
|
|
|
|
| rng = np.random.default_rng(0)
|
| n = 200000
|
| p = 0.3
|
| r = rng.random(n)
|
| ch = rng.random(n)
|
| fire = r < p
|
| third = 1.0 / 3.0
|
| x_gate = fire & (ch < third)
|
| y_gate = fire & (ch >= third) & (ch < 2 * third)
|
| z_gate = fire & (ch >= 2 * third)
|
| n_fire = fire.sum()
|
| assert x_gate.sum() / n_fire == pytest.approx(third, abs=0.01)
|
| assert y_gate.sum() / n_fire == pytest.approx(third, abs=0.01)
|
| assert z_gate.sum() / n_fire == pytest.approx(third, abs=0.01)
|
|
|
| def test_combined_model_depolarizing_subchannel_also_fixed(self):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| rng = np.random.default_rng(7)
|
| sv1 = np.array([0.0, 1.0], dtype=complex)
|
| n_shots = 30000
|
| p = 0.6
|
| p_dep = p * 0.5
|
| p_damp = p * 0.333333
|
| counts = np.zeros(2)
|
| for _ in range(n_shots):
|
| sv = NoiseModel.apply_to_sv(sv1.copy(), n=1, model='combined', p=p, rng=rng)
|
| probs_ = np.abs(sv) ** 2
|
| probs_ /= probs_.sum()
|
| counts[rng.choice(2, p=probs_)] += 1
|
| freq = counts / n_shots
|
| expected0 = 2 * p_dep / 3 + (1 - 2 * p_dep / 3) * p_damp
|
| assert freq[0] == pytest.approx(expected0, abs=0.02)
|
|
|
| def test_phaseflip_jax_array_branch(self):
|
|
|
|
|
|
|
|
|
| sv = jnp.array([1.0, 0.0], dtype=jnp.complex128)
|
| rng = np.random.default_rng(11)
|
| sv_out = NoiseModel.apply_to_sv(sv, n=1, model='phaseflip', p=0.5, rng=rng)
|
| assert isinstance(sv_out, jnp.ndarray)
|
| assert abs(float(jnp.linalg.norm(sv_out)) - 1.0) < 1e-6
|
|
|
| def test_combined_model_jax_array_branch(self):
|
| sv = jnp.array([0.0, 1.0], dtype=jnp.complex128)
|
| rng = np.random.default_rng(13)
|
| sv_out = NoiseModel.apply_to_sv(sv, n=1, model='combined', p=0.4, rng=rng)
|
| assert isinstance(sv_out, jnp.ndarray)
|
| assert abs(float(jnp.linalg.norm(sv_out)) - 1.0) < 1e-6
|
|
|
| def test_apply_to_sv_numpy_path_without_rng_uses_fresh_entropy(self):
|
|
|
|
|
| sv = np.array([1.0, 0.0], dtype=complex)
|
| sv_out = NoiseModel.apply_to_sv(sv, n=1, model='bitflip', p=0.5, rng=None)
|
| assert abs(np.linalg.norm(sv_out) - 1.0) < 1e-10
|
|
|
|
|
| class TestQuantumHardwareRegistry:
|
| def test_print_diagnostics_runs(self, capsys):
|
| from dense_evolution.registry import QuantumHardwareRegistry
|
| reg = QuantumHardwareRegistry()
|
| reg.print_diagnostics()
|
| captured = capsys.readouterr()
|
| assert "MAX_DENSE" in captured.out
|
|
|
| def test_detect_gpu_true_branch(self, monkeypatch):
|
| from dense_evolution.registry import QuantumHardwareRegistry
|
| import subprocess
|
| monkeypatch.setattr(subprocess, "check_output", lambda *a, **kw: b"fake gpu output")
|
| reg = QuantumHardwareRegistry()
|
| assert reg.has_gpu is True
|
|
|
| def test_noise_spec_repr(self):
|
| from dense_evolution import NoiseSpec
|
| spec = NoiseSpec(model="depolarizing", p=0.1, jax_key=jax.random.PRNGKey(0))
|
| r = repr(spec)
|
| assert "NoiseSpec" in r and "depolarizing" in r
|
|
|
|
|
| class TestNoiseModelRngJaxKeyUnification:
|
| """Issue #7: apply_to_sv used to pick rng vs jax_key based on the
|
| input array's type, not on which one was actually passed -- a JAX
|
| statevector silently ignored `rng` entirely and drew from OS entropy
|
| instead, so a caller seeding `rng` for reproducibility got a
|
| different, non-reproducible result every call with no signal
|
| anything was wrong. Fixed: `rng` (when given and `jax_key` isn't)
|
| now deterministically derives the JAX key too."""
|
|
|
| def test_seeded_rng_is_reproducible_on_jax_statevector(self):
|
| sv = jnp.array([0.5, 0.5, 0.5, 0.5], dtype=jnp.complex128)
|
| rng_a = np.random.default_rng(42)
|
| out_a = NoiseModel().apply_to_sv(sv, n=2, model='depolarizing', p=0.3, rng=rng_a)
|
| rng_b = np.random.default_rng(42)
|
| out_b = NoiseModel().apply_to_sv(sv, n=2, model='depolarizing', p=0.3, rng=rng_b)
|
| np.testing.assert_allclose(np.asarray(out_a), np.asarray(out_b))
|
|
|
| def test_seeded_rng_sequence_reproducible_across_multiple_calls(self):
|
| sv = jnp.array([0.5, 0.5, 0.5, 0.5], dtype=jnp.complex128)
|
|
|
| def run_sequence(seed):
|
| rng = np.random.default_rng(seed)
|
| o1 = NoiseModel().apply_to_sv(sv, n=2, model='depolarizing', p=0.3, rng=rng)
|
| o2 = NoiseModel().apply_to_sv(sv, n=2, model='depolarizing', p=0.3, rng=rng)
|
| o3 = NoiseModel().apply_to_sv(sv, n=2, model='depolarizing', p=0.3, rng=rng)
|
| return o1, o2, o3
|
|
|
| run1 = run_sequence(42)
|
| run2 = run_sequence(42)
|
| for a, b in zip(run1, run2):
|
| np.testing.assert_allclose(np.asarray(a), np.asarray(b))
|
|
|
| def test_explicit_jax_key_still_takes_precedence_over_rng(self):
|
| sv = jnp.array([0.5, 0.5, 0.5, 0.5], dtype=jnp.complex128)
|
| key = jax.random.PRNGKey(7)
|
| out1 = NoiseModel().apply_to_sv(
|
| sv, n=2, model='depolarizing', p=0.3, rng=np.random.default_rng(1), jax_key=key
|
| )
|
| out2 = NoiseModel().apply_to_sv(
|
| sv, n=2, model='depolarizing', p=0.3, rng=np.random.default_rng(999), jax_key=key
|
| )
|
| np.testing.assert_allclose(np.asarray(out1), np.asarray(out2))
|
|
|
| def test_numpy_path_unaffected(self):
|
| sv = np.array([0.5, 0.5, 0.5, 0.5], dtype=complex)
|
| out_a = NoiseModel().apply_to_sv(sv, n=2, model='depolarizing', p=0.3, rng=np.random.default_rng(42))
|
| out_b = NoiseModel().apply_to_sv(sv, n=2, model='depolarizing', p=0.3, rng=np.random.default_rng(42))
|
| np.testing.assert_allclose(out_a, out_b)
|
|
|
|
|
| class TestNoiseSpecPyTree:
|
| """NoiseSpec (issue #8): registers NoiseModel's parameters as a real
|
| JAX PyTree -- model/qubits static (aux_data), p/jax_key as leaves
|
| (children) -- so it can be passed through jax.jit/grad/vmap/scan the
|
| same way any other JAX-native value can, e.g. as circuit_to_energy_fn's
|
| `noise=` argument (see test_autodiff.py::TestEnergyFnNoiseSpec)."""
|
|
|
| def test_tree_flatten_unflatten_roundtrip(self):
|
| key = jax.random.PRNGKey(5)
|
| spec = NoiseSpec(model='depolarizing', p=0.2, jax_key=key, qubits=[0, 2])
|
| leaves, treedef = jax.tree_util.tree_flatten(spec)
|
| rebuilt = jax.tree_util.tree_unflatten(treedef, leaves)
|
| assert rebuilt.model == 'depolarizing'
|
| assert rebuilt.qubits == (0, 2)
|
| assert rebuilt.p == 0.2
|
| assert bool(jnp.array_equal(rebuilt.jax_key, key))
|
|
|
| def test_leaves_are_p_and_jax_key_only(self):
|
|
|
|
|
| spec = NoiseSpec(model='bitflip', p=0.1, jax_key=jax.random.PRNGKey(1), qubits=[0])
|
| leaves = jax.tree_util.tree_leaves(spec)
|
| assert len(leaves) == 2
|
|
|
| def test_jax_tree_map_transforms_leaves(self):
|
|
|
|
|
|
|
|
|
| spec = NoiseSpec(model='depolarizing', p=0.1, jax_key=jax.random.PRNGKey(0))
|
| doubled = jax.tree_util.tree_map(lambda x: x * 2 if jnp.ndim(x) == 0 else x, spec)
|
| assert doubled.p == pytest.approx(0.2)
|
|
|
|
|
|
|
|
|
|
|
|
|
| class TestTranspiler:
|
|
|
| def test_ccx_decomposition_length(self):
|
| result = QuantumTranspiler.decompose_toffoli(0, 1, 2)
|
| assert len(result) == 15
|
|
|
| def test_swap_decomposition_length(self):
|
| result = QuantumTranspiler.decompose_swap(0, 1)
|
| assert len(result) == 3
|
|
|
| def test_transpile_passes_through_basic_gates(self):
|
| circuit = [('h', 0), ('x', 1), ('cx', 0, 1)]
|
| result = QuantumTranspiler.transpile(circuit)
|
| assert result == circuit
|
|
|
| def test_transpile_expands_ccx(self):
|
| circuit = [('ccx', 0, 1, 2)]
|
| result = QuantumTranspiler.transpile(circuit)
|
| assert len(result) == 15
|
| assert all(op[0] in ('h', 'cx', 't', 'tdg') for op in result)
|
|
|
| def test_toffoli_correctness(self, sim3):
|
| """CCX|110⟩ = |111⟩"""
|
| sim3.apply_gate_1q(GATES['x'], 0)
|
| sim3.apply_gate_1q(GATES['x'], 1)
|
| sim3.run_circuit([('ccx', 0, 1, 2)])
|
| p = probs(sim3)
|
| assert p[7] > 0.99
|
|
|
| def test_toffoli_no_flip_without_both_controls(self, sim3):
|
| """CCX|100⟩ = |100⟩ (only one control active)"""
|
| sim3.apply_gate_1q(GATES['x'], 0)
|
| sim3.run_circuit([('ccx', 0, 1, 2)])
|
| p = probs(sim3)
|
| assert p[4] > 0.99
|
|
|
|
|
| class TestQASMRangeSyntax:
|
| """Regression guard for audit finding #2: `gate q[a:b]` on an inherently
|
| single-qubit gate used to attach all resolved qubits to ONE op, so only
|
| the first qubit was ever actually gated — the rest were silently dropped
|
| with no error, and probabilities still summed to 1. The parser's own
|
| docstring already promised "range syntax expanded to individual qubits";
|
| parse() now honors that by emitting one op per qubit instead of one op
|
| carrying the whole list."""
|
|
|
| def test_range_syntax_expands_to_separate_ops(self):
|
| qasm = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[4]; h q[0:3];'
|
| circ = QASMParser().parse(qasm)
|
| assert len(circ.ops) == 3
|
| assert [op['qubits'] for op in circ.ops] == [[0], [1], [2]]
|
| assert all(op['name'] == 'h' for op in circ.ops)
|
|
|
| def test_range_syntax_produces_correct_superposition(self, sim4):
|
| qasm = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[4]; h q[0:3];'
|
| circ = QASMParser().parse(qasm)
|
| sim4.run_circuit_jit_beast_mode([[op['name'], op['qubits'][0], -1] for op in circ.ops])
|
| p = probs(sim4)
|
|
|
| nonzero = np.where(p > 1e-9)[0]
|
| assert len(nonzero) == 8
|
| assert np.allclose(p[nonzero], 1.0 / 8, atol=1e-9)
|
|
|
| def test_two_qubit_gate_qubit_list_is_not_expanded(self):
|
|
|
|
|
| qasm = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[2]; cx q[0],q[1];'
|
| circ = QASMParser().parse(qasm)
|
| assert len(circ.ops) == 1
|
| assert circ.ops[0]['qubits'] == [0, 1]
|
|
|
|
|
| class TestQASMForLoop:
|
| """QASM 3.0 `for`-loops are brace-delimited, not ';'-terminated — the
|
| parser used to split statements on ';' alone, so a `for ... { ... }`
|
| block both lost its own body (never extracted) AND corrupted whatever
|
| real statement followed it on the same line (the stray closing '}'
|
| merged with the next statement's text into one garbage op). Verified
|
| directly: `for int i in [0:2] { h q[i]; } cx q[0],q[1];` used to produce
|
| a single ghost op named '}' and silently drop both the loop body and
|
| the real cx — the executed circuit stayed |000> at 100% probability
|
| with no error. _process_block_constructs now unrolls resolvable `for`
|
| loops and cleanly strips `if`/`while`/`def` blocks before the ';'-split
|
| ever runs, needed for VQE ansätze written with a loop over qubits."""
|
|
|
| def test_for_loop_body_extracted_and_following_gate_preserved(self):
|
| qasm = '''
|
| qreg q[3];
|
| for int i in [0:2] { h q[i]; }
|
| cx q[0], q[1];
|
| '''
|
| circ = QASMParser().parse(qasm)
|
| assert len(circ.ops) == 4
|
| assert [op['name'] for op in circ.ops] == ['h', 'h', 'h', 'cx']
|
| assert [op['qubits'] for op in circ.ops] == [[0], [1], [2], [0, 1]]
|
|
|
| def test_for_loop_executes_to_real_ghz_not_ghost_op(self, sim3):
|
| qasm = '''
|
| qreg q[3];
|
| for int i in [0:2] { h q[i]; }
|
| cx q[0], q[1];
|
| '''
|
| circ = QASMParser().parse(qasm)
|
| sim3.run_circuit(circ.to_tuples())
|
| p = probs(sim3)
|
|
|
| assert p[0] < 0.99
|
|
|
| def test_for_loop_bound_resolved_from_declared_int_variable(self):
|
| qasm = '''
|
| int n = 3;
|
| qreg q[3];
|
| for int i in [0:n-1] { rx(0.5) q[i]; }
|
| '''
|
| circ = QASMParser().parse(qasm)
|
| assert len(circ.ops) == 3
|
| assert all(op['name'] == 'rx' and op['params'] == [0.5] for op in circ.ops)
|
| assert [op['qubits'] for op in circ.ops] == [[0], [1], [2]]
|
|
|
| def test_for_range_is_inclusive_of_end_bound(self):
|
|
|
|
|
| qasm = 'qreg q[3]; for i in [0:2] { x q[i]; }'
|
| circ = QASMParser().parse(qasm)
|
| assert [op['qubits'][0] for op in circ.ops] == [0, 1, 2]
|
|
|
| def test_for_loop_body_with_multiple_statements_expands_all(self):
|
| qasm = 'qreg q[2]; for i in [0:1] { h q[i]; x q[i]; }'
|
| circ = QASMParser().parse(qasm)
|
| assert [op['name'] for op in circ.ops] == ['h', 'x', 'h', 'x']
|
| assert [op['qubits'][0] for op in circ.ops] == [0, 0, 1, 1]
|
|
|
| def test_if_block_does_not_corrupt_following_statement(self):
|
| qasm = 'qreg q[2]; if (c==1) { x q[0]; } h q[1];'
|
| circ = QASMParser().parse(qasm)
|
| assert len(circ.ops) == 1
|
| assert circ.ops[0] == {'type': 'gate', 'name': 'h', 'qubits': [1], 'params': []}
|
|
|
| def test_no_block_constructs_is_a_no_op(self):
|
|
|
|
|
| qasm = 'qreg q[2]; h q[0]; cx q[0], q[1];'
|
| circ = QASMParser().parse(qasm)
|
| assert [op['name'] for op in circ.ops] == ['h', 'cx']
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| def test_unresolvable_for_loop_bound_stripped_following_gate_preserved(self):
|
|
|
|
|
|
|
|
|
|
|
| qasm = '''
|
| qreg q[3];
|
| for int i in [0:n] { h q[i]; }
|
| cx q[0], q[1];
|
| '''
|
| circ = QASMParser().parse(qasm)
|
| assert [op['name'] for op in circ.ops] == ['cx']
|
| assert circ.ops[0]['qubits'] == [0, 1]
|
|
|
| def test_unresolvable_for_loop_stripped_execution_matches_bare_circuit(self, sim3):
|
|
|
|
|
|
|
| qasm_with_loop = '''
|
| qreg q[3];
|
| for int i in [0:n] { h q[i]; }
|
| cx q[0], q[1];
|
| '''
|
| circ = QASMParser().parse(qasm_with_loop)
|
| sim3.run_circuit(circ.to_tuples())
|
| p_with_loop = probs(sim3)
|
|
|
| ref = DenseSVSimulator(n_qubits=3, use_gpu=False, use_float32=False)
|
| ref_circ = QASMParser().parse('qreg q[3]; cx q[0], q[1];')
|
| ref.run_circuit(ref_circ.to_tuples())
|
| p_ref = probs(ref)
|
|
|
| np.testing.assert_allclose(p_with_loop, p_ref, atol=1e-12)
|
|
|
| def test_while_block_does_not_corrupt_following_statement(self):
|
| qasm = 'qreg q[2]; while (c==1) { x q[0]; } h q[1];'
|
| circ = QASMParser().parse(qasm)
|
| assert len(circ.ops) == 1
|
| assert circ.ops[0] == {'type': 'gate', 'name': 'h', 'qubits': [1], 'params': []}
|
|
|
| def test_multiple_unresolvable_constructs_in_sequence(self):
|
|
|
|
|
| qasm = '''
|
| qreg q[3];
|
| h q[0];
|
| for int i in [0:n] { x q[i]; }
|
| x q[1];
|
| if (c==1) { y q[0]; }
|
| y q[2];
|
| while (c==1) { z q[0]; }
|
| cx q[0], q[2];
|
| '''
|
| circ = QASMParser().parse(qasm)
|
| assert [op['name'] for op in circ.ops] == ['h', 'x', 'y', 'cx']
|
| assert [op['qubits'] for op in circ.ops] == [[0], [1], [2], [0, 2]]
|
|
|
| def test_resolvable_for_then_unresolvable_if_then_valid_code(self):
|
|
|
|
|
|
|
|
|
|
|
|
|
| qasm = '''
|
| qreg q[3];
|
| for int i in [0:1] { h q[i]; }
|
| if (some_undeclared_condition) { x q[2]; }
|
| cx q[0], q[1];
|
| '''
|
| circ = QASMParser().parse(qasm)
|
| assert [op['name'] for op in circ.ops] == ['h', 'h', 'cx']
|
| assert [op['qubits'] for op in circ.ops] == [[0], [1], [0, 1]]
|
|
|
|
|
| class TestQASMParserValidateAndEdgeCases:
|
| """QASMParser.validate() was never called by any existing test (only
|
| parse() itself), plus a handful of parser edge-case branches (QASM3
|
| `bit[N]` classical register syntax, unbalanced braces/parentheses,
|
| multi-parameter gates, and out-of-declared-range indexed qubits)."""
|
|
|
| def test_validate_accepts_well_formed_circuit(self):
|
| qasm = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[2]; h q[0]; cx q[0],q[1];'
|
| circ = QASMParser().parse(qasm)
|
| ok, msg = QASMParser().validate(circ)
|
| assert ok is True
|
| assert msg == 'OK'
|
|
|
| def test_validate_rejects_zero_qubits(self):
|
| empty = QASMCircuit(0, 0, [])
|
| ok, msg = QASMParser().validate(empty)
|
| assert ok is False
|
| assert 'n_qubits' in msg
|
|
|
| def test_validate_rejects_no_ops(self):
|
| no_ops = QASMCircuit(2, 0, [])
|
| ok, msg = QASMParser().validate(no_ops)
|
| assert ok is False
|
| assert 'No gate operations' in msg
|
|
|
| def test_validate_rejects_out_of_range_qubit_reference(self):
|
| bad = QASMCircuit(2, 0, [{'type': 'gate', 'name': 'h', 'qubits': [5], 'params': []}])
|
| ok, msg = QASMParser().validate(bad)
|
| assert ok is False
|
| assert 'qubit 5' in msg
|
|
|
| def test_qasm3_bit_declaration(self):
|
| qasm = 'OPENQASM 3.0; qreg q[2]; bit[2] c; h q[0]; cx q[0],q[1];'
|
| circ = QASMParser().parse(qasm)
|
| assert circ.n_cbits == 2
|
| assert [op['name'] for op in circ.ops] == ['h', 'cx']
|
|
|
| def test_unbalanced_parentheses_in_gate_call_skipped(self):
|
|
|
|
|
| qasm = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[2]; rx(0.5 q[0]; h q[1];'
|
| circ = QASMParser().parse(qasm)
|
| assert [op['name'] for op in circ.ops] == ['h']
|
|
|
| def test_unbalanced_braces_in_for_loop_bail_out(self):
|
|
|
|
|
|
|
| qasm = '''
|
| OPENQASM 3.0;
|
| qreg q[2];
|
| for int i in [0:1] { h q[i];
|
| cx q[0], q[1];
|
| '''
|
| circ = QASMParser().parse(qasm)
|
| assert isinstance(circ.ops, list)
|
|
|
| def test_multi_parameter_gate_comma_split(self):
|
|
|
|
|
|
|
| qasm = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[1]; u2(0.1,0.2) q[0];'
|
| circ = QASMParser().parse(qasm)
|
| assert circ.ops[0]['name'] == 'u2'
|
| assert circ.ops[0]['params'] == pytest.approx([0.1, 0.2])
|
|
|
| def test_indexed_qubit_beyond_declared_range_uses_numeric_fallback(self):
|
|
|
|
|
|
|
| qasm = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[2]; h q[5];'
|
| circ = QASMParser().parse(qasm)
|
| assert circ.ops[0]['qubits'] == [5]
|
|
|
|
|
| class TestQASMCircuitIterable:
|
| """Found via a user's own Colab testing: Chunk.run_chunk(circuit) (and
|
| QuantumTranspiler.transpile, which it calls internally) iterates
|
| directly over its `circuit` argument — `for cmd in circuit`. Passing a
|
| QASMCircuit straight from QASMParser().parse(...) (instead of calling
|
| .to_tuples() first) used to raise `TypeError: 'QASMCircuit' object is
|
| not iterable`, a real usability gap for a very natural usage pattern.
|
| Fixed by adding __iter__, duck-typing QASMCircuit as an iterable of the
|
| same tuples to_tuples() already returns — no existing call site inside
|
| dense_evolution relied on QASMCircuit being non-iterable."""
|
|
|
| def test_iterating_a_qasmcircuit_matches_to_tuples(self):
|
| circ = QASMParser().parse('qreg q[2]; h q[0]; cx q[0],q[1]; rz(0.5) q[1];')
|
| assert list(circ) == circ.to_tuples()
|
|
|
| def test_chunk_run_chunk_accepts_a_bare_qasmcircuit(self):
|
| circ = QASMParser().parse('qreg q[2]; h q[0]; cx q[0],q[1];')
|
| ch = Chunk(2)
|
| ch.run_chunk(circ)
|
| probs_ = np.asarray(ch.get_probabilities())
|
| assert abs(probs_.sum() - 1.0) < 1e-9
|
|
|
| def test_transpile_accepts_a_bare_qasmcircuit(self):
|
| circ = QASMParser().parse('qreg q[3]; ccx q[0],q[1],q[2];')
|
| expanded = QuantumTranspiler.transpile(circ)
|
| assert len(expanded) == 15
|
|
|
|
|
| class TestParserEvalSecurity:
|
| """_eval_param (gate parameters) and _resolve_int_expr (for-loop bounds)
|
| used to call raw eval() with only `{'__builtins__': {}}` as protection —
|
| that blocks bare builtin names (open, len, __import__...) but does
|
| NOT block attribute/dunder traversal of the live object graph, which
|
| needs no builtin name at all. Verified directly: a gate parameter of
|
| `().__class__.__bases__[0].__subclasses__().__len__()`, passed through
|
| the public QASMParser.parse() entry point, executed successfully and
|
| returned a real value (2200.0, the live subclass count) before this
|
| fix — a genuine code-execution vulnerability, not a hypothetical one.
|
| Both now go through _eval_ast_node, an AST node-type whitelist with no
|
| eval()/exec() anywhere — an attribute access is an ast.Attribute node,
|
| which is never one of the handled cases, so '.' in an expression always
|
| lands in the rejection branch structurally, not via a blocklist."""
|
|
|
| _ESCAPE_PAYLOADS = [
|
| '().__class__.__bases__[0].__subclasses__().__len__()',
|
| '__import__("os").system("echo pwned")',
|
| 'getattr(1, "__class__")',
|
| '[x for x in range(10)][0]',
|
| '(lambda: 1)()',
|
| 'exec("1")',
|
| 'globals()',
|
| '().__class__.__init__.__globals__',
|
| ]
|
|
|
| @pytest.mark.parametrize('payload', _ESCAPE_PAYLOADS)
|
| def test_eval_param_blocks_sandbox_escapes(self, payload):
|
|
|
|
|
|
|
|
|
|
|
|
|
| with pytest.raises(ValueError):
|
| QASMParser()._eval_param(payload)
|
|
|
| @pytest.mark.parametrize('payload', _ESCAPE_PAYLOADS)
|
| def test_resolve_int_expr_blocks_sandbox_escapes(self, payload):
|
| assert QASMParser()._resolve_int_expr(payload, {}) is None
|
|
|
| def test_original_exploit_through_full_parse_raises(self):
|
|
|
|
|
| qasm = ('OPENQASM 3.0; qubit[1] q; '
|
| 'rx(().__class__.__bases__[0].__subclasses__().__len__()) q[0];')
|
| with pytest.raises(ValueError):
|
| QASMParser().parse(qasm)
|
|
|
| def test_original_exploit_in_for_loop_bound_yields_no_ops(self):
|
| qasm = ('OPENQASM 3.0; qubit[1] q; '
|
| 'for int i in [0:().__class__.__bases__[0].__subclasses__().__len__()] '
|
| '{ x q[0]; }')
|
| circ = QASMParser().parse(qasm)
|
| assert circ.ops == []
|
|
|
| @pytest.mark.parametrize('expr,expected', [
|
| ('pi', np.pi), ('pi/2', np.pi / 2), ('-pi/4', -np.pi / 4),
|
| ('pi/8', np.pi / 8), ('0.5', 0.5), ('-0.5', -0.5),
|
| ('sqrt(2)', np.sqrt(2)), ('cos(0.3)', np.cos(0.3)),
|
| ('sin(pi/4)*2', np.sin(np.pi / 4) * 2),
|
| ('2*pi/3', 2 * np.pi / 3), ('1+2*3', 7.0),
|
| ])
|
| def test_legitimate_expressions_unchanged(self, expr, expected):
|
| assert QASMParser()._eval_param(expr) == pytest.approx(expected)
|
|
|
| @pytest.mark.parametrize('malformed', [
|
| 'pi * / 2', 'pi +', '(pi', 'pi 2', '**pi', 'pi // 2',
|
| ])
|
| def test_malformed_expression_raises_instead_of_silent_zero(self, malformed):
|
|
|
|
|
|
|
|
|
| with pytest.raises(ValueError):
|
| QASMParser()._eval_param(malformed)
|
|
|
| def test_malformed_expression_through_full_parse_raises(self):
|
| qasm = 'OPENQASM 2.0; include "qelib1.inc"; qreg q[1]; rx(pi * / 2) q[0];'
|
| with pytest.raises(ValueError):
|
| QASMParser().parse(qasm)
|
|
|
| def test_legitimate_for_loop_bounds_unchanged(self):
|
| qasm = 'qreg q[3]; int n = 3; for int i in [0:n-1] { x q[i]; }'
|
| circ = QASMParser().parse(qasm)
|
| assert [op['qubits'][0] for op in circ.ops] == [0, 1, 2]
|
|
|
|
|
|
|
|
|
|
|
|
|
| class TestCircuitChunking:
|
|
|
| def test_chunking_preserves_norm(self):
|
| """5000 H + 5000 CNOT on 4 qubits: norm must stay 1.0"""
|
| sim = DenseSVSimulator(n_qubits=4, use_gpu=False, use_float32=False)
|
| n_gates = 500
|
| circuit = [('h', i % 4) for i in range(n_gates // 2)]
|
| circuit += [('cx', i % 3, (i % 3) + 1) for i in range(n_gates // 2)]
|
| sim.run_circuit(circuit)
|
| assert abs(norm(sim) - 1.0) < 1e-10
|
|
|
| def test_run_circuit_with_chunking_exists(self):
|
| sim = DenseSVSimulator(n_qubits=2, use_gpu=False, use_float32=False)
|
| assert hasattr(sim, 'run_circuit_with_chunking') or hasattr(sim, 'run_circuit')
|
|
|
|
|
| class TestChunkPublicAPI:
|
| """Regression guard for audit finding #3: the README's own Anti-OOM
|
| quick-start (`from dense_evolution import Chunk`) raised ImportError —
|
| Chunk was never re-exported from the package root, only reachable via
|
| `dense_evolution.chunk`. Chunk also lacked get_probabilities()/
|
| get_statevector(), unlike DenseSVSimulator, so even the right import
|
| path needed an undocumented `np.abs(sim.sv)**2` workaround."""
|
|
|
| def test_chunk_importable_from_package_root(self):
|
|
|
| from dense_evolution import Chunk as ChunkFromRoot
|
| assert ChunkFromRoot is Chunk
|
|
|
| def test_chunk_get_probabilities_matches_manual_computation(self):
|
| sim = Chunk(6)
|
| sim.run_chunk([['h', i] for i in range(6)], 500)
|
| probs = np.asarray(sim.get_probabilities())
|
| manual = np.abs(np.asarray(sim.sv)) ** 2
|
| np.testing.assert_allclose(probs, manual, atol=1e-12)
|
| assert abs(probs.sum() - 1.0) < 1e-9
|
| assert np.allclose(probs, 1.0 / 64, atol=1e-9)
|
|
|
|
|
| class TestChunkMultiPiece:
|
| """Chunk(n_qubits) beyond the RAM-safe budget used to silently simulate
|
| FEWER qubits than requested (inner simulator sized to
|
| min(n_qubits, chunk_size_bits)) instead of real multi-chunk splitting —
|
| num_chunks/chunk_dim were computed but never acted on. Found testing
|
| Chunk(n_qubits=28) directly: get_probabilities() returned 2**27 elements,
|
| not 2**28.
|
|
|
| These tests force a small chunk_size_bits via monkeypatching
|
| get_dynamic_chunk (so num_chunks>1 is cheap to test) and cross-check the
|
| multi-chunk dispatch against a plain DenseSVSimulator(n_qubits) running
|
| the identical circuit — the only real correctness bar here, since this
|
| is bit-manipulation-heavy code where a plausible-looking-but-wrong
|
| formula is easy to miss by inspection alone."""
|
|
|
| @pytest.fixture
|
| def force_chunk_bits(self, monkeypatch):
|
| """Returns a function to force MemoryChunker's safe budget to a
|
| fixed small value, so num_chunks>1 can be tested without needing
|
| real multi-GB allocations."""
|
| import dense_evolution.chunk as chunk_mod
|
|
|
| def _force(bits):
|
| monkeypatch.setattr(chunk_mod, "get_dynamic_chunk", lambda dtype_target: bits)
|
|
|
| return _force
|
|
|
| def _compare_to_reference(self, n_qubits, circuit):
|
| c = Chunk(n_qubits)
|
| c.run_chunk(circuit)
|
| sv_chunk = np.asarray(c.get_statevector())
|
|
|
| ref = DenseSVSimulator(n_qubits)
|
| ref.run_circuit(circuit, transpile=True)
|
| sv_ref = np.asarray(ref.get_statevector())
|
|
|
| return sv_chunk, sv_ref
|
|
|
| def test_empty_circuit_canary(self, force_chunk_bits):
|
|
|
|
|
| force_chunk_bits(4)
|
| c = Chunk(6)
|
| assert c.num_chunks == 4
|
| probs = np.asarray(c.get_probabilities())
|
| assert probs.shape == (64,)
|
| assert abs(probs.sum() - 1.0) < 1e-9
|
| assert abs(probs[0] - 1.0) < 1e-9
|
|
|
| ref = DenseSVSimulator(6)
|
| np.testing.assert_allclose(np.asarray(c.get_statevector()), ref.get_statevector(), atol=1e-12)
|
|
|
| @pytest.mark.parametrize("qubit", [3])
|
| def test_1q_local(self, force_chunk_bits, qubit):
|
| force_chunk_bits(4)
|
| sv_chunk, sv_ref = self._compare_to_reference(6, [('h', qubit)])
|
| np.testing.assert_allclose(sv_chunk, sv_ref, atol=1e-9)
|
|
|
| @pytest.mark.parametrize("qubit", [0, 1])
|
| def test_1q_chunk_select(self, force_chunk_bits, qubit):
|
| force_chunk_bits(4)
|
| sv_chunk, sv_ref = self._compare_to_reference(6, [('h', qubit)])
|
| np.testing.assert_allclose(sv_chunk, sv_ref, atol=1e-9)
|
|
|
| def test_2q_local_local(self, force_chunk_bits):
|
| force_chunk_bits(4)
|
| sv_chunk, sv_ref = self._compare_to_reference(6, [('h', 3), ('cx', 3, 4)])
|
| np.testing.assert_allclose(sv_chunk, sv_ref, atol=1e-9)
|
|
|
| @pytest.mark.parametrize("gate", ["cx", "cz", "cy"])
|
| def test_2q_control_chunk_select_target_local(self, force_chunk_bits, gate):
|
| force_chunk_bits(4)
|
| sv_chunk, sv_ref = self._compare_to_reference(6, [('h', 0), (gate, 0, 3)])
|
| np.testing.assert_allclose(sv_chunk, sv_ref, atol=1e-9)
|
|
|
| @pytest.mark.parametrize("gate", ["cx", "cz", "cy"])
|
| def test_2q_control_local_target_chunk_select(self, force_chunk_bits, gate):
|
| force_chunk_bits(4)
|
| sv_chunk, sv_ref = self._compare_to_reference(6, [('h', 3), (gate, 3, 0)])
|
| np.testing.assert_allclose(sv_chunk, sv_ref, atol=1e-9)
|
|
|
| @pytest.mark.parametrize("gate", ["cx", "cz", "cy"])
|
| def test_2q_control_chunk_select_target_chunk_select(self, force_chunk_bits, gate):
|
| force_chunk_bits(4)
|
| sv_chunk, sv_ref = self._compare_to_reference(6, [('h', 0), ('h', 1), (gate, 0, 1)])
|
| np.testing.assert_allclose(sv_chunk, sv_ref, atol=1e-9)
|
|
|
| def test_parametric_2q_gates_all_four_locations(self, force_chunk_bits):
|
|
|
|
|
| force_chunk_bits(4)
|
| cases = [
|
| [('h', 0), ('cp', 0, 3, 0.7)],
|
| [('h', 3), ('crz', 3, 0, 1.1)],
|
| [('h', 0), ('h', 1), ('cp', 0, 1, 0.9)],
|
| [('h', 3), ('h', 4), ('crz', 3, 4, 0.4)],
|
| ]
|
| for circuit in cases:
|
| sv_chunk, sv_ref = self._compare_to_reference(6, circuit)
|
| np.testing.assert_allclose(sv_chunk, sv_ref, atol=1e-9)
|
|
|
| def test_random_mixed_circuits_num_chunks_8(self, force_chunk_bits):
|
|
|
|
|
|
|
|
|
| force_chunk_bits(4)
|
| n = 7
|
| rng = np.random.default_rng(1234)
|
| gates_1q = ['h', 'x', 'y', 'z', 's', 'sdg', 't', 'tdg']
|
| gates_1q_param = ['rx', 'ry', 'rz', 'p']
|
| gates_2q = ['cx', 'cz', 'cy']
|
| gates_2q_param = ['cp', 'crz']
|
|
|
| for _trial in range(8):
|
| circuit = []
|
| for _ in range(20):
|
| kind = rng.integers(0, 4)
|
| if kind == 0:
|
| circuit.append((rng.choice(gates_1q), int(rng.integers(0, n))))
|
| elif kind == 1:
|
| circuit.append((rng.choice(gates_1q_param), int(rng.integers(0, n)),
|
| float(rng.uniform(-3.14, 3.14))))
|
| elif kind == 2:
|
| q1, q2 = rng.choice(n, size=2, replace=False)
|
| circuit.append((rng.choice(gates_2q), int(q1), int(q2)))
|
| else:
|
| q1, q2 = rng.choice(n, size=2, replace=False)
|
| circuit.append((rng.choice(gates_2q_param), int(q1), int(q2),
|
| float(rng.uniform(-3.14, 3.14))))
|
| sv_chunk, sv_ref = self._compare_to_reference(n, circuit)
|
| np.testing.assert_allclose(sv_chunk, sv_ref, atol=1e-8,
|
| err_msg=f"circuit={circuit}")
|
|
|
| def test_num_chunks_1_path_untouched(self, force_chunk_bits):
|
|
|
|
|
| force_chunk_bits(16)
|
| c = Chunk(6)
|
| assert c.num_chunks == 1
|
| c.run_chunk([('h', i) for i in range(6)])
|
| probs = np.asarray(c.get_probabilities())
|
| assert np.allclose(probs, 1.0 / 64, atol=1e-9)
|
|
|
| def test_chunk_get_statevector_matches_sv(self):
|
| sim = Chunk(4)
|
| sim.run_chunk([['h', 0]], 500)
|
| np.testing.assert_array_equal(np.asarray(sim.get_statevector()), np.asarray(sim.sv))
|
|
|
| def test_2q_gate_spanning_first_to_last_qubit(self, force_chunk_bits):
|
|
|
|
|
|
|
|
|
|
|
| force_chunk_bits(4)
|
| sv_chunk, sv_ref = self._compare_to_reference(
|
| 6, [('h', 0), ('h', 5), ('cx', 0, 5)])
|
| np.testing.assert_allclose(sv_chunk, sv_ref, atol=1e-9)
|
|
|
|
|
|
|
| def test_memory_pressure_error_fires_on_insufficient_ram_num_chunks_1(self, monkeypatch):
|
| import dense_evolution.chunk as chunk_mod
|
|
|
| class _FakeVM:
|
| total = 8 * 1024 ** 3
|
| available = 0.05 * 8 * 1024 ** 3
|
| percent = 95.0
|
|
|
| monkeypatch.setattr(chunk_mod.psutil, "virtual_memory", lambda: _FakeVM())
|
| with pytest.raises(chunk_mod.MemoryPressureError, match="MEMORIA CRITICA"):
|
| Chunk(10, memory_threshold=0.15)
|
|
|
| def test_memory_pressure_error_fires_on_insufficient_ram_num_chunks_gt_1(self, monkeypatch, force_chunk_bits):
|
| import dense_evolution.chunk as chunk_mod
|
|
|
| force_chunk_bits(4)
|
|
|
| class _FakeVM:
|
| total = 8 * 1024 ** 3
|
| available = 0.05 * 8 * 1024 ** 3
|
| percent = 95.0
|
|
|
| monkeypatch.setattr(chunk_mod.psutil, "virtual_memory", lambda: _FakeVM())
|
| with pytest.raises(chunk_mod.MemoryPressureError, match="MEMORIA INSUFFICIENTE"):
|
| Chunk(6, memory_threshold=0.15)
|
|
|
| def test_memory_pressure_error_not_raised_with_ample_ram(self, monkeypatch, force_chunk_bits):
|
|
|
|
|
|
|
|
|
|
|
| import dense_evolution.chunk as chunk_mod
|
|
|
| force_chunk_bits(4)
|
|
|
| class _FakeVM:
|
| total = 8 * 1024 ** 3
|
| available = 0.90 * 8 * 1024 ** 3
|
| percent = 10.0
|
|
|
| monkeypatch.setattr(chunk_mod.psutil, "virtual_memory", lambda: _FakeVM())
|
| c = Chunk(6, memory_threshold=0.15)
|
| assert c.num_chunks == 4
|
|
|
|
|
| class TestChunkMultiPieceJIT:
|
| """dense_evolution.chunk's multi-chunk dispatch (num_chunks>1) used to
|
| apply gates one at a time via a Python loop calling
|
| sim.apply_gate_1q/apply_gate_2q (neither @jax.jit'd, no jax.lax.scan) —
|
| 6x slower than run_circuit_jit_beast_mode on an identical workload
|
| (10 qubits/200 gates/4 forced chunks: 2.2s vs 0.37s). Replaced with a
|
| single jax.lax.scan over the whole circuit, operating on the stacked
|
| (num_chunks, chunk_dim) representation directly (never materializing a
|
| (2**n_qubits,) array — the anti-OOM property Chunk exists for). The 6
|
| gate/qubit-location cases were ported formula-for-formula from the old
|
| Python-loop implementation (dense_evolution/chunk.py git history) and
|
| verified against it case-by-case before it was removed — TestChunkMultiPiece
|
| above (unchanged, all 18 tests still pass against the new kernel) is
|
| that cross-check. This class covers what's new: dtype and the actual
|
| measured speedup."""
|
|
|
| @pytest.fixture
|
| def force_chunk_bits(self, monkeypatch):
|
| import dense_evolution.chunk as chunk_mod
|
|
|
| def _force(bits):
|
| monkeypatch.setattr(chunk_mod, "get_dynamic_chunk", lambda dtype_target: bits)
|
|
|
| return _force
|
|
|
| @pytest.mark.parametrize("use_float32", [False, True])
|
| def test_dtype_consistency_across_all_cases(self, force_chunk_bits, use_float32):
|
|
|
|
|
|
|
|
|
|
|
| force_chunk_bits(4)
|
| n = 6
|
| circuit = [
|
| ('h', 3), ('rx', 4, 0.6),
|
| ('h', 0), ('ry', 1, 0.4),
|
| ('cx', 3, 4),
|
| ('cp', 0, 4, 0.7),
|
| ('crz', 4, 1, 1.1),
|
| ('cy', 0, 1),
|
| ]
|
| chunk_sim = Chunk(n, use_float32=use_float32)
|
| chunk_sim.run_chunk(circuit)
|
| ref = DenseSVSimulator(n, use_float32=use_float32)
|
| ref.run_circuit(circuit, transpile=True)
|
| atol = 1e-5 if use_float32 else 1e-9
|
| np.testing.assert_allclose(
|
| np.asarray(chunk_sim.get_statevector()), np.asarray(ref.get_statevector()), atol=atol)
|
|
|
| def test_measured_speedup_over_python_loop_dispatch(self, force_chunk_bits, capsys):
|
|
|
|
|
|
|
|
|
|
|
|
|
| import time
|
| force_chunk_bits(8)
|
| n_qubits = 10
|
| circuit = ([('h', i % n_qubits) for i in range(100)]
|
| + [('cx', i % (n_qubits - 1), (i % (n_qubits - 1)) + 1) for i in range(100)])
|
|
|
| chunk_sim = Chunk(n_qubits=n_qubits, memory_threshold=0.01)
|
| assert chunk_sim.num_chunks == 4
|
| t0 = time.perf_counter()
|
| chunk_sim.run_chunk(circuit)
|
| elapsed = time.perf_counter() - t0
|
|
|
| with capsys.disabled():
|
| print(f"\n[multi-chunk JIT speed] n_qubits={n_qubits} num_chunks=4 "
|
| f"200 gates: {elapsed:.4f}s (pre-fix Python-loop dispatch: ~2.2s)")
|
|
|
|
|
| class TestChunkUtilities:
|
| """Coverage-driven tests for chunk.py's smaller utility surfaces --
|
| get_dynamic_chunk's non-default dtype branches, SafeMemoryGuard/
|
| MemoryChunker's __repr__/geometry/validation, _compile_multi_chunk_ops's
|
| unknown-gate-skip and empty-circuit paths, CircuitChunker's no-simulator
|
| guard, and Chunk's property forwarders in num_chunks>1 mode (never
|
| exercised by TestChunkMultiPiece's own tests, which only ever compare
|
| statevectors/probabilities against DenseSVSimulator, not these
|
| introspection properties directly)."""
|
|
|
| def test_get_dynamic_chunk_numpy_complex128(self):
|
| from dense_evolution.chunk import get_dynamic_chunk
|
| bits = get_dynamic_chunk(np.complex128)
|
| assert 16 <= bits <= 27
|
|
|
| def test_get_dynamic_chunk_other_dtype(self):
|
| from dense_evolution.chunk import get_dynamic_chunk
|
| bits = get_dynamic_chunk(np.float32)
|
| assert 16 <= bits <= 27
|
|
|
| def test_safe_memory_guard_rejects_invalid_threshold(self):
|
| from dense_evolution.chunk import SafeMemoryGuard
|
| with pytest.raises(ValueError):
|
| SafeMemoryGuard(threshold_pct=0.0)
|
| with pytest.raises(ValueError):
|
| SafeMemoryGuard(threshold_pct=1.0)
|
| with pytest.raises(ValueError):
|
| SafeMemoryGuard(threshold_pct=-0.1)
|
|
|
| def test_safe_memory_guard_repr(self):
|
| from dense_evolution.chunk import SafeMemoryGuard
|
| guard = SafeMemoryGuard(threshold_pct=0.15)
|
| r = repr(guard)
|
| assert "SafeMemoryGuard" in r and "threshold=15%" in r
|
|
|
| def test_memory_chunker_geometry_and_repr(self):
|
| from dense_evolution.chunk import MemoryChunker
|
| mc = MemoryChunker(n_qubits=4)
|
| num_chunks, chunk_dim, chunk_size_bits = mc.geometry()
|
| assert num_chunks == mc.num_chunks
|
| assert chunk_dim == mc.chunk_dim
|
| assert chunk_size_bits == mc.chunk_size_bits
|
| r = repr(mc)
|
| assert "MemoryChunker" in r and "num_chunks=" in r
|
|
|
| def test_compile_multi_chunk_ops_skips_unknown_gate(self):
|
| from dense_evolution.chunk import _compile_multi_chunk_ops
|
|
|
|
|
|
|
| rows = _compile_multi_chunk_ops([('not_a_real_gate', 0), ('h', 0)])
|
| rows_np = np.asarray(rows)
|
| assert rows_np.shape[0] == 1
|
|
|
| def test_compile_multi_chunk_ops_empty_circuit(self):
|
| from dense_evolution.chunk import _compile_multi_chunk_ops
|
| rows = _compile_multi_chunk_ops([])
|
| rows_np = np.asarray(rows)
|
| assert rows_np.shape == (0, 4)
|
|
|
| def test_circuit_chunker_requires_simulator_instance(self):
|
| from dense_evolution.chunk import CircuitChunker
|
| chunker = CircuitChunker()
|
| with pytest.raises(RuntimeError, match="no simulator instance"):
|
| chunker.split_circuit([('h', 0)])
|
|
|
| def test_chunk_repr(self):
|
| c = Chunk(n_qubits=3)
|
| r = repr(c)
|
| assert "Chunk(" in r and "num_chunks=" in r
|
|
|
| def test_chunk_multi_piece_property_forwarding(self, monkeypatch):
|
|
|
|
|
|
|
|
|
| import dense_evolution.chunk as chunk_mod
|
| monkeypatch.setattr(chunk_mod, "get_dynamic_chunk", lambda dtype_target: 4)
|
|
|
| c = Chunk(n_qubits=6)
|
| assert c.num_chunks == 4
|
| assert c.chunk_size_bits == 4
|
| assert c.chunk_dim == 2 ** 4
|
| assert c.dtype is not None
|
| assert c.memory_geometry.num_chunks == 4
|
| assert c.memory_mb() > 0
|
|
|
|
|
| original_sv = np.asarray(c.sv).copy()
|
| c.sv = original_sv
|
| np.testing.assert_allclose(np.asarray(c.sv), original_sv, atol=1e-12)
|
|
|
|
|
| class TestChunkDistributed:
|
| """Chunk.run_chunk_distributed (issue #1): dispatches the multi-chunk
|
| kernel across a real JAX device mesh (jax.shard_map + jax.lax.ppermute
|
| for the cross-chunk edge exchange) instead of one process's RAM, one
|
| physical chunk per device (v1 scope). JAX's device count is fixed at
|
| process start, so these tests need >= 8 devices to exercise the
|
| interesting (num_chunks>1) cases and are skipped otherwise -- run
|
| with XLA_FLAGS=--xla_force_host_platform_device_count=8 (or more) to
|
| actually execute them; see CI, which sets this for a dedicated step.
|
|
|
| Correctness bar is the same one TestChunkMultiPiece already
|
| established: cross-check against a plain DenseSVSimulator running the
|
| identical circuit -- not against the single-process multi-chunk path,
|
| to avoid two implementations of the same bug looking like agreement."""
|
|
|
| MIN_DEVICES = 8
|
|
|
| @pytest.fixture(autouse=True)
|
| def _require_devices(self):
|
| if jax.device_count() < self.MIN_DEVICES:
|
| pytest.skip(
|
| f"needs >= {self.MIN_DEVICES} JAX devices, only "
|
| f"{jax.device_count()} available -- run with XLA_FLAGS="
|
| f"--xla_force_host_platform_device_count={self.MIN_DEVICES}"
|
| )
|
|
|
| @pytest.fixture
|
| def force_chunk_bits(self, monkeypatch):
|
| import dense_evolution.chunk as chunk_mod
|
|
|
| def _force(bits):
|
| monkeypatch.setattr(chunk_mod, "get_dynamic_chunk", lambda dtype_target: bits)
|
|
|
| return _force
|
|
|
| def _compare_to_reference(self, n_qubits, circuit):
|
| c = Chunk(n_qubits)
|
| c.run_chunk_distributed(circuit)
|
| sv_dist = np.asarray(c.get_statevector())
|
|
|
| ref = DenseSVSimulator(n_qubits)
|
| ref.run_circuit(circuit, transpile=True)
|
| sv_ref = np.asarray(ref.get_statevector())
|
| return sv_dist, sv_ref
|
|
|
| def test_1q_local(self, force_chunk_bits):
|
| force_chunk_bits(3)
|
| sv_dist, sv_ref = self._compare_to_reference(6, [('h', 3), ('rx', 4, 0.4), ('rz', 5, 0.9)])
|
| np.testing.assert_allclose(sv_dist, sv_ref, atol=1e-9)
|
|
|
| def test_1q_chunk_select(self, force_chunk_bits):
|
| force_chunk_bits(3)
|
| sv_dist, sv_ref = self._compare_to_reference(6, [('h', 0), ('h', 1), ('h', 2)])
|
| np.testing.assert_allclose(sv_dist, sv_ref, atol=1e-9)
|
|
|
| def test_2q_local_local(self, force_chunk_bits):
|
| force_chunk_bits(3)
|
| sv_dist, sv_ref = self._compare_to_reference(6, [('h', 3), ('cx', 3, 4), ('cx', 4, 5)])
|
| np.testing.assert_allclose(sv_dist, sv_ref, atol=1e-9)
|
|
|
| @pytest.mark.parametrize("gate", ["cx", "cz", "cy"])
|
| def test_2q_control_chunk_select_target_local(self, force_chunk_bits, gate):
|
| force_chunk_bits(3)
|
| circuit = [('h', 0), ('h', 1), ('h', 2), ('h', 5), (gate, 0, 5)]
|
| sv_dist, sv_ref = self._compare_to_reference(6, circuit)
|
| np.testing.assert_allclose(sv_dist, sv_ref, atol=1e-9)
|
|
|
| @pytest.mark.parametrize("gate", ["cx", "cz", "cy"])
|
| def test_2q_control_local_target_chunk_select(self, force_chunk_bits, gate):
|
| force_chunk_bits(3)
|
| circuit = [('h', 5), ('h', 0), (gate, 5, 0)]
|
| sv_dist, sv_ref = self._compare_to_reference(6, circuit)
|
| np.testing.assert_allclose(sv_dist, sv_ref, atol=1e-9)
|
|
|
| @pytest.mark.parametrize("gate", ["cx", "cz", "cy"])
|
| def test_2q_control_chunk_select_target_chunk_select(self, force_chunk_bits, gate):
|
| force_chunk_bits(3)
|
| circuit = [('h', 0), ('h', 1), (gate, 0, 1)]
|
| sv_dist, sv_ref = self._compare_to_reference(6, circuit)
|
| np.testing.assert_allclose(sv_dist, sv_ref, atol=1e-9)
|
|
|
| def test_parametric_2q_gates(self, force_chunk_bits):
|
| force_chunk_bits(3)
|
| circuit = [('h', q) for q in range(6)] + [('crz', 0, 1, 0.5), ('cp', 3, 4, 0.7)]
|
| sv_dist, sv_ref = self._compare_to_reference(6, circuit)
|
| np.testing.assert_allclose(sv_dist, sv_ref, atol=1e-9)
|
|
|
| def test_random_mixed_circuit(self, force_chunk_bits):
|
| force_chunk_bits(3)
|
| rng = np.random.default_rng(11)
|
| pool_1q, pool_2q = ['h', 'x', 'y', 'z', 's'], ['cx', 'cz', 'cy']
|
| circuit = []
|
| for _ in range(30):
|
| if rng.random() < 0.5:
|
| circuit.append((rng.choice(pool_1q), int(rng.integers(0, 6))))
|
| else:
|
| q1, q2 = rng.choice(6, size=2, replace=False)
|
| circuit.append((rng.choice(pool_2q), int(q1), int(q2)))
|
| sv_dist, sv_ref = self._compare_to_reference(6, circuit)
|
| np.testing.assert_allclose(sv_dist, sv_ref, atol=1e-9)
|
|
|
| def test_dtype_float32(self, force_chunk_bits):
|
| force_chunk_bits(2)
|
| circuit = [('h', 0), ('h', 1), ('cx', 0, 2), ('rz', 3, 0.6)]
|
| c = Chunk(4, use_float32=True)
|
| c.run_chunk_distributed(circuit)
|
| sv_dist = np.asarray(c.get_statevector())
|
| ref = DenseSVSimulator(4, use_float32=True)
|
| ref.run_circuit(circuit, transpile=True)
|
| sv_ref = np.asarray(ref.get_statevector())
|
| np.testing.assert_allclose(sv_dist, sv_ref, atol=1e-4)
|
| assert sv_dist.dtype == np.complex64
|
|
|
| def test_num_chunks_1_raises_clear_error(self):
|
|
|
|
|
|
|
| c = Chunk(2)
|
| assert c.num_chunks == 1
|
| with pytest.raises(RuntimeError, match="num_chunks"):
|
| c.run_chunk_distributed([('h', 0)])
|
|
|
| def test_insufficient_devices_raises_clear_error(self, force_chunk_bits, monkeypatch):
|
|
|
|
|
| force_chunk_bits(3)
|
| c = Chunk(6)
|
| monkeypatch.setattr(jax, "device_count", lambda: 4)
|
| with pytest.raises(RuntimeError, match="devices"):
|
| c.run_chunk_distributed([('h', 0)])
|
|
|
|
|
|
|
|
|
|
|
|
|
| class TestMemory:
|
|
|
| def test_memory_mb_12_qubits(self):
|
| sim = DenseSVSimulator(n_qubits=12, use_gpu=False, use_float32=False)
|
| mb = sim.memory_mb()
|
| expected = (2**12 * 16) / 1e6
|
| assert abs(mb - expected) < 0.01
|
|
|
| def test_memory_mb_float32(self):
|
| sim = DenseSVSimulator(n_qubits=12, use_gpu=False, use_float32=True)
|
| mb = sim.memory_mb()
|
| expected = (2**12 * 8) / 1e6
|
| assert abs(mb - expected) < 0.01
|
|
|
|
|
|
|
|
|
|
|
| class TestFullPipelineIntegration:
|
| """Converted from dense_evolution/test2.py and dense_evolution/stress_test.py
|
| (audit finding #5): two byte-identical, assertion-free print-and-eyeball
|
| debug scripts that shipped inside every `pip install dense-evolution`
|
| (via the package-data "*.py" glob), were 0% covered, and never ran in CI.
|
| The one real signal they checked — parser -> transpiler -> simulator ->
|
| noise model wired together end to end, and Kraus noise application being
|
| genuinely stochastic across independent runs — is preserved here as a
|
| real, CI-enforced test; both original scripts have been deleted."""
|
|
|
| def test_parse_transpile_simulate_and_apply_noise(self):
|
| qasm_bench = """
|
| OPENQASM 2.0;
|
| include "qelib1.inc";
|
| qreg q[6];
|
| h q[0];
|
| cx q[0], q[1];
|
| cx q[1], q[2];
|
| cx q[2], q[3];
|
| cx q[3], q[4];
|
| cx q[4], q[5];
|
| rx(1.570796) q[0];
|
| ry(0.785398) q[1];
|
| rz(0.392699) q[2];
|
| """
|
| parser = QASMParser()
|
| circ = parser.parse(qasm_bench)
|
| tuples = QuantumTranspiler.transpile(circ.to_tuples())
|
| n_qubits = circ.n_qubits
|
| assert n_qubits == 6
|
| assert len(tuples) == 9
|
|
|
| sim_ideale = DenseSVSimulator(n_qubits)
|
| sim_ideale.run_circuit_jit_beast_mode(tuples)
|
| prob_ideale = sim_ideale.get_probabilities()
|
| assert abs(float(np.sum(prob_ideale)) - 1.0) < 1e-9
|
|
|
| sim_noisy1 = DenseSVSimulator(n_qubits)
|
| sim_noisy1.run_circuit_jit_beast_mode(tuples)
|
| sim_noisy1.sv = NoiseModel.apply_to_sv(sim_noisy1.sv, n_qubits, model='amplitude_damping', p=0.15)
|
| prob_noisy1 = sim_noisy1.get_probabilities()
|
|
|
| sim_noisy2 = DenseSVSimulator(n_qubits)
|
| sim_noisy2.run_circuit_jit_beast_mode(tuples)
|
| sim_noisy2.sv = NoiseModel.apply_to_sv(sim_noisy2.sv, n_qubits, model='amplitude_damping', p=0.15)
|
| prob_noisy2 = sim_noisy2.get_probabilities()
|
|
|
|
|
|
|
|
|
| stochastic_spread = float(np.linalg.norm(prob_noisy1 - prob_noisy2))
|
| assert stochastic_spread > 1e-12
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| class TestPredictiveHealingCore:
|
|
|
| def test_advanced_sigma_is_product_of_inputs(self):
|
| s = healing.calculate_advanced_sigma(
|
| jnp.array(2.0), jnp.array(3.0), jnp.array(1.0), jnp.array(1.0), jnp.array(1.0))
|
| assert float(s) == pytest.approx(6.0)
|
|
|
| def test_phi_ab_identical_states_returns_baseline_0_7(self):
|
|
|
|
|
|
|
| a = jnp.array([1.0, 0.0])
|
| phi = healing.calculate_phi_ab(a, a, jnp.array([1.0, 0.0]))
|
| assert float(phi) == pytest.approx(0.7, abs=1e-6)
|
|
|
| def test_phi_ab_clipped_to_unit_interval(self):
|
| a = jnp.array([1.0, 0.0])
|
| b = jnp.array([-1.0, 0.0])
|
| phi = healing.calculate_phi_ab(a, b, jnp.array([1.0, 0.0]))
|
| assert 0.0 <= float(phi) <= 1.0
|
|
|
| def test_phi_ab_handles_complex_statevectors_no_crash(self):
|
|
|
|
|
|
|
| sim = DenseSVSimulator(n_qubits=3)
|
| sim.apply_gate_1q(GATES['h'], 0)
|
| sim.apply_gate_2q(GATES['cx'], 0, 1)
|
| sv = jnp.array(sim.get_statevector())
|
| assert jnp.iscomplexobj(sv)
|
| phi = healing.calculate_phi_ab(sv, sv, sv)
|
| assert not jnp.iscomplexobj(phi)
|
| assert 0.0 <= float(phi) <= 1.0
|
|
|
| def test_phi_ab_complex_alignment_matches_manual_hermitian_dot(self):
|
|
|
|
|
|
|
|
|
| sim_a = DenseSVSimulator(n_qubits=1)
|
| sim_a.apply_gate_1q(GATES['h'], 0)
|
| state_A = jnp.array(sim_a.get_statevector())
|
|
|
| sim_b = DenseSVSimulator(n_qubits=1)
|
| sim_b.apply_gate_1q(GATES['h'], 0)
|
| sim_b.apply_gate_1q(GATES['s'], 0)
|
| state_B = jnp.array(sim_b.get_statevector())
|
| assert np.any(np.abs(np.imag(np.array(state_B))) > 1e-9)
|
|
|
| phi = healing.calculate_phi_ab(state_A, state_B, state_B)
|
| assert 0.0 <= float(phi) <= 1.0
|
|
|
| sc = np.array(state_B) - np.array(state_A)
|
| expected_alignment = np.real(np.vdot(sc, np.array(state_B))) / (
|
| np.linalg.norm(sc) * np.linalg.norm(state_B))
|
| expected_semantic_alignment = (expected_alignment + 1.0) / 2.0
|
| dist = np.linalg.norm(np.array(state_A) - np.array(state_B))
|
| expected_coherence = 1.0 - dist / float(healing.GLOBAL_CONSTANTS['MAX_SEMANTIC_DISTANCE'])
|
| expected_phi = np.clip(
|
| expected_semantic_alignment * healing.GLOBAL_CONSTANTS['WEIGHT_SEMANTIC']
|
| + expected_coherence * healing.GLOBAL_CONSTANTS['WEIGHT_COHERENCE'], 0.0, 1.0)
|
| assert float(phi) == pytest.approx(float(expected_phi), abs=1e-9)
|
|
|
| def test_vettore_dinamico_zero_energy_is_guarded(self):
|
|
|
|
|
| vd = healing.calculate_vettore_dinamico(jnp.array(0.0), jnp.array(5.0), jnp.array(0.7))
|
| assert float(vd) == pytest.approx(0.0)
|
| assert not np.isnan(float(vd))
|
|
|
| def test_vettore_dinamico_equal_energies_is_zero(self):
|
| vd = healing.calculate_vettore_dinamico(jnp.array(5.0), jnp.array(5.0), jnp.array(0.7))
|
| assert float(vd) == pytest.approx(0.0, abs=1e-6)
|
|
|
| def test_vettore_statico_growing_vs_static_branches(self):
|
| growing = healing.calculate_vettore_statico(jnp.array(0.5))
|
| static = healing.calculate_vettore_statico(jnp.array(0.001))
|
| assert float(growing) == pytest.approx(0.0)
|
| assert float(static) == pytest.approx(1.0)
|
|
|
| def test_delta_preemp_zero_deviation(self):
|
| d = healing.calculate_delta_preemp(jnp.array(10.0), 10.0)
|
| assert float(d) == pytest.approx(0.0)
|
|
|
| def test_delta_preemp_half_deviation(self):
|
| d = healing.calculate_delta_preemp(jnp.array(5.0), 10.0)
|
| assert float(d) == pytest.approx(0.5)
|
|
|
| def test_delta_preemp_nonpositive_target_falls_back_to_1(self):
|
|
|
| d = healing.calculate_delta_preemp(jnp.array(5.0), -1.0)
|
| assert float(d) == pytest.approx(6.0)
|
|
|
| def test_phi_trigger_active_branch(self):
|
| trigger, lam, eps = healing.evaluate_phi_trigger(jnp.array(0.5))
|
| assert float(trigger) == pytest.approx(1.0)
|
| assert float(lam) == pytest.approx(0.05)
|
| assert float(eps) == pytest.approx(0.01)
|
|
|
| def test_phi_trigger_stasis_branch(self):
|
| trigger, lam, eps = healing.evaluate_phi_trigger(jnp.array(0.001))
|
| assert float(trigger) == pytest.approx(0.0)
|
| assert float(lam) == pytest.approx(0.15)
|
| assert float(eps) == pytest.approx(0.11)
|
|
|
| def test_jax_reflection_empty_arrays_returns_zero_not_nan(self):
|
|
|
|
|
|
|
| avg_c, var_c, avg_n = healing.calculate_jax_reflection(
|
| jnp.array([], dtype=jnp.float64), jnp.array([], dtype=jnp.float64))
|
| assert float(avg_c) == 0.0 and not np.isnan(float(avg_c))
|
| assert float(var_c) == 0.0 and not np.isnan(float(var_c))
|
| assert float(avg_n) == 0.0 and not np.isnan(float(avg_n))
|
|
|
| def test_jax_reflection_nonempty_matches_plain_numpy(self):
|
| coh = jnp.array([0.8, 0.6, 0.9], dtype=jnp.float64)
|
| noise = jnp.array([0.1, 0.2], dtype=jnp.float64)
|
| avg_c, var_c, avg_n = healing.calculate_jax_reflection(coh, noise)
|
| assert float(avg_c) == pytest.approx(np.mean([0.8, 0.6, 0.9]))
|
| assert float(var_c) == pytest.approx(np.var([0.8, 0.6, 0.9]))
|
| assert float(avg_n) == pytest.approx(np.mean([0.1, 0.2]))
|
|
|
|
|
| class TestMemoryReflectionEngine:
|
|
|
| def test_reflect_aggregates_events_correctly(self):
|
| eng = healing.MemoryReflectionEngine()
|
| eng.record_event('coherence', 0.8, 'run1')
|
| eng.record_event('coherence', 0.6, 'run2')
|
| eng.record_event('noise', 0.1, 'run1')
|
| eng.record_event('intervention', 1.0, 'fallback triggered')
|
|
|
| report = eng.reflect()
|
| assert report['average_coherence'] == pytest.approx(0.7)
|
| assert report['variance_coherence'] == pytest.approx(np.var([0.8, 0.6]))
|
| assert report['interventions_count'] == 1
|
| assert report['average_noise'] == pytest.approx(0.1)
|
| assert report['total_events'] == 4
|
|
|
| def test_reflect_on_empty_engine_returns_none_not_nan(self):
|
| eng = healing.MemoryReflectionEngine()
|
| report = eng.reflect()
|
| assert report == {
|
| 'average_coherence': None,
|
| 'variance_coherence': None,
|
| 'interventions_count': 0,
|
| 'average_noise': None,
|
| 'total_events': 0,
|
| }
|
|
|
| def test_export_and_load_memory_round_trip(self, tmp_path):
|
| eng = healing.MemoryReflectionEngine()
|
| eng.record_event('coherence', 0.8, 'run1')
|
| eng.record_event('noise', 0.1, 'run1')
|
|
|
| path = tmp_path / "memory.json"
|
| eng.export_memory(str(path))
|
|
|
| eng2 = healing.MemoryReflectionEngine()
|
| eng2.load_memory(str(path))
|
|
|
| assert eng2.memory == eng.memory
|
| assert eng2.reflect() == eng.reflect()
|
|
|
