Spaces:
Running
Running
| """Unit tests for SuperAdmin TOTP enrollment.""" | |
| import sys | |
| from unittest.mock import AsyncMock, MagicMock, patch | |
| import pytest | |
| from fastapi import HTTPException | |
| sys.modules.setdefault("qrcode", MagicMock()) | |
| from app.services.totp_setup_service import TotpSetupService | |
| def actor(): | |
| user = MagicMock() | |
| user.id = "sa-uuid" | |
| user.email = "super@example.com" | |
| user.totp_secret = None | |
| return user | |
| def service(): | |
| db = AsyncMock() | |
| redis = AsyncMock() | |
| svc = TotpSetupService(db, redis) | |
| svc._users = AsyncMock() | |
| svc._audit = AsyncMock() | |
| return svc | |
| class TestTotpSetupService: | |
| async def test_status_not_configured(self, service, actor): | |
| assert (await service.status(actor)) == {"configured": False} | |
| async def test_begin_rejects_when_already_configured(self, service, actor): | |
| actor.totp_secret = "EXISTING" | |
| with pytest.raises(HTTPException) as exc: | |
| await service.begin_setup(actor) | |
| assert exc.value.status_code == 400 | |
| async def test_begin_returns_qr(self, service, actor): | |
| with patch("app.services.totp_setup_service.generate_totp_secret", return_value="JBSWY3DPEHPK3PXP"), \ | |
| patch("app.services.totp_setup_service.generate_qr_code_base64", return_value="data:image/png;base64,abc"): | |
| result = await service.begin_setup(actor) | |
| assert result["qr_code_data_uri"].startswith("data:image/png") | |
| assert result["manual_entry_key"] == "JBSWY3DPEHPK3PXP" | |
| service._redis.set.assert_awaited_once() | |
| async def test_confirm_persists_secret(self, service, actor): | |
| service._load_pending = AsyncMock(return_value="JBSWY3DPEHPK3PXP") | |
| with patch("app.services.totp_setup_service.verify_totp", return_value=True): | |
| result = await service.confirm_setup(actor, "123456") | |
| assert result == {"configured": True} | |
| service._users.set_totp_secret.assert_awaited_once_with("sa-uuid", "JBSWY3DPEHPK3PXP") | |
| service._audit.log.assert_awaited_once() | |
| async def test_confirm_rejects_invalid_code(self, service, actor): | |
| service._load_pending = AsyncMock(return_value="JBSWY3DPEHPK3PXP") | |
| with patch("app.services.totp_setup_service.verify_totp", return_value=False): | |
| with pytest.raises(HTTPException) as exc: | |
| await service.confirm_setup(actor, "000000") | |
| assert exc.value.status_code == 400 | |