104 lines
4.0 KiB
Python
104 lines
4.0 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
from typing import Callable, Dict, Tuple, Any
|
|
|
|
import requests
|
|
|
|
from hysteria_panel.api import HysteriaAPI
|
|
|
|
|
|
class FauxResponse:
|
|
def __init__(self, status: int = 200, headers: Dict[str, str] | None = None, json_data: Any = None, text: str = "OK"):
|
|
self.status_code = status
|
|
self.headers = headers or {}
|
|
self._json_data = json_data
|
|
self.text = text
|
|
|
|
def raise_for_status(self):
|
|
if 400 <= self.status_code:
|
|
raise requests.HTTPError(f"HTTP {self.status_code}")
|
|
|
|
def json(self):
|
|
return self._json_data
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, responder: Callable[[str, str], FauxResponse]):
|
|
self.headers: Dict[str, str] = {}
|
|
self._responder = responder
|
|
|
|
def request(self, method: str, url: str, headers=None, json=None, params=None, timeout=None):
|
|
# Extract only the path for routing in tests
|
|
# URL format: scheme://host:port/path
|
|
path = url.split("//", 1)[-1]
|
|
path = path[path.find(":") + 1:] # remove host:port
|
|
path = path[path.find("/"):] # keep leading /path
|
|
return self._responder(method.lower(), path)
|
|
|
|
|
|
class TestHysteriaAPI(unittest.TestCase):
|
|
def _make_responder(self, mapping: Dict[Tuple[str, str], FauxResponse]):
|
|
def responder(method: str, path: str) -> FauxResponse:
|
|
# Default to 200 OK text response
|
|
return mapping.get((method, path), FauxResponse())
|
|
return responder
|
|
|
|
def test_init_sets_base_url_and_headers_http(self):
|
|
mapping = {('get', '/'): FauxResponse(status=200, text='ok')}
|
|
|
|
with patch('requests.Session', lambda: FakeSession(self._make_responder(mapping))):
|
|
api = HysteriaAPI('example.com', 9999, 'secret', use_https=False)
|
|
self.assertEqual(api.base_url, 'http://example.com:9999')
|
|
# Authorization header is set
|
|
self.assertIn('Authorization', api._session.headers)
|
|
self.assertEqual(api._session.headers['Authorization'], 'secret')
|
|
# User-Agent prefix
|
|
ua = api._session.headers.get('User-Agent')
|
|
# Support both str and bytes user-agent values
|
|
if isinstance(ua, (bytes, bytearray)):
|
|
self.assertTrue(ua.startswith(b'HysteriaPanel/'))
|
|
else:
|
|
self.assertTrue(str(ua).startswith('HysteriaPanel/'))
|
|
|
|
def test_init_sets_base_url_https(self):
|
|
mapping = {('get', '/'): FauxResponse(status=200, text='ok')}
|
|
|
|
with patch('requests.Session', lambda: FakeSession(self._make_responder(mapping))):
|
|
api = HysteriaAPI('example.com', 443, 'secret', use_https=True)
|
|
self.assertEqual(api.base_url, 'https://example.com:443')
|
|
|
|
def test_request_json_and_kick(self):
|
|
mapping = {
|
|
('get', '/'): FauxResponse(status=200, text='ok'),
|
|
('get', '/traffic'): FauxResponse(status=200, headers={'Content-Type': 'application/json'}, json_data={'alice': {'tx': 1, 'rx': 2}}),
|
|
('post', '/kick'): FauxResponse(status=204, headers={}),
|
|
}
|
|
|
|
with patch('requests.Session', lambda: FakeSession(self._make_responder(mapping))):
|
|
api = HysteriaAPI('h', 1, 's')
|
|
ok, data = api.get_traffic()
|
|
self.assertTrue(ok)
|
|
self.assertIsInstance(data, dict)
|
|
self.assertIn('alice', data)
|
|
|
|
ok, data = api.kick_users(['alice'])
|
|
self.assertTrue(ok)
|
|
self.assertIsNone(data)
|
|
|
|
def test_request_error(self):
|
|
def responder(method: str, path: str) -> FauxResponse:
|
|
if path == '/':
|
|
return FauxResponse(status=200, text='ok')
|
|
raise requests.RequestException("boom")
|
|
|
|
with patch('requests.Session', lambda: FakeSession(responder)):
|
|
api = HysteriaAPI('h', 1, 's')
|
|
ok, data = api._request('get', '/error')
|
|
self.assertFalse(ok)
|
|
self.assertIn('boom', data)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|