FusionAGI/tests/test_tensorflow_adapter.py

"""Tests for fusionagi.adapters.tensorflow_adapter (uses NumPy backend, no TF required)."""

import pytest

from fusionagi.gpu.backend import get_backend, reset_backend


@pytest.fixture(autouse=True)
def _use_numpy():
    reset_backend()
    get_backend(force="numpy")
    yield
    reset_backend()


class TestTensorFlowAdapterImport:
    """Test that TensorFlowAdapter is importable (may be None without TF)."""

    def test_import(self):
        pass
        # TensorFlowAdapter is None when tensorflow is not installed
        # This is by design — GPU is an optional dependency


class TestGPUMemorySearch:
    """Test GPU-accelerated memory search."""

    def test_semantic_search(self):
        from fusionagi.memory.gpu_search import semantic_search
        from fusionagi.schemas.atomic import AtomicSemanticUnit, AtomicUnitType

        units = [
            AtomicSemanticUnit(
                unit_id="u1",
                content="the sky is blue",
                type=AtomicUnitType.FACT,
                confidence=1.0,
            ),
            AtomicSemanticUnit(
                unit_id="u2",
                content="water is wet",
                type=AtomicUnitType.FACT,
                confidence=1.0,
            ),
            AtomicSemanticUnit(
                unit_id="u3",
                content="python programming language",
                type=AtomicUnitType.FACT,
                confidence=1.0,
            ),
        ]
        results = semantic_search("sky color", units, top_k=2)
        assert len(results) <= 2
        assert all(isinstance(r, tuple) for r in results)
        assert all(isinstance(r[0], AtomicSemanticUnit) for r in results)
        assert all(isinstance(r[1], float) for r in results)

    def test_semantic_search_empty(self):
        from fusionagi.memory.gpu_search import semantic_search

        results = semantic_search("query", [], top_k=5)
        assert results == []

    def test_batch_embed_units(self):
        from fusionagi.memory.gpu_search import batch_embed_units
        from fusionagi.schemas.atomic import AtomicSemanticUnit, AtomicUnitType

        units = [
            AtomicSemanticUnit(
                unit_id="u1",
                content="test content",
                type=AtomicUnitType.FACT,
                confidence=1.0,
            ),
        ]
        result = batch_embed_units(units)
        assert result is not None