mirror of
https://github.com/rangermix/TwitchDropsMiner.git
synced 2026-05-26 07:08:04 +00:00
feat: Add proxy support with verification and benefit type filters (#13)
* feat: Add proxy configuration and verification functionality with dedicated API and UI.
* no header needed
* batch render
* feat: Add inventory filters for drop benefit types and refine proxy settings UI.
* debug message
* benefit type filter fix
* avoid duplicated update
* claude md update
* sync change from 5d7a697c2b
---------
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
This commit is contained in:
63
tests/test_proxy_settings.py
Normal file
63
tests/test_proxy_settings.py
Normal file
@@ -0,0 +1,63 @@
|
||||
|
||||
import unittest
|
||||
import asyncio
|
||||
from unittest.mock import MagicMock
|
||||
from yarl import URL
|
||||
|
||||
# Mock the imports that depend on application structure if needed,
|
||||
# or just import them if PYTHONPATH is set correctly.
|
||||
# Assuming run from root, imports should work.
|
||||
from src.config.settings import Settings
|
||||
from src.web.managers.settings import SettingsManager
|
||||
from src.web.managers.console import ConsoleOutputManager
|
||||
|
||||
class TestProxySettings(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.mock_broadcaster = MagicMock()
|
||||
# Mock emit to be awaitable
|
||||
f = asyncio.Future()
|
||||
f.set_result(None)
|
||||
self.mock_broadcaster.emit = MagicMock(return_value=f)
|
||||
|
||||
self.mock_settings = MagicMock(spec=Settings)
|
||||
# Setup properties
|
||||
self.mock_settings.proxy = URL()
|
||||
self.mock_settings.language = "en"
|
||||
self.mock_settings.dark_mode = False
|
||||
self.mock_settings.games_to_watch = []
|
||||
self.mock_settings.connection_quality = 1
|
||||
self.mock_settings.minimum_refresh_interval_minutes = 30
|
||||
|
||||
self.mock_console = MagicMock(spec=ConsoleOutputManager)
|
||||
|
||||
# Mock asyncio.create_task
|
||||
self.create_task_patcher = unittest.mock.patch('asyncio.create_task')
|
||||
self.mock_create_task = self.create_task_patcher.start()
|
||||
|
||||
def tearDown(self):
|
||||
self.create_task_patcher.stop()
|
||||
|
||||
def test_update_proxy_setting(self):
|
||||
manager = SettingsManager(self.mock_broadcaster, self.mock_settings, self.mock_console)
|
||||
|
||||
# Test setting a proxy
|
||||
proxy_url = "http://user:pass@localhost:8080"
|
||||
manager.update_settings({"proxy": proxy_url})
|
||||
|
||||
self.assertEqual(self.mock_settings.proxy, URL(proxy_url))
|
||||
self.mock_console.print.assert_called_with(f"Proxy set to: {proxy_url}")
|
||||
|
||||
# Test clearing a proxy
|
||||
manager.update_settings({"proxy": ""})
|
||||
self.assertEqual(self.mock_settings.proxy, URL())
|
||||
self.mock_console.print.assert_called_with("Proxy cleared")
|
||||
|
||||
def test_proxy_persistence_trigger(self):
|
||||
manager = SettingsManager(self.mock_broadcaster, self.mock_settings, self.mock_console)
|
||||
manager.update_settings({"proxy": "http://1.2.3.4:8080"})
|
||||
|
||||
self.mock_settings.alter.assert_called()
|
||||
self.mock_settings.save.assert_called()
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
86
tests/test_verify_proxy.py
Normal file
86
tests/test_verify_proxy.py
Normal file
@@ -0,0 +1,86 @@
|
||||
|
||||
import asyncio
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch, AsyncMock
|
||||
from src.web.app import verify_proxy
|
||||
from src.web.app import ProxyVerifyRequest
|
||||
|
||||
class MockResponseContext:
|
||||
def __init__(self, response_or_exc):
|
||||
self.response_or_exc = response_or_exc
|
||||
|
||||
async def __aenter__(self):
|
||||
if isinstance(self.response_or_exc, Exception):
|
||||
raise self.response_or_exc
|
||||
return self.response_or_exc
|
||||
|
||||
async def __aexit__(self, exc_type, exc, tb):
|
||||
pass
|
||||
|
||||
class TestVerifyProxy(unittest.TestCase):
|
||||
def setUp(self):
|
||||
# Patch aiohttp.ClientSession
|
||||
self.session_patcher = patch('aiohttp.ClientSession')
|
||||
self.mock_session_cls = self.session_patcher.start()
|
||||
# session object itself is not async, it has async methods/CMs
|
||||
self.mock_session = MagicMock()
|
||||
# Ensure the session context manager returns our mock session
|
||||
# ClientSession() -> CM -> __aenter__ -> session
|
||||
self.mock_session_cls.return_value.__aenter__.return_value = self.mock_session
|
||||
|
||||
def tearDown(self):
|
||||
self.session_patcher.stop()
|
||||
|
||||
def test_verify_proxy_success(self):
|
||||
# Mock successful response
|
||||
mock_response = AsyncMock()
|
||||
mock_response.status = 200
|
||||
|
||||
# Configure get to return our custom context manager
|
||||
self.mock_session.get.side_effect = lambda *args, **kwargs: MockResponseContext(mock_response)
|
||||
|
||||
request = ProxyVerifyRequest(proxy="http://valid-proxy:8080")
|
||||
|
||||
# Run async function
|
||||
result = asyncio.run(verify_proxy(request))
|
||||
|
||||
self.assertTrue(result['success'])
|
||||
self.assertIn("Connected!", result['message'])
|
||||
self.assertIn("latency", result)
|
||||
|
||||
def test_verify_proxy_failure_status(self):
|
||||
# Mock error status response
|
||||
mock_response = AsyncMock()
|
||||
mock_response.status = 503
|
||||
|
||||
self.mock_session.get.side_effect = lambda *args, **kwargs: MockResponseContext(mock_response)
|
||||
|
||||
request = ProxyVerifyRequest(proxy="http://bad-proxy:8080")
|
||||
|
||||
result = asyncio.run(verify_proxy(request))
|
||||
|
||||
self.assertFalse(result['success'])
|
||||
|
||||
# The expected message in app.py is: f"Proxy reachable but returned {response.status}"
|
||||
self.assertIn("Proxy reachable but returned 503", result['message'])
|
||||
|
||||
def test_verify_proxy_connection_error(self):
|
||||
# Mock connection error
|
||||
error = Exception("Connection refused")
|
||||
self.mock_session.get.side_effect = lambda *args, **kwargs: MockResponseContext(error)
|
||||
|
||||
request = ProxyVerifyRequest(proxy="http://down-proxy:8080")
|
||||
|
||||
result = asyncio.run(verify_proxy(request))
|
||||
|
||||
self.assertFalse(result['success'])
|
||||
self.assertIn("Connection failed", result['message'])
|
||||
|
||||
def test_verify_proxy_empty(self):
|
||||
request = ProxyVerifyRequest(proxy="")
|
||||
result = asyncio.run(verify_proxy(request))
|
||||
self.assertFalse(result['success'])
|
||||
self.assertEqual(result['message'], "Proxy URL is empty")
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user