164 lines
5.4 KiB
Python
164 lines
5.4 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from javis.config.settings import Settings
|
|
from javis.core.chat_service import ChatService
|
|
from javis.core.provider_router import HybridProvider
|
|
from javis.interface.cli import build_runtime, run_chat, run_gemini_configuration
|
|
from javis.memory.sqlite_store import SQLiteSessionStore
|
|
from javis.providers.base import ChatMessage
|
|
from javis.security.privacy import CloudPolicy
|
|
|
|
|
|
class _FakeProvider:
|
|
name = "ollama"
|
|
model = "test-model"
|
|
|
|
def chat(self, messages: list[ChatMessage]) -> str:
|
|
return f"Echo: {messages[-1].content}"
|
|
|
|
|
|
class _FakeHybridProvider(_FakeProvider):
|
|
name = "hybrid"
|
|
model = "auto"
|
|
|
|
def __init__(self) -> None:
|
|
self.mode = "auto"
|
|
self.last_route = None
|
|
|
|
def set_mode(self, mode: str) -> None:
|
|
self.mode = mode
|
|
|
|
|
|
class _LazySecrets:
|
|
def __init__(self) -> None:
|
|
self.reads = 0
|
|
|
|
def get_gemini_key(self) -> str | None:
|
|
self.reads += 1
|
|
return None
|
|
|
|
def gemini_key_available(self) -> bool:
|
|
return False
|
|
|
|
|
|
class CliTests(unittest.TestCase):
|
|
def test_basic_chat_commands(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3")
|
|
service = ChatService(store, _FakeProvider())
|
|
inputs = iter(["Hallo", "/sessions", "/new", "/clear", "/exit"])
|
|
output: list[str] = []
|
|
|
|
result = run_chat(
|
|
service,
|
|
input_fn=lambda _prompt: next(inputs),
|
|
output=output.append,
|
|
)
|
|
|
|
self.assertEqual(result, 0)
|
|
self.assertTrue(any(line == "Javis: Echo: Hallo" for line in output))
|
|
self.assertTrue(any("Nachrichten" in line for line in output))
|
|
self.assertEqual(output[-1], "Chat beendet.")
|
|
|
|
def test_provider_privacy_and_status_commands_are_sanitized(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3")
|
|
provider = _FakeHybridProvider()
|
|
provider.last_route = SimpleNamespace(
|
|
privacy_policy=CloudPolicy.ASK,
|
|
provider="ollama",
|
|
cloud_suppressed_by_local_mode=True,
|
|
technical_fallback=False,
|
|
)
|
|
service = ChatService(store, provider)
|
|
inputs = iter(
|
|
[
|
|
"/provider",
|
|
"/provider local",
|
|
"/privacy",
|
|
"/status",
|
|
"/exit",
|
|
]
|
|
)
|
|
output: list[str] = []
|
|
|
|
result = run_chat(
|
|
service,
|
|
input_fn=lambda _prompt: next(inputs),
|
|
output=output.append,
|
|
status_fn=lambda: [
|
|
"Gemini-Schlüssel vorhanden: ja",
|
|
"Heutige Cloudanfragen: 0",
|
|
],
|
|
)
|
|
|
|
self.assertEqual(result, 0)
|
|
self.assertIn("Aktiver Providermodus: auto", output)
|
|
self.assertIn("Providermodus: local", output)
|
|
self.assertTrue(
|
|
any(
|
|
"Datenschutzentscheidung: ask; Provider: ollama; "
|
|
"Cloud durch Modus local unterdrückt: ja; "
|
|
"Technischer Fallback: nein" in line
|
|
for line in output
|
|
)
|
|
)
|
|
self.assertIn("Gemini-Schlüssel vorhanden: ja", output)
|
|
|
|
def test_gemini_configuration_defaults_to_no(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "javis.toml"
|
|
output: list[str] = []
|
|
|
|
result = run_gemini_configuration(
|
|
path,
|
|
input_fn=lambda _prompt: "",
|
|
output=output.append,
|
|
)
|
|
|
|
self.assertEqual(result, 1)
|
|
self.assertFalse(path.exists())
|
|
self.assertIn("deaktiviert", output[-1])
|
|
|
|
def test_gemini_configuration_requires_both_confirmations(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "javis.toml"
|
|
answers = iter(["ja", "ja"])
|
|
|
|
result = run_gemini_configuration(
|
|
path,
|
|
input_fn=lambda _prompt: next(answers),
|
|
output=lambda _line: None,
|
|
)
|
|
|
|
self.assertEqual(result, 0)
|
|
self.assertTrue(path.exists())
|
|
content = path.read_text(encoding="utf-8")
|
|
self.assertIn("billing_confirmed_disabled = true", content)
|
|
self.assertNotIn("api_key", content.lower())
|
|
|
|
def test_runtime_wires_hybrid_provider_without_loading_key_at_start(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
settings = Settings(data_dir=Path(directory))
|
|
secrets = _LazySecrets()
|
|
|
|
service, _status = build_runtime(
|
|
settings,
|
|
input_fn=lambda _prompt: "",
|
|
output=lambda _line: None,
|
|
secret_provider=secrets,
|
|
)
|
|
session = service.new_session()
|
|
|
|
self.assertIsInstance(service.provider, HybridProvider)
|
|
self.assertEqual(session.provider, "hybrid")
|
|
self.assertEqual(session.model, "auto")
|
|
self.assertEqual(secrets.reads, 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|