AI Code Review Crisis - 45% Security Flaws in Generated Code

# The AI Coding Paradox
implementation_time = 10  # minutes with AI
review_time = 45  # minutes to safely review AI code
test_time = 30  # minutes to verify correctness

total_time = implementation_time + review_time + test_time  # 85 minutes
# vs 60 minutes for human-written code with proper review

# AI makes coding faster but INCREASES time-to-production
# if review process isn't optimized

import re
import ast
from typing import List, Dict, Tuple
from dataclasses import dataclass
from enum import Enum

class Severity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

@dataclass
class SecurityIssue:
    file_path: str
    line_number: int
    issue_type: str
    severity: Severity
    description: str
    code_snippet: str
    fix_suggestion: str

class AICodeSecurityScanner:
    """
    Security scanner specifically designed for AI-generated code
    Catches common vulnerabilities that AI models frequently generate
    """

    def __init__(self):
        self.issues: List[SecurityIssue] = []

        # Patterns that indicate security vulnerabilities
        self.sql_injection_patterns = [
            r'execute\s*\(\s*f["\'].*?\{.*?\}',  # f-string in SQL
            r'execute\s*\(\s*["\'].*?\%\s*\(',  # % formatting in SQL
            r'execute\s*\(\s*.*?\+\s*',  # String concatenation in SQL
            r'\.format\s*\(.*?\).*?execute',  # .format() with execute
        ]

        self.hardcoded_secret_patterns = [
            r'password\s*=\s*["\'][^"\']{8,}["\']',
            r'api[_-]?key\s*=\s*["\'][^"\']{20,}["\']',
            r'secret\s*=\s*["\'][^"\']{20,}["\']',
            r'token\s*=\s*["\'][^"\']{20,}["\']',
            r'aws[_-]?secret[_-]?access[_-]?key',
        ]

        self.xss_patterns = [
            r'innerHTML\s*=',
            r'document\.write\s*\(',
            r'eval\s*\(',
            r'dangerouslySetInnerHTML',
        ]

    def scan_file(self, file_path: str, content: str) -> List[SecurityIssue]:
        """Scan a single file for security vulnerabilities"""
        self.issues = []
        lines = content.split('\n')

        # Check each line for vulnerabilities
        for line_num, line in enumerate(lines, 1):
            # SQL Injection checks
            for pattern in self.sql_injection_patterns:
                if re.search(pattern, line, re.IGNORECASE):
                    self.issues.append(SecurityIssue(
                        file_path=file_path,
                        line_number=line_num,
                        issue_type="SQL Injection",
                        severity=Severity.CRITICAL,
                        description="Potential SQL injection vulnerability detected. Never use string formatting/concatenation with SQL queries.",
                        code_snippet=line.strip(),
                        fix_suggestion="Use parameterized queries: cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))"
                    ))

            # Hardcoded secrets check
            for pattern in self.hardcoded_secret_patterns:
                if re.search(pattern, line, re.IGNORECASE):
                    # Exclude obvious test/example cases
                    if not any(word in line.lower() for word in ['example', 'test', 'demo', 'placeholder']):
                        self.issues.append(SecurityIssue(
                            file_path=file_path,
                            line_number=line_num,
                            issue_type="Hardcoded Credentials",
                            severity=Severity.CRITICAL,
                            description="Hardcoded credentials detected. Credentials must never be in source code.",
                            code_snippet=line.strip()[:50] + "...",  # Truncate for security
                            fix_suggestion="Use environment variables: api_key = os.getenv('API_KEY')"
                        ))

            # XSS vulnerability check
            for pattern in self.xss_patterns:
                if re.search(pattern, line, re.IGNORECASE):
                    self.issues.append(SecurityIssue(
                        file_path=file_path,
                        line_number=line_num,
                        issue_type="XSS Vulnerability",
                        severity=Severity.HIGH,
                        description="Potential XSS vulnerability. User input must be sanitized before rendering.",
                        code_snippet=line.strip(),
                        fix_suggestion="Use safe DOM methods or sanitize input with DOMPurify"
                    ))

        # Check for missing input validation
        self._check_input_validation(file_path, content, lines)

        # Check for insecure dependencies
        if file_path.endswith(('requirements.txt', 'package.json')):
            self._check_dependencies(file_path, content)

        return self.issues

    def _check_input_validation(self, file_path: str, content: str, lines: List[str]):
        """Check if user input is validated"""
        # Look for request parameter access without validation
        request_patterns = [
            r'request\.(GET|POST|args|form|json)\[',
            r'request\.(GET|POST|args|form|json)\.get\(',
        ]

        for line_num, line in enumerate(lines, 1):
            for pattern in request_patterns:
                if re.search(pattern, line):
                    # Check if there's validation nearby (within 5 lines)
                    context_start = max(0, line_num - 6)
                    context_end = min(len(lines), line_num + 5)
                    context = '\n'.join(lines[context_start:context_end])

                    # Look for validation keywords
                    has_validation = any(keyword in context.lower() for keyword in [
                        'validate', 'validator', 'isinstance', 'type(',
                        'assert', 'raise', 'if not', 'try'
                    ])

                    if not has_validation:
                        self.issues.append(SecurityIssue(
                            file_path=file_path,
                            line_number=line_num,
                            issue_type="Missing Input Validation",
                            severity=Severity.HIGH,
                            description="User input accessed without apparent validation",
                            code_snippet=line.strip(),
                            fix_suggestion="Validate input: if not isinstance(user_id, int): raise ValueError()"
                        ))
                        break  # Only report once per line

    def _check_dependencies(self, file_path: str, content: str):
        """Check for known vulnerable dependencies"""
        vulnerable_packages = {
            'requests': ['2.25.0', '2.26.0'],  # Example vulnerable versions
            'flask': ['1.0.0', '1.1.0'],
            'django': ['2.2.0', '3.0.0'],
        }

        lines = content.split('\n')
        for line_num, line in enumerate(lines, 1):
            for package, vulnerable_versions in vulnerable_packages.items():
                for version in vulnerable_versions:
                    if f'{package}=={version}' in line.lower():
                        self.issues.append(SecurityIssue(
                            file_path=file_path,
                            line_number=line_num,
                            issue_type="Vulnerable Dependency",
                            severity=Severity.HIGH,
                            description=f"Known vulnerable version of {package} detected",
                            code_snippet=line.strip(),
                            fix_suggestion=f"Upgrade {package} to latest stable version"
                        ))

    def generate_report(self) -> str:
        """Generate human-readable security report"""
        if not self.issues:
            return "✅ No security issues detected"

        # Group by severity
        by_severity = {
            Severity.CRITICAL: [],
            Severity.HIGH: [],
            Severity.MEDIUM: [],
            Severity.LOW: []
        }

        for issue in self.issues:
            by_severity[issue.severity].append(issue)

        report = "🚨 SECURITY SCAN RESULTS 🚨\n\n"
        report += f"Total Issues Found: {len(self.issues)}\n"
        report += f"Critical: {len(by_severity[Severity.CRITICAL])}\n"
        report += f"High: {len(by_severity[Severity.HIGH])}\n"
        report += f"Medium: {len(by_severity[Severity.MEDIUM])}\n"
        report += f"Low: {len(by_severity[Severity.LOW])}\n\n"

        # Detail critical and high issues
        for severity in [Severity.CRITICAL, Severity.HIGH]:
            if by_severity[severity]:
                report += f"\n{'='*60}\n"
                report += f"{severity.value.upper()} SEVERITY ISSUES\n"
                report += f"{'='*60}\n\n"

                for issue in by_severity[severity]:
                    report += f"[{issue.severity.value.upper()}] {issue.issue_type}\n"
                    report += f"File: {issue.file_path}:{issue.line_number}\n"
                    report += f"Description: {issue.description}\n"
                    report += f"Code: {issue.code_snippet}\n"
                    report += f"Fix: {issue.fix_suggestion}\n\n"

        return report

