Alexander Whitestone
7ac8d0268f
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m13s
92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
"""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
|