forust
d9f1c8325a
ci / lint-prettier (push) Failing after 8s
ci / lint-ruff (push) Successful in 4s
ci / lint-yaml (push) Failing after 7s
ci / lint-dockerfiles (push) Successful in 4s
ci / validate (push) Successful in 5s
deploy / redeploy (push) Failing after 0s
ci / build (push) Has been skipped
ci / deploy-userbot-panel (push) Has been skipped
- ensure_prerequisites runs on startup, not per-request; kube config errors surface as 503 PanelError - serialize provisioning with a lock; drop per-endpoint prereq checks - guard SPA fallback against path traversal (relative_to) - add backend tests for auth flow, k8s service, spa routing; ci comment for legacy userbot deployments
102 lines
2.9 KiB
Python
102 lines
2.9 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from app.auth_service import TelegramAuthService
|
|
from app.models import PhoneStart, StringSessionStart
|
|
|
|
|
|
class FakeClient:
|
|
instances = []
|
|
requires_password = False
|
|
|
|
def __init__(self, *_args, **kwargs):
|
|
self.kwargs = kwargs
|
|
self.is_connected = False
|
|
self.disconnected = False
|
|
self.__class__.instances.append(self)
|
|
|
|
async def connect(self):
|
|
self.is_connected = True
|
|
|
|
async def disconnect(self):
|
|
self.is_connected = False
|
|
self.disconnected = True
|
|
|
|
async def send_code(self, _phone):
|
|
return SimpleNamespace(phone_code_hash="hash")
|
|
|
|
async def sign_in(self, *_args):
|
|
if self.requires_password:
|
|
from pyrogram.errors import SessionPasswordNeeded
|
|
|
|
raise SessionPasswordNeeded()
|
|
|
|
async def check_password(self, _password):
|
|
return None
|
|
|
|
async def get_me(self):
|
|
return SimpleNamespace(id=1)
|
|
|
|
async def export_session_string(self):
|
|
return "exported-session"
|
|
|
|
|
|
def phone_payload() -> PhoneStart:
|
|
return PhoneStart(
|
|
instance_id="test",
|
|
display_name="Test",
|
|
api_id=123,
|
|
api_hash="0123456789abcdef",
|
|
phone="+421900000000",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phone_code_success_closes_and_forgets_client(monkeypatch) -> None:
|
|
monkeypatch.setattr("app.auth_service.Client", FakeClient)
|
|
auth = TelegramAuthService()
|
|
|
|
flow_id = await auth.start_phone(phone_payload())
|
|
result = await auth.submit_code(flow_id, "12345")
|
|
|
|
assert result.session_string == "exported-session"
|
|
assert flow_id not in auth.flows
|
|
assert FakeClient.instances[-1].disconnected is True
|
|
assert FakeClient.instances[-1].kwargs["in_memory"] is True
|
|
assert FakeClient.instances[-1].kwargs["no_updates"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_string_session_is_validated_and_closed(monkeypatch) -> None:
|
|
monkeypatch.setattr("app.auth_service.Client", FakeClient)
|
|
auth = TelegramAuthService()
|
|
payload = StringSessionStart(
|
|
instance_id="test",
|
|
display_name="Test",
|
|
api_id=123,
|
|
api_hash="0123456789abcdef",
|
|
session_string="x" * 64,
|
|
)
|
|
|
|
result = await auth.validate_string_session(payload)
|
|
|
|
assert result.session_string == "exported-session"
|
|
assert FakeClient.instances[-1].disconnected is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_phone_rejects_overflow(monkeypatch) -> None:
|
|
from app.errors import PanelError
|
|
|
|
monkeypatch.setattr("app.auth_service.Client", FakeClient)
|
|
auth = TelegramAuthService(ttl_seconds=3600, max_flows=2)
|
|
|
|
await auth.start_phone(phone_payload())
|
|
await auth.start_phone(phone_payload())
|
|
|
|
with pytest.raises(PanelError) as error:
|
|
await auth.start_phone(phone_payload())
|
|
|
|
assert error.value.status_code == 429
|
|
assert "Too many pending" in error.value.detail
|