chore: сделал
This commit is contained in:
protokey committed
1 parent
602199d510
commit
45e7a1d22f
25 files changed
+1164
-113
No files matched your search
+142
-3
@@ -5,11 +5,11 @@ from Telegram import interfaces
|
||||
from Telegram.models import Bot, Chat, User as TelegramUser
|
||||
from Telegram.testing import Conversation, FakeApi
|
||||
|
||||
from . import tasks
|
||||
from . import fields, tasks
|
||||
from .bot import texts
|
||||
from .bot.interface import MyPointInterface, is_proxy_url, parse_credentials
|
||||
from .models import Point
|
||||
from .tests_services import FakeRunner, create_point, patch_runner
|
||||
from .models import Credentials, Point, Server
|
||||
from .tests_services import FakeRunner, FakeXuiApi, create_point, patch_runner, patch_xui_api
|
||||
|
||||
|
||||
class ParsersTests(TestCase):
|
||||
@@ -189,6 +189,132 @@ class PointMenuTests(BotTestCase):
|
||||
self.assertIn(self.point.name, self.say(texts.SERVER_MENU))
|
||||
|
||||
|
||||
class KeysMenuTests(BotTestCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.say('/start')
|
||||
self.point = create_point(TelegramUser.objects.get(), mask=None)
|
||||
self.point.panel_inbound_id = 1
|
||||
self.point.status = Point.Status.READY
|
||||
self.point.save()
|
||||
self.panel_api = FakeXuiApi()
|
||||
patcher = patch_xui_api(self.panel_api)
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
self.say(texts.MANAGE)
|
||||
self.say(self.point.name)
|
||||
|
||||
def test_issue_show_and_revoke_key(self):
|
||||
self.assertIn('Ключей пока нет', self.say(texts.CLIENTS))
|
||||
self.assertEqual(self.keyboard, texts.KEYS_KEYBOARD)
|
||||
self.assertEqual(self.say(texts.ADD_KEY), texts.KEY_NAME_PROMPT)
|
||||
self.assertIn('латиница', self.say('Мама'))
|
||||
card = self.say('Mom-Phone')
|
||||
self.assertIn(self.point.subscription_url('sub-mom-phone'), card)
|
||||
self.assertIn('mom-phone', self.say(texts.CLIENTS))
|
||||
self.assertIn('↓ 3.0 МБ', self.say('mom-phone'))
|
||||
self.assertEqual(self.say('ghost'), texts.KEY_NOT_FOUND)
|
||||
self.say(texts.REVOKE_KEY)
|
||||
self.assertIn('Ключей пока нет', self.say('mom-phone'))
|
||||
self.assertEqual(self.panel_api.clients, {})
|
||||
|
||||
def test_panel_error(self):
|
||||
self.say(texts.CLIENTS)
|
||||
self.say(texts.ADD_KEY)
|
||||
self.panel_api.down = True
|
||||
self.assertEqual(self.say('mom'), texts.PANEL_ERROR.format(error='connection refused'))
|
||||
|
||||
def test_not_ready_point(self):
|
||||
self.point.set_status(Point.Status.ERROR)
|
||||
self.assertEqual(self.say(texts.CLIENTS), texts.POINT_NOT_READY)
|
||||
|
||||
def test_card_counts_keys(self):
|
||||
self.say(texts.CLIENTS)
|
||||
self.say(texts.ADD_KEY)
|
||||
self.say('mom')
|
||||
self.assertIn('Ключи: 1', self.say(texts.SERVER_MENU))
|
||||
|
||||
|
||||
class PointSettingsTests(BotTestCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.say('/start')
|
||||
self.point = create_point(TelegramUser.objects.get(), mask=None)
|
||||
self.point.panel_inbound_id = 1
|
||||
self.point.status = Point.Status.READY
|
||||
self.point.save()
|
||||
self.say(texts.MANAGE)
|
||||
self.say(self.point.name)
|
||||
|
||||
@mock.patch.object(tasks.set_domain, 'send')
|
||||
def test_change_domain(self, send):
|
||||
self.assertIn('10.0.0.1', self.say(texts.POINT_DOMAIN))
|
||||
self.assertEqual(self.keyboard, [[texts.CANCEL, texts.DELETE]])
|
||||
self.assertEqual(self.say('not a domain'), texts.DOMAIN_INVALID)
|
||||
self.assertEqual(self.say('VPN.Example.com.'), texts.CHANGE_STARTED)
|
||||
send.assert_called_once_with(self.point.pk, mock.ANY, 'vpn.example.com')
|
||||
self.point.refresh_from_db()
|
||||
self.assertEqual(self.point.status, Point.Status.UPDATING)
|
||||
self.assertEqual(self.say(texts.SERVER_MENU).count('настраивается'), 1)
|
||||
self.assertEqual(self.say(texts.POINT_DOMAIN), texts.POINT_BUSY.format(status='настраивается'))
|
||||
|
||||
@mock.patch.object(tasks.set_domain, 'send')
|
||||
def test_mask_point_cannot_drop_domain(self, send):
|
||||
Server.objects.create_server(self.point, Server.Role.MASK, Credentials('10.0.0.3', 'pw'))
|
||||
self.point.domain = 'vpn.example.com'
|
||||
self.point.save()
|
||||
self.assertIn('10.0.0.3', self.say(texts.POINT_DOMAIN))
|
||||
self.assertEqual(self.say(texts.DELETE), texts.DOMAIN_REQUIRED_BY_MASK)
|
||||
send.assert_not_called()
|
||||
|
||||
@mock.patch.object(tasks.set_domain, 'send')
|
||||
def test_drop_domain(self, send):
|
||||
self.point.domain = 'vpn.example.com'
|
||||
self.point.save()
|
||||
self.say(texts.POINT_DOMAIN)
|
||||
self.say(texts.DELETE)
|
||||
send.assert_called_once_with(self.point.pk, mock.ANY, '')
|
||||
|
||||
@mock.patch.object(tasks.set_outbound_proxy, 'send')
|
||||
def test_change_proxy(self, send):
|
||||
self.say(texts.POINT_PROXY)
|
||||
self.assertEqual(self.keyboard, [[texts.CF_WARP, texts.CANCEL, texts.DELETE]])
|
||||
self.assertEqual(self.say('ftp://x:21'), texts.PROXY_INVALID)
|
||||
self.assertEqual(self.say('socks5://user:secret@proxy.example:1080'), texts.CHANGE_STARTED)
|
||||
# сообщение с паролем удалено, в брокер прокси уходит зашифрованным
|
||||
self.assertEqual(len(self.api.deleted), 2)
|
||||
encrypted = send.call_args.args[2]
|
||||
self.assertNotIn('secret', encrypted)
|
||||
self.assertEqual(fields.decrypt(encrypted), 'socks5://user:secret@proxy.example:1080')
|
||||
|
||||
@mock.patch.object(tasks.set_outbound_proxy, 'send')
|
||||
def test_warp_and_direct(self, send):
|
||||
self.say(texts.POINT_PROXY)
|
||||
self.say(texts.CF_WARP)
|
||||
self.assertEqual(fields.decrypt(send.call_args.args[2]), Point.CF_WARP)
|
||||
self.point.set_status(Point.Status.READY)
|
||||
self.say(texts.SERVER_MENU)
|
||||
self.say(texts.POINT_PROXY)
|
||||
self.say(texts.DELETE)
|
||||
self.assertEqual(send.call_args.args[2], '')
|
||||
|
||||
def test_cancel_returns_to_point(self):
|
||||
self.say(texts.POINT_PROXY)
|
||||
self.assertIn('Состояние: активен', self.say(texts.CANCEL))
|
||||
self.assertEqual(self.keyboard, texts.POINT_KEYBOARD)
|
||||
|
||||
def test_not_ready_point(self):
|
||||
self.point.set_status(Point.Status.ERROR)
|
||||
self.assertEqual(self.say(texts.POINT_PROXY), texts.POINT_NOT_READY)
|
||||
|
||||
def test_card_hides_proxy_password(self):
|
||||
self.point.outbound_proxy = 'socks5://user:secret@proxy.example:1080'
|
||||
self.point.save()
|
||||
card = self.say(texts.SERVER_MENU)
|
||||
self.assertIn('socks5://user:***@proxy.example:1080', card)
|
||||
self.assertNotIn('secret', card)
|
||||
|
||||
|
||||
class TasksTests(TestCase):
|
||||
def setUp(self):
|
||||
bot = Bot.objects.create(token='1:test')
|
||||
@@ -219,3 +345,16 @@ class TasksTests(TestCase):
|
||||
text = self.run_task(tasks.release_point, self.point.pk, self.chat.pk)
|
||||
self.assertIn('10.0.0.1 - пароль root', text)
|
||||
self.assertFalse(Point.objects.exists())
|
||||
|
||||
def test_change_notifies_result(self):
|
||||
self.point.panel_inbound_id = 1
|
||||
self.point.status = Point.Status.READY
|
||||
self.point.save()
|
||||
proxy = fields.encrypt('socks5://user:secret@proxy.example:1080')
|
||||
text = self.run_task(tasks.set_outbound_proxy, self.point.pk, self.chat.pk, proxy)
|
||||
self.assertIn('user:***@proxy.example', text)
|
||||
with self.assertLogs('App.tasks', 'ERROR'):
|
||||
text = self.run_task(tasks.set_domain, self.point.pk, self.chat.pk, 'vpn.example.com', runner=FakeRunner(fail='xui_cert'))
|
||||
self.assertIn('прежними настройками', text)
|
||||
self.point.refresh_from_db()
|
||||
self.assertEqual((self.point.status, self.point.domain), (Point.Status.READY, ''))
|
||||
Reference in new issue
Block a user