# Usage Example
scanner = AICodeSecurityScanner()

# Example: Scan AI-generated code with vulnerabilities
vulnerable_code = '''
import sqlite3

def get_user(user_id):
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    # AI-generated SQL injection vulnerability
    cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")
    return cursor.fetchone()

# Hardcoded API key (common AI mistake)
api_key = "sk-proj-1234567890abcdef"

def fetch_data():
    response = requests.get(f"https://api.example.com?key={api_key}")
    return response.json()
'''

issues = scanner.scan_file("api.py", vulnerable_code)
print(scanner.generate_report())
print(f"\n⚠️  Found {len(issues)} security issues requiring immediate attention")

from dataclasses import dataclass
from typing import List, Dict
from enum import Enum

class ReviewResult(Enum):
    APPROVED = "approved"
    NEEDS_CHANGES = "needs_changes"
    REJECTED = "rejected"

@dataclass
class CodeReviewChecklist:
    """
    Structured checklist for reviewing AI-generated code
    Based on production incidents from 1000+ teams
    """

    # Intent & Context
    what_does_this_do: str  # 1-2 sentence explanation
    why_this_approach: str  # Justify architecture choice
    ai_generated_sections: List[str]  # Which parts are AI-written

    # Correctness
    logic_verified: bool  # Does it implement the right algorithm?
    edge_cases_tested: bool  # What about None, empty, negative?
    business_rules_correct: bool  # Matches requirements?

    # Security
    input_validated: bool  # All user input checked?
    sql_injection_safe: bool  # Parameterized queries only?
    xss_prevented: bool  # Output sanitized?
    secrets_removed: bool  # No hardcoded credentials?
    auth_enforced: bool  # Endpoints protected?

    # Performance
    complexity_analyzed: bool  # O(n²) acceptable for this use case?
    database_queries_optimized: bool  # N+1 query problem?
    memory_usage_reasonable: bool  # Will it OOM on production data?

    # Testing
    unit_tests_written: bool
    integration_tests_pass: bool
    test_coverage_acceptable: bool  # >70% for critical paths

    # Observability
    logging_adequate: bool  # Can we debug production issues?
    metrics_emitted: bool  # Can we measure performance?
    error_handling_present: bool  # Graceful degradation?

    # Maintainability
    code_readable: bool  # Will someone understand this in 6 months?
    comments_explain_why: bool  # Not just "what"
    no_magic_numbers: bool  # Constants named and explained
    follows_style_guide: bool

