tegwick
1d86bf1bbd
Comprehensive fix for test suite warnings across multiple issue test files: ### SQLite3 Date Adapter Warnings (Python 3.12) - Fixed 101 warnings in Issue 113 (activity_tracker.py) - Fixed 55 warnings in Issue 114 (allocation_engine.py) - Fixed 148 warnings in Issue 122 (worktime_tracker.py + test file) - Fixed 18 warnings in Issue 124 (day_wrapup_commands.py + worktime_tracker.py) ### Pytest-asyncio Configuration - Added asyncio_default_fixture_loop_scope = function to pytest.ini - Eliminates pytest-asyncio deprecation warning ### Runtime Warnings for Unawaited Coroutines - Fixed 2 warnings in Issue 59 (gitea plugin async mocking) - Enhanced AsyncTestCase with better coroutine cleanup - Improved async mock management in test utilities ### Technical Changes - Convert Python date/datetime objects to ISO strings before SQLite queries - Use .isoformat() with defensive hasattr() checks for backward compatibility - Simplified async test mocking to avoid coroutine creation - Enhanced cleanup_async_mocks() function for comprehensive cleanup ### Results - Before: ~324 warnings across test suite - After: 0 warnings - completely clean test suite - All 216+ tests pass with zero warning noise 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
371 lines
15 KiB
Python
371 lines
15 KiB
Python
"""
|
|
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 |