Fix Tests (#24)

* Fix generation of random tensors

In some edge cases, `tnco.tests.utils.generate_random_tensors` is unable
to get all the requested number of tensors at the first attempt. In the
rare event this happens, the routine now discards the invalid set of
tensors and try to create a new one

* `tnco.tests` --> `tnco.testing`

* Fix `pytest` fixtures

Author: s-mandra

.github/workflows/run_tests.yml

@@ -75,7 +75,7 @@ jobs:
         == '3.13.x'
       run: |
         pip install pytest-sugar
-        PYTEST_FRACTION_N_TESTS=2 pytest --verbose --timeout=180 -n auto tests/*.py
+        PYTEST_FRACTION_N_TESTS=0.2 pytest --verbose --timeout=180 -n auto tests/*.py
 
     - name: Run CLI for ${{ matrix.os }}, with ${{ matrix.compiler }} and python==${{
         matrix.python-version }} (${{ matrix.target }})

examples/Optimization.ipynb

@@ -406,7 +406,7 @@
    ],
    "source": [
     "# An arbitrary tensor tensor can be specified by using 'tnco.app.TensorNetwork'\n",
-    "from tnco.tests.utils import generate_random_tensors\n",
+    "from tnco.testing.utils import generate_random_tensors\n",
     "\n",
     "# Get a list of random indices for each tensor. The resulting tensor network\n",
     "# will have two disconnected components, 2 * 10 tensors (10 tensors for each\n",

pyproject.toml

@@ -67,6 +67,7 @@ testing = [
   "numpy",
   "ply",
   "pytest",
+  "pytest-repeat",
   "pytest-timeout",
   "pytest-xdist",
   "qiskit",

tests/conftest.py

@@ -13,32 +13,9 @@
 # limitations under the License.
 
 import signal
-from os import environ
-from random import randrange
 
 import pytest
 
-# Global seed
-global_seed = None
-fraction_n_tests = 0.1
-
-
-def pytest_configure(config):
-    # Randomize hash seed
-    environ["PYTEST_SEED"] = environ.get("PYTEST_SEED", str(randrange(2**32)))
-
-
-def pytest_sessionstart(session):
-    # Assign global seed
-    global global_seed
-    global_seed = environ["PYTEST_SEED"]
-    print(f'seed: {global_seed}')
-
-    # Set maximum number of tests
-    global fraction_n_tests
-    fraction_n_tests = float(environ.get('PYTEST_FRACTION_N_TESTS', 10)) / 100
-    print(f'fraction_n_tests: {fraction_n_tests * 100:1.0f}%')
-
 
 @pytest.fixture
 def timeout(request):

tests/test_app.py

@@ -17,6 +17,7 @@
 import json
 import pickle
 from decimal import Decimal
+from os import environ
 from random import Random
 from tempfile import NamedTemporaryFile
 
@@ -26,24 +27,36 @@
 import pytest
 from cirq.contrib.qasm_import import circuit_from_qasm
 from quimb.tensor import Tensor, TensorNetwork
-
-from conftest import fraction_n_tests, global_seed  # Get global seed
 from tnco.app import Optimizer
 from tnco.app import Tensor as TS
 from tnco.app import TensorNetwork as TN
 from tnco.app import load_tn
-from tnco.tests.utils import generate_random_tensors
+from tnco.testing.utils import generate_random_tensors
+
+# Initialize RNG
+rng = Random(
+    environ.get('PYTEST_SEED') +
+    environ.get('PYTEST_XDIST_WORKER') if 'PYTEST_SEED' in environ else None)
+
+# Fix max number of repetitions
+max_repeat = max(1, float(environ.get('PYTEST_MAX_REPEAT', 'inf')))
+
+# Fix ratio of number of tests
+fraction_n_tests = max(
+    min(float(environ.get('PYTEST_FRACTION_N_TESTS', '1')), 1), 0)
+
 
-rng = Random(global_seed)
+def repeat(n: int):
+    return pytest.mark.repeat(max(min(n * fraction_n_tests, max_repeat), 1))
 
 
-def sample_seeds(k, /):
-    k = int(max(1, k * fraction_n_tests))
-    return rng.sample(range(2**32), k=k)
+@pytest.fixture
+def random_seed():
+    return rng.randrange(2**32)
 
 
-@pytest.mark.parametrize('seed', sample_seeds(200))
-def test_LoadTN_CirqCircuit(seed):
+@repeat(20)
+def test_LoadTN_CirqCircuit(random_seed):
 
     def check_tn(U, tn):
         tn = TensorNetwork(map(Tensor, tn.arrays,
@@ -52,7 +65,7 @@ def check_tn(U, tn):
         np.testing.assert_allclose(U.ravel(), tn.data.ravel(), atol=1e-5)
 
     # Get RNG
-    rng = Random(seed)
+    rng = Random(random_seed)
 
     # Get random seed
     circuit_seed = rng.randrange(2**30)
@@ -99,8 +112,8 @@ def check_tn(U, tn):
              load_tn_(circuit.to_qasm()))
 
 
-@pytest.mark.parametrize('seed', sample_seeds(400))
-def test_OptimizeTN(seed, **kwargs):
+@repeat(40)
+def test_OptimizeTN(random_seed, **kwargs):
     # How to convert inds
     def convert_index(x):
         if isinstance(x, (str, int, frozenset)):
@@ -119,10 +132,10 @@ def to_tuple(x):
             return x
 
     # Get rng
-    rng = Random(seed)
+    rng = Random(random_seed)
 
     # Get numpy rng
-    rng_np = np.random.default_rng(seed)
+    rng_np = np.random.default_rng(random_seed)
 
     # Initialize variables
     n_tensors = kwargs.get('n_tensors', rng.randint(5, 10))
@@ -164,7 +177,7 @@ def to_tuple(x):
         n_cc=n_cc,
         n_output_inds=n_output_inds,
         randomize_names=randomize_names,
-        seed=seed)
+        seed=random_seed)
 
     # Get random dimensions
     dims = dict(

tests/test_circuit.py

@@ -14,6 +14,7 @@
 
 import itertools as its
 from math import sqrt
+from os import environ
 from random import Random
 
 import cirq
@@ -23,20 +24,32 @@
 import qiskit
 from qiskit.circuit.random import random_circuit
 from quimb.tensor import Tensor, TensorNetwork
-
-from conftest import fraction_n_tests, global_seed  # Get global seed
 from tnco.utils.circuit import commute, load, same
 
-rng = Random(global_seed)
+# Initialize RNG
+rng = Random(
+    environ.get('PYTEST_SEED') +
+    environ.get('PYTEST_XDIST_WORKER') if 'PYTEST_SEED' in environ else None)
+
+# Fix max number of repetitions
+max_repeat = max(1, float(environ.get('PYTEST_MAX_REPEAT', 'inf')))
+
+# Fix ratio of number of tests
+fraction_n_tests = max(
+    min(float(environ.get('PYTEST_FRACTION_N_TESTS', '1')), 1), 0)
+
+
+def repeat(n: int):
+    return pytest.mark.repeat(max(min(n * fraction_n_tests, max_repeat), 1))
 
 
-def sample_seeds(k, /):
-    k = int(max(1, k * fraction_n_tests))
-    return rng.sample(range(2**32), k=k)
+@pytest.fixture
+def random_seed():
+    return rng.randrange(2**32)
 
 
-@pytest.mark.parametrize('seed', sample_seeds(200))
-def test_Commute(seed, **kwargs):
+@repeat(20)
+def test_Commute(random_seed, **kwargs):
     # Commutation using cirq
     def commute_(gate_A, gate_B, *, use_matrix_commutation=True, atol=1e-5):
         if use_matrix_commutation:
@@ -48,7 +61,7 @@ def commute_(gate_A, gate_B, *, use_matrix_commutation=True, atol=1e-5):
         return not frozenset(gate_A[1]) & frozenset(gate_B[1])
 
     # Initialize RNG
-    rng = Random(seed)
+    rng = Random(random_seed)
 
     # Set parameters
     n_qubits = kwargs.get('n_qubits', 5)
@@ -58,10 +71,9 @@ def commute_(gate_A, gate_B, *, use_matrix_commutation=True, atol=1e-5):
     circuit = tuple(
         map(
             lambda gate: (cirq.unitary(gate), gate.qubits),
-            cirq.testing.random_circuit(n_qubits,
-                                        n_moments,
-                                        op_density=1,
-                                        random_state=seed).all_operations()))
+            cirq.testing.random_circuit(
+                n_qubits, n_moments, op_density=1,
+                random_state=random_seed).all_operations()))
 
     # Check commutation
     assert all(
@@ -76,8 +88,8 @@ def commute_(gate_A, gate_B, *, use_matrix_commutation=True, atol=1e-5):
             zip(rng.choices(circuit, k=200), rng.choices(circuit, k=200))))
 
 
-@pytest.mark.parametrize('seed', sample_seeds(200))
-def test_Same(seed, **kwargs):
+@repeat(20)
+def test_Same(random_seed, **kwargs):
     # Same using quimb
     def same_(gate_A, gate_B, atol=1e-5):
         gate_A = Tensor(
@@ -92,7 +104,7 @@ def same_(gate_A, gate_B, atol=1e-5):
             gate_A.data, gate_B.transpose_like(gate_A).data, atol=1e-5)
 
     # Initialize RNG
-    rng = Random(seed)
+    rng = Random(random_seed)
 
     # Set parameters
     n_qubits = kwargs.get('n_qubits', 5)
@@ -102,10 +114,9 @@ def same_(gate_A, gate_B, atol=1e-5):
     circuit = tuple(
         map(
             lambda gate: (cirq.unitary(gate), gate.qubits),
-            cirq.testing.random_circuit(n_qubits,
-                                        n_moments,
-                                        op_density=1,
-                                        random_state=seed).all_operations()))
+            cirq.testing.random_circuit(
+                n_qubits, n_moments, op_density=1,
+                random_state=random_seed).all_operations()))
 
     # Check same
     assert all(
@@ -116,11 +127,11 @@ def same_(gate_A, gate_B, atol=1e-5):
         map(lambda g: same_(g, g) and same(g, g), rng.choices(circuit, k=200)))
 
 
-@pytest.mark.parametrize('seed', sample_seeds(200))
-def test_LoadArbitraryInitialFinalState(seed, **kwargs):
+@repeat(20)
+def test_LoadArbitraryInitialFinalState(random_seed, **kwargs):
     # Get rng
-    rng = Random(seed)
-    np_rng = np.random.default_rng(seed)
+    rng = Random(random_seed)
+    np_rng = np.random.default_rng(random_seed)
 
     # Get parameters
     n_qubits = kwargs.get('n_qubits', rng.randint(5, 10))
@@ -151,7 +162,7 @@ def random_qubit():
         circuit = cirq.testing.random_circuit(n_qubits,
                                               n_moments,
                                               op_density=op_density,
-                                              random_state=seed)
+                                              random_state=random_seed)
     else:
         circuit = cirq.testing.random_circuit(n_qubits,
                                               n_moments,
@@ -163,7 +174,7 @@ def random_qubit():
                                                   cirq.CCZPowGate(): 3,
                                                   cirq.ISWAP: 2
                                               },
-                                              random_state=seed)
+                                              random_state=random_seed)
     circuit = map((cirq.H**0.5).on, circuit.all_qubits()) + circuit + map(
         (cirq.H**0.5).on, circuit.all_qubits())
 
@@ -188,7 +199,7 @@ def random_qubit():
         decompose_hyper_inds=decompose_hyper_inds,
         fuse=fuse,
         use_matrix_commutation=use_matrix_commutation,
-        seed=seed)
+        seed=random_seed)
 
     def get_state(x):
         valid_token = {
@@ -243,10 +254,10 @@ def get_state(x):
     np.testing.assert_allclose(ex, ts, atol=1e-5)
 
 
-@pytest.mark.parametrize('seed', sample_seeds(200))
-def test_LoadUnitary(seed, **kwargs):
+@repeat(20)
+def test_LoadUnitary(random_seed, **kwargs):
     # Get rng
-    rng = Random(seed)
+    rng = Random(random_seed)
 
     # Get parameters
     n_qubits = kwargs.get('n_qubits', rng.randint(5, 10))
@@ -265,7 +276,7 @@ def test_LoadUnitary(seed, **kwargs):
         circuit = cirq.testing.random_circuit(n_qubits,
                                               n_moments,
                                               op_density=op_density,
-                                              random_state=seed)
+                                              random_state=random_seed)
     else:
         circuit = cirq.testing.random_circuit(n_qubits,
                                               n_moments,
@@ -277,7 +288,7 @@ def test_LoadUnitary(seed, **kwargs):
                                                   cirq.CCZPowGate(): 3,
                                                   cirq.ISWAP: 2
                                               },
-                                              random_state=seed)
+                                              random_state=random_seed)
     circuit = map((cirq.H**0.5).on, circuit.all_qubits()) + circuit + map(
         (cirq.H**0.5).on, circuit.all_qubits())
 
@@ -291,7 +302,7 @@ def test_LoadUnitary(seed, **kwargs):
         decompose_hyper_inds=decompose_hyper_inds,
         fuse=fuse,
         use_matrix_commutation=use_matrix_commutation,
-        seed=seed)
+        seed=random_seed)
 
     # Get exact unitary
     U = cirq.unitary(circuit)
@@ -337,8 +348,8 @@ def test_LoadUnitary(seed, **kwargs):
     assert not arrays and not ts_inds and not output_inds
 
 
-@pytest.mark.parametrize('seed', sample_seeds(200))
-def test_LoadQisKit(seed):
+@repeat(20)
+def test_LoadQisKit(random_seed):
     # Set number of qubits and depth of the circuit
     n_qubits = 6
     circuit_depth = 12
@@ -350,8 +361,8 @@ def test_LoadQisKit(seed):
     # Get a random circuit
     circuit = qiskit.QuantumCircuit(n_qubits)
     mit.consume(map(lambda i: circuit.append(sqrt_H, [i]), range(n_qubits)))
-    circuit = circuit.compose(random_circuit(n_qubits, circuit_depth,
-                                             seed=seed))
+    circuit = circuit.compose(
+        random_circuit(n_qubits, circuit_depth, seed=random_seed))
     mit.consume(map(lambda i: circuit.append(sqrt_H, [i]), range(n_qubits)))
 
     # Get unitary