This is an automated email from the ASF dual-hosted git repository.
hcr pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/mahout.git
The following commit(s) were added to refs/heads/main by this push:
new a74d5a2de MAHOUT-823: [QDP] test: add & refactor NumPy encoding tests
for improved clarity and coverage (#824)
a74d5a2de is described below
commit a74d5a2de178093bd19cd74a41aa91f28a46b5d5
Author: Vic Wen <[email protected]>
AuthorDate: Thu Jan 15 16:30:44 2026 +0800
MAHOUT-823: [QDP] test: add & refactor NumPy encoding tests for improved
clarity and coverage (#824)
* test: add & refactor NumPy encoding tests for improved clarity and
coverage
- Updated test descriptions to better reflect functionality.
- Consolidated encoding tests into parameterized cases for basic, large,
and single sample inputs.
- Introduced helper function to verify tensor properties, enhancing code
reuse.
- Added tests for different encoding methods and precision settings.
- Implemented error handling tests for invalid inputs and unsupported
dimensions.
* test: enhance and parameterize NumPy encoding tests
---
qdp/qdp-python/tests/test_numpy.py | 200 +++++++++++++++++++++++--------------
1 file changed, 123 insertions(+), 77 deletions(-)
diff --git a/qdp/qdp-python/tests/test_numpy.py
b/qdp/qdp-python/tests/test_numpy.py
index 1616247bc..c5f53ecbd 100644
--- a/qdp/qdp-python/tests/test_numpy.py
+++ b/qdp/qdp-python/tests/test_numpy.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python3
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
@@ -15,70 +14,51 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-"""Test NumPy file format support in Mahout QDP Python bindings"""
+"""Test NumPy file format and array input support in Mahout QDP Python
bindings"""
import tempfile
import os
+
import numpy as np
+import pytest
import torch
-from _qdp import QdpEngine
-
-
-def test_encode_from_numpy_basic():
- """Test basic NumPy file encoding"""
- engine = QdpEngine(device_id=0)
-
- # Create test data
- num_samples = 10
- num_qubits = 3
- sample_size = 2**num_qubits # 8
-
- # Generate normalized data
- data = np.random.randn(num_samples, sample_size).astype(np.float64)
- # Normalize each row
- norms = np.linalg.norm(data, axis=1, keepdims=True)
- data = data / norms
-
- # Save to temporary .npy file
- with tempfile.NamedTemporaryFile(suffix=".npy", delete=False) as f:
- npy_path = f.name
-
- try:
- np.save(npy_path, data)
- qtensor = engine.encode(npy_path, num_qubits)
- tensor = torch.from_dlpack(qtensor)
+from _qdp import QdpEngine
- # Verify shape
- assert tensor.shape == (num_samples, sample_size), (
- f"Expected shape {(num_samples, sample_size)}, got {tensor.shape}"
- )
- # Verify it's on GPU
- assert tensor.is_cuda, "Tensor should be on CUDA device"
+def _verify_tensor(tensor, expected_shape, check_normalization=False):
+ """Helper function to verify tensor properties"""
+ assert tensor.shape == expected_shape, (
+ f"Expected shape {expected_shape}, got {tensor.shape}"
+ )
+ assert tensor.is_cuda, "Tensor should be on CUDA device"
- # Verify normalization (amplitude encoding normalizes)
+ if check_normalization:
norms = tensor.abs().pow(2).sum(dim=1).sqrt()
assert torch.allclose(norms, torch.ones_like(norms), atol=1e-5), (
"States should be normalized"
)
- print("ā test_encode_from_numpy_basic passed")
-
- finally:
- if os.path.exists(npy_path):
- os.remove(npy_path)
[email protected]
[email protected](
+ "num_samples,num_qubits,check_norm",
+ [
+ (10, 3, True), # Basic: 10 samples, 3 qubits, check normalization
+ (100, 6, False), # Large: 100 samples, 6 qubits
+ (1, 4, False), # Single sample: 1 sample, 4 qubits
+ ],
+)
+def test_encode_from_numpy_file(num_samples, num_qubits, check_norm):
+ """Test NumPy file encoding"""
+ pytest.importorskip("torch")
+ if not torch.cuda.is_available():
+ pytest.skip("GPU required for QdpEngine")
-def test_encode_from_numpy_large():
- """Test NumPy encoding with larger dataset"""
engine = QdpEngine(device_id=0)
+ sample_size = 2**num_qubits
- num_samples = 100
- num_qubits = 6
- sample_size = 2**num_qubits # 64
-
- # Generate test data
+ # Generate normalized data
data = np.random.randn(num_samples, sample_size).astype(np.float64)
norms = np.linalg.norm(data, axis=1, keepdims=True)
data = data / norms
@@ -89,53 +69,119 @@ def test_encode_from_numpy_large():
try:
np.save(npy_path, data)
-
qtensor = engine.encode(npy_path, num_qubits)
tensor = torch.from_dlpack(qtensor)
- # Verify
- assert tensor.shape == (num_samples, sample_size)
- assert tensor.is_cuda
-
- print("ā test_encode_from_numpy_large passed")
+ _verify_tensor(tensor, (num_samples, sample_size), check_norm)
finally:
if os.path.exists(npy_path):
os.remove(npy_path)
-def test_encode_from_numpy_single_sample():
- """Test NumPy encoding with single sample"""
[email protected]
[email protected]("num_qubits", [1, 2, 3, 4])
+def test_encode_numpy_array_1d(num_qubits):
+ """Test 1D NumPy array encoding (single sample)"""
+ pytest.importorskip("torch")
+ if not torch.cuda.is_available():
+ pytest.skip("GPU required for QdpEngine")
+
engine = QdpEngine(device_id=0)
+ sample_size = 2**num_qubits
+ data = np.random.randn(sample_size).astype(np.float64)
+ data = data / np.linalg.norm(data)
- num_qubits = 4
- sample_size = 2**num_qubits # 16
+ qtensor = engine.encode(data, num_qubits)
+ tensor = torch.from_dlpack(qtensor)
+ _verify_tensor(tensor, (1, sample_size), check_normalization=True)
- # Single sample
- data = np.random.randn(1, sample_size).astype(np.float64)
- data = data / np.linalg.norm(data)
- with tempfile.NamedTemporaryFile(suffix=".npy", delete=False) as f:
- npy_path = f.name
[email protected]
[email protected]("num_samples,num_qubits", [(5, 2), (10, 3)])
+def test_encode_numpy_array_2d(num_samples, num_qubits):
+ """Test 2D NumPy array encoding (batch)"""
+ pytest.importorskip("torch")
+ if not torch.cuda.is_available():
+ pytest.skip("GPU required for QdpEngine")
- try:
- np.save(npy_path, data)
+ engine = QdpEngine(device_id=0)
+ sample_size = 2**num_qubits
+ data = np.random.randn(num_samples, sample_size).astype(np.float64)
+ norms = np.linalg.norm(data, axis=1, keepdims=True)
+ data = data / norms
- qtensor = engine.encode(npy_path, num_qubits)
- tensor = torch.from_dlpack(qtensor)
+ qtensor = engine.encode(data, num_qubits)
+ tensor = torch.from_dlpack(qtensor)
+ _verify_tensor(tensor, (num_samples, sample_size),
check_normalization=True)
- assert tensor.shape == (1, sample_size)
- assert tensor.is_cuda
- print("ā test_encode_from_numpy_single_sample passed")
[email protected]
[email protected]("encoding_method", ["amplitude"])
+def test_encode_numpy_encoding_methods(encoding_method):
+ """Test different encoding methods"""
+ pytest.importorskip("torch")
+ if not torch.cuda.is_available():
+ pytest.skip("GPU required for QdpEngine")
- finally:
- if os.path.exists(npy_path):
- os.remove(npy_path)
+ # TODO: Add angle and basis encoding tests when implemented
+ engine = QdpEngine(device_id=0)
+ num_qubits = 2
+ sample_size = 2**num_qubits
+ data = np.array([1.0, 2.0, 3.0, 4.0], dtype=np.float64)
+
+ qtensor = engine.encode(data, num_qubits, encoding_method=encoding_method)
+ tensor = torch.from_dlpack(qtensor)
+ _verify_tensor(tensor, (1, sample_size))
+
+
[email protected]
[email protected](
+ "precision,expected_dtype",
+ [
+ ("float32", torch.complex64),
+ ("float64", torch.complex128),
+ ],
+)
+def test_encode_numpy_precision(precision, expected_dtype):
+ """Test different precision settings"""
+ pytest.importorskip("torch")
+ if not torch.cuda.is_available():
+ pytest.skip("GPU required for QdpEngine")
+
+ engine = QdpEngine(device_id=0, precision=precision)
+ num_qubits = 2
+ data = np.array([1.0, 2.0, 3.0, 4.0], dtype=np.float64)
+
+ qtensor = engine.encode(data, num_qubits)
+ tensor = torch.from_dlpack(qtensor)
+ assert tensor.dtype == expected_dtype, (
+ f"Expected {expected_dtype}, got {tensor.dtype}"
+ )
+
+
[email protected]
[email protected](
+ "data,error_match",
+ [
+ (
+ np.array([1.0, 2.0, 3.0, 4.0], dtype=np.float32),
+ None, # Wrong dtype - will raise RuntimeError or TypeError
+ ),
+ (
+ np.array([[[1.0, 2.0], [3.0, 4.0]]], dtype=np.float64),
+ None, # 3D array - will raise RuntimeError or TypeError
+ ),
+ ],
+)
+def test_encode_numpy_errors(data, error_match):
+ """Test error handling for invalid inputs"""
+ pytest.importorskip("torch")
+ if not torch.cuda.is_available():
+ pytest.skip("GPU required for QdpEngine")
+ engine = QdpEngine(device_id=0)
+ num_qubits = 2 if data.ndim == 1 else 1
-if __name__ == "__main__":
- test_encode_from_numpy_basic()
- test_encode_from_numpy_large()
- test_encode_from_numpy_single_sample()
- print("\nā
All NumPy encoding tests passed!")
+ with pytest.raises((RuntimeError, TypeError)):
+ engine.encode(data, num_qubits)