class AICodeReviewer:
    """Framework for reviewing AI-generated code safely"""

    def __init__(self):
        self.review_history = []

    def review_pr(
        self,
        pr_title: str,
        files_changed: int,
        ai_percentage: float,  # % of code AI-generated
        checklist: CodeReviewChecklist
    ) -> ReviewResult:
        """
        Determine if PR is safe to merge
        Uses weighted scoring based on incident data
        """

        # Calculate risk score
        risk_score = 0

        # Critical security checks (blocking)
        critical_checks = [
            checklist.input_validated,
            checklist.sql_injection_safe,
            checklist.xss_prevented,
            checklist.secrets_removed,
            checklist.auth_enforced
        ]

        if not all(critical_checks):
            return ReviewResult.REJECTED

        # High-priority checks (should pass)
        high_priority = [
            checklist.logic_verified,
            checklist.edge_cases_tested,
            checklist.business_rules_correct,
            checklist.unit_tests_written,
            checklist.error_handling_present
        ]

        high_priority_pass_rate = sum(high_priority) / len(high_priority)

        if high_priority_pass_rate < 0.8:  # Less than 80% passing
            return ReviewResult.NEEDS_CHANGES

        # Medium-priority checks (nice to have)
        medium_priority = [
            checklist.complexity_analyzed,
            checklist.logging_adequate,
            checklist.code_readable,
            checklist.follows_style_guide
        ]

        medium_priority_pass_rate = sum(medium_priority) / len(medium_priority)

        # AI code requires stricter review
        if ai_percentage > 0.5:  # More than 50% AI-generated
            if medium_priority_pass_rate < 0.75:
                return ReviewResult.NEEDS_CHANGES

        # Size matters - large PRs need extra scrutiny
        if files_changed > 10 and ai_percentage > 0.3:
            if not checklist.integration_tests_pass:
                return ReviewResult.NEEDS_CHANGES

        # All checks passed
        self.review_history.append({
            'pr_title': pr_title,
            'files_changed': files_changed,
            'ai_percentage': ai_percentage,
            'result': ReviewResult.APPROVED
        })

        return ReviewResult.APPROVED

    def generate_review_comment(
        self,
        result: ReviewResult,
        checklist: CodeReviewChecklist
    ) -> str:
        """Generate helpful review feedback"""

        if result == ReviewResult.APPROVED:
            return "✅ LGTM! All safety checks passed."

        feedback = "Review Feedback:\n\n"

        # Critical issues first
        if not checklist.input_validated:
            feedback += "🔴 CRITICAL: Input validation missing\n"
        if not checklist.sql_injection_safe:
            feedback += "🔴 CRITICAL: SQL injection vulnerability\n"
        if not checklist.secrets_removed:
            feedback += "🔴 CRITICAL: Hardcoded credentials detected\n"

        # High-priority issues
        if not checklist.logic_verified:
            feedback += "⚠️  Logic verification incomplete\n"
        if not checklist.unit_tests_written:
            feedback += "⚠️  Unit tests required\n"
        if not checklist.edge_cases_tested:
            feedback += "⚠️  Edge cases not covered\n"

        feedback += "\nPlease address these issues before re-requesting review."
        return feedback

