diff --git a/tests/test_cron_cloud_context.py b/tests/test_cron_cloud_context.py new file mode 100644 index 000000000..79933996a --- /dev/null +++ b/tests/test_cron_cloud_context.py @@ -0,0 +1,91 @@ +"""Tests for cloud-context warning injection (#378, #456).""" + +import pytest +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from cron.scheduler import ( + _LOCAL_SERVICE_PATTERNS, + _detect_local_service_refs, + _inject_cloud_context, +) + + +class TestDetectLocalServiceRefs: + """Pattern detection for local service references in prompts.""" + + def test_localhost_with_port(self): + refs = _detect_local_service_refs("Check localhost:11434 is up") + assert len(refs) >= 1 + assert any("11434" in r for r in refs) + + def test_127_with_port(self): + refs = _detect_local_service_refs("curl http://127.0.0.1:8080/health") + assert len(refs) >= 1 + + def test_check_ollama(self): + refs = _detect_local_service_refs("Check Ollama is responding") + assert len(refs) >= 1 + + def test_ollama_responding(self): + refs = _detect_local_service_refs("Verify Ollama responding on this machine") + assert len(refs) >= 1 + + def test_curl_localhost(self): + refs = _detect_local_service_refs("curl localhost and report status") + assert len(refs) >= 1 + + def test_ping_localhost(self): + refs = _detect_local_service_refs("ping localhost to check connectivity") + assert len(refs) >= 1 + + def test_no_false_positive_cloud(self): + refs = _detect_local_service_refs("Check the weather in Paris today") + assert len(refs) == 0 + + def test_no_false_positive_api(self): + refs = _detect_local_service_refs("Call the OpenRouter API endpoint") + assert len(refs) == 0 + + def test_multiple_refs(self): + refs = _detect_local_service_refs("curl localhost:11434 then ping localhost") + assert len(refs) >= 2 + + +class TestInjectCloudContext: + """Cloud-context warning injection.""" + + def test_prepends_warning(self): + prompt = "Check Ollama is responding" + result = _inject_cloud_context(prompt, ["Check Ollama"], "nous") + assert result.startswith("[SYSTEM NOTE") + assert "nous" in result + assert prompt in result + + def test_preserves_original_prompt(self): + prompt = "Check Ollama at localhost:11434" + result = _inject_cloud_context(prompt, ["localhost:11434"], "openrouter") + assert prompt in result + + def test_mentions_cannot_reach(self): + prompt = "curl localhost" + result = _inject_cloud_context(prompt, ["curl localhost"], "nous") + assert "CANNOT reach" in result or "cannot reach" in result + + def test_suggests_local_provider(self): + prompt = "Check Ollama" + result = _inject_cloud_context(prompt, ["Check Ollama"], "nous") + assert "local" in result.lower() + + +class TestCloudBypassLocal: + """Local endpoints should not trigger injection.""" + + def test_local_endpoint_skips(self): + # The caller checks _is_cloud before calling _detect_local_service_refs + # so this is tested at integration level. Here we verify detection + # still finds refs (the bypass is the caller\'s responsibility). + refs = _detect_local_service_refs("Check Ollama at localhost:11434") + assert len(refs) > 0 # Detection works, caller decides whether to inject