568a735f65
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m11s
When a cron job runs on a cloud endpoint but its prompt references local services (Ollama, localhost, etc.), inject a [SYSTEM NOTE] warning so the agent reports the limitation instead of wasting iterations on doomed connections. Detection: 12 regex patterns for localhost, 127.x, ollama, curl/wget localhost, http://localhost, local model references. Also adds missing ModelContextError and CRON_MIN_CONTEXT_TOKENS. 14 tests added (all passing). Closes #500 (Fixes #378, Closes #456)
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""Tests for cron cloud-context warning injection (#468)."""
|
|
|
|
import pytest
|
|
|
|
from cron.scheduler import (
|
|
_LOCAL_SERVICE_PATTERNS,
|
|
_detect_local_service_refs,
|
|
_inject_cloud_context,
|
|
_CLOUD_CONTEXT_WARNING,
|
|
)
|
|
|
|
|
|
class TestDetectLocalServiceRefs:
|
|
"""Test local service reference detection."""
|
|
|
|
def test_detects_localhost_with_port(self):
|
|
refs = _detect_local_service_refs("Connect to localhost:11434")
|
|
assert len(refs) > 0
|
|
|
|
def test_detects_127_address(self):
|
|
refs = _detect_local_service_refs("Check http://127.0.0.1:8080/health")
|
|
assert len(refs) > 0
|
|
|
|
def test_detects_ollama(self):
|
|
refs = _detect_local_service_refs("Run ollama pull gemma4")
|
|
assert len(refs) > 0
|
|
|
|
def test_detects_curl_localhost(self):
|
|
refs = _detect_local_service_refs("curl localhost:11434/api/tags")
|
|
assert len(refs) > 0
|
|
|
|
def test_detects_wget_localhost(self):
|
|
refs = _detect_local_service_refs("wget localhost:8080/data")
|
|
assert len(refs) > 0
|
|
|
|
def test_detects_http_localhost(self):
|
|
refs = _detect_local_service_refs("http://localhost:3000")
|
|
assert len(refs) > 0
|
|
|
|
def test_detects_local_model(self):
|
|
refs = _detect_local_service_refs("Use the local model for inference")
|
|
assert len(refs) > 0
|
|
|
|
def test_no_refs_returns_empty(self):
|
|
refs = _detect_local_service_refs("Search the web for Python tutorials")
|
|
assert len(refs) == 0
|
|
|
|
def test_case_insensitive(self):
|
|
refs = _detect_local_service_refs("OLLAMA is running on LocalHost:11434")
|
|
assert len(refs) > 0
|
|
|
|
|
|
class TestInjectCloudContext:
|
|
"""Test cloud context warning injection."""
|
|
|
|
def test_no_warning_on_local_endpoint(self):
|
|
prompt = "Check ollama on localhost:11434"
|
|
result = _inject_cloud_context(prompt, "http://localhost:11434/v1")
|
|
assert result == prompt # No injection for local endpoints
|
|
|
|
def test_no_warning_when_no_local_refs(self):
|
|
prompt = "Search the web for news"
|
|
result = _inject_cloud_context(prompt, "https://api.openai.com/v1")
|
|
assert result == prompt
|
|
|
|
def test_injects_warning_on_cloud_with_local_refs(self):
|
|
prompt = "Check ollama status on localhost:11434"
|
|
result = _inject_cloud_context(prompt, "https://api.openai.com/v1")
|
|
assert _CLOUD_CONTEXT_WARNING in result
|
|
assert prompt in result
|
|
assert result.startswith(_CLOUD_CONTEXT_WARNING)
|
|
|
|
def test_nous_cloud_injects_warning(self):
|
|
prompt = "curl localhost:11434/api/tags"
|
|
result = _inject_cloud_context(prompt, "https://inference-api.nousresearch.com/v1")
|
|
assert _CLOUD_CONTEXT_WARNING in result
|
|
|
|
def test_warning_content(self):
|
|
prompt = "local model check"
|
|
result = _inject_cloud_context(prompt, "https://api.example.com/v1")
|
|
assert "CLOUD" in result
|
|
assert "NOT accessible" in result
|
|
assert "localhost" in result
|