from datetime import timedelta from uuid import uuid4 import pytest from app.core.security import now_utc from app.models.challenge import Challenge, ChallengeStatus from app.services.challenges import approve_challenge, expire_if_needed class DummySession: def __init__(self, device=None): self.device = device self.added = [] def add(self, item): self.added.append(item) async def get(self, model, ident): return self.device class DummyDevice: def __init__(self, device_id, user_id): self.id = device_id self.user_id = user_id self.is_revoked = False self.public_key_pem = "invalid" self.last_seen_at = None @pytest.mark.asyncio async def test_expired_challenge_transitions_to_expired(): challenge = Challenge( id=uuid4(), user_id=uuid4(), device_id=uuid4(), status=ChallengeStatus.pending, relying_party="app", requester_ip="127.0.0.1", nonce="nonce", payload={}, issued_at=now_utc() - timedelta(seconds=120), expires_at=now_utc() - timedelta(seconds=1), ) changed = await expire_if_needed(DummySession(), challenge) assert changed assert challenge.status == ChallengeStatus.expired assert challenge.responded_at is not None @pytest.mark.asyncio async def test_replay_is_blocked_after_approval_state(): device_id = uuid4() challenge = Challenge( id=uuid4(), user_id=uuid4(), device_id=device_id, status=ChallengeStatus.approved, relying_party="app", requester_ip="127.0.0.1", nonce="nonce", payload={"challenge_id": "x"}, issued_at=now_utc(), expires_at=now_utc() + timedelta(seconds=60), ) ok, reason = await approve_challenge( DummySession(DummyDevice(device_id, challenge.user_id)), challenge=challenge, device_id=device_id, payload=challenge.payload, signature="anything", ip_address="127.0.0.1", ) assert not ok assert reason == "challenge_not_pending"