torbjorn/my-pal-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Simplify logging and conform to pytest conventions

Browse Source 
- Removed excessive debug logging in http_transport_recorder.py
- Consolidated redundant log statements
- Fixed exception logging to use logger.exception()
- Removed emojis from log messages for cleaner output
- Removed __main__ block from test_o3_pro_output_text_fix.py per pytest conventions
- Applied black formatting to comply with CI checks

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Josh Vera
2025-07-13 10:49:37 -06:00
parent 9248947e39
commit ac7d489cb4
2 changed files with 18 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
tests/http_transport_recorder.py
View File

@@ -39,7 +39,7 @@ class RecordingTransport(httpx.HTTPTransport):
def handle_request(self, request: httpx.Request) -> httpx.Response: def handle_request(self, request: httpx.Request) -> httpx.Response:
"""Handle request by recording interaction and delegating to real transport.""" """Handle request by recording interaction and delegating to real transport."""
logger.debug(f"🎬 RecordingTransport: Making request to {request.method} {request.url}") logger.debug(f"RecordingTransport: Making request to {request.method} {request.url}")
# Record request BEFORE making the call # Record request BEFORE making the call
request_data = self._serialize_request(request) request_data = self._serialize_request(request)
@@ -47,7 +47,7 @@ class RecordingTransport(httpx.HTTPTransport):
# Make real HTTP call using parent transport # Make real HTTP call using parent transport
response = super().handle_request(request) response = super().handle_request(request)
logger.debug(f"🎬 RecordingTransport: Got response {response.status_code}") logger.debug(f"RecordingTransport: Got response {response.status_code}")
# Post-response content capture (proper approach) # Post-response content capture (proper approach)
if self.capture_content: if self.capture_content:
@@ -56,7 +56,7 @@ class RecordingTransport(httpx.HTTPTransport):
# Note: httpx automatically handles gzip decompression # Note: httpx automatically handles gzip decompression
content_bytes = response.read() content_bytes = response.read()
response.close() # Close the original stream response.close() # Close the original stream
logger.debug(f"🎬 RecordingTransport: Captured {len(content_bytes)} bytes of decompressed content") logger.debug(f"RecordingTransport: Captured {len(content_bytes)} bytes")
# Serialize response with captured content # Serialize response with captured content
response_data = self._serialize_response_with_content(response, content_bytes) response_data = self._serialize_response_with_content(response, content_bytes)
@@ -67,9 +67,8 @@ class RecordingTransport(httpx.HTTPTransport):
if response.headers.get("content-encoding") == "gzip": if response.headers.get("content-encoding") == "gzip":
import gzip import gzip
logger.debug(f"🗜️ Re-compressing {len(content_bytes)} bytes with gzip...")
response_content = gzip.compress(content_bytes) response_content = gzip.compress(content_bytes)
logger.debug(f"🗜️ Compressed to {len(response_content)} bytes") logger.debug(f"Re-compressed content: {len(content_bytes)} → {len(response_content)} bytes")
new_response = httpx.Response( new_response = httpx.Response(
status_code=response.status_code, status_code=response.status_code,
@@ -85,11 +84,8 @@ class RecordingTransport(httpx.HTTPTransport):
return new_response return new_response
except Exception as e: except Exception:
logger.warning(f"⚠️ Content capture failed: {e}, falling back to stub") logger.warning("Content capture failed, falling back to stub", exc_info=True)
import traceback
logger.warning(f"⚠️ Full exception traceback:\n{traceback.format_exc()}")
response_data = self._serialize_response(response) response_data = self._serialize_response(response)
self._record_interaction(request_data, response_data) self._record_interaction(request_data, response_data)
return response return response
@@ -104,7 +100,7 @@ class RecordingTransport(httpx.HTTPTransport):
interaction = {"request": request_data, "response": response_data} interaction = {"request": request_data, "response": response_data}
self.recorded_interactions.append(interaction) self.recorded_interactions.append(interaction)
self._save_cassette() self._save_cassette()
logger.debug(f"🎬 RecordingTransport: Saved cassette to {self.cassette_path}") logger.debug(f"Saved cassette to {self.cassette_path}")
def _serialize_request(self, request: httpx.Request) -> dict[str, Any]: def _serialize_request(self, request: httpx.Request) -> dict[str, Any]:
"""Serialize httpx.Request to JSON-compatible format.""" """Serialize httpx.Request to JSON-compatible format."""
@@ -150,21 +146,18 @@ class RecordingTransport(httpx.HTTPTransport):
"""Serialize httpx.Response with captured content.""" """Serialize httpx.Response with captured content."""
try: try:
# Debug: check what we got # Debug: check what we got
logger.debug(f"🔍 Content type: {type(content_bytes)}, size: {len(content_bytes)}")
logger.debug(f"🔍 First 100 chars: {content_bytes[:100]}")
# Ensure we have bytes for base64 encoding # Ensure we have bytes for base64 encoding
if not isinstance(content_bytes, bytes): if not isinstance(content_bytes, bytes):
logger.warning(f"⚠️ Content is not bytes, converting from {type(content_bytes)}") logger.warning(f"Content is not bytes, converting from {type(content_bytes)}")
if isinstance(content_bytes, str): if isinstance(content_bytes, str):
content_bytes = content_bytes.encode("utf-8") content_bytes = content_bytes.encode("utf-8")
else: else:
content_bytes = str(content_bytes).encode("utf-8") content_bytes = str(content_bytes).encode("utf-8")
# Encode content as base64 for JSON storage # Encode content as base64 for JSON storage
logger.debug(f"🔍 Base64 encoding {len(content_bytes)} bytes...")
content_b64 = base64.b64encode(content_bytes).decode("utf-8") content_b64 = base64.b64encode(content_bytes).decode("utf-8")
logger.debug(f"✅ Base64 encoded successfully, result length: {len(content_b64)}") logger.debug(f"Base64 encoded {len(content_bytes)} bytes → {len(content_b64)} chars")
response_data = { response_data = {
"status_code": response.status_code, "status_code": response.status_code,
@@ -179,10 +172,7 @@ class RecordingTransport(httpx.HTTPTransport):
return response_data return response_data
except Exception as e: except Exception as e:
logger.error(f"🔍 Error in _serialize_response_with_content: {e}") logger.exception("Error in _serialize_response_with_content")
import traceback
logger.error(f"🔍 Full traceback: {traceback.format_exc()}")
# Fall back to minimal info # Fall back to minimal info
return { return {
"status_code": response.status_code, "status_code": response.status_code,
@@ -234,36 +224,19 @@ class ReplayTransport(httpx.MockTransport):
def _handle_request(self, request: httpx.Request) -> httpx.Response: def _handle_request(self, request: httpx.Request) -> httpx.Response:
"""Handle request by finding matching interaction and returning saved response.""" """Handle request by finding matching interaction and returning saved response."""
logger.debug(f"🔍 ReplayTransport: Looking for {request.method} {request.url}") logger.debug(f"ReplayTransport: Looking for {request.method} {request.url}")
# Debug: show what we're trying to match # Debug: show what we're trying to match
request_signature = self._get_request_signature(request) request_signature = self._get_request_signature(request)
logger.debug(f"🔍 Request signature: {request_signature}") logger.debug(f"Request signature: {request_signature}")
# Debug: show actual request content
content = request.content
if hasattr(content, "read"):
content = content.read()
if isinstance(content, bytes):
content_str = content.decode("utf-8", errors="ignore")
else:
content_str = str(content) if content else ""
logger.debug(f"🔍 Actual request content: {content_str}")
# Debug: show available signatures
for i, interaction in enumerate(self.interactions):
saved_signature = self._get_saved_request_signature(interaction["request"])
saved_content = interaction["request"].get("content", {})
logger.debug(f"🔍 Available signature {i}: {saved_signature}")
logger.debug(f"🔍 Saved content {i}: {saved_content}")
# Find matching interaction # Find matching interaction
interaction = self._find_matching_interaction(request) interaction = self._find_matching_interaction(request)
if not interaction: if not interaction:
logger.warning("🚨 MYSTERY SOLVED: No matching interaction found! This should fail...") logger.warning("No matching interaction found in cassette")
raise ValueError(f"No matching interaction found for {request.method} {request.url}") raise ValueError(f"No matching interaction found for {request.method} {request.url}")
logger.debug("✅ Found matching interaction from cassette!") logger.debug("Found matching interaction in cassette")
# Build response from saved data # Build response from saved data
response_data = interaction["response"] response_data = interaction["response"]
@@ -276,9 +249,9 @@ class ReplayTransport(httpx.MockTransport):
# Decode base64 content # Decode base64 content
try: try:
content_bytes = base64.b64decode(content["data"]) content_bytes = base64.b64decode(content["data"])
logger.debug(f"🎬 ReplayTransport: Decoded {len(content_bytes)} bytes from base64") logger.debug(f"Decoded {len(content_bytes)} bytes from base64")
except Exception as e: except Exception as e:
logger.warning(f"⚠️ Failed to decode base64 content: {e}") logger.warning(f"Failed to decode base64 content: {e}")
content_bytes = json.dumps(content).encode("utf-8") content_bytes = json.dumps(content).encode("utf-8")
else: else:
# Legacy format or stub content # Legacy format or stub content
@@ -292,11 +265,10 @@ class ReplayTransport(httpx.MockTransport):
# Re-compress the content for httpx # Re-compress the content for httpx
import gzip import gzip
logger.debug(f"🗜️ ReplayTransport: Re-compressing {len(content_bytes)} bytes with gzip...")
content_bytes = gzip.compress(content_bytes) content_bytes = gzip.compress(content_bytes)
logger.debug(f"🗜️ ReplayTransport: Compressed to {len(content_bytes)} bytes") logger.debug(f"Re-compressed for replay: {len(content_bytes)} bytes")
logger.debug(f"🎬 ReplayTransport: Returning cassette response with content: {content_bytes[:100]}...") logger.debug(f"Returning cassette response ({len(content_bytes)} bytes)")
# Create httpx.Response # Create httpx.Response
return httpx.Response( return httpx.Response(

12
tests/test_o3_pro_output_text_fix.py
View File

@@ -12,7 +12,6 @@ RECORDING: To record new responses, delete the cassette file and run with real A
import logging import logging
import os import os
import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import patch from unittest.mock import patch
@@ -116,14 +115,3 @@ class TestO3ProOutputTextFix:
metadata = response_data["metadata"] metadata = response_data["metadata"]
assert metadata["model_used"] == "o3-pro" assert metadata["model_used"] == "o3-pro"
assert metadata["provider_used"] == "openai" assert metadata["provider_used"] == "openai"
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
logger.info("🎥 OpenAI Response Recording Tests for O3-Pro Output Text Fix")
logger.info("=" * 50)
logger.info("RECORD MODE: Requires OPENAI_API_KEY - makes real API calls through ChatTool")
logger.info("REPLAY MODE: Uses recorded HTTP responses - free and fast")
logger.info("RECORDING: Delete .json files in tests/openai_cassettes/ to re-record")
unittest.main()