Merge pull request #44 from AnkushMalaker/faster-tests

Fix crucial bug in processor status endpoint and also make tests faster

Author: AnkushMalaker

backends/advanced-backend/Docs/quickstart.md

@@ -24,6 +24,11 @@ At the moment, the basic functionalities are:
 
 ## Quick Start
 
+### 0. Forbidden Knowledge
+This is a secret command for the daring. It runs a full end to end tests with a lot of logging. Folling that is probably THE quickest way to understand whats happening end to end.
+`source .env && export DEEPGRAM_API_KEY OPENAI_API_KEY && uv run pytest tests/test_integration.py -vv -s --log-cli-level=INFO --log-cli-format='%(asctime)s - %(levelname)s - %(message)s'`   
+
+
 ### 1. Environment Setup
 
 Copy the `.env.template` file to `.env` and configure the required values:

backends/advanced-backend/src/advanced_omi_backend/processors.py

@@ -680,6 +680,15 @@ async def _memory_processor(self):
     async def _process_memory_item(self, item: MemoryProcessingItem):
         """Process a single memory item."""
         start_time = time.time()
+        audio_logger.info(f"🚀 MEMORY PROCESSING STARTED for {item.audio_uuid} at {start_time}")
+
+        # Track memory processing start
+        self.track_processing_stage(
+            item.client_id,
+            "memory",
+            "started",
+            {"audio_uuid": item.audio_uuid, "started_at": start_time},
+        )
 
         # Debug tracking removed for cleaner architecture
         # tracker = get_debug_tracker()
@@ -777,6 +786,18 @@ async def _process_memory_item(self, item: MemoryProcessingItem):
                         )
                     except Exception as e:
                         audio_logger.warning(f"Failed to update memory status: {e}")
+
+                    # Track memory processing completion
+                    self.track_processing_stage(
+                        item.client_id,
+                        "memory",
+                        "completed",
+                        {
+                            "audio_uuid": item.audio_uuid,
+                            "memories_created": len(created_memory_ids),
+                            "processing_time": time.time() - start_time,
+                        },
+                    )
                 elif success and not created_memory_ids:
                     # Successful processing but no memories created (likely empty transcript)
                     audio_logger.info(
@@ -793,6 +814,19 @@ async def _process_memory_item(self, item: MemoryProcessingItem):
                         )
                     except Exception as e:
                         audio_logger.warning(f"Failed to update memory status: {e}")
+
+                    # Track memory processing completion (even though no memories created)
+                    self.track_processing_stage(
+                        item.client_id,
+                        "memory",
+                        "completed",
+                        {
+                            "audio_uuid": item.audio_uuid,
+                            "memories_created": 0,
+                            "processing_time": time.time() - start_time,
+                            "status": "skipped",
+                        },
+                    )
                 else:
                     # This shouldn't happen, but handle it gracefully
                     audio_logger.warning(
@@ -813,6 +847,18 @@ async def _process_memory_item(self, item: MemoryProcessingItem):
                             f"📝 Updated memory processing status to FAILED for {item.audio_uuid}"
                         )
 
+                    # Track memory processing failure
+                    self.track_processing_stage(
+                        item.client_id,
+                        "memory",
+                        "failed",
+                        {
+                            "audio_uuid": item.audio_uuid,
+                            "error": f"Unexpected result: success={success}, ids={created_memory_ids}",
+                            "processing_time": time.time() - start_time,
+                        },
+                    )
+
                 # Debug tracking removed for cleaner architecture
                 # tracker.track_event(
                 #     transaction_id,
@@ -836,6 +882,18 @@ async def _process_memory_item(self, item: MemoryProcessingItem):
                         f"📝 Updated memory processing status to FAILED for {item.audio_uuid}"
                     )
 
+                # Track memory processing failure
+                self.track_processing_stage(
+                    item.client_id,
+                    "memory",
+                    "failed",
+                    {
+                        "audio_uuid": item.audio_uuid,
+                        "error": "Memory service returned False",
+                        "processing_time": time.time() - start_time,
+                    },
+                )
+
                 # Debug tracking removed for cleaner architecture
                 # tracker.track_event(
                 #     transaction_id,
@@ -857,6 +915,18 @@ async def _process_memory_item(self, item: MemoryProcessingItem):
             except Exception as e:
                 audio_logger.warning(f"Failed to update memory status: {e}")
 
+            # Track memory processing timeout failure
+            self.track_processing_stage(
+                item.client_id,
+                "memory",
+                "failed",
+                {
+                    "audio_uuid": item.audio_uuid,
+                    "error": "Processing timeout (5 minutes)",
+                    "processing_time": time.time() - start_time,
+                },
+            )
+
             # Debug tracking removed for cleaner architecture
             # tracker.track_event(
             #     transaction_id,
