r2rpy/py/tests/integration/test_indices.py

import pytest

from r2r import R2RClient, R2RException


@pytest.fixture(scope="session")
def config():

    class TestConfig:
        base_url = "http://localhost:7272"
        superuser_email = "admin@example.com"
        superuser_password = "change_me_immediately"

    return TestConfig()


@pytest.fixture(scope="session")
def client(config):
    """Create a client instance and log in as superuser."""
    client = R2RClient(config.base_url)
    client.users.login(config.superuser_email, config.superuser_password)
    return client


# def test_create_and_get_index(client: R2RClient):
#     index_name = f"test_index_{uuid.uuid4().hex[:8]}"
#     config = {
#         "table_name": "chunks",
#         "index_method": "hnsw",
#         "index_measure": "cosine_distance",
#         "index_arguments": {"m": 16, "ef_construction": 64, "ef": 40},
#         "index_name": index_name,
#         "index_column": "vec",
#         "concurrently": True,
#     }

#     # Create the index
#     create_resp = client.indices.create(
#         config=config, run_with_orchestration=True
#     ).results
#     assert create_resp.message is not None, "No message in create response"

#     # Get the index details
#     results = client.indices.retrieve(
#         index_name=index_name, table_name="chunks"
#     ).results
#     assert results.index is not None, "No index in get response"
#     assert results.index["name"] == index_name, "Index name mismatch"


def test_list_indices(client: R2RClient):
    try:
        resp = client.indices.list(limit=5)
        results = resp.results
    except Exception as e:
        print(f"Error: {e}")
    assert results.indices is not None, "Indices field is None"
    # Just ensure we get a list without error. Detailed checks depend on data availability.
    assert isinstance(results.indices, list), "Indices field is not a list"


# def test_delete_index(client: R2RClient):
#     # Create an index to delete
#     index_name = f"test_delete_index_{uuid.uuid4().hex[:8]}"
#     config = {
#         "table_name": "chunks",
#         "index_method": "hnsw",
#         "index_measure": "cosine_distance",
#         "index_arguments": {"m": 16, "ef_construction": 64, "ef": 40},
#         "index_name": index_name,
#         "index_column": "vec",
#         "concurrently": True,
#     }

#     client.indices.create(config=config, run_with_orchestration=True).results

#     # Delete the index
#     delete_resp = client.indices.delete(
#         index_name=index_name, table_name="chunks"
#     ).results
#     assert delete_resp.message is not None, "No message in delete response"

#     # Verify deletion by attempting to retrieve the index
#     with pytest.raises(R2RException) as exc_info:
#         client.indices.retrieve(index_name=index_name, table_name="chunks")
#     assert (
#         "not found" in str(exc_info.value).lower()
#     ), "Unexpected error message for deleted index"


def test_error_handling(client: R2RClient):
    # Try to get a non-existent index
    with pytest.raises(R2RException) as exc_info:
        client.indices.retrieve(index_name="nonexistent_index",
                                table_name="chunks")
    assert "not found" in str(exc_info.value).lower(), (
        "Unexpected error message for non-existent index")