283 lines
10 KiB
Python
283 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock, patch
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
from javis.core.provider_router import ApprovalChoice
|
|
from javis.providers.base import ChatMessage
|
|
from javis.security.privacy import CloudPolicy, PrivacyDecision
|
|
from javis.ui.main_window import MainWindow
|
|
from javis.ui.models import DesktopStatus, GenerationState, StreamEvent, StreamEventKind
|
|
from javis.ui.workers import ApprovalBridge, StreamWorker
|
|
|
|
|
|
class _Controller:
|
|
def __init__(self) -> None:
|
|
self.active_session_id = None
|
|
self.state = GenerationState.READY
|
|
self.session = SimpleNamespace(
|
|
id="session-one",
|
|
title="Desktop-Test",
|
|
last_provider="ollama",
|
|
message_count=2,
|
|
updated_at="2026-07-30T20:00:00+00:00",
|
|
)
|
|
self.messages = [
|
|
ChatMessage("user", "Hallo"),
|
|
ChatMessage("assistant", "Guten Tag"),
|
|
]
|
|
self.approval: PrivacyDecision | None = None
|
|
self.queued_approval: ApprovalChoice | None = None
|
|
self.started = threading.Event()
|
|
self.release = threading.Event()
|
|
self.release.set()
|
|
|
|
def list_sessions(self, search: str = ""):
|
|
return [self.session] if not search or search in self.session.title else []
|
|
|
|
def load_session(self, session_id: str):
|
|
self.active_session_id = session_id
|
|
return SimpleNamespace(
|
|
session=self.session,
|
|
messages=list(self.messages),
|
|
)
|
|
|
|
def new_session(self):
|
|
self.active_session_id = self.session.id
|
|
return self.session
|
|
|
|
def rename_active_session(self, title: str):
|
|
self.session.title = title
|
|
return self.session
|
|
|
|
def set_mode(self, _mode: str) -> None:
|
|
return
|
|
|
|
def approval_decision(self, _text: str) -> PrivacyDecision | None:
|
|
return self.approval
|
|
|
|
def queue_approval(self, choice: ApprovalChoice) -> None:
|
|
self.queued_approval = choice
|
|
|
|
def stream_message(self, text: str, cancel_event: threading.Event):
|
|
self.started.set()
|
|
self.state = GenerationState.THINKING
|
|
yield StreamEvent(StreamEventKind.STATE, state=self.state)
|
|
while not self.release.wait(0.005):
|
|
if cancel_event.is_set():
|
|
self.state = GenerationState.ABORTED
|
|
yield StreamEvent(StreamEventKind.STATE, state=self.state)
|
|
return
|
|
if cancel_event.is_set():
|
|
self.state = GenerationState.ABORTED
|
|
yield StreamEvent(StreamEventKind.STATE, state=self.state)
|
|
return
|
|
self.state = GenerationState.ANSWERING
|
|
yield StreamEvent(StreamEventKind.STATE, state=self.state)
|
|
yield StreamEvent(StreamEventKind.CHUNK, text="Testantwort")
|
|
self.messages.extend(
|
|
[
|
|
ChatMessage("user", text),
|
|
ChatMessage("assistant", "Testantwort"),
|
|
]
|
|
)
|
|
self.session.message_count = len(self.messages)
|
|
self.state = GenerationState.READY
|
|
yield StreamEvent(StreamEventKind.FINISHED, state=self.state)
|
|
|
|
def status(self) -> DesktopStatus:
|
|
return DesktopStatus(
|
|
state=self.state,
|
|
mode="Lokal",
|
|
provider="ollama",
|
|
ollama_reachable=True,
|
|
local_model="test-model",
|
|
privacy_policy="never",
|
|
cloud_requests_today=0,
|
|
local_fallback=False,
|
|
data_dir="D:\\Testdaten",
|
|
)
|
|
|
|
|
|
class _MessageBox:
|
|
Question = object()
|
|
AcceptRole = object()
|
|
RejectRole = object()
|
|
DestructiveRole = object()
|
|
selected_label = ""
|
|
|
|
def __init__(self, _parent) -> None:
|
|
self.buttons: dict[str, object] = {}
|
|
|
|
def setWindowTitle(self, _text: str) -> None:
|
|
return
|
|
|
|
def setIcon(self, _icon: object) -> None:
|
|
return
|
|
|
|
def setText(self, _text: str) -> None:
|
|
return
|
|
|
|
def setInformativeText(self, _text: str) -> None:
|
|
return
|
|
|
|
def addButton(self, label: str, _role: object) -> object:
|
|
button = object()
|
|
self.buttons[label] = button
|
|
return button
|
|
|
|
def exec(self) -> None:
|
|
return
|
|
|
|
def clickedButton(self) -> object | None:
|
|
return self.buttons.get(type(self).selected_label)
|
|
|
|
|
|
class DesktopGuiTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.application = QApplication.instance() or QApplication([])
|
|
|
|
def setUp(self) -> None:
|
|
self.bridge = ApprovalBridge()
|
|
self.controller = _Controller()
|
|
self.window = MainWindow(self.controller, self.bridge)
|
|
self.window.show()
|
|
self.application.processEvents()
|
|
|
|
def tearDown(self) -> None:
|
|
self.controller.release.set()
|
|
self.window._stop_generation()
|
|
self._process_until(lambda: self.window._thread is None)
|
|
self.window.close()
|
|
self.application.processEvents()
|
|
|
|
def _process_until(self, condition, timeout: float = 2.0) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
while not condition() and time.monotonic() < deadline:
|
|
self.application.processEvents()
|
|
time.sleep(0.005)
|
|
self.application.processEvents()
|
|
self.assertTrue(condition())
|
|
|
|
def test_offscreen_window_loads_session_and_core_widgets(self) -> None:
|
|
self.assertEqual(self.window.windowTitle(), "Javis Desktop")
|
|
self.assertEqual(self.window.session_list.count(), 1)
|
|
self.assertIn("Hallo", self.window.chat_view.toPlainText())
|
|
self.assertTrue(self.window.send_button.isEnabled())
|
|
self.assertFalse(self.window.stop_button.isEnabled())
|
|
self.assertEqual(self.window.status_labels["provider"].text(), "ollama")
|
|
|
|
def test_approval_dialog_maps_all_three_choices(self) -> None:
|
|
decision = PrivacyDecision(CloudPolicy.ASK, "persönliche Information")
|
|
self.bridge.resolve = Mock()
|
|
expected = {
|
|
"Einmal erlauben": ApprovalChoice.ALLOW,
|
|
"Lokal beantworten": ApprovalChoice.LOCAL,
|
|
"Abbrechen": ApprovalChoice.CANCEL,
|
|
}
|
|
|
|
with patch("javis.ui.main_window.QMessageBox", _MessageBox):
|
|
for label, choice in expected.items():
|
|
with self.subTest(label=label):
|
|
_MessageBox.selected_label = label
|
|
self.assertIs(self.window._show_approval(decision), choice)
|
|
|
|
def test_local_completion_and_second_question_keep_gui_reusable(self) -> None:
|
|
for text in ("Erste Frage", "Zweite Frage"):
|
|
self.window.message_edit.setPlainText(text)
|
|
self.window._send()
|
|
self._process_until(lambda: self.window._thread is None)
|
|
|
|
self.assertEqual(self.controller.session.message_count, 6)
|
|
self.assertIn("Zweite Frage", self.window.chat_view.toPlainText())
|
|
self.assertTrue(self.window.send_button.isEnabled())
|
|
|
|
def test_approval_is_decided_before_worker_starts(self) -> None:
|
|
self.controller.approval = PrivacyDecision(CloudPolicy.ASK, "privat")
|
|
self.window._show_approval = Mock(return_value=ApprovalChoice.ALLOW)
|
|
|
|
self.window.message_edit.setPlainText("Private Testfrage")
|
|
self.window._send()
|
|
|
|
self.assertIs(self.controller.queued_approval, ApprovalChoice.ALLOW)
|
|
self.window._show_approval.assert_called_once()
|
|
self._process_until(lambda: self.window._thread is None)
|
|
|
|
def test_local_approval_choice_is_queued_before_worker_starts(self) -> None:
|
|
self.controller.approval = PrivacyDecision(CloudPolicy.ASK, "privat")
|
|
self.window._show_approval = Mock(return_value=ApprovalChoice.LOCAL)
|
|
|
|
self.window.message_edit.setPlainText("Lokal beantworten")
|
|
self.window._send()
|
|
|
|
self.assertIs(self.controller.queued_approval, ApprovalChoice.LOCAL)
|
|
self._process_until(lambda: self.window._thread is None)
|
|
|
|
def test_stop_during_streaming_discards_partial_exchange(self) -> None:
|
|
self.controller.release.clear()
|
|
before = list(self.controller.messages)
|
|
self.window.message_edit.setPlainText("Nicht speichern")
|
|
self.window._send()
|
|
self._process_until(self.controller.started.is_set)
|
|
|
|
self.window._stop_generation()
|
|
self._process_until(lambda: self.window._thread is None)
|
|
|
|
self.assertEqual(self.controller.messages, before)
|
|
self.assertEqual(self.controller.state, GenerationState.ABORTED)
|
|
|
|
def test_stop_immediately_after_start_is_safe(self) -> None:
|
|
self.controller.release.clear()
|
|
self.window.message_edit.setPlainText("Sofort stoppen")
|
|
self.window._send()
|
|
self.window._stop_generation()
|
|
|
|
self._process_until(lambda: self.window._thread is None)
|
|
self.assertEqual(self.controller.session.message_count, 2)
|
|
|
|
def test_completion_is_idempotent_and_retains_qt_references(self) -> None:
|
|
worker = object()
|
|
thread = object()
|
|
self.window._worker = worker
|
|
self.window._thread = thread
|
|
self.window._generation_finalized = False
|
|
|
|
self.window._generation_complete()
|
|
self.window._generation_complete()
|
|
|
|
self.assertIs(self.window._worker, worker)
|
|
self.assertIs(self.window._thread, thread)
|
|
self.window._thread_finished()
|
|
self.assertIsNone(self.window._worker)
|
|
self.assertIsNone(self.window._thread)
|
|
|
|
def test_worker_emits_completion_only_once(self) -> None:
|
|
worker = StreamWorker(self.controller, "Nur einmal")
|
|
completions: list[bool] = []
|
|
worker.completed.connect(lambda: completions.append(True))
|
|
|
|
worker.run()
|
|
worker.run()
|
|
|
|
self.assertEqual(completions, [True])
|
|
|
|
def test_close_during_generation_cancels_then_closes(self) -> None:
|
|
self.controller.release.clear()
|
|
self.window.message_edit.setPlainText("Beim Schließen stoppen")
|
|
self.window._send()
|
|
self._process_until(self.controller.started.is_set)
|
|
|
|
self.window.close()
|
|
self.assertTrue(self.window._close_when_finished)
|
|
self._process_until(lambda: self.window._thread is None)
|
|
self._process_until(lambda: not self.window.isVisible())
|