@@ -877,6 +947,18 @@ async def _process_memory_item(self, item: MemoryProcessingItem):
             except Exception as repo_e:
                 audio_logger.warning(f"Failed to update memory status: {repo_e}")
 
+            # Track memory processing exception failure
+            self.track_processing_stage(
+                item.client_id,
+                "memory",
+                "failed",
+                {
+                    "audio_uuid": item.audio_uuid,
+                    "error": f"Exception: {str(e)}",
+                    "processing_time": time.time() - start_time,
+                },
+            )
+
             # Debug tracking removed for cleaner architecture
             # tracker.track_event(
             #     transaction_id,
@@ -886,9 +968,10 @@ async def _process_memory_item(self, item: MemoryProcessingItem):
             #     metadata={"processing_time": time.time() - start_time},
             # )
 
-        processing_time_ms = (time.time() - start_time) * 1000
+        end_time = time.time()
+        processing_time_ms = (end_time - start_time) * 1000
         audio_logger.info(
-            f"🔄 Completed memory processing for {item.audio_uuid} in {processing_time_ms:.1f}ms"
+            f"🏁 MEMORY PROCESSING COMPLETED for {item.audio_uuid} in {processing_time_ms:.1f}ms (end time: {end_time})"
         )
 
     async def _cropping_processor(self):

backends/advanced-backend/src/advanced_omi_backend/task_manager.py

@@ -111,7 +111,13 @@ def _task_done(self, task_id: str):
             except Exception:
                 task_info.error = "Unknown error"
         else:
-            logger.debug(f"Task completed: {task_info.name}")
+            # DEBUG: Add more visible logging for memory task completion
+            if "memory_" in task_info.name:
+                logger.info(
+                    f"✅ MEMORY TASK COMPLETED: {task_info.name} at {task_info.completed_at}"
+                )
+            else:
+                logger.debug(f"Task completed: {task_info.name}")
 
         # Move to completed list
         del self.tasks[task_id]
@@ -181,8 +187,18 @@ def get_active_tasks(self) -> List[TaskInfo]:
         return list(self.tasks.values())
 
     def get_task_info(self, task_id: str) -> Optional[TaskInfo]:
-        """Get task info by task ID."""
-        return self.tasks.get(task_id)
+        """Get task info by task ID from both active and completed tasks."""
+        # First check active tasks
+        task_info = self.tasks.get(task_id)
+        if task_info:
+            return task_info
+
+        # Then check completed tasks
+        for completed_task in self.completed_tasks:
+            if f"{completed_task.name}_{id(completed_task.task)}" == task_id:
+                return completed_task
+
+        return None
 
     def get_task_count_by_type(self) -> Dict[str, int]:
         """Get count of active tasks grouped by type."""

backends/advanced-backend/tests/test_integration.py

@@ -58,7 +58,7 @@
 # Test constants
 BACKEND_URL = "http://localhost:8001"  # Test backend port
 TEST_AUDIO_PATH = tests_dir.parent.parent.parent / "extras/test-audios/DIY Experts Glass Blowing_16khz_mono_4min.wav"
-MAX_STARTUP_WAIT = 120  # seconds
+MAX_STARTUP_WAIT = 60  # seconds
 PROCESSING_TIMEOUT = 300  # seconds for audio processing (5 minutes)
 
 
@@ -851,10 +851,16 @@ def wait_for_memory_processing(self, client_id: str, timeout: int = 120):
 
                 if response.status_code == 200:
                     data = response.json()
+                    
+                    # DEBUG: Log full API response to see exactly what we're getting
+                    logger.info(f"🔍 Full processor status API response: {data}")
+                    
                     stages = data.get("stages", {})
 
                     # Check if memory stage is complete
                     memory_stage = stages.get("memory", {})
+                    logger.info(f"🧠 Memory stage data: {memory_stage}")
+                    
                     if memory_stage.get("completed", False):
                         logger.info(f"✅ Memory processing completed for client_id: {client_id}")
                         memory_processing_complete = True
@@ -873,6 +879,8 @@ def wait_for_memory_processing(self, client_id: str, timeout: int = 120):
                             error = stage_info.get("error")
                             status = "✅" if completed else "❌" if error else "⏳"
                             logger.info(f"  {status} {stage_name}: {'completed' if completed else 'error' if error else 'processing'}")
+                            # DEBUG: Show all fields in memory stage
+                            logger.info(f"    All memory stage fields: {stage_info}")
 
                 else:
                     logger.warning(f"❌ Processor status API call failed with status: {response.status_code}")
@@ -937,6 +945,10 @@ def test_runner():
 def test_full_pipeline_integration(test_runner):
     """Test the complete audio processing pipeline."""
     try:
