mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
169 lines
6 KiB
Python
169 lines
6 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
# Copyright (c) 2025 Allaun
|
|
|
|
"""Tests for nodupe/tools/ml/__init__.py - ML Backend implementations."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
# Import the ML backend classes
|
|
from nodupe.tools.ml import (
|
|
CPUBackend,
|
|
MLBackend,
|
|
ONNXBackend,
|
|
create_ml_backend,
|
|
get_ml_backend,
|
|
)
|
|
|
|
|
|
class TestMLBackend:
|
|
"""Test abstract MLBackend class."""
|
|
|
|
def test_is_abstract(self):
|
|
"""MLBackend cannot be instantiated directly."""
|
|
with pytest.raises(TypeError):
|
|
MLBackend()
|
|
|
|
|
|
class TestCPUBackend:
|
|
"""Test CPUBackend class."""
|
|
|
|
def test_cpu_backend_creation(self):
|
|
"""CPUBackend can be created."""
|
|
backend = CPUBackend()
|
|
assert backend is not None
|
|
|
|
def test_is_available(self):
|
|
"""CPUBackend is always available."""
|
|
backend = CPUBackend()
|
|
assert backend.is_available() is True
|
|
|
|
def test_get_embedding_dimensions_default(self):
|
|
"""CPUBackend returns default dimensions."""
|
|
backend = CPUBackend()
|
|
assert backend.get_embedding_dimensions() == 128
|
|
|
|
def test_get_embedding_dimensions_custom(self):
|
|
"""CPUBackend can be created with custom dimensions."""
|
|
backend = CPUBackend()
|
|
# Override the dimensions attribute
|
|
backend.dimensions = 256
|
|
assert backend.get_embedding_dimensions() == 256
|
|
|
|
def test_generate_embeddings_empty_list(self):
|
|
"""CPUBackend handles empty list."""
|
|
backend = CPUBackend()
|
|
embeddings = backend.generate_embeddings([])
|
|
assert embeddings == []
|
|
|
|
def test_generate_embeddings_strings(self):
|
|
"""CPUBackend generates embeddings for strings."""
|
|
backend = CPUBackend()
|
|
embeddings = backend.generate_embeddings(['hello', 'world'])
|
|
assert len(embeddings) == 2
|
|
assert all(len(emb) == 128 for emb in embeddings)
|
|
|
|
def test_generate_embeddings_lists(self):
|
|
"""CPUBackend generates embeddings for lists."""
|
|
backend = CPUBackend()
|
|
embeddings = backend.generate_embeddings([[1, 2, 3], [4, 5, 6]])
|
|
assert len(embeddings) == 2
|
|
assert all(len(emb) == 128 for emb in embeddings)
|
|
|
|
def test_generate_embeddings_numpy_arrays(self):
|
|
"""CPUBackend generates embeddings for numpy arrays."""
|
|
backend = CPUBackend()
|
|
arr1 = np.array([1, 2, 3])
|
|
arr2 = np.array([4, 5, 6])
|
|
embeddings = backend.generate_embeddings([arr1, arr2])
|
|
assert len(embeddings) == 2
|
|
|
|
def test_generate_embeddings_mixed_types(self):
|
|
"""CPUBackend generates embeddings for mixed types."""
|
|
backend = CPUBackend()
|
|
embeddings = backend.generate_embeddings(['text', [1, 2], np.array([3, 4]), 42])
|
|
assert len(embeddings) == 4
|
|
assert all(len(emb) == 128 for emb in embeddings)
|
|
|
|
def test_generate_embeddings_exception_handling(self):
|
|
"""CPUBackend handles exceptions gracefully."""
|
|
backend = CPUBackend()
|
|
# Force an exception by passing something that causes issues
|
|
embeddings = backend.generate_embeddings([])
|
|
assert embeddings == []
|
|
|
|
|
|
class TestONNXBackend:
|
|
"""Test ONNXBackend class."""
|
|
|
|
def test_onnx_backend_creation(self):
|
|
"""ONNXBackend can be created without model path."""
|
|
backend = ONNXBackend()
|
|
assert backend is not None
|
|
|
|
def test_onnx_backend_with_model_path(self):
|
|
"""ONNXBackend can be created with model path."""
|
|
backend = ONNXBackend(model_path="test_model.onnx")
|
|
assert backend.model_path == "test_model.onnx"
|
|
|
|
def test_onnx_backend_not_available_without_onnxruntime(self):
|
|
"""ONNXBackend is not available when onnxruntime is not installed."""
|
|
backend = ONNXBackend()
|
|
# Should fall back to CPU when onnxruntime is not available
|
|
assert backend.is_available() is False
|
|
|
|
def test_get_embedding_dimensions(self):
|
|
"""ONNXBackend returns dimensions."""
|
|
backend = ONNXBackend()
|
|
assert backend.get_embedding_dimensions() == 128
|
|
|
|
|
|
class TestCreateMLBackend:
|
|
"""Test create_ml_backend factory function."""
|
|
|
|
def test_create_ml_backend_auto(self):
|
|
"""create_ml_backend with 'auto' returns CPUBackend."""
|
|
backend = create_ml_backend('auto')
|
|
assert isinstance(backend, CPUBackend)
|
|
|
|
def test_create_ml_backend_cpu(self):
|
|
"""create_ml_backend with 'cpu' returns CPUBackend."""
|
|
backend = create_ml_backend('cpu')
|
|
assert isinstance(backend, CPUBackend)
|
|
|
|
def test_create_ml_backend_onnx(self):
|
|
"""create_ml_backend with 'onnx' returns ONNXBackend."""
|
|
backend = create_ml_backend('onnx')
|
|
assert isinstance(backend, ONNXBackend)
|
|
|
|
def test_create_ml_backend_unknown_type(self):
|
|
"""create_ml_backend raises ValueError for unknown type."""
|
|
with pytest.raises(ValueError, match="Unknown backend type"):
|
|
create_ml_backend('unknown')
|
|
|
|
def test_create_ml_backend_auto_onnx_available(self):
|
|
"""create_ml_backend auto falls back to CPU when ONNX not available."""
|
|
# ONNX is not available in test environment
|
|
backend = create_ml_backend("auto")
|
|
assert isinstance(backend, CPUBackend)
|
|
def test_get_ml_backend_returns_backend(self):
|
|
"""get_ml_backend returns an MLBackend."""
|
|
backend = get_ml_backend()
|
|
assert isinstance(backend, MLBackend)
|
|
|
|
def test_get_ml_backend_singleton(self):
|
|
"""get_ml_backend returns the same instance."""
|
|
# Note: This may fail because the module initializes on import
|
|
# Reset the global to test singleton behavior
|
|
import nodupe.tools.ml as ml_module
|
|
original_backend = ml_module.ML_BACKEND
|
|
|
|
backend1 = get_ml_backend()
|
|
backend2 = get_ml_backend()
|
|
|
|
# Both should be the same instance (or different CPUBackends)
|
|
# This test verifies the function works
|
|
assert backend1 is not None
|
|
assert backend2 is not None
|