markitect-main/tests/utils/assertions.py

"""
Custom assertions and test utilities for MarkiTect tests.
"""

import json
import asyncio
from typing import Any, Dict, List, Optional, Union, Callable, Awaitable
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import AsyncMock
import pytest


def assert_issue_equal(actual, expected, ignore_timestamps: bool = False):
    """Assert that two Issue objects are equal."""
    assert actual.number == expected.number, f"Issue numbers don't match: {actual.number} != {expected.number}"
    assert actual.title == expected.title, f"Issue titles don't match: {actual.title} != {expected.title}"
    assert actual.state == expected.state, f"Issue states don't match: {actual.state} != {expected.state}"
    assert len(actual.labels) == len(expected.labels), f"Label counts don't match: {len(actual.labels)} != {len(expected.labels)}"

    # Compare labels
    actual_label_names = {label.name for label in actual.labels}
    expected_label_names = {label.name for label in expected.labels}
    assert actual_label_names == expected_label_names, f"Labels don't match: {actual_label_names} != {expected_label_names}"

    if not ignore_timestamps:
        assert actual.created_at == expected.created_at, f"Created timestamps don't match"
        assert actual.updated_at == expected.updated_at, f"Updated timestamps don't match"
        assert actual.closed_at == expected.closed_at, f"Closed timestamps don't match"


def assert_project_equal(actual, expected, ignore_timestamps: bool = False):
    """Assert that two Project objects are equal."""
    assert actual.name == expected.name, f"Project names don't match: {actual.name} != {expected.name}"
    assert actual.description == expected.description, f"Project descriptions don't match"
    assert actual.state == expected.state, f"Project states don't match: {actual.state} != {expected.state}"
    assert actual.kanban_columns == expected.kanban_columns, f"Kanban columns don't match"
    assert len(actual.milestones) == len(expected.milestones), f"Milestone counts don't match"

    if not ignore_timestamps:
        assert actual.created_at == expected.created_at, f"Created timestamps don't match"
        assert actual.updated_at == expected.updated_at, f"Updated timestamps don't match"
        assert actual.archived_at == expected.archived_at, f"Archived timestamps don't match"


def assert_milestone_equal(actual, expected):
    """Assert that two Milestone objects are equal."""
    assert actual.id == expected.id, f"Milestone IDs don't match: {actual.id} != {expected.id}"
    assert actual.title == expected.title, f"Milestone titles don't match: {actual.title} != {expected.title}"
    assert actual.description == expected.description, f"Milestone descriptions don't match"
    assert actual.state == expected.state, f"Milestone states don't match: {actual.state} != {expected.state}"
    assert actual.open_issues == expected.open_issues, f"Open issue counts don't match"
    assert actual.closed_issues == expected.closed_issues, f"Closed issue counts don't match"
    assert actual.due_date == expected.due_date, f"Due dates don't match"


def assert_json_equal(actual: Union[str, Dict], expected: Union[str, Dict]):
    """Assert that two JSON objects are equal."""
    if isinstance(actual, str):
        actual = json.loads(actual)
    if isinstance(expected, str):
        expected = json.loads(expected)

    assert actual == expected, f"JSON objects don't match:\nActual: {json.dumps(actual, indent=2)}\nExpected: {json.dumps(expected, indent=2)}"


def assert_markdown_structure_equal(actual: str, expected: str):
    """Assert that two markdown documents have the same structure (ignoring whitespace differences)."""
    actual_lines = [line.strip() for line in actual.split('\n') if line.strip()]
    expected_lines = [line.strip() for line in expected.split('\n') if line.strip()]

    assert len(actual_lines) == len(expected_lines), f"Line count mismatch: {len(actual_lines)} != {len(expected_lines)}"

    for i, (actual_line, expected_line) in enumerate(zip(actual_lines, expected_lines)):
        assert actual_line == expected_line, f"Line {i+1} mismatch:\nActual: {actual_line}\nExpected: {expected_line}"


def assert_file_exists(file_path: Union[str, Path], message: str = None):
    """Assert that a file exists."""
    path = Path(file_path)
    assert path.exists(), message or f"File does not exist: {path}"
    assert path.is_file(), message or f"Path is not a file: {path}"


def assert_directory_exists(dir_path: Union[str, Path], message: str = None):
    """Assert that a directory exists."""
    path = Path(dir_path)
    assert path.exists(), message or f"Directory does not exist: {path}"
    assert path.is_dir(), message or f"Path is not a directory: {path}"


def assert_file_contains(file_path: Union[str, Path], content: str, message: str = None):
    """Assert that a file contains specific content."""
    path = Path(file_path)
    assert_file_exists(path)

    file_content = path.read_text()
    assert content in file_content, message or f"File {path} does not contain: {content}"


