294 lines
10 KiB
Python
294 lines
10 KiB
Python
import sqlite3
|
|
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}"
|
|
|
|
def stream_chat(self, messages: list[ChatMessage]):
|
|
yield "Echo: "
|
|
yield messages[-1].content
|
|
|
|
|
|
class _MarkdownProvider(_FakeProvider):
|
|
def stream_chat(self, messages: list[ChatMessage]):
|
|
yield "## **Ant"
|
|
yield "wort**\n`SQLite`"
|
|
|
|
|
|
class _InterruptingProvider(_FakeProvider):
|
|
def stream_chat(self, messages: list[ChatMessage]):
|
|
yield "angefangene Antwort"
|
|
raise KeyboardInterrupt
|
|
|
|
|
|
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.assertIn("Javis: Echo: Hallo\n", "".join(output))
|
|
self.assertTrue(any("Nachrichten" in line for line in output))
|
|
self.assertEqual(output[-1], "Chat beendet.")
|
|
|
|
def test_streaming_cleans_terminal_markdown_but_persists_raw_answer(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3")
|
|
service = ChatService(store, _MarkdownProvider())
|
|
inputs = iter(["Frage", "/exit"])
|
|
output: list[str] = []
|
|
streamed: list[str] = []
|
|
|
|
result = run_chat(
|
|
service,
|
|
input_fn=lambda _prompt: next(inputs),
|
|
output=output.append,
|
|
stream_output=streamed.append,
|
|
)
|
|
session = service.list_sessions()[0]
|
|
messages = service.load_session(session.id).messages
|
|
|
|
self.assertEqual(result, 0)
|
|
self.assertEqual("".join(streamed), "Javis: Antwort\nSQLite\n")
|
|
self.assertEqual(messages[-1].content, "## **Antwort**\n`SQLite`")
|
|
|
|
def test_keyboard_interrupt_drops_partial_answer_and_returns_to_prompt(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3")
|
|
service = ChatService(store, _InterruptingProvider())
|
|
inputs = iter(["Frage", "/exit"])
|
|
output: list[str] = []
|
|
streamed: list[str] = []
|
|
|
|
result = run_chat(
|
|
service,
|
|
input_fn=lambda _prompt: next(inputs),
|
|
output=output.append,
|
|
stream_output=streamed.append,
|
|
)
|
|
session = service.list_sessions()[0]
|
|
messages = service.load_session(session.id).messages
|
|
|
|
self.assertEqual(result, 0)
|
|
self.assertEqual(messages, [])
|
|
self.assertTrue(any("nicht gespeichert" 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_numbered_sessions_load_rename_and_search(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3")
|
|
service = ChatService(store, _FakeProvider())
|
|
first = service.new_session()
|
|
service.send(first.id, "Erste Frage")
|
|
second = service.new_session()
|
|
service.send(second.id, "Zweite Frage")
|
|
shown_before = service.list_sessions()
|
|
target = shown_before[1]
|
|
inputs = iter(
|
|
[
|
|
"/sessions",
|
|
"/load 2",
|
|
"/rename Gefundene Sitzung",
|
|
"/search Gefundene",
|
|
"/exit",
|
|
]
|
|
)
|
|
output: list[str] = []
|
|
|
|
result = run_chat(
|
|
service,
|
|
session_id=first.id,
|
|
input_fn=lambda _prompt: next(inputs),
|
|
output=output.append,
|
|
)
|
|
renamed_title = store.get_session(target.id).title
|
|
|
|
self.assertEqual(result, 0)
|
|
self.assertTrue(any(line.startswith("1. ") and "Frage" in line for line in output))
|
|
self.assertIn(f"Sitzung geladen: {target.id}", output)
|
|
self.assertEqual(renamed_title, "Gefundene Sitzung")
|
|
self.assertTrue(any("Gefundene Sitzung" in line for line in output))
|
|
|
|
def test_load_accepts_unique_prefix_and_rejects_ambiguous_prefix(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
database = Path(directory) / "sessions.sqlite3"
|
|
store = SQLiteSessionStore(database)
|
|
service = ChatService(store, _FakeProvider())
|
|
active = service.new_session()
|
|
with sqlite3.connect(database) as connection:
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO sessions
|
|
(id, created_at, updated_at, provider, model, title, last_provider)
|
|
VALUES (?, '2026-01-01', '2026-01-01',
|
|
'ollama', 'test-model', ?, 'ollama')
|
|
""",
|
|
(
|
|
("shared-111", "Eins"),
|
|
("shared-222", "Zwei"),
|
|
("unique-333", "Drei"),
|
|
),
|
|
)
|
|
connection.close()
|
|
inputs = iter(["/load unique", "/load shared", "/exit"])
|
|
output: list[str] = []
|
|
|
|
result = run_chat(
|
|
service,
|
|
session_id=active.id,
|
|
input_fn=lambda _prompt: next(inputs),
|
|
output=output.append,
|
|
)
|
|
|
|
self.assertEqual(result, 0)
|
|
self.assertIn("Sitzung geladen: unique-333", output)
|
|
self.assertTrue(any("nicht eindeutig" in line for line in 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()
|