Update result analyzer and remaining changes

documentdb_tests/compatibility/result_analyzer/analyzer.py

@@ -13,25 +13,7 @@
 
 
 # Module-level constants
-INFRA_EXCEPTIONS = {
-    # Python built-in connection errors
-    "ConnectionError",
-    "ConnectionRefusedError",
-    "ConnectionResetError",
-    "ConnectionAbortedError",
-    # Python timeout errors
-    "TimeoutError",
-    "socket.timeout",
-    "socket.error",
-    # PyMongo connection errors
-    "pymongo.errors.ConnectionFailure",
-    "pymongo.errors.ServerSelectionTimeoutError",
-    "pymongo.errors.NetworkTimeout",
-    "pymongo.errors.AutoReconnect",
-    "pymongo.errors.ExecutionTimeout",
-    # Generic network/OS errors
-    "OSError",
-}
+from documentdb_tests.framework.infra_exceptions import INFRA_EXCEPTION_NAMES as INFRA_EXCEPTIONS
 
 
 # Mapping from TestOutcome to counter key names
@@ -79,10 +61,10 @@ def categorize_outcome(test_result: Dict[str, Any]) -> str:
 def extract_exception_type(crash_message: str) -> str:
     """
     Extract exception type from pytest crash message.
-    
+
     Args:
         crash_message: Message like "module.Exception: error details"
-        
+
     Returns:
         Full exception type (e.g., "pymongo.errors.OperationFailure")
         or empty string if not found
@@ -92,125 +74,149 @@ def extract_exception_type(crash_message: str) -> str:
     match = re.match(r'^([a-zA-Z0-9_.]+):\s', crash_message)
     if match:
         return match.group(1)
-
+
+    return ""
+
+
+def extract_failure_tag(test_result: Dict[str, Any]) -> str:
+    """
+    Extract failure tag (e.g. RESULT_MISMATCH) from assertion message.
+
+    The framework assertions prefix errors with tags like:
+    [RESULT_MISMATCH], [UNEXPECTED_ERROR], [UNEXPECTED_SUCCESS],
+    [ERROR_MISMATCH], [TEST_EXCEPTION]
+
+    Args:
+        test_result: Full test result dict from pytest JSON
+
+    Returns:
+        Tag string without brackets, or empty string if not found
+    """
+    call_info = test_result.get("call", {})
+    crash_info = call_info.get("crash", {})
+    crash_message = crash_info.get("message", "")
+
+    match = re.search(r'\[([A-Z_]+)\]', crash_message)
+    if match:
+        return match.group(1)
     return ""
 
 
 def is_infrastructure_error(test_result: Dict[str, Any]) -> bool:
     """
     Check if error is infrastructure-related based on exception type.
-    
+
     This checks the actual exception type rather than keywords in error messages,
     preventing false positives from error messages that happen to contain
     infrastructure-related words (e.g., "host" in an assertion message).
-    
+
     Args:
         test_result: Full test result dict from pytest JSON
-        
+
     Returns:
         True if error is infrastructure-related, False otherwise
     """
     # Get the crash info from call
     call_info = test_result.get("call", {})
     crash_info = call_info.get("crash", {})
     crash_message = crash_info.get("message", "")
-    
+
     if not crash_message:
         return False
-    
+
     # Extract exception type from "module.ExceptionClass: message" format
     exception_type = extract_exception_type(crash_message)
-    
+
     if not exception_type:
         return False
-    
+
     # Check against module-level constant
     return exception_type in INFRA_EXCEPTIONS
 
 
 def load_registered_markers(pytest_ini_path: str = "pytest.ini") -> set:
     """
     Load registered markers from pytest.ini.
-    
+
     Parses the markers section to extract marker names, ensuring we only
     use markers that are explicitly registered in pytest configuration.
-    
+
     Args:
         pytest_ini_path: Path to pytest.ini file (defaults to "pytest.ini")
-        
+
     Returns:
         Set of registered marker names
     """
     # Check if pytest.ini exists
     if not Path(pytest_ini_path).exists():
         return set()
-    
+
     registered_markers = set()
-    
+
     try:
         with open(pytest_ini_path, 'r') as f:
             in_markers_section = False
-            
+
             for line in f:
                 # Check if we're entering the markers section
                 if line.strip() == "markers =":
                     in_markers_section = True
                     continue
-                
+
                 if in_markers_section:
                     # Marker lines are indented, config keys are not
                     if line and not line[0].isspace():
                         # Non-indented line means we left the markers section
                         break
-                    
+
                     # Parse indented marker lines like "    find: Find operation tests"
                     match = re.match(r'^\s+([a-zA-Z0-9_]+):', line)
                     if match:
                         registered_markers.add(match.group(1))
-    
+
     except Exception:
         # If parsing fails, return empty set
         pass
-    
+
     return registered_markers
 
 
 class ResultAnalyzer:
     """
     Analyzer for pytest JSON test results.
-    
+
     This class provides stateful analysis with configurable pytest.ini path,
     making it easier to test and use in multiple contexts.
-    
+
     Args:
         pytest_ini_path: Path to pytest.ini file for marker configuration
-    
+
     Example:
         analyzer = ResultAnalyzer("pytest.ini")
         results = analyzer.analyze_results("report.json")
     """
-    
+
     def __init__(self, pytest_ini_path: str = "pytest.ini"):
         """
         Initialize the result analyzer.
