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())