def assert_file_not_contains(file_path: Union[str, Path], content: str, message: str = None):
    """Assert that a file does not contain specific content."""
    path = Path(file_path)
    assert_file_exists(path)

    file_content = path.read_text()
    assert content not in file_content, message or f"File {path} unexpectedly contains: {content}"


def assert_time_approximately_equal(actual: datetime, expected: datetime, tolerance_seconds: int = 1):
    """Assert that two datetime objects are approximately equal within tolerance."""
    diff = abs((actual - expected).total_seconds())
    assert diff <= tolerance_seconds, f"Times differ by {diff} seconds, tolerance is {tolerance_seconds}"


def assert_performance_within_bounds(execution_time: float, max_time: float, operation: str = "operation"):
    """Assert that an operation completed within performance bounds."""
    assert execution_time <= max_time, f"{operation} took {execution_time:.3f}s, expected <= {max_time:.3f}s"


def assert_memory_usage_within_bounds(memory_usage_mb: float, max_memory_mb: float, operation: str = "operation"):
    """Assert that memory usage is within bounds."""
    assert memory_usage_mb <= max_memory_mb, f"{operation} used {memory_usage_mb:.2f}MB, expected <= {max_memory_mb:.2f}MB"


def assert_mock_called_with_pattern(mock, pattern: Callable[[Any], bool], message: str = None):
    """Assert that a mock was called with arguments matching a pattern."""
    found_match = False
    for call in mock.call_args_list:
        if pattern(call):
            found_match = True
            break

    assert found_match, message or f"Mock was not called with expected pattern. Calls: {mock.call_args_list}"


def assert_sequence_equal(actual: List[Any], expected: List[Any], compare_fn: Optional[Callable[[Any, Any], bool]] = None):
    """Assert that two sequences are equal using optional custom comparison."""
    assert len(actual) == len(expected), f"Sequence lengths don't match: {len(actual)} != {len(expected)}"

    for i, (actual_item, expected_item) in enumerate(zip(actual, expected)):
        if compare_fn:
            assert compare_fn(actual_item, expected_item), f"Items at index {i} don't match"
        else:
            assert actual_item == expected_item, f"Items at index {i} don't match: {actual_item} != {expected_item}"


def assert_contains_all(container: Union[List, Dict, str], items: List[Any], message: str = None):
    """Assert that a container contains all specified items."""
    missing_items = []
    for item in items:
        if item not in container:
            missing_items.append(item)

    assert not missing_items, message or f"Container missing items: {missing_items}"


def assert_contains_none(container: Union[List, Dict, str], items: List[Any], message: str = None):
    """Assert that a container contains none of the specified items."""
    found_items = []
    for item in items:
        if item in container:
            found_items.append(item)

    assert not found_items, message or f"Container unexpectedly contains items: {found_items}"


def assert_label_categories_valid(categories):
    """Assert that label categories are valid and properly separated."""
    from domain.issues.models import LabelCategories

    assert isinstance(categories, LabelCategories), "Categories must be LabelCategories instance"

    # Check for overlaps between categories
    all_labels = (
        categories.state_labels +
        categories.priority_labels +
        categories.type_labels +
        categories.other_labels
    )

    # No label should appear in multiple categories
    seen_labels = set()
    for label in all_labels:
        assert label not in seen_labels, f"Label '{label}' appears in multiple categories"
        seen_labels.add(label)


def assert_kanban_column_valid(column: str, valid_columns: List[str]):
    """Assert that a kanban column is valid."""
    assert column in valid_columns, f"Invalid kanban column '{column}'. Valid columns: {valid_columns}"


def assert_business_rule_violated(exception_type: type, exception_message_pattern: str = None):
    """Context manager to assert that a business rule violation occurs."""
    return pytest.raises(exception_type, match=exception_message_pattern)


def assert_async_operation_succeeds(async_func: Callable, timeout: float = 30.0):
    """Assert that an async operation succeeds within timeout."""
    import asyncio

    async def run_with_timeout():
        return await asyncio.wait_for(async_func(), timeout=timeout)

    try:
        result = asyncio.run(run_with_timeout())
        return result
    except asyncio.TimeoutError:
        pytest.fail(f"Async operation timed out after {timeout} seconds")
    except Exception as e:
        pytest.fail(f"Async operation failed: {e}")


