pdf-quiz-generator/backend/tests/test_captcha.py

"""The Cap gate itself: what a missing secret, a bad solve and an outage each mean.

Real service and real contact route; only the call to Cap is replaced.
"""
import unittest
from unittest.mock import AsyncMock, patch

import httpx
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool

from app.config import settings
from app.database import get_db
from app.routers import contact
from app.services import captcha


class FakeResponse:
    def __init__(self, payload): self.payload = payload
    def json(self): return self.payload


def fake_cap(payload=None, error=None):
    """Stand in for httpx.AsyncClient and record what Cap was asked."""
    posts = []

    class FakeClient:
        def __init__(self, **kwargs): self.kwargs = kwargs
        async def __aenter__(self): return self
        async def __aexit__(self, *exc): return False
        async def post(self, url, json=None):
            posts.append({'url': url, 'data': json, 'client': self.kwargs})
            if error:
                raise error
            return FakeResponse(payload)

    return FakeClient, posts


class VerifyTests(unittest.IsolatedAsyncioTestCase):
    def configured(self, secret='synthetic-configured-key'):
        # Cap needs all three before it is switched on: a secret to verify
        # with, a site key to address, and somewhere to address it.
        for name, value in (('CAP_SECRET_KEY', secret),
                            ('CAP_SITE_KEY', 'synthetic-site' if secret else ''),
                            ('CAP_API_URL', 'http://cap:3000')):
            patcher = patch.object(settings, name, value)
            patcher.start()
            self.addCleanup(patcher.stop)

    async def test_an_unset_secret_skips_the_check_without_calling_out(self):
        self.configured('')
        client, posts = fake_cap({'success': False})
        with patch.object(httpx, 'AsyncClient', client):
            self.assertTrue(await captcha.verify('any-token', fail_open=False))
            await captcha.require_human(None, fail_open=False)
        self.assertFalse(captcha.configured())
        self.assertEqual(posts, [])

    async def test_a_solve_is_posted_to_cap_as_secret_and_response(self):
        self.configured()
        client, posts = fake_cap({'success': True})
        with patch.object(httpx, 'AsyncClient', client):
            self.assertTrue(await captcha.verify('solved-test-token', fail_open=False))
        self.assertEqual(len(posts), 1)
        self.assertEqual(posts[0]['url'], 'http://cap:3000/synthetic-site/siteverify')
        self.assertEqual(posts[0]['data'],
                         {'secret': 'synthetic-configured-key', 'response': 'solved-test-token'})
        self.assertEqual(posts[0]['client']['timeout'], 10)

    async def test_cap_saying_no_is_a_rejection(self):
        self.configured()
        client, _ = fake_cap({'success': False, 'error-codes': ['invalid-input-response']})
        with patch.object(httpx, 'AsyncClient', client):
            self.assertFalse(await captcha.verify('spent-test-token', fail_open=True))

    async def test_an_unreachable_cap_answers_whichever_way_the_caller_asked(self):
        self.configured()
        client, _ = fake_cap(error=httpx.ConnectError('synthetic network failure'))
        with patch.object(httpx, 'AsyncClient', client):
            self.assertTrue(await captcha.verify('any-token', fail_open=True))
            self.assertFalse(await captcha.verify('any-token', fail_open=False))

    async def test_require_human_reports_a_missing_and_a_bad_solve_differently(self):
        self.configured()
        with patch.object(captcha, 'verify', new_callable=AsyncMock, return_value=False):
            with self.assertRaises(HTTPException) as missing:
                await captcha.require_human('', fail_open=True)
            with self.assertRaises(HTTPException) as bad:
                await captcha.require_human('spent-test-token', fail_open=True)
        self.assertEqual((missing.exception.status_code, missing.exception.detail),
                         (400, 'Bot verification required'))
        self.assertEqual((bad.exception.status_code, bad.exception.detail),
                         (400, 'Bot verification failed — please try again'))


class ContactFormTests(unittest.TestCase):
    def setUp(self):
        self.engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
        self.db = Session(self.engine)
        # Created by raw DDL at startup in production, so there is no model to build it from.
        self.db.execute(text("""CREATE TABLE contact_submissions (
            id INTEGER PRIMARY KEY, name TEXT, email TEXT, type TEXT, message TEXT,
            read INTEGER DEFAULT 0, created_at TIMESTAMP)"""))
        self.db.commit()
        app = FastAPI()
        app.include_router(contact.router, prefix='/contact')
        app.dependency_overrides[get_db] = lambda: self.db
        self.client = TestClient(app)
        for target, value in [('CAP_SECRET_KEY', 'synthetic-configured-key'), ('ADMIN_EMAIL', '')]:
            patcher = patch.object(settings, target, value)
            patcher.start()
            self.addCleanup(patcher.stop)
        verifier = patch.object(captcha, 'verify', new_callable=AsyncMock, return_value=False)
        self.verify = verifier.start()
        self.addCleanup(verifier.stop)

    def tearDown(self):
        self.db.close()

    def submit(self, **overrides):
        return self.client.post('/contact', json={
            'name': 'Asker', 'email': 'asker@example.com', 'type': 'question',
            'message': 'Synthetic enquiry', **overrides})

    def stored(self):
        return self.db.execute(text('SELECT name, message FROM contact_submissions')).fetchall()

    def test_an_unsolved_form_is_turned_away_before_anything_is_written(self):
        self.assertEqual(self.submit().json()['detail'], 'Bot verification required')
        self.assertEqual(self.submit(captcha_token='spent-test-token').status_code, 400)
        self.assertEqual(self.stored(), [])
        self.verify.assert_awaited_once_with('spent-test-token', fail_open=False)

    def test_a_solved_form_is_stored(self):
        self.verify.return_value = True
        self.assertEqual(self.submit(captcha_token='solved-test-token').status_code, 200)
        self.assertEqual(self.stored(), [('Asker', 'Synthetic enquiry')])
        # Unreachable Cap must not become a way past the contact gate.
        self.verify.assert_awaited_once_with('solved-test-token', fail_open=False)

    def test_an_unconfigured_site_accepts_the_form_with_no_challenge(self):
        with patch.object(settings, 'CAP_SECRET_KEY', ''):
            self.assertEqual(self.submit().status_code, 200)
        self.assertEqual(self.stored(), [('Asker', 'Synthetic enquiry')])
        self.verify.assert_not_awaited()


if __name__ == '__main__':
    unittest.main()