Method test_concurrent_database_access

tests/test_cache.py:431–448

Test concurrent database access.

(self, cache_manager, sample_kernel)

            # This test ensures no unhandled exception

    def test_concurrent_database_access(self, cache_manager, sample_kernel):
        """Test concurrent database access."""
        import threading
        
        def cache_kernel(model_num):
            cache_manager.cache_kernel(f"model{model_num}", "op", sample_kernel)
        
        threads = []
        for i in range(5):
            t = threading.Thread(target=cache_kernel, args=(i,))
            threads.append(t)
            t.start()
        
        for t in threads:
            t.join()
        
        # All kernels should be cached
        assert cache_manager.count_cached_kernels() == 5