# Usage Example
checklist = CodeReviewChecklist(
    what_does_this_do="Adds user authentication endpoint with JWT tokens",
    why_this_approach="JWT allows stateless auth, scales better than sessions",
    ai_generated_sections=["JWT validation logic", "Error handling"],
    logic_verified=True,
    edge_cases_tested=True,
    business_rules_correct=True,
    input_validated=True,
    sql_injection_safe=True,
    xss_prevented=True,
    secrets_removed=True,  # ❌ Let's say this is False in reality
    auth_enforced=True,
    complexity_analyzed=True,
    database_queries_optimized=True,
    memory_usage_reasonable=True,
    unit_tests_written=True,
    integration_tests_pass=True,
    test_coverage_acceptable=True,
    logging_adequate=True,
    metrics_emitted=True,
    error_handling_present=True,
    code_readable=True,
    comments_explain_why=True,
    no_magic_numbers=True,
    follows_style_guide=True
)

# Simulate: AI wrote 60% of this PR
reviewer = AICodeReviewer()
result = reviewer.review_pr(
    pr_title="Add JWT authentication",
    files_changed=5,
    ai_percentage=0.6,
    checklist=checklist
)

print(f"Review Result: {result.value}")
print(reviewer.generate_review_comment(result, checklist))

# ❌ AI-generated code: Looks correct but has logic error
def calculate_discount(price: float, user_type: str) -> float:
    """Calculate discounted price"""
    if user_type == "premium":
        return price * 0.8  # 20% off
    elif user_type == "gold":
        return price * 0.7  # 30% off
    else:
        return price * 0.9  # 10% off

# ❓ What if price is negative? What if user_type is None?
# AI didn't consider edge cases!

# ✅ Production-ready version with validation
def calculate_discount_safe(price: float, user_type: str) -> float:
    """
    Calculate discounted price with validation

    Args:
        price: Original price (must be positive)
        user_type: Customer tier ('premium', 'gold', 'standard')

    Returns:
        Discounted price

    Raises:
        ValueError: If price is invalid or user_type unknown
    """
    # Validate inputs
    if price < 0:
        raise ValueError(f"Price must be positive, got {price}")

    if price == 0:
        return 0.0  # No discount needed on free items

    # Define discounts as constants
    DISCOUNTS = {
        'premium': 0.20,  # 20% off
        'gold': 0.30,     # 30% off
        'standard': 0.10  # 10% off
    }

    # Normalize user_type
    user_type_normalized = user_type.lower().strip() if user_type else 'standard'

    # Get discount with fallback
    discount_rate = DISCOUNTS.get(user_type_normalized, DISCOUNTS['standard'])

    # Apply discount
    discounted_price = price * (1 - discount_rate)

    # Log for monitoring
    print(f"Applied {discount_rate:.0%} discount for {user_type_normalized}: ${price:.2f} -> ${discounted_price:.2f}")

    return round(discounted_price, 2)

# Test edge cases
assert calculate_discount_safe(100, "premium") == 80.0
assert calculate_discount_safe(100, "PREMIUM") == 80.0  # Case insensitive
assert calculate_discount_safe(0, "premium") == 0.0  # Free items
assert calculate_discount_safe(100, None) == 90.0  # Default to standard
assert calculate_discount_safe(100, "invalid") == 90.0  # Unknown tier

try:
    calculate_discount_safe(-50, "premium")
    assert False, "Should have raised ValueError"
except ValueError:
    pass  # Expected

# ❌ AI-generated: O(n²) when O(n) is possible
def find_duplicates_slow(items: List[str]) -> List[str]:
    """AI might generate this inefficient version"""
    duplicates = []
    for i in range(len(items)):
        for j in range(i + 1, len(items)):
            if items[i] == items[j] and items[i] not in duplicates:
                duplicates.append(items[i])
    return duplicates

# ✅ Human-optimized: O(n) with set
def find_duplicates_fast(items: List[str]) -> List[str]:
    """Efficient version that scales to millions of items"""
    seen = set()
    duplicates = set()

    for item in items:
        if item in seen:
            duplicates.add(item)
        else:
            seen.add(item)

    return list(duplicates)

