"""Pytest plugin for enhanced diagnostic testing.
This plugin extends pytest with capabilities for detailed diagnostic reporting
while maintaining standard test pass/fail behavior.
"""
import json
import time
import traceback
from json import JSONEncoder
from pathlib import Path
from typing import Any, Dict, Generator, List, Optional
import pytest
# Custom JSON Encoder that can handle binary data
class DiagnosticJSONEncoder(JSONEncoder):
"""Custom JSON encoder that can handle bytes and other non-serializable types."""
def default(self, obj: Any) -> Any:
"""Convert bytes and other types to JSON-serializable objects."""
if isinstance(obj, bytes):
# Convert bytes to base64 string for JSON serialization
import base64
return {"__bytes__": True, "value": base64.b64encode(obj).decode("ascii")}
# Handle Path objects
if isinstance(obj, Path):
return str(obj)
# Handle tree-sitter specific types
if hasattr(obj, "start_point") and hasattr(obj, "end_point") and hasattr(obj, "type"):
# Probably a tree-sitter Node
return {
"type": obj.type,
"start_point": obj.start_point,
"end_point": obj.end_point,
"_tsnode": True,
}
# Handle types with custom __dict__ but no standard serialization
if hasattr(obj, "__dict__"):
try:
return obj.__dict__
except (TypeError, AttributeError):
pass
# Let the base class handle any other types
return super().default(obj)
# Global storage for test context and diagnostic results
_DIAGNOSTICS: Dict[str, "DiagnosticData"] = {}
_CURRENT_TEST: Dict[str, Any] = {}
class DiagnosticData:
"""Container for diagnostic information."""
def __init__(self, test_id: str):
"""Initialize with test ID."""
self.test_id = test_id
self.start_time = time.time()
self.end_time: Optional[float] = None
self.status = "pending"
self.details: Dict[str, Any] = {}
self.errors: List[Dict[str, Any]] = []
self.artifacts: Dict[str, Any] = {}
def add_error(self, error_type: str, message: str, tb: Optional[str] = None) -> None:
"""Add an error to the diagnostic data."""
error_info = {
"type": error_type,
"message": message,
}
if tb:
error_info["traceback"] = tb
self.errors.append(error_info)
self.status = "error"
def add_detail(self, key: str, value: Any) -> None:
"""Add a detail to the diagnostic data."""
self.details[key] = value
def add_artifact(self, name: str, content: Any) -> None:
"""Add an artifact to the diagnostic data."""
self.artifacts[name] = content
def finalize(self, status: str = "completed") -> None:
"""Mark the diagnostic as complete."""
self.end_time = time.time()
if not self.errors:
self.status = status
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"test_id": self.test_id,
"status": self.status,
"start_time": self.start_time,
"end_time": self.end_time,
"duration": self.end_time - self.start_time if self.end_time else None,
"details": self.details,
"errors": self.errors,
"artifacts": self.artifacts,
}
@pytest.fixture
def diagnostic(request: Any) -> Generator[DiagnosticData, None, None]:
"""Fixture to provide diagnostic functionality to tests."""
# Get the current test ID
test_id = f"{request.path}::{request.node.name}"
# Create a diagnostic data instance
diag = DiagnosticData(test_id)
_DIAGNOSTICS[test_id] = diag
yield diag
# Finalize the diagnostic when the test is done
diag.finalize()
def pytest_configure(config: Any) -> None:
"""Set up the plugin when pytest starts."""
# Register additional markers
config.addinivalue_line("markers", "diagnostic: mark test as producing diagnostic information")
def pytest_runtest_protocol(item: Any, nextitem: Any) -> Optional[bool]:
"""Custom test protocol that captures detailed diagnostics."""
# Use the standard protocol
return None
def pytest_runtest_setup(item: Any) -> None:
"""Set up the test environment."""
# This is no longer needed as we use the request fixture
pass
def pytest_runtest_teardown(item: Any) -> None:
"""Clean up after a test."""
# This is no longer needed as we use the request fixture
pass
def pytest_terminal_summary(terminalreporter: Any, exitstatus: Any, config: Any) -> None:
"""Add diagnostic summary to the terminal output."""
if _DIAGNOSTICS:
terminalreporter.write_sep("=", "Diagnostic Summary")
error_count = sum(1 for d in _DIAGNOSTICS.values() if d.status == "error")
terminalreporter.write_line(f"Collected {len(_DIAGNOSTICS)} diagnostics, {error_count} with errors")
# If there are errors, show details
if error_count:
terminalreporter.write_sep("-", "Error Details")
for test_id, diag in _DIAGNOSTICS.items():
if diag.status == "error":
terminalreporter.write_line(f"- {test_id}")
for i, error in enumerate(diag.errors):
terminalreporter.write_line(f" Error {i + 1}: {error['type']}: {error['message']}")
def pytest_sessionfinish(session: Any, exitstatus: Any) -> None:
"""Generate JSON reports at the end of the test session."""
output_dir = Path("diagnostic_results")
output_dir.mkdir(exist_ok=True)
timestamp = time.strftime("%Y%m%d_%H%M%S")
output_file = output_dir / f"diagnostic_results_{timestamp}.json"
# Convert diagnostics to JSON-serializable dict
diagnostics_dict = {k: v.to_dict() for k, v in _DIAGNOSTICS.items()}
# Write the results to a file
with open(output_file, "w") as f:
json.dump(
{
"timestamp": timestamp,
"diagnostics": diagnostics_dict,
"summary": {
"total": len(diagnostics_dict),
"errors": sum(1 for d in diagnostics_dict.values() if d["status"] == "error"),
"completed": sum(1 for d in diagnostics_dict.values() if d["status"] == "completed"),
},
},
f,
indent=2,
cls=DiagnosticJSONEncoder,
)
print(f"\nDiagnostic results saved to {output_file}")
@pytest.hookimpl(tryfirst=True)
def pytest_exception_interact(node: Any, call: Any, report: Any) -> None:
"""Capture exception details for diagnostics."""
if call.excinfo:
try:
test_id = f"{node.path}::{node.name}"
if test_id in _DIAGNOSTICS:
diag = _DIAGNOSTICS[test_id]
exc_type = call.excinfo.type.__name__
exc_value = str(call.excinfo.value)
tb_str = "\n".join(traceback.format_tb(call.excinfo.tb))
diag.add_error(exc_type, exc_value, tb_str)
except Exception as e:
print(f"Error recording diagnostic info: {e}")
