Coverage for jinja2_async_environment/testing/fixtures.py: 0%

1"""Test fixtures and utilities for jinja2-async-environment testing.""" 

2 

3import typing as t 

4 

5try: 

6 from ..caching import CacheManager 

7except ImportError: 

8 # Fallback during refactoring 

9 CacheManager = None 

10 

11from ..environment import AsyncEnvironment 

12 

13 

14def create_test_cache_manager( 

15 smaller_caches: bool = True, short_ttl: bool = True 

16) -> CacheManager: 

17 """Create a cache manager optimized for testing. 

18 

19 Args: 

20 smaller_caches: Use smaller cache sizes for testing 

21 short_ttl: Use shorter TTL values for testing 

22 

23 Returns: 

24 CacheManager configured for testing 

25 """ 

26 if smaller_caches: 

27 # Smaller caches for faster tests 

28 package_size = 50 

29 template_size = 100 

30 compilation_size = 200 

31 module_size = 20 

32 else: 

33 # Standard sizes 

34 package_size = 500 

35 template_size = 1000 

36 compilation_size = 2000 

37 module_size = 200 

38 

39 if short_ttl: 

40 # Short TTL for testing cache expiration 

41 default_ttl = 5 

42 else: 

43 # Standard TTL 

44 default_ttl = 300 

45 

46 return CacheManager( 

47 package_cache_size=package_size, 

48 template_cache_size=template_size, 

49 compilation_cache_size=compilation_size, 

50 module_cache_size=module_size, 

51 default_ttl=default_ttl, 

52 ) 

53 

54 

55def create_test_environment( 

56 cache_manager: CacheManager | None = None, **kwargs: t.Any 

57) -> AsyncEnvironment: 

58 """Create an AsyncEnvironment configured for testing. 

59 

60 Args: 

61 cache_manager: Cache manager to use (creates test one if None) 

62 **kwargs: Additional arguments for AsyncEnvironment 

63 

64 Returns: 

65 AsyncEnvironment configured for testing 

66 """ 

67 if cache_manager is None: 

68 cache_manager = create_test_cache_manager() 

69 

70 # Set testing-friendly defaults 

71 test_defaults = { 

72 "cache_size": 100, # Smaller cache for tests 

73 "auto_reload": True, # Enable auto-reload for test changes 

74 } 

75 

76 # Merge with provided kwargs 

77 test_defaults.update(kwargs) 

78 

79 # Add cache manager if the environment supports it 

80 # Note: This will be implemented in Phase 2 

81 return AsyncEnvironment(**test_defaults) 

82 

83 

84def clear_all_test_caches() -> None: 

85 """Clear all caches that might be used during testing. 

86 

87 This is useful for ensuring test isolation. 

88 """ 

89 # Clear default cache manager 

90 default_manager = CacheManager.get_default() 

91 default_manager.clear_all() 

92 

93 # Clear any global unified cache if it exists 

94 try: 

95 from ..loaders import _unified_cache 

96 

97 _unified_cache.clear_all() 

98 except (ImportError, AttributeError): 

99 pass 

100 

101 

102def get_cache_statistics() -> dict[str, t.Any]: 

103 """Get cache statistics for test verification. 

104 

105 Returns: 

106 Dictionary with cache statistics 

107 """ 

108 try: 

109 default_manager = CacheManager.get_default() 

110 return default_manager.get_statistics() 

111 except Exception: 

112 return {"error": "Could not retrieve cache statistics"} 

113 

114 

115class TestCacheManager(CacheManager): 

116 """Cache manager with additional testing utilities.""" 

117 

118 def __init__(self, *args: t.Any, **kwargs: t.Any): 

119 """Initialize test cache manager with test-friendly defaults.""" 

120 # Override defaults for testing 

121 test_kwargs = { 

122 "package_cache_size": 50, 

123 "template_cache_size": 100, 

124 "compilation_cache_size": 200, 

125 "module_cache_size": 20, 

126 "default_ttl": 5, 

127 } 

128 test_kwargs.update(kwargs) 

129 super().__init__(*args, **test_kwargs) 

130 

131 # Track operations for testing 

132 self._operation_log: list[str] = [] 

133 

134 def get(self, cache_type: str, key: str) -> t.Any: 

135 """Get with operation logging.""" 

136 self._operation_log.append(f"get:{cache_type}:{key}") 

137 return super().get(cache_type, key) 

138 

139 def set( 

140 self, cache_type: str, key: str, value: t.Any, ttl: int | None = None 

141 ) -> None: 

142 """Set with operation logging.""" 

143 self._operation_log.append(f"set:{cache_type}:{key}") 

144 super().set(cache_type, key, value, ttl) 

145 

146 def get_operation_log(self) -> list[str]: 

147 """Get log of cache operations for test verification.""" 

148 return list(self._operation_log) 

149 

150 def clear_operation_log(self) -> None: 

151 """Clear the operation log.""" 

152 self._operation_log.clear()