.
This commit is contained in:
lyuksovannyy
@@ -0,0 +1,694 @@
|
||||
"""
|
||||
Comprehensive simulation and unit test suite for AI Chat Summaries Plugin.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
# Configure sys.path
|
||||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
if BASE_DIR not in sys.path:
|
||||
sys.path.insert(0, BASE_DIR)
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from src.config import (
|
||||
CONTEXT_WINDOW_BUDGETS,
|
||||
DEFAULT_CONTEXT_WINDOW,
|
||||
DEFAULT_MODEL_CUSTOM,
|
||||
DEFAULT_MODEL_OAUTH_FREE,
|
||||
DEFAULT_MODEL_OAUTH_PAID,
|
||||
MAX_SINGLE_MESSAGE_CHARS,
|
||||
OAUTH_FREE_MODELS,
|
||||
OAUTH_PAID_MODELS,
|
||||
OAUTH_TIER_FREE,
|
||||
OAUTH_TIER_PLUS,
|
||||
OAUTH_TIER_PRO,
|
||||
PROVIDER_ANTHROPIC,
|
||||
PROVIDER_CHATGPT_OAUTH,
|
||||
PROVIDER_CUSTOM,
|
||||
PROVIDER_GEMINI,
|
||||
PROVIDER_OLLAMA,
|
||||
PROVIDER_OPENAI,
|
||||
SETTING_ENABLE_THINKING_STREAM,
|
||||
SETTING_OAUTH_ACCESS_TOKEN,
|
||||
SETTING_OAUTH_MODEL,
|
||||
SETTING_OAUTH_USER_EMAIL,
|
||||
SETTING_OAUTH_USER_TIER,
|
||||
SETTING_PROVIDER,
|
||||
STYLE_BRIEF,
|
||||
STYLE_BULLETS,
|
||||
STYLE_CUSTOM,
|
||||
STYLE_DETAILED,
|
||||
)
|
||||
from src.localization import LocalizationManager, locali
|
||||
from src.providers.base import (
|
||||
build_debug_log,
|
||||
build_system_prompt,
|
||||
mask_sensitive_data,
|
||||
)
|
||||
from src.providers.builtin import (
|
||||
AnthropicDirectHandler,
|
||||
GeminiDirectHandler,
|
||||
OllamaDirectHandler,
|
||||
OpenAIDirectHandler,
|
||||
)
|
||||
from src.providers.custom import CustomAIHandler
|
||||
from src.providers.dispatcher import UnifiedDispatcher
|
||||
from src.providers.oauth import ChatGPTOAuthHandler
|
||||
from src.services.message_fetcher import FormattedMessage, MessageFetcher
|
||||
from src.ui.progress_widget import PinnedProgressManager
|
||||
|
||||
|
||||
class MockPlugin:
|
||||
"""Mock exteraGram plugin environment for testing."""
|
||||
def __init__(self):
|
||||
self._settings = {}
|
||||
self.oauth_handler = ChatGPTOAuthHandler(self)
|
||||
self.custom_handler = CustomAIHandler(self)
|
||||
self.dispatcher = UnifiedDispatcher(self)
|
||||
self.message_fetcher = MessageFetcher(self)
|
||||
self.progress_manager = PinnedProgressManager(self)
|
||||
|
||||
def get_setting(self, key, default=None):
|
||||
return self._settings.get(key, default)
|
||||
|
||||
def set_setting(self, key, value):
|
||||
self._settings[key] = value
|
||||
|
||||
def hook_method(self, target, hook):
|
||||
return "mock_hook_ref"
|
||||
|
||||
def unhook_method(self, ref):
|
||||
pass
|
||||
|
||||
class TestPluginMetadata(unittest.TestCase):
|
||||
"""Validates that plugin metadata contains non-empty id, name, and standard attributes."""
|
||||
|
||||
def test_config_metadata(self):
|
||||
from src import config
|
||||
self.assertTrue(bool(config.__id__))
|
||||
self.assertTrue(bool(config.__name__))
|
||||
self.assertTrue(bool(config.id))
|
||||
self.assertTrue(bool(config.name))
|
||||
self.assertEqual(config.__id__, "ai_chat_summaries")
|
||||
self.assertEqual(config.__name__, "AI Chat Summaries")
|
||||
|
||||
def test_plugin_entrypoint_metadata(self):
|
||||
from importlib.machinery import SourceFileLoader
|
||||
plugin_path = os.path.join(BASE_DIR, "ai_chat_summary.plugin")
|
||||
mod = SourceFileLoader("ai_chat_summaries_bundle", plugin_path).load_module()
|
||||
self.assertTrue(bool(getattr(mod, "__id__", None)))
|
||||
self.assertTrue(bool(getattr(mod, "__name__", None)))
|
||||
self.assertTrue(bool(getattr(mod, "id", None)))
|
||||
self.assertTrue(bool(getattr(mod, "name", None)))
|
||||
self.assertEqual(mod.__id__, "ai_chat_summaries")
|
||||
self.assertEqual(mod.__name__, "AI Chat Summaries")
|
||||
|
||||
def test_plugin_lifecycle_and_settings(self):
|
||||
from importlib.machinery import SourceFileLoader
|
||||
plugin_path = os.path.join(BASE_DIR, "ai_chat_summary.plugin")
|
||||
mod = SourceFileLoader("ai_chat_summaries_bundle", plugin_path).load_module()
|
||||
plugin = mod.AIChatSummariesPlugin()
|
||||
|
||||
# exteraGram invokes on_plugin_load / on_plugin_unload
|
||||
self.assertTrue(hasattr(plugin, "on_plugin_load"))
|
||||
self.assertTrue(hasattr(plugin, "on_plugin_unload"))
|
||||
self.assertTrue(hasattr(plugin, "on_menu_click"))
|
||||
self.assertTrue(plugin.has_settings())
|
||||
|
||||
plugin.on_plugin_load()
|
||||
self.assertTrue(len(plugin.create_settings()) > 0)
|
||||
plugin.on_plugin_unload()
|
||||
|
||||
|
||||
|
||||
class TestDiagnostics(unittest.TestCase):
|
||||
"""Tests structured error retention and copyable reports."""
|
||||
|
||||
def setUp(self):
|
||||
from src import diagnostics
|
||||
self.diagnostics = diagnostics
|
||||
diagnostics.clear()
|
||||
|
||||
def tearDown(self):
|
||||
self.diagnostics.uninstall_uncaught_exception_hooks()
|
||||
self.diagnostics.clear()
|
||||
|
||||
def test_structured_error_report_contains_full_context(self):
|
||||
try:
|
||||
raise ValueError("broken provider")
|
||||
except ValueError as exc:
|
||||
self.diagnostics.record_error("provider.test", exc)
|
||||
|
||||
errors = self.diagnostics.get_errors()
|
||||
self.assertEqual(len(errors), 1)
|
||||
self.assertEqual(errors[0]["where"], "provider.test")
|
||||
self.assertEqual(errors[0]["type"], "ValueError")
|
||||
self.assertEqual(errors[0]["message"], "broken provider")
|
||||
self.assertIn("Traceback", errors[0]["traceback"])
|
||||
self.assertTrue(errors[0]["thread"])
|
||||
|
||||
report = self.diagnostics.build_errors_report()
|
||||
self.assertIn("provider.test", report)
|
||||
self.assertIn("ValueError", report)
|
||||
self.assertIn("broken provider", report)
|
||||
self.assertIn("Traceback", report)
|
||||
|
||||
def test_logs_and_full_report_include_errors(self):
|
||||
self.diagnostics.record_error_message(
|
||||
"pipeline.fetch",
|
||||
"history failed",
|
||||
"TLRPC: HISTORY_FAILED",
|
||||
)
|
||||
logs = self.diagnostics.build_logs_report()
|
||||
full = self.diagnostics.build_report()
|
||||
|
||||
self.assertIn("pipeline.fetch", logs)
|
||||
self.assertIn("history failed", logs)
|
||||
self.assertIn("pipeline.fetch", full)
|
||||
self.assertIn("TLRPC: HISTORY_FAILED", full)
|
||||
|
||||
def test_uncaught_hooks_install_and_restore(self):
|
||||
old_sys = sys.excepthook
|
||||
old_thread = getattr(__import__("threading"), "excepthook", None)
|
||||
self.diagnostics.install_uncaught_exception_hooks()
|
||||
self.assertIsNot(sys.excepthook, old_sys)
|
||||
self.diagnostics.uninstall_uncaught_exception_hooks()
|
||||
self.assertIs(sys.excepthook, old_sys)
|
||||
if old_thread is not None:
|
||||
import threading
|
||||
self.assertIs(threading.excepthook, old_thread)
|
||||
|
||||
|
||||
|
||||
class TestLocalization(unittest.TestCase):
|
||||
"""Tests for internationalization dictionaries and fallback behavior."""
|
||||
|
||||
def test_english_strings(self):
|
||||
loc = LocalizationManager("en")
|
||||
title = loc.get("PLUGIN_NAME")
|
||||
self.assertEqual(title, "AI Chat Summaries")
|
||||
self.assertIn("ChatGPT", loc.get("OAUTH_HEADER"))
|
||||
|
||||
def test_russian_strings(self):
|
||||
loc = LocalizationManager("ru")
|
||||
title = loc.get("PLUGIN_NAME")
|
||||
self.assertEqual(title, "AI Саммари Чатов")
|
||||
self.assertIn("ChatGPT", loc.get("OAUTH_HEADER"))
|
||||
|
||||
def test_interpolation_and_fallback(self):
|
||||
loc = LocalizationManager("en")
|
||||
msg = loc.get("OAUTH_CONNECTED_STATUS", email="user@test.com", tier="PRO")
|
||||
self.assertIn("user@test.com", msg)
|
||||
self.assertIn("PRO", msg)
|
||||
|
||||
# Fallback to key itself if not found
|
||||
non_existent = loc.get("NON_EXISTENT_KEY_12345")
|
||||
self.assertEqual(non_existent, "NON_EXISTENT_KEY_12345")
|
||||
|
||||
|
||||
class TestBaseProviderUtilities(unittest.TestCase):
|
||||
"""Tests for prompt building, sensitive data masking, and debug log construction."""
|
||||
|
||||
def test_sensitive_masking(self):
|
||||
raw_header = {"Authorization": "Bearer sk-1234567890abcdef12345678", "User-Agent": "test"}
|
||||
masked = mask_sensitive_data(raw_header)
|
||||
self.assertTrue("sk-" in str(masked["Authorization"]))
|
||||
self.assertTrue("..." in str(masked["Authorization"]))
|
||||
self.assertNotIn("abcdef12345678", str(masked["Authorization"]))
|
||||
|
||||
raw_str = "Key: AIzaSyD9876543210abcdef and sk-proj-1122334455667788"
|
||||
masked_str = mask_sensitive_data(raw_str)
|
||||
self.assertNotIn("9876543210abcdef", masked_str)
|
||||
self.assertNotIn("1122334455667788", masked_str)
|
||||
|
||||
def test_debug_log_building(self):
|
||||
log = build_debug_log(
|
||||
url="https://api.openai.com/v1/chat/completions",
|
||||
method="POST",
|
||||
headers={"Authorization": "Bearer sk-testsecret12345"},
|
||||
payload={"model": "gpt-4o", "messages": [{"role": "user", "content": "Hello"}]},
|
||||
status_code=401,
|
||||
response_body={"error": {"message": "Invalid API key"}},
|
||||
)
|
||||
self.assertIn("=== AI Chat Summaries Debug Log ===", log)
|
||||
self.assertIn("Status Code: 401", log)
|
||||
self.assertIn("Invalid API key", log)
|
||||
self.assertNotIn("sk-testsecret12345", log)
|
||||
|
||||
def test_system_prompt_builder(self):
|
||||
prompt_brief = build_system_prompt(
|
||||
style=STYLE_BRIEF,
|
||||
lang=1, # Russian
|
||||
chat_title="Team Alpha",
|
||||
custom_prompt="Focus on deadlines",
|
||||
)
|
||||
self.assertIn("Team Alpha", prompt_brief)
|
||||
self.assertIn("Russian", prompt_brief)
|
||||
self.assertIn("Focus on deadlines", prompt_brief)
|
||||
self.assertIn("Brief overview", prompt_brief)
|
||||
|
||||
prompt_custom = build_system_prompt(
|
||||
style=STYLE_CUSTOM,
|
||||
lang=2, # English
|
||||
custom_prompt="Provide a 3-bullet list of blockers",
|
||||
)
|
||||
self.assertIn("Provide a 3-bullet list of blockers", prompt_custom)
|
||||
self.assertIn("English", prompt_custom)
|
||||
|
||||
|
||||
class TestChatGPTOAuth(unittest.TestCase):
|
||||
"""Tests for OAuth PKCE generation, code parsing, and tier-based model selection."""
|
||||
|
||||
def setUp(self):
|
||||
self.plugin = MockPlugin()
|
||||
self.oauth = self.plugin.oauth_handler
|
||||
|
||||
def test_auth_url_generation(self):
|
||||
url, verifier, state = self.oauth.generate_auth_url()
|
||||
self.assertTrue(url.startswith("https://auth0.openai.com/authorize?"))
|
||||
self.assertIn("code_challenge=", url)
|
||||
self.assertIn("code_challenge_method=S256", url)
|
||||
self.assertIn("client_id=exteragram_ai_summaries", url)
|
||||
self.assertTrue(len(verifier) > 40)
|
||||
self.assertTrue(len(state) > 10)
|
||||
|
||||
def test_code_extraction(self):
|
||||
raw_code = "4/0AeanS0abcdef12345"
|
||||
self.assertEqual(self.oauth.extract_code_from_input(raw_code), raw_code)
|
||||
|
||||
redirect_url = "exteragram://oauth/chatgpt?code=test_code_xyz123&state=state_abc"
|
||||
self.assertEqual(self.oauth.extract_code_from_input(redirect_url), "test_code_xyz123")
|
||||
|
||||
frag_url = "https://example.com/oauth/callback#code=frag_code_999"
|
||||
self.assertEqual(self.oauth.extract_code_from_input(frag_url), "frag_code_999")
|
||||
|
||||
def test_tier_model_mapping(self):
|
||||
# Free tier
|
||||
self.plugin.set_setting(SETTING_OAUTH_USER_TIER, OAUTH_TIER_FREE)
|
||||
free_models = self.oauth.get_tier_models()
|
||||
self.assertEqual(free_models, OAUTH_FREE_MODELS)
|
||||
self.assertIn("luna", free_models)
|
||||
self.assertEqual(self.oauth.get_active_model(), DEFAULT_MODEL_OAUTH_FREE)
|
||||
|
||||
# Plus / Pro tier
|
||||
self.plugin.set_setting(SETTING_OAUTH_USER_TIER, OAUTH_TIER_PLUS)
|
||||
paid_models = self.oauth.get_tier_models()
|
||||
self.assertEqual(paid_models, OAUTH_PAID_MODELS)
|
||||
self.assertIn("terra", paid_models)
|
||||
self.assertIn("sol", paid_models)
|
||||
self.assertEqual(self.oauth.get_active_model(), DEFAULT_MODEL_OAUTH_PAID)
|
||||
|
||||
@patch("requests.post")
|
||||
@patch("requests.get")
|
||||
def test_token_exchange_and_profile_inspection(self, mock_get, mock_post):
|
||||
# Mock token exchange response
|
||||
mock_post_resp = MagicMock()
|
||||
mock_post_resp.status_code = 200
|
||||
mock_post_resp.json.return_value = {
|
||||
"access_token": "mock_access_token_123",
|
||||
"refresh_token": "mock_refresh_token_456",
|
||||
"expires_in": 3600,
|
||||
}
|
||||
mock_post.return_value = mock_post_resp
|
||||
|
||||
# Mock user info profile response (Pro tier)
|
||||
mock_get_resp = MagicMock()
|
||||
mock_get_resp.status_code = 200
|
||||
mock_get_resp.json.return_value = {
|
||||
"email": "dev@company.com",
|
||||
"subscription": "pro_tier",
|
||||
}
|
||||
mock_get.return_value = mock_get_resp
|
||||
|
||||
success, data, err = self.oauth.exchange_code_for_tokens("code_test_123")
|
||||
self.assertTrue(success)
|
||||
self.assertIsNotNone(data)
|
||||
self.assertEqual(data["email"], "dev@company.com")
|
||||
self.assertEqual(data["tier"], OAUTH_TIER_PLUS)
|
||||
|
||||
# Verify saved in plugin settings
|
||||
self.assertEqual(self.plugin.get_setting(SETTING_OAUTH_ACCESS_TOKEN), "mock_access_token_123")
|
||||
self.assertEqual(self.plugin.get_setting(SETTING_OAUTH_USER_EMAIL), "dev@company.com")
|
||||
self.assertEqual(self.plugin.get_setting(SETTING_OAUTH_MODEL), DEFAULT_MODEL_OAUTH_PAID)
|
||||
|
||||
|
||||
class TestCustomAIProvider(unittest.TestCase):
|
||||
"""Tests for Custom AI Provider multi-stage verification and context budgeting."""
|
||||
|
||||
def setUp(self):
|
||||
self.plugin = MockPlugin()
|
||||
self.custom = self.plugin.custom_handler
|
||||
|
||||
@patch("requests.head")
|
||||
def test_stage1_reachability(self, mock_head):
|
||||
# Invalid scheme
|
||||
ok, msg = self.custom.verify_stage1_reachability("ftp://example.com")
|
||||
self.assertFalse(ok)
|
||||
|
||||
# Reachable HTTP 200
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_head.return_value = mock_resp
|
||||
|
||||
ok, msg = self.custom.verify_stage1_reachability("https://api.deepseek.com/v1")
|
||||
self.assertTrue(ok)
|
||||
self.assertIn("HTTP 200", msg)
|
||||
|
||||
@patch("requests.post")
|
||||
def test_stage2_probe_test(self, mock_post):
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.json.return_value = {
|
||||
"choices": [{"message": {"role": "assistant", "content": "Hello!"}}]
|
||||
}
|
||||
mock_post.return_value = mock_resp
|
||||
|
||||
ok, msg, debug_log, elapsed = self.custom.verify_stage2_probe_test(
|
||||
base_url="https://api.deepseek.com/v1",
|
||||
api_key="sk-test-12345",
|
||||
model_name="deepseek-chat",
|
||||
)
|
||||
self.assertTrue(ok)
|
||||
self.assertIsNone(debug_log)
|
||||
|
||||
# Verify request payload has thinking disabled & 'Hey there'
|
||||
call_kwargs = mock_post.call_args[1]
|
||||
json_payload = call_kwargs["json"]
|
||||
self.assertEqual(json_payload["model"], "deepseek-chat")
|
||||
self.assertEqual(json_payload["messages"][0]["content"], "Hey there")
|
||||
self.assertEqual(json_payload["max_tokens"], 10)
|
||||
self.assertEqual(json_payload["thinking"], {"type": "disabled"})
|
||||
|
||||
def test_context_window_budgeting(self):
|
||||
self.plugin.set_setting("custom_context_window", "8k")
|
||||
budget_8k = self.custom.get_context_character_budget()
|
||||
self.assertEqual(budget_8k, CONTEXT_WINDOW_BUDGETS["8k"])
|
||||
|
||||
# Generate long message (e.g. 10,000 chars) -> must clamp to 8,000 chars per message
|
||||
long_msg = "X" * 10_000
|
||||
short_msg = "Normal message 123"
|
||||
messages = [long_msg, short_msg]
|
||||
|
||||
transcript = self.custom.budget_and_chunk_transcript(messages)
|
||||
self.assertIn("Normal message 123", transcript)
|
||||
self.assertIn("[... truncated long message content ...]", transcript)
|
||||
self.assertTrue(len(transcript) <= budget_8k)
|
||||
|
||||
|
||||
class TestMessageFetcher(unittest.TestCase):
|
||||
"""Tests for message conversion, sender formatting, and chronological sorting."""
|
||||
|
||||
def test_formatted_message_transcript_line(self):
|
||||
msg = FormattedMessage(
|
||||
id=101,
|
||||
date=1700000000,
|
||||
time_str="2023-11-14 22:13",
|
||||
sender_name="Alice Wonderland",
|
||||
sender_username="@alice",
|
||||
text="Let's schedule our release for Monday 10am UTC.",
|
||||
reply_to_id=98,
|
||||
forward_from="Bob",
|
||||
media_info="[Photo: System architecture]",
|
||||
)
|
||||
line = msg.to_transcript_line()
|
||||
self.assertIn("[2023-11-14 22:13] #101 Alice Wonderland (fwd: Bob, reply-to #98):", line)
|
||||
self.assertIn("[Photo: System architecture]", line)
|
||||
self.assertIn("Let's schedule our release for Monday 10am UTC.", line)
|
||||
|
||||
def test_chronological_ordering(self):
|
||||
fetcher = MessageFetcher()
|
||||
raw_msgs = [
|
||||
MagicMock(id=3, date=1700000300, message="Third message", from_id=None, peer_id=None, reply_to=None, fwd_from=None, media=None, action=None),
|
||||
MagicMock(id=1, date=1700000100, message="First message", from_id=None, peer_id=None, reply_to=None, fwd_from=None, media=None, action=None),
|
||||
MagicMock(id=2, date=1700000200, message="Second message", from_id=None, peer_id=None, reply_to=None, fwd_from=None, media=None, action=None),
|
||||
]
|
||||
processed = fetcher._process_messages(raw_msgs, {}, {})
|
||||
self.assertEqual(len(processed), 3)
|
||||
self.assertEqual(processed[0].id, 1)
|
||||
self.assertEqual(processed[1].id, 2)
|
||||
self.assertEqual(processed[2].id, 3)
|
||||
|
||||
|
||||
class TestThinkingStreamDispatcher(unittest.TestCase):
|
||||
"""Tests for UnifiedDispatcher streaming token accumulation and thinking parsing."""
|
||||
|
||||
def test_streaming_token_accumulator(self):
|
||||
plugin = MockPlugin()
|
||||
dispatcher = plugin.dispatcher
|
||||
dispatcher.reset_stream_state("Custom endpoint", "deepseek-r1", "General Chat")
|
||||
|
||||
accumulated_thoughts = []
|
||||
accumulated_content = []
|
||||
|
||||
def listener(t_delta, c_delta):
|
||||
if t_delta:
|
||||
accumulated_thoughts.append(t_delta)
|
||||
if c_delta:
|
||||
accumulated_content.append(c_delta)
|
||||
|
||||
dispatcher.add_stream_listener(listener)
|
||||
|
||||
# Simulate streaming chunk delivery
|
||||
dispatcher._on_stream_chunk("Thinking step 1... ", "")
|
||||
dispatcher._on_stream_chunk("Thinking step 2.", "")
|
||||
dispatcher._on_stream_chunk("", "Summary result: ")
|
||||
dispatcher._on_stream_chunk("", "Everything is completed successfully.")
|
||||
|
||||
self.assertEqual(dispatcher.active_stream_state["thinking"], "Thinking step 1... Thinking step 2.")
|
||||
self.assertEqual(dispatcher.active_stream_state["content"], "Summary result: Everything is completed successfully.")
|
||||
self.assertEqual(len(accumulated_thoughts), 2)
|
||||
|
||||
class TestBuiltinDirectProviders(unittest.TestCase):
|
||||
"""Tests for OpenAI, Anthropic, Gemini, and Ollama direct API handlers and retry backoff."""
|
||||
|
||||
def setUp(self):
|
||||
self.plugin = MockPlugin()
|
||||
|
||||
@patch("requests.post")
|
||||
def test_openai_direct_summarize(self, mock_post):
|
||||
self.plugin.set_setting("openai_api_key", "sk-proj-test12345")
|
||||
self.plugin.set_setting("openai_model", "gpt-4o")
|
||||
handler = OpenAIDirectHandler(self.plugin)
|
||||
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.json.return_value = {
|
||||
"choices": [{"message": {"role": "assistant", "content": "Summary of chat"}}]
|
||||
}
|
||||
mock_post.return_value = mock_resp
|
||||
|
||||
resp = handler.summarize(
|
||||
messages_transcript="Alice: Hi\nBob: Hello",
|
||||
style=STYLE_BRIEF,
|
||||
lang=0,
|
||||
enable_stream=False,
|
||||
)
|
||||
self.assertTrue(resp.success)
|
||||
self.assertEqual(resp.text, "Summary of chat")
|
||||
self.assertEqual(resp.model, "gpt-4o")
|
||||
|
||||
@patch("requests.post")
|
||||
def test_anthropic_direct_with_thinking(self, mock_post):
|
||||
self.plugin.set_setting("anthropic_api_key", "sk-ant-test12345")
|
||||
self.plugin.set_setting("anthropic_model", "claude-3-7-sonnet-latest")
|
||||
handler = AnthropicDirectHandler(self.plugin)
|
||||
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.json.return_value = {
|
||||
"content": [
|
||||
{"type": "thinking", "thinking": "Analyzing topics..."},
|
||||
{"type": "text", "text": "Claude Summary of chat"}
|
||||
]
|
||||
}
|
||||
mock_post.return_value = mock_resp
|
||||
|
||||
resp = handler.summarize(
|
||||
messages_transcript="Alice: Project roadmap",
|
||||
style=STYLE_DETAILED,
|
||||
lang=2,
|
||||
enable_stream=False,
|
||||
)
|
||||
self.assertTrue(resp.success)
|
||||
self.assertEqual(resp.text, "Claude Summary of chat")
|
||||
self.assertEqual(resp.reasoning, "Analyzing topics...")
|
||||
|
||||
@patch("requests.post")
|
||||
def test_gemini_direct_summarize(self, mock_post):
|
||||
self.plugin.set_setting("gemini_api_key", "AIzaSyTest12345")
|
||||
self.plugin.set_setting("gemini_model", "gemini-2.0-flash")
|
||||
handler = GeminiDirectHandler(self.plugin)
|
||||
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.json.return_value = {
|
||||
"candidates": [
|
||||
{
|
||||
"content": {
|
||||
"parts": [
|
||||
{"thought": True, "text": "Gemini thought"},
|
||||
{"text": "Gemini summary text"}
|
||||
]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
mock_post.return_value = mock_resp
|
||||
|
||||
resp = handler.summarize(
|
||||
messages_transcript="User: Update?",
|
||||
style=STYLE_BULLETS,
|
||||
lang=1,
|
||||
enable_stream=False,
|
||||
)
|
||||
self.assertTrue(resp.success)
|
||||
self.assertEqual(resp.text, "Gemini summary text")
|
||||
self.assertEqual(resp.reasoning, "Gemini thought")
|
||||
|
||||
@patch("time.sleep")
|
||||
def test_exponential_backoff_retry(self, mock_sleep):
|
||||
handler = OpenAIDirectHandler(self.plugin)
|
||||
attempts = [0]
|
||||
|
||||
def flaky_call():
|
||||
attempts[0] += 1
|
||||
if attempts[0] < 3:
|
||||
# Simulate 429 Rate Limit
|
||||
resp = MagicMock()
|
||||
resp.status_code = 429
|
||||
import requests
|
||||
err = requests.exceptions.HTTPError("429 Rate Limit")
|
||||
err.response = resp
|
||||
raise err
|
||||
return "success_after_retry"
|
||||
|
||||
result = handler.execute_with_retry(flaky_call, max_retries=3, base_delay=0.1)
|
||||
self.assertEqual(result, "success_after_retry")
|
||||
self.assertEqual(attempts[0], 3)
|
||||
self.assertEqual(mock_sleep.call_count, 2)
|
||||
|
||||
|
||||
class TestUIWidgetsAndLifecycle(unittest.TestCase):
|
||||
"""Tests for ProgressManager, PreRequest configuration, and unread count calculations."""
|
||||
|
||||
def test_pinned_view_padding_shrink_and_restore(self):
|
||||
manager = PinnedProgressManager()
|
||||
mock_pinned_view = MagicMock()
|
||||
mock_pinned_view.getPaddingRight.return_value = 16
|
||||
mock_pinned_view.getPaddingLeft.return_value = 16
|
||||
mock_pinned_view.getPaddingTop.return_value = 8
|
||||
mock_pinned_view.getPaddingBottom.return_value = 8
|
||||
|
||||
manager.attach_pinned_view(mock_pinned_view)
|
||||
self.assertEqual(manager.original_right_padding, 16)
|
||||
|
||||
# Simulate show_progress
|
||||
mock_activity = MagicMock()
|
||||
manager.show_progress(mock_activity, 12345, "Tech Chat")
|
||||
self.assertTrue(manager.is_showing)
|
||||
|
||||
# Hide progress -> must restore padding
|
||||
manager.hide_progress()
|
||||
self.assertFalse(manager.is_showing)
|
||||
mock_pinned_view.setPadding.assert_called_with(16, 8, 16, 8)
|
||||
|
||||
def test_unread_count_and_offset_calculation(self):
|
||||
from src.hooks.unread_hook import UnreadBadgeHook
|
||||
hook = UnreadBadgeHook()
|
||||
|
||||
# Case 1: 0 or empty unread -> count 100, offset 0
|
||||
raw_unread = 0
|
||||
target_count = min(1000, max(1, raw_unread)) if raw_unread > 0 else 100
|
||||
target_offset = max(0, raw_unread - target_count) if raw_unread > 0 else 0
|
||||
self.assertEqual(target_count, 100)
|
||||
self.assertEqual(target_offset, 0)
|
||||
|
||||
# Case 2: 250 unread messages -> count 250, offset 0 (all unread)
|
||||
raw_unread = 250
|
||||
target_count = min(1000, max(1, raw_unread)) if raw_unread > 0 else 100
|
||||
target_offset = max(0, raw_unread - target_count) if raw_unread > 0 else 0
|
||||
self.assertEqual(target_count, 250)
|
||||
self.assertEqual(target_offset, 0)
|
||||
|
||||
# Case 3: 1500 unread messages -> count 1000, offset 500 (oldest 1000 unread batch)
|
||||
raw_unread = 1500
|
||||
target_count = min(1000, max(1, raw_unread)) if raw_unread > 0 else 100
|
||||
target_offset = max(0, raw_unread - target_count) if raw_unread > 0 else 0
|
||||
self.assertEqual(target_count, 1000)
|
||||
self.assertEqual(target_offset, 500)
|
||||
|
||||
def test_scroll_to_latest_message_navigation(self):
|
||||
from src.ui.summary_dialog import scroll_to_message_in_chat
|
||||
mock_activity = MagicMock()
|
||||
mock_activity.scrollToMessageId = MagicMock()
|
||||
|
||||
# Navigate to message #54321
|
||||
success = scroll_to_message_in_chat(mock_activity, message_id=54321, topic_id=0)
|
||||
self.assertTrue(success)
|
||||
mock_activity.scrollToMessageId.assert_called_with(54321, 0, True, 0, False, 0)
|
||||
|
||||
def test_page_down_long_press_timing_contract(self):
|
||||
from src.hooks import unread_hook
|
||||
|
||||
class Event:
|
||||
def __init__(self, action):
|
||||
self._action = action
|
||||
def getAction(self):
|
||||
return self._action
|
||||
|
||||
# Short press: below threshold, should remain native.
|
||||
with patch.object(unread_hook.time, "monotonic", side_effect=[10.0, 10.2, 10.2]):
|
||||
unread_hook._gesture_before_dispatch(Event(0))
|
||||
unread_hook._gesture_before_dispatch(Event(1))
|
||||
self.assertLess(unread_hook._held_duration(), unread_hook.LONG_PRESS_SECONDS)
|
||||
|
||||
# Long press: above threshold, should enter our interception path.
|
||||
with patch.object(unread_hook.time, "monotonic", side_effect=[20.0, 20.6, 20.6]):
|
||||
unread_hook._gesture_before_dispatch(Event(0))
|
||||
unread_hook._gesture_before_dispatch(Event(1))
|
||||
self.assertGreaterEqual(unread_hook._held_duration(), unread_hook.LONG_PRESS_SECONDS)
|
||||
|
||||
class TestPluginEndToEndSimulation(unittest.TestCase):
|
||||
"""End-to-end simulation of summarization pipeline from trigger to result."""
|
||||
|
||||
@patch("requests.post")
|
||||
def test_full_pipeline_run(self, mock_post):
|
||||
plugin = MockPlugin()
|
||||
plugin.set_setting(SETTING_PROVIDER, PROVIDER_CUSTOM)
|
||||
plugin.set_setting("custom_base_url", "https://api.deepseek.com/v1")
|
||||
plugin.set_setting("custom_api_key", "sk-deepseek-test")
|
||||
plugin.set_setting("custom_model", "deepseek-chat")
|
||||
plugin.set_setting(SETTING_ENABLE_THINKING_STREAM, False)
|
||||
|
||||
# Mock AI completion response
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.status_code = 200
|
||||
mock_resp.json.return_value = {
|
||||
"choices": [{"message": {"role": "assistant", "content": "Summary: Key topics discussed."}}]
|
||||
}
|
||||
mock_post.return_value = mock_resp
|
||||
|
||||
# Verify unified dispatcher routes properly
|
||||
provider = plugin.dispatcher.get_active_provider()
|
||||
self.assertEqual(provider.name, "Custom endpoint")
|
||||
|
||||
# Dispatch summarization
|
||||
response = plugin.dispatcher.run_summary(
|
||||
messages_transcript="Alice: Deployment ready.\nBob: Approved.",
|
||||
chat_title="Release Channel",
|
||||
style=STYLE_BRIEF,
|
||||
lang=0,
|
||||
)
|
||||
|
||||
self.assertTrue(response.success)
|
||||
self.assertEqual(response.text, "Summary: Key topics discussed.")
|
||||
self.assertEqual(response.model, "deepseek-chat")
|
||||
self.assertEqual(response.provider_name, "Custom endpoint")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user