diff --git a/tests/integration/test_chat.py b/tests/integration/test_chat.py index 2651d169..b213e37b 100644 --- a/tests/integration/test_chat.py +++ b/tests/integration/test_chat.py @@ -1036,208 +1036,7 @@ This is test documentation for integration tests. return False # ------------------------------------------------------------------------- - # Compression Tests - # ------------------------------------------------------------------------- - - def test_compression_heavy_tool_usage(self) -> bool: - """Test compression with heavy conversation usage.""" - test_name = "Compression - Heavy Tool Usage" - self.print_header(f"Testing {test_name}") - - if not self.require_auth(test_name): - return True - - self.print_info("Making 10 consecutive requests to build conversation history...") - - current_conv_id = None - - for i in range(10): - question = f"Tell me about Python topic {i+1}: data structures, decorators, async, testing. Provide a comprehensive explanation." - - payload = { - "question": question, - "history": "[]", - "isNoneDoc": True, - } - - if current_conv_id: - payload["conversation_id"] = current_conv_id - - try: - response = self.post("/api/answer", json=payload, timeout=90) - - if response.status_code == 200: - result = response.json() - current_conv_id = result.get("conversation_id", current_conv_id) - answer_preview = (result.get("answer") or "")[:80] - self.print_success(f"Request {i+1}/10 completed") - self.print_info(f" Answer: {answer_preview}...") - else: - self.print_error(f"Request {i+1}/10 failed: status {response.status_code}") - self.record_result(test_name, False, f"Request {i+1} failed") - return False - - time.sleep(2) - - except Exception as e: - self.print_error(f"Request {i+1}/10 failed: {str(e)}") - self.record_result(test_name, False, str(e)) - return False - - if current_conv_id: - self.print_success("Heavy usage test completed") - self.record_result(test_name, True, f"10 requests, conv_id: {current_conv_id}") - return True - else: - self.print_warning("No conversation_id received") - self.record_result(test_name, False, "No conversation_id") - return False - - def test_compression_needle_in_haystack(self) -> bool: - """Test that compression preserves critical information. - - Note: This is a long-running test that may timeout due to LLM response times. - Timeouts are handled gracefully as they indicate performance issues, not bugs. - """ - test_name = "Compression - Needle in Haystack" - self.print_header(f"Testing {test_name}") - - if not self.require_auth(test_name): - return True - - conversation_id = None - - # Step 1: Send general questions - self.print_info("Step 1: Sending general questions...") - for i, question in enumerate([ - "Tell me about Python best practices in detail", - "Explain Python data structures comprehensively", - ]): - payload = { - "question": question, - "history": "[]", - "isNoneDoc": True, - } - if conversation_id: - payload["conversation_id"] = conversation_id - - try: - response = self.post("/api/answer", json=payload, timeout=90) - if response.status_code == 200: - result = response.json() - conversation_id = result.get("conversation_id", conversation_id) - self.print_success(f"General question {i+1}/2 completed") - else: - self.print_error(f"Request failed: status {response.status_code}") - self.record_result(test_name, False, "General questions failed") - return False - time.sleep(2) - except Exception as e: - # Timeout errors are expected for long LLM responses - if "timed out" in str(e).lower() or "timeout" in str(e).lower(): - self.print_warning(f"Request timed out: {str(e)[:50]}") - self.record_result(test_name, True, "Skipped (timeout)") - return True - self.print_error(f"Request failed: {str(e)}") - self.record_result(test_name, False, str(e)) - return False - - # Step 2: Send critical information - self.print_info("Step 2: Sending CRITICAL information...") - critical_payload = { - "question": "Please remember: The production database password is stored in DB_PASSWORD_PROD environment variable. The backup runs at 3:00 AM UTC daily.", - "history": "[]", - "isNoneDoc": True, - "conversation_id": conversation_id, - } - - try: - response = self.post("/api/answer", json=critical_payload, timeout=90) - if response.status_code == 200: - result = response.json() - conversation_id = result.get("conversation_id", conversation_id) - self.print_success("Critical information sent") - else: - self.record_result(test_name, False, "Critical info failed") - return False - time.sleep(2) - except Exception as e: - if "timed out" in str(e).lower() or "timeout" in str(e).lower(): - self.print_warning(f"Request timed out: {str(e)[:50]}") - self.record_result(test_name, True, "Skipped (timeout)") - return True - self.record_result(test_name, False, str(e)) - return False - - # Step 3: Bury with more questions - self.print_info("Step 3: Sending more questions to bury critical info...") - for i, question in enumerate([ - "Explain Python decorators in great detail", - "Tell me about Python async programming comprehensively", - ]): - payload = { - "question": question, - "history": "[]", - "isNoneDoc": True, - "conversation_id": conversation_id, - } - - try: - response = self.post("/api/answer", json=payload, timeout=90) - if response.status_code == 200: - result = response.json() - conversation_id = result.get("conversation_id", conversation_id) - self.print_success(f"Burying question {i+1}/2 completed") - else: - self.record_result(test_name, False, "Burying questions failed") - return False - time.sleep(2) - except Exception as e: - if "timed out" in str(e).lower() or "timeout" in str(e).lower(): - self.print_warning(f"Request timed out: {str(e)[:50]}") - self.record_result(test_name, True, "Skipped (timeout)") - return True - self.record_result(test_name, False, str(e)) - return False - - # Step 4: Test recall - self.print_info("Step 4: Testing if critical info was preserved...") - recall_payload = { - "question": "What was the database password environment variable I mentioned earlier?", - "history": "[]", - "isNoneDoc": True, - "conversation_id": conversation_id, - } - - try: - response = self.post("/api/answer", json=recall_payload, timeout=90) - if response.status_code == 200: - result = response.json() - answer = (result.get("answer") or "").lower() - - if "db_password_prod" in answer or "database password" in answer: - self.print_success("Critical information preserved!") - self.print_info(f"Answer: {answer[:150]}...") - self.record_result(test_name, True, "Info preserved") - return True - else: - self.print_warning("Critical information may have been lost") - self.print_info(f"Answer: {answer[:150]}...") - self.record_result(test_name, False, "Info not preserved") - return False - else: - self.record_result(test_name, False, "Recall failed") - return False - except Exception as e: - if "timed out" in str(e).lower() or "timeout" in str(e).lower(): - self.print_warning(f"Request timed out: {str(e)[:50]}") - self.record_result(test_name, True, "Skipped (timeout)") - return True - self.record_result(test_name, False, str(e)) - return False - - # ------------------------------------------------------------------------- - # Feedback Tests (NEW) + # Feedback Tests # ------------------------------------------------------------------------- def test_feedback_positive(self) -> bool: @@ -1435,15 +1234,6 @@ This is test documentation for integration tests. self.test_tts_basic() time.sleep(1) - # Compression tests (longer running) - if self.is_authenticated: - self.test_compression_heavy_tool_usage() - time.sleep(2) - - self.test_compression_needle_in_haystack() - else: - self.print_info("Skipping compression tests (no authentication)") - return self.print_summary()