# Performance comparison
import time

large_list = ["item" + str(i % 1000) for i in range(10000)]

start = time.time()
find_duplicates_slow(large_list[:1000])  # Only 1000 items
slow_time = time.time() - start

start = time.time()
find_duplicates_fast(large_list)  # Full 10000 items
fast_time = time.time() - start

print(f"Slow (O(n²)): {slow_time:.3f}s for 1000 items")
print(f"Fast (O(n)):  {fast_time:.3f}s for 10000 items")
print(f"Fast version is {slow_time/fast_time:.0f}x faster despite 10x more data")

import subprocess
import json
from typing import Dict, List

class CICDCodeReviewPipeline:
    """
    Automated code review pipeline for CI/CD
    Combines static analysis, security scans, and AI review
    """

    def __init__(self, repo_path: str):
        self.repo_path = repo_path
        self.results = {}

    def run_static_analysis(self) -> Dict:
        """Run static code analysis (pylint, mypy, etc.)"""
        print("Running static analysis...")

        # Example: Run pylint
        try:
            result = subprocess.run(
                ['pylint', '--output-format=json', self.repo_path],
                capture_output=True,
                text=True,
                timeout=300
            )

            issues = json.loads(result.stdout) if result.stdout else []

            return {
                'tool': 'pylint',
                'issues_found': len(issues),
                'critical_issues': len([i for i in issues if i.get('type') == 'error']),
                'passed': len(issues) == 0
            }
        except Exception as e:
            return {'tool': 'pylint', 'error': str(e), 'passed': False}

    def run_security_scan(self) -> Dict:
        """Run security vulnerability scan"""
        print("Running security scan...")

        # Use our custom security scanner
        scanner = AICodeSecurityScanner()

        # In real implementation, scan all files
        # For demo, we'll simulate
        critical_vulns = 0
        high_vulns = 0

        return {
            'tool': 'security_scanner',
            'critical_vulnerabilities': critical_vulns,
            'high_vulnerabilities': high_vulns,
            'passed': critical_vulns == 0
        }

    def run_test_suite(self) -> Dict:
        """Run automated tests"""
        print("Running test suite...")

        try:
            result = subprocess.run(
                ['pytest', '--cov', '--json-report'],
                capture_output=True,
                text=True,
                timeout=600,
                cwd=self.repo_path
            )

            # Parse test results
            tests_passed = result.returncode == 0

            return {
                'tool': 'pytest',
                'passed': tests_passed,
                'coverage_threshold_met': True  # Check actual coverage
            }
        except Exception as e:
            return {'tool': 'pytest', 'error': str(e), 'passed': False}

    def check_pr_size(self, files_changed: int, lines_changed: int) -> Dict:
        """Check if PR is reviewable size"""
        # Large PRs are harder to review safely
        MAX_FILES = 15
        MAX_LINES = 500

        too_large = files_changed > MAX_FILES or lines_changed > MAX_LINES

        return {
            'check': 'pr_size',
            'files_changed': files_changed,
            'lines_changed': lines_changed,
            'passed': not too_large,
            'warning': 'PR too large for safe review' if too_large else None
        }

    def run_full_pipeline(
        self,
        files_changed: int,
        lines_changed: int
    ) -> Dict:
        """Run complete code review pipeline"""

        print("=" * 60)
        print("CI/CD CODE REVIEW PIPELINE")
        print("=" * 60)

        # 1. PR size check
        self.results['pr_size'] = self.check_pr_size(files_changed, lines_changed)

        # 2. Static analysis
        self.results['static_analysis'] = self.run_static_analysis()

        # 3. Security scan
        self.results['security'] = self.run_security_scan()

        # 4. Test suite
        self.results['tests'] = self.run_test_suite()

        # Determine overall result
        all_passed = all(
            result.get('passed', False)
            for result in self.results.values()
        )

        # Block merge if critical issues found
        blocking_issues = []

        if self.results['security']['critical_vulnerabilities'] > 0:
            blocking_issues.append("Critical security vulnerabilities detected")

        if not self.results['tests']['passed']:
            blocking_issues.append("Test suite failing")

        return {
            'pipeline_passed': all_passed and len(blocking_issues) == 0,
            'results': self.results,
            'blocking_issues': blocking_issues,
            'can_merge': len(blocking_issues) == 0
        }

