"""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()