-        
+
         Args:
             pytest_ini_path: Path to pytest.ini file (default: "pytest.ini")
         """
         self.pytest_ini_path = pytest_ini_path
         self._markers_cache: set = None
-    
+
     def _get_registered_markers(self) -> set:
         """
         Get registered markers (cached per instance).
-        
+
         Returns:
             Set of registered marker names
         """
         if self._markers_cache is None:
             self._markers_cache = load_registered_markers(self.pytest_ini_path)
         return self._markers_cache
-    
+
     def extract_markers(self, test_result: Dict[str, Any]) -> List[str]:
         """
         Extract pytest markers (tags) from a test result.
@@ -331,11 +337,14 @@ def analyze_results(self, json_report_path: str) -> Dict[str, Any]:
                 "tags": tags,
             }
 
-            # Add error information and infra error flag for failed tests
+            # Add error information for failed tests
             if test_outcome == TestOutcome.FAIL:
                 call_info = test.get("call", {})
                 test_detail["error"] = call_info.get("longrepr", "")
-                test_detail["is_infra_error"] = is_infrastructure_error(test)
+                if is_infrastructure_error(test):
+                    test_detail["failure_type"] = "INFRA_ERROR"
+                else:
+                    test_detail["failure_type"] = extract_failure_tag(test) or "UNKNOWN"
 
             tests_details.append(test_detail)
 

documentdb_tests/compatibility/result_analyzer/cli.py

@@ -95,18 +95,6 @@ def main():
             if not args.quiet:
                 print(f"\nReport saved to: {args.output}")
 
-        # If no output file and quiet mode, print to stdout
-        elif not args.quiet:
-            print("\nResults by Tag:")
-            print("-" * 60)
-            for tag, stats in sorted(
-                analysis["by_tag"].items(), key=lambda x: x[1]["pass_rate"], reverse=True
-            ):
-                passed = stats["passed"]
-                total = stats["total"]
-                rate = stats["pass_rate"]
-                print(f"{tag:30s} | {passed:3d}/{total:3d} passed ({rate:5.1f}%)")
-
         # Return exit code based on test results
         if analysis["summary"]["failed"] > 0:
             return 1

documentdb_tests/compatibility/result_analyzer/report_generator.py

@@ -99,12 +99,23 @@ def generate_text_report(analysis: Dict[str, Any], output_path: str):
         lines.append("-" * 80)
         for test in failed_tests:
             lines.append(f"\n{test['name']}")
+            failure_type = test.get("failure_type", "UNKNOWN")
+            lines.append(f"  Type: {failure_type}")
             lines.append(f"  Tags: {', '.join(test['tags'])}")
             lines.append(f"  Duration: {test['duration']:.2f}s")
             if "error" in test:
                 error_preview = test["error"][:200]
                 lines.append(f"  Error: {error_preview}...")
 
+    # Skipped tests
+    skipped_tests = [t for t in analysis["tests"] if t["outcome"] == "SKIPPED"]
+    if skipped_tests:
+        lines.append("")
+        lines.append("SKIPPED TESTS")
+        lines.append("-" * 80)
+        for test in skipped_tests:
+            lines.append(f"  {test['name']}")
+
     lines.append("")
     lines.append("=" * 80)
 
@@ -128,4 +139,40 @@ def print_summary(analysis: Dict[str, Any]):
     print(f"Passed:  {summary['passed']} ({summary['pass_rate']}%)")
     print(f"Failed:  {summary['failed']}")
     print(f"Skipped: {summary['skipped']}")
+    print("=" * 60)
+
+    # By tag
+    by_tag = analysis.get("by_tag", {})
+    if by_tag:
+        print("\nResults by Tag:")
+        print("-" * 60)
+        sorted_tags = sorted(by_tag.items(), key=lambda x: x[1]["pass_rate"])
+        for tag, stats in sorted_tags:
+            print(f"  {tag:<30s} | {stats['passed']:>3}/{stats['total']:>3} passed ({stats['pass_rate']:>5.1f}%)")
+
+    # Failed tests
+    failed_tests = [t for t in analysis["tests"] if t["outcome"] == "FAIL"]
+    if failed_tests:
+        # Count by failure_type
+        from collections import Counter
+        type_counts = Counter(t.get("failure_type", "UNKNOWN") for t in failed_tests)
+
+        print(f"\nFailed Tests ({len(failed_tests)}):")
+        print("-" * 60)
+        for ft, count in sorted(type_counts.items()):
+            print(f"\n  {ft} ({count}):")
+            for test in failed_tests:
+                if test.get("failure_type", "UNKNOWN") == ft:
+                    name = test["name"].split("::")[-1]
+                    print(f"    {name}")
+
+    # Skipped tests
+    skipped_tests = [t for t in analysis["tests"] if t["outcome"] == "SKIPPED"]
+    if skipped_tests:
+        print(f"\nSkipped Tests ({len(skipped_tests)}):")
+        print("-" * 60)
+        for test in skipped_tests:
+            name = test["name"].split("::")[-1]
+            print(f"  {name}")
+
     print("=" * 60 + "\n")

documentdb_tests/compatibility/tests/collection/__init__.py

@@ -0,0 +1 @@
+"""Collection management tests."""