+        # Test timing tracking
+        test_start_time = time.time()
+        phase_times = {}
+        
         # Immediate logging to debug environment
         logger.info("=" * 80)
         logger.info("🚀 STARTING INTEGRATION TEST")
@@ -946,18 +958,54 @@ def test_full_pipeline_integration(test_runner):
         logger.info(f"CI environment: {os.environ.get('CI', 'NOT SET')}")
         logger.info(f"GITHUB_ACTIONS: {os.environ.get('GITHUB_ACTIONS', 'NOT SET')}")
 
-        # Setup
+        # Phase 1: Environment setup
+        phase_start = time.time()
+        logger.info("📋 Phase 1: Setting up test environment...")
         test_runner.setup_environment()
+        phase_times['env_setup'] = time.time() - phase_start
+        logger.info(f"✅ Environment setup completed in {phase_times['env_setup']:.2f}s")
+        
+        # Phase 2: Service startup
+        phase_start = time.time()
+        logger.info("🐳 Phase 2: Starting services...")
         test_runner.start_services()
+        phase_times['service_startup'] = time.time() - phase_start
+        logger.info(f"✅ Service startup completed in {phase_times['service_startup']:.2f}s")
+        
+        # Phase 3: Wait for services
+        phase_start = time.time()
+        logger.info("⏳ Phase 3: Waiting for services to be ready...")
         test_runner.wait_for_services()
+        phase_times['service_readiness'] = time.time() - phase_start
+        logger.info(f"✅ Service readiness check completed in {phase_times['service_readiness']:.2f}s")
+        
+        # Phase 4: Authentication
+        phase_start = time.time()
+        logger.info("🔑 Phase 4: Authentication...")
         test_runner.authenticate()
+        phase_times['authentication'] = time.time() - phase_start
+        logger.info(f"✅ Authentication completed in {phase_times['authentication']:.2f}s")
 
-        # Test audio processing
+        # Phase 5: Audio upload and processing
+        phase_start = time.time()
+        logger.info("📤 Phase 5: Audio upload...")
         client_id = test_runner.upload_test_audio()
+        phase_times['audio_upload'] = time.time() - phase_start
+        logger.info(f"✅ Audio upload completed in {phase_times['audio_upload']:.2f}s")
+        
+        # Phase 6: Transcription processing
+        phase_start = time.time()
+        logger.info("🎤 Phase 6: Transcription processing...")
         conversation, transcription = test_runner.verify_processing_results(client_id)
+        phase_times['transcription_processing'] = time.time() - phase_start
+        logger.info(f"✅ Transcription processing completed in {phase_times['transcription_processing']:.2f}s")
 
-        # Validate memory extraction
+        # Phase 7: Memory extraction
+        phase_start = time.time()
+        logger.info("🧠 Phase 7: Memory extraction...")
         memories = test_runner.validate_memory_extraction(client_id)
+        phase_times['memory_extraction'] = time.time() - phase_start
+        logger.info(f"✅ Memory extraction completed in {phase_times['memory_extraction']:.2f}s")
 
         # Basic assertions
         assert conversation is not None
@@ -1003,10 +1051,25 @@ def test_full_pipeline_integration(test_runner):
 """
                 assert False, error_msg
 
-        # Log success  
+        # Calculate total test time
+        total_test_time = time.time() - test_start_time
+        phase_times['total_test'] = total_test_time
+        
+        # Log success with detailed timing
         logger.info("=" * 80)
         logger.info("🎉 INTEGRATION TEST PASSED!")
         logger.info("=" * 80)
+        logger.info(f"⏱️  TIMING BREAKDOWN:")
+        logger.info(f"  📋 Environment Setup:      {phase_times['env_setup']:>6.2f}s")
+        logger.info(f"  🐳 Service Startup:        {phase_times['service_startup']:>6.2f}s")
+        logger.info(f"  ⏳ Service Readiness:      {phase_times['service_readiness']:>6.2f}s")
+        logger.info(f"  🔑 Authentication:         {phase_times['authentication']:>6.2f}s")
+        logger.info(f"  📤 Audio Upload:           {phase_times['audio_upload']:>6.2f}s")
+        logger.info(f"  🎤 Transcription:          {phase_times['transcription_processing']:>6.2f}s")
+        logger.info(f"  🧠 Memory Extraction:      {phase_times['memory_extraction']:>6.2f}s")
+        logger.info(f"  {'─' * 35}")
+        logger.info(f"  🏁 TOTAL TEST TIME:        {total_test_time:>6.2f}s ({total_test_time/60:.1f}m)")
+        logger.info("")
         logger.info(f"📊 Test Results:")
         logger.info(f"  ✅ Audio file processed successfully")
         logger.info(f"  ✅ Transcription generated: {len(transcription)} characters")