import os from pathlib import Path from src.core.prompt_manager import get_prompt, _prompt_cache from src.core.config import settings import time def test_prompt_manager(): print("Starting Prompt Manager Verification...") local_p = "LOCAL PROMPT" prompt_name = "test_prompt" # 1. Test Local Fallback (Langfuse Disabled) settings.USE_LANGFUSE_PROMPTS = False p = get_prompt(prompt_name, local_p) assert p == local_p print("PASS: Local fallback when USE_LANGFUSE_PROMPTS=False") # 2. Test Fetching (Langfuse Enabled) - This will likely fail or return fallback if not configured settings.USE_LANGFUSE_PROMPTS = True print("\nTesting with USE_LANGFUSE_PROMPTS=True...") # We expect this to either work (if keys are valid and prompt exists) # or fall back gracefully (if keys are invalid or prompt doesn't exist) p = get_prompt(prompt_name, local_p) print(f"Result for '{prompt_name}': {p[:50]}...") # 3. Test Cache print("\nTesting TTL Cache...") # Manually inject into cache _prompt_cache[prompt_name] = ("CACHED CONTENT", time.time()) # TTL is 600 by default. This should return cached content. p = get_prompt(prompt_name, local_p) assert p == "CACHED CONTENT" print("PASS: Cache hit before TTL expiry") # Test TTL Expiry print("\nTesting Cache Expiry...") _prompt_cache[prompt_name] = ("EXPIRED CONTENT", time.time() - 1000) # Should attempt to fetch (and likely fall back to local if test_prompt doesn't exist in Langfuse) p = get_prompt(prompt_name, local_p) # If fetch fails, it returns local_p. If it hits Langfuse, it returns that. # The important part is that it didn't return "EXPIRED CONTENT". assert p != "EXPIRED CONTENT" print(f"PASS: Cache expiry handled (Returned: {p})") if __name__ == "__main__": try: test_prompt_manager() print("\nAll logical checks passed!") except Exception as e: print(f"\nVerification FAILED: {e}")