213 lines
9.0 KiB
Python
213 lines
9.0 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from urllib.parse import parse_qs, urlparse
|
|
from unittest.mock import Mock, patch
|
|
|
|
import auth_webview
|
|
import main
|
|
from vk_api.exceptions import VkApiError
|
|
|
|
|
|
class AuthFallbackTests(unittest.TestCase):
|
|
def test_working_client_is_tried_first_with_vk_ru_redirect(self):
|
|
manager = main.VkChatManager.__new__(main.VkChatManager)
|
|
url = manager._build_vk_auth_url()
|
|
params = parse_qs(urlparse(url).query)
|
|
self.assertEqual(params["client_id"], ["6287487"])
|
|
self.assertEqual(params["redirect_uri"], ["https://oauth.vk.ru/blank.html"])
|
|
|
|
def test_expired_session_reopens_embedded_auth(self):
|
|
manager = main.VkChatManager.__new__(main.VkChatManager)
|
|
manager._auth_relogin_in_progress = False
|
|
manager._last_auth_relogin_ts = 0
|
|
manager._vk_error_code = Mock(return_value=5)
|
|
manager._format_vk_error = Mock(return_value="Session expired")
|
|
manager._log_event = Mock()
|
|
manager._clear_auth_state = Mock()
|
|
manager.start_auth = Mock()
|
|
with patch.object(main.QMessageBox, "warning"):
|
|
manager._force_relogin(Exception("Session expired"), "загрузки чатов")
|
|
manager.start_auth.assert_called_once_with(keep_status_text=True)
|
|
|
|
def test_parses_token_from_full_vk_response_link(self):
|
|
token, expires_in = auth_webview.extract_token(
|
|
"https://oauth.vk.ru/blank.html#access_token=test-token&expires_in=0&user_id=123&email=private@example.test"
|
|
)
|
|
self.assertEqual((token, expires_in), ("test-token", 0))
|
|
|
|
def test_reads_redirect_before_failed_page_load(self):
|
|
result = auth_webview.extract_redirect_result(
|
|
"https://oauth.vk.ru/blank.html#access_token=test-token&expires_in=0"
|
|
)
|
|
self.assertEqual(result, {"token": "test-token", "expires_in": 0})
|
|
self.assertIsNone(
|
|
auth_webview.extract_redirect_result(
|
|
"https://example.test/blank.html#access_token=test-token"
|
|
)
|
|
)
|
|
|
|
def test_embedded_window_captures_redirect_on_navigation_start(self):
|
|
class Event:
|
|
def __init__(self):
|
|
self.handler = None
|
|
|
|
def __iadd__(self, handler):
|
|
self.handler = handler
|
|
return self
|
|
|
|
navigation = Event()
|
|
finished = auth_webview.threading.Event()
|
|
window = SimpleNamespace(
|
|
native=SimpleNamespace(webview=SimpleNamespace(NavigationStarting=navigation)),
|
|
events=SimpleNamespace(before_show=Event(), loaded=Event(), closed=Event()),
|
|
get_current_url=lambda: "https://vk.ru/feed",
|
|
evaluate_js=Mock(),
|
|
destroy=Mock(side_effect=finished.set),
|
|
)
|
|
|
|
def start(_callback, **_kwargs):
|
|
window.events.before_show.handler()
|
|
window.events.loaded.handler()
|
|
args = SimpleNamespace(
|
|
Uri="https://oauth.vk.ru/blank.html#access_token=test-token&expires_in=0",
|
|
Cancel=False,
|
|
)
|
|
navigation.handler(None, args)
|
|
self.assertTrue(args.Cancel)
|
|
_callback()
|
|
self.assertTrue(finished.wait(2))
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir, patch.object(
|
|
auth_webview.webview, "create_window", return_value=window
|
|
) as create_window, patch.object(auth_webview.webview, "start", side_effect=start):
|
|
result_path = Path(temp_dir) / "auth.json"
|
|
auth_webview.main_auth("https://oauth.vk.com/authorize", str(result_path), login_first=True)
|
|
create_window.assert_called_once_with("VK Авторизация", "https://vk.ru/")
|
|
self.assertIn("https://oauth.vk.com/authorize", window.evaluate_js.call_args.args[0])
|
|
self.assertEqual(
|
|
json.loads(result_path.read_text(encoding="utf-8")),
|
|
{"token": "test-token", "expires_in": 0},
|
|
)
|
|
window.destroy.assert_called_once()
|
|
|
|
def test_parses_vk_oauth_error_from_redirect_url(self):
|
|
error, description = auth_webview.extract_auth_error(
|
|
"https://oauth.vk.com/blank.html#error=invalid_client&error_description=Bad+client"
|
|
)
|
|
self.assertEqual((error, description), ("invalid_client", "Bad client"))
|
|
|
|
def test_parses_vk_oauth_error_from_json_page(self):
|
|
error, description = auth_webview.extract_auth_error_from_page(
|
|
'{"error":"invalid_request","error_description":"incorrect app. Unavailable for apps with direct auth."}'
|
|
)
|
|
self.assertEqual(error, "invalid_request")
|
|
self.assertIn("Unavailable for apps with direct auth", description)
|
|
|
|
def test_next_client_is_tried_after_oauth_error(self):
|
|
manager = main.VkChatManager.__new__(main.VkChatManager)
|
|
manager._auth_client_index = 0
|
|
manager._auth_process_error_text = None
|
|
manager._auth_relogin_in_progress = False
|
|
manager._set_auth_ui_state = Mock()
|
|
manager._log_event = Mock()
|
|
manager.start_auth = Mock()
|
|
manager.status_label = Mock()
|
|
manager.set_ui_state = Mock()
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
result_path = Path(directory) / "auth_result.json"
|
|
result_path.write_text(json.dumps({"error": "invalid_client"}), encoding="utf-8")
|
|
manager.auth_output_path = str(result_path)
|
|
with patch.object(main, "VK_CLIENT_IDS", ("2274003", "3140623")):
|
|
manager._on_auth_process_finished(0, None)
|
|
|
|
manager.start_auth.assert_called_once_with(client_index=1)
|
|
self.assertFalse(result_path.exists())
|
|
|
|
def test_user_denial_does_not_open_another_client(self):
|
|
manager = main.VkChatManager.__new__(main.VkChatManager)
|
|
manager._auth_client_index = 0
|
|
manager._auth_process_error_text = None
|
|
manager._auth_relogin_in_progress = False
|
|
manager._set_auth_ui_state = Mock()
|
|
manager._log_event = Mock()
|
|
manager.start_auth = Mock()
|
|
manager.status_label = Mock()
|
|
manager.set_ui_state = Mock()
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
result_path = Path(directory) / "auth_result.json"
|
|
result_path.write_text(json.dumps({"error": "access_denied"}), encoding="utf-8")
|
|
manager.auth_output_path = str(result_path)
|
|
manager._on_auth_process_finished(0, None)
|
|
|
|
manager.start_auth.assert_not_called()
|
|
manager.set_ui_state.assert_called_once_with(False)
|
|
|
|
def test_api_access_denied_tries_next_client_without_saving_token(self):
|
|
manager = main.VkChatManager.__new__(main.VkChatManager)
|
|
manager._log_event = Mock()
|
|
manager._auth_relogin_in_progress = False
|
|
manager.status_label = Mock()
|
|
manager.start_auth = Mock()
|
|
manager.set_ui_state = Mock()
|
|
manager.token = None
|
|
|
|
class DeniedService:
|
|
def __init__(self):
|
|
self.api = SimpleNamespace(messages=SimpleNamespace(getConversations=Mock(side_effect=VkApiError("Access denied"))))
|
|
|
|
def set_token(self, token):
|
|
pass
|
|
|
|
def vk_error_code(self, exc):
|
|
return 15
|
|
|
|
with patch.object(main, "VkService", DeniedService), patch.object(main, "token_store_save_token") as save:
|
|
manager.handle_new_auth_token("candidate-token", 0, client_index=0)
|
|
|
|
manager.start_auth.assert_called_once_with(client_index=1)
|
|
save.assert_not_called()
|
|
self.assertIsNone(manager.token)
|
|
|
|
def test_flood_control_keeps_token_and_delays_chat_loading(self):
|
|
manager = main.VkChatManager.__new__(main.VkChatManager)
|
|
manager._log_event = Mock()
|
|
manager._auth_relogin_in_progress = False
|
|
manager.status_label = Mock()
|
|
manager.token_input = Mock()
|
|
manager.token_countdown_timer = Mock()
|
|
manager.token_countdown_timer.isActive.return_value = False
|
|
manager.vk_service = Mock()
|
|
manager.set_ui_state = Mock()
|
|
manager.load_chats = Mock()
|
|
manager.token = None
|
|
|
|
class LimitedService:
|
|
def __init__(self):
|
|
self.api = SimpleNamespace(messages=SimpleNamespace(getConversations=Mock(side_effect=VkApiError("Flood control"))))
|
|
|
|
def set_token(self, token):
|
|
pass
|
|
|
|
def vk_error_code(self, exc):
|
|
return 9
|
|
|
|
with patch.object(main, "VkService", LimitedService), patch.object(main, "token_store_save_token", return_value=0) as save, patch.object(main.QTimer, "singleShot") as delay:
|
|
manager.handle_new_auth_token("candidate-token", 0)
|
|
|
|
save.assert_called_once()
|
|
delay.assert_called_once()
|
|
self.assertEqual(delay.call_args.args[0], 60_000)
|
|
manager.load_chats.assert_not_called()
|
|
self.assertEqual(manager.token, "candidate-token")
|
|
delay.call_args.args[1]()
|
|
manager.load_chats.assert_called_once()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|