# Custom pytest markers for different types of assertions
def mark_performance_test(max_time: float = None, max_memory_mb: float = None):
    """Mark a test as a performance test with optional bounds."""
    markers = [pytest.mark.performance]
    if max_time:
        markers.append(pytest.mark.parametrize("max_execution_time", [max_time]))
    if max_memory_mb:
        markers.append(pytest.mark.parametrize("max_memory_usage", [max_memory_mb]))
    return markers


def mark_integration_test(external_service: str = None):
    """Mark a test as an integration test."""
    markers = [pytest.mark.integration]
    if external_service:
        markers.append(pytest.mark.parametrize("external_service", [external_service]))
    return markers


# Async testing utilities
def create_async_mock_that_returns(return_value: Any) -> AsyncMock:
    """Create an AsyncMock that returns a specific value."""
    mock = AsyncMock()
    mock.return_value = return_value
    return mock


def create_async_mock_that_raises(exception: Exception) -> AsyncMock:
    """Create an AsyncMock that raises a specific exception."""
    mock = AsyncMock()
    mock.side_effect = exception
    return mock


def await_safely(coro: Awaitable[Any]) -> Any:
    """Safely await a coroutine in a test context.

    This function properly handles the async context and ensures
    no coroutine warnings are generated.
    """
    try:
        loop = asyncio.get_event_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)

    try:
        return loop.run_until_complete(coro)
    finally:
        # Clean up any pending tasks to avoid warnings
        pending = asyncio.all_tasks(loop)
        if pending:
            loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))


def cleanup_async_mocks(*mocks):
    """Cleanup async mocks to prevent coroutine warnings.

    This should be called in test teardown to properly cleanup
    any async mocks that might have created unawaited coroutines.
    """
    for mock in mocks:
        if hasattr(mock, '_mock_children'):
            for child in mock._mock_children.values():
                if asyncio.iscoroutine(child):
                    child.close()
        if hasattr(mock, 'return_value') and asyncio.iscoroutine(mock.return_value):
            mock.return_value.close()
        # Clean up any pending coroutines from the mock itself
        if hasattr(mock, '_mock_call_signature'):
            # AsyncMock can create coroutines internally, we need to close them
            if hasattr(mock, '_mock_return_value') and asyncio.iscoroutine(mock._mock_return_value):
                mock._mock_return_value.close()
        # Close any coroutine that might be stored as the mock object itself
        if asyncio.iscoroutine(mock):
            mock.close()


class AsyncTestCase:
    """Base class for async test cases with proper cleanup."""

    def setup_method(self):
        """Setup for each test method."""
        self.async_mocks = []
        self.tracked_objects = []  # Track objects that may contain async mocks

    def teardown_method(self):
        """Cleanup after each test method."""
        cleanup_async_mocks(*self.async_mocks)
        # Clean up any async mocks in tracked objects
        for obj in self.tracked_objects:
            if hasattr(obj, '__dict__'):
                for attr_name, attr_value in obj.__dict__.items():
                    if hasattr(attr_value, '_mock_children') or asyncio.iscoroutinefunction(attr_value):
                        cleanup_async_mocks(attr_value)
        self.async_mocks.clear()
        self.tracked_objects.clear()

    def create_async_mock(self, return_value: Any = None, side_effect: Any = None) -> AsyncMock:
        """Create an async mock and track it for cleanup."""
        mock = AsyncMock()
        if return_value is not None:
            mock.return_value = return_value
        if side_effect is not None:
            mock.side_effect = side_effect
        self.async_mocks.append(mock)
        return mock

    def track_for_cleanup(self, obj: Any) -> Any:
        """Track an object that may contain async mocks for cleanup."""
        self.tracked_objects.append(obj)
        return obj


# Test data validation helpers
def validate_issue_data(data: Dict[str, Any]) -> bool:
    """Validate that data represents a valid issue."""
    required_fields = ["number", "title", "state", "labels", "created_at", "updated_at"]
    for field in required_fields:
        if field not in data:
            return False

    if not isinstance(data["number"], int) or data["number"] <= 0:
        return False

    if not isinstance(data["title"], str) or not data["title"].strip():
        return False

    if data["state"] not in ["open", "closed"]:
        return False

    if not isinstance(data["labels"], list):
        return False

    return True


def validate_project_data(data: Dict[str, Any]) -> bool:
    """Validate that data represents a valid project."""
    required_fields = ["name", "state", "milestones", "kanban_columns", "created_at", "updated_at"]
    for field in required_fields:
        if field not in data:
            return False

    if not isinstance(data["name"], str) or not data["name"].strip():
        return False

    if data["state"] not in ["active", "archived"]:
        return False

    if not isinstance(data["milestones"], list):
        return False

    if not isinstance(data["kanban_columns"], list):
        return False

    return True