# Usage in GitHub Actions / GitLab CI
pipeline = CICDCodeReviewPipeline(repo_path='/path/to/repo')
result = pipeline.run_full_pipeline(files_changed=8, lines_changed=320)

if result['can_merge']:
    print("\n✅ All checks passed - PR approved for merge")
else:
    print("\n❌ Pipeline failed - address these issues:")
    for issue in result['blocking_issues']:
        print(f"  - {issue}")
    exit(1)  # Fail CI build

import pytest
from typing import List, Optional

# ❌ AI-generated function (no tests)
def process_orders(orders: List[dict]) -> float:
    """Calculate total revenue from orders"""
    total = 0
    for order in orders:
        total += order['price'] * order['quantity']
        if order['discount']:
            total -= order['discount']
    return total

# ✅ Production-ready with comprehensive tests
def process_orders_safe(orders: List[dict]) -> float:
    """
    Calculate total revenue from orders with validation

    Args:
        orders: List of order dicts with keys: price, quantity, discount (optional)

    Returns:
        Total revenue after discounts

    Raises:
        ValueError: If order data is invalid
    """
    if not orders:
        return 0.0

    total = 0.0

    for idx, order in enumerate(orders):
        # Validate required fields
        if 'price' not in order or 'quantity' not in order:
            raise ValueError(f"Order {idx} missing required fields")

        price = order['price']
        quantity = order['quantity']
        discount = order.get('discount', 0)

        # Validate types and ranges
        if not isinstance(price, (int, float)) or price < 0:
            raise ValueError(f"Order {idx}: Invalid price {price}")

        if not isinstance(quantity, int) or quantity < 0:
            raise ValueError(f"Order {idx}: Invalid quantity {quantity}")

        if not isinstance(discount, (int, float)) or discount < 0:
            raise ValueError(f"Order {idx}: Invalid discount {discount}")

        # Calculate line total
        line_total = price * quantity - discount

        # Discount can't exceed line total
        if line_total < 0:
            raise ValueError(f"Order {idx}: Discount exceeds line total")

        total += line_total

    return round(total, 2)

# Comprehensive test suite
class TestProcessOrders:
    """Test suite covering all edge cases"""

    def test_empty_orders(self):
        """Test with no orders"""
        assert process_orders_safe([]) == 0.0

    def test_single_order_no_discount(self):
        """Test basic case"""
        orders = [{'price': 10.0, 'quantity': 2, 'discount': 0}]
        assert process_orders_safe(orders) == 20.0

    def test_multiple_orders_with_discounts(self):
        """Test multiple orders"""
        orders = [
            {'price': 100, 'quantity': 2, 'discount': 10},  # 190
            {'price': 50, 'quantity': 1, 'discount': 5},    # 45
        ]
        assert process_orders_safe(orders) == 235.0

    def test_missing_price_field(self):
        """Test error handling for missing data"""
        orders = [{'quantity': 2}]
        with pytest.raises(ValueError, match="missing required fields"):
            process_orders_safe(orders)

    def test_negative_price(self):
        """Test validation of negative price"""
        orders = [{'price': -10, 'quantity': 2}]
        with pytest.raises(ValueError, match="Invalid price"):
            process_orders_safe(orders)

    def test_negative_quantity(self):
        """Test validation of negative quantity"""
        orders = [{'price': 10, 'quantity': -2}]
        with pytest.raises(ValueError, match="Invalid quantity"):
            process_orders_safe(orders)

    def test_discount_exceeds_total(self):
        """Test that discount can't be more than line total"""
        orders = [{'price': 10, 'quantity': 1, 'discount': 20}]
        with pytest.raises(ValueError, match="Discount exceeds line total"):
            process_orders_safe(orders)

    def test_optional_discount_field(self):
        """Test that discount is optional"""
        orders = [{'price': 10, 'quantity': 2}]  # No discount key
        assert process_orders_safe(orders) == 20.0

    def test_zero_quantity(self):
        """Test edge case of zero quantity"""
        orders = [{'price': 100, 'quantity': 0}]
        assert process_orders_safe(orders) == 0.0

    def test_floating_point_precision(self):
        """Test rounding for floating point math"""
        orders = [{'price': 10.99, 'quantity': 3}]
        # 10.99 * 3 = 32.97
        assert process_orders_safe(orders) == 32.97

# Run tests with coverage
# pytest --cov=your_module --cov-report=term-missing
# Aim for >70% coverage for AI-generated code