Repository: mem0ai/mem0
File: tests/test_memory.py (Python)

Code before:
import threading
from unittest.mock import MagicMock, patch

import pytest

from mem0 import Memory
from mem0.configs.base import MemoryConfig


class MockFieldInfo:
    """Mock pydantic field info."""
    def __init__(self, default=None):
        self.default = default


class MockOpenSearchConfig:
    
    model_fields = {
        'collection_name': MockFieldInfo(default="default_collection"),
        'host': MockFieldInfo(default="localhost"),
        'port': MockFieldInfo(default=9200),
        'embedding_model_dims': MockFieldInfo(default=1536),
        'http_auth': MockFieldInfo(default=None),
        'auth': MockFieldInfo(default=None),
        'credentials': MockFieldInfo(default=None),
        'connection_class': MockFieldInfo(default=None),
        'use_ssl': MockFieldInfo(default=False),
        'verify_certs': MockFieldInfo(default=False),
    }
    
    def __init__(self, collection_name="test_collection", include_auth=True, **kwargs):
        self.collection_name = collection_name
        self.host = kwargs.get("host", "localhost")
        self.port = kwargs.get("port", 9200)
        self.embedding_model_dims = kwargs.get("embedding_model_dims", 1536)
        self.use_ssl = kwargs.get("use_ssl", True)
        self.verify_certs = kwargs.get("verify_certs", True)
        
        if any(field in kwargs for field in ["http_auth", "auth", "credentials", "connection_class"]):
            self.http_auth = kwargs.get("http_auth")
            self.auth = kwargs.get("auth")
            self.credentials = kwargs.get("credentials")

Review comment (suggestion):
Hey @lsvishaal can you add this tests to opensearch test? This test is for memory and not related to any specific vector store here. 

Code after:
from unittest.mock import MagicMock, patch

import pytest

from mem0 import Memory
from mem0.configs.base import MemoryConfig


class MockVectorMemory:
    """Mock memory object for testing incomplete payloads."""
    
    def __init__(self, memory_id: str, payload: dict, score: float = 0.8):
        self.id = memory_id
        self.payload = payload
        self.score = score


@pytest.fixture
def memory_client():
    with patch.object(Memory, "__init__", return_value=None):
        client = Memory()
        client.add = MagicMock(return_value={"results": [{"id": "1", "memory": "Name is John Doe.", "event": "ADD"}]})
        client.get = MagicMock(return_value={"id": "1", "memory": "Name is John Doe."})
        client.update = MagicMock(return_value={"message": "Memory updated successfully!"})
        client.delete = MagicMock(return_value={"message": "Memory deleted successfully!"})
        client.history = MagicMock(return_value=[{"memory": "I like Indian food."}, {"memory": "I like Italian food."}])
        client.get_all = MagicMock(return_value=["Name is John Doe.", "Name is John Doe. I like to code in Python."])
        yield client


def test_create_memory(memory_client):
    data = "Name is John Doe."
    result = memory_client.add([{"role": "user", "content": data}], user_id="test_user")
    assert result["results"][0]["memory"] == data


def test_get_memory(memory_client):
    data = "Name is John Doe."
    memory_client.add([{"role": "user", "content": data}], user_id="test_user")
    result = memory_client.get("1")
    assert result["memory"] == data