mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-14 06:17:23 +00:00
fix: remove bad tests
This commit is contained in:
Alex committed
1 parent
e6ac4008fe
commit
6381f7dd4e
1 file changed
+1
-211
@@ -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()
|
||||
|
||||
|
||||
|
||||
Reference in new issue
Block a user