12 18 test env fix

backends/advanced/src/advanced_omi_backend/services/memory/providers/chronicle.py

@@ -290,6 +290,62 @@ async def count_memories(self, user_id: str) -> Optional[int]:
             memory_logger.error(f"Count memories failed: {e}")
             return None
 
+    async def get_memory(self, memory_id: str, user_id: Optional[str] = None) -> Optional[MemoryEntry]:
+        """Get a specific memory by ID.
+
+        Retrieves a single memory from Qdrant by its ID and verifies user ownership.
+
+        Args:
+            memory_id: Unique identifier of the memory to retrieve
+            user_id: Optional user ID for access control
+
+        Returns:
+            MemoryEntry object if found and authorized, None otherwise
+        """
+        if not self._initialized:
+            await self.initialize()
+
+        try:
+            # Retrieve the point from Qdrant
+            points = await self.vector_store.client.retrieve(
+                collection_name=self.vector_store.collection_name,
+                ids=[memory_id],
+                with_payload=True,
+                with_vectors=False
+            )
+
+            if not points:
+                memory_logger.debug(f"Memory {memory_id} not found in vector store")
+                return None
+
+            point = points[0]
+            payload = point.payload
+
+            # Check user ownership if user_id is provided
+            if user_id:
+                memory_user_id = payload.get("metadata", {}).get("user_id")
+                if memory_user_id != user_id:
+                    memory_logger.warning(
+                        f"Access denied: Memory {memory_id} belongs to {memory_user_id}, requested by {user_id}"
+                    )
+                    return None
+
+            # Convert to MemoryEntry
+            memory_entry = MemoryEntry(
+                id=str(point.id),
+                content=payload.get("content", ""),
+                metadata=payload.get("metadata", {}),
+                created_at=payload.get("created_at"),
+                embedding=None  # Don't return embeddings for single memory retrieval
+            )
+
+            memory_logger.info(f"📖 Retrieved memory {memory_id} for user {user_id}")
+            return memory_entry
+
+        except Exception as e:
+            memory_logger.error(f"Get memory {memory_id} failed: {e}", exc_info=True)
+            return None
+
     async def delete_memory(self, memory_id: str, user_id: Optional[str] = None, user_email: Optional[str] = None) -> bool:
         """Delete a specific memory by ID.
 
@@ -428,7 +484,7 @@ def _create_memory_entries(
                         "extraction_enabled": self.config.extraction_enabled,
                     },
                     embedding=embedding,
-                    created_at=str(int(time.time())),
+                    created_at=int(time.time()),
                 )
             )
 
@@ -626,15 +682,15 @@ async def _apply_memory_actions(
                 if emb is None:
                     memory_logger.warning(f"Skipping ADD action due to missing embedding: {action_text}")
                     continue
-                    
+
                 memory_id = str(uuid.uuid4())
                 memory_entries.append(
                     MemoryEntry(
                         id=memory_id,
                         content=action_text,
                         metadata=base_metadata,
                         embedding=emb,
-                        created_at=str(int(time.time())),
+                        created_at=int(time.time()),
                     )
                 )
                 memory_logger.info(f"➕ Added new memory: {memory_id} - {action_text[:50]}...")

tests/setup/test_env.py

@@ -1,25 +1,26 @@
 # Test Environment Configuration
 import os
 from pathlib import Path
+from dotenv import load_dotenv
 
 # Load .env file from backends/advanced directory if it exists
 # This allows tests to work when run from VSCode or command line
-def load_env_file():
-    """Load environment variables from .env file if it exists."""
-    # Look for .env in backends/advanced directory
-    env_file = Path(__file__).parent.parent.parent / "backends" / "advanced" / ".env"
-    if env_file.exists():
-        with open(env_file) as f:
-            for line in f:
-                line = line.strip()
-                if line and not line.startswith('#') and '=' in line:
-                    key, value = line.split('=', 1)
-                    # Only set if not already in environment (CI takes precedence)
-                    if key not in os.environ:
-                        os.environ[key] = value
+# def load_env_file():
+#     """Load environment variables from .env file if it exists."""
+#     # Look for .env in backends/advanced directory
+#     env_file = Path(__file__).parent.parent.parent / "backends" / "advanced" / ".env"
+#     if env_file.exists():
+#         with open(env_file) as f:
+#             for line in f:
+#                 line = line.strip()
+#                 if line and not line.startswith('#') and '=' in line:
+#                     key, value = line.split('=', 1)
+#                     # Only set if not already in environment (CI takes precedence)
+#                     if key not in os.environ:
+#                         os.environ[key] = value
 
 # Load .env file (CI environment variables take precedence)
-load_env_file()
+# load_env_file()
 
 # Load .env from backends/advanced directory to get COMPOSE_PROJECT_NAME
 backend_env_path = Path(__file__).resolve().parents[2] / "backends" / "advanced" / ".env"