import base64 import importlib import json import os from pathlib import Path import tempfile import unittest from datetime import datetime, timedelta, timezone from unittest.mock import patch from fastapi.testclient import TestClient from itsdangerous import TimestampSigner class PollTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.test_root = Path(__file__).resolve().parent cls.data = tempfile.TemporaryDirectory(prefix='.test-data-', dir=cls.test_root) environment = os.environ.copy() for key in ['DATA_REPO', 'DATA_TOKEN', 'SPACE_HOST', 'OAUTH_CLIENT_ID', 'OAUTH_CLIENT_SECRET']: environment.pop(key, None) environment.update(DATA_DIR=cls.data.name, SESSION_SECRET='test-only-secret', BASE_URL='http://testserver') with patch.dict(os.environ, environment, clear=True): cls.module = importlib.import_module('app') @classmethod def tearDownClass(cls): if not Path(cls.data.name).resolve().is_relative_to(cls.test_root): raise RuntimeError('Test data must remain inside the tests directory') cls.data.cleanup() def setUp(self): self.module.state.clear() self.module.state.update(polls={}) self.at = datetime(2026, 10, 28, 12, tzinfo=timezone.utc) self.clock = patch.object(self.module, 'utc_now', return_value=self.at).start() self.addCleanup(patch.stopall) self.owner = TestClient(self.module.app) self.guest = TestClient(self.module.app) self.addCleanup(self.owner.close) self.addCleanup(self.guest.close) session = base64.b64encode(json.dumps({'user': {'id': 'owner', 'name': 'Tester'}}).encode()) self.owner.cookies.set('session', TimestampSigner(self.module.SECRET).sign(session).decode()) self.headers = {'X-HuggingPolls': '1'} def create(self, **overrides): body = {'question': 'Favorite snack?', 'description': 'Choose for Friday.', 'options': ['Fruit', 'Cookies']} body.update(overrides) response = self.owner.post('/api/polls', json=body, headers=self.headers) self.assertEqual(response.status_code, 200, response.text) return response.json()['id'] def test_default_duration_and_public_summary(self): poll_id = self.create() poll = self.guest.get(f'/api/polls/{poll_id}').json() self.assertEqual(datetime.fromisoformat(poll['expires_at']), self.at + timedelta(days=7)) self.assertTrue(poll['listed']) self.assertFalse(poll['closed']) self.assertIsNone(poll['results']) summary = self.guest.get('/api/browse').json()['items'][0] self.assertEqual(summary['id'], poll_id) self.assertEqual(summary['option_count'], 2) self.assertTrue({'votes', 'owner_id', 'options', 'results'}.isdisjoint(summary)) def test_expiry_boundary_enforces_votes_and_prevents_reopen(self): poll_id = self.create(duration_hours=1) original = self.module.path.read_bytes() self.clock.return_value = self.at + timedelta(hours=1) - timedelta(microseconds=1) self.assertFalse(self.guest.get(f'/api/polls/{poll_id}').json()['closed']) self.clock.return_value = self.at + timedelta(hours=1) response = self.guest.post(f'/api/polls/{poll_id}/vote', json={'choice': 0}, headers=self.headers) self.assertEqual(response.status_code, 409) self.assertIn('expired', response.json()['detail']) poll = self.guest.get(f'/api/polls/{poll_id}').json() self.assertTrue(poll['expired']) self.assertTrue(poll['closed']) self.assertEqual(poll['results'], [0, 0]) self.assertEqual(self.guest.get('/api/browse').json()['total'], 0) self.assertEqual(self.guest.get('/api/browse?status=closed').json()['total'], 1) self.assertTrue(self.owner.get('/api/polls').json()[0]['expired']) response = self.owner.patch(f'/api/polls/{poll_id}', json={'closed': False}, headers=self.headers) self.assertEqual(response.status_code, 409) self.assertEqual(self.module.path.read_bytes(), original, 'Expiry reads must not write storage') self.assertEqual(self.module.state['polls'][poll_id]['votes'], []) def test_manual_close_and_reopen_before_expiry(self): poll_id = self.create(duration_hours=24) self.assertEqual(self.owner.patch(f'/api/polls/{poll_id}', json={'closed': True}, headers=self.headers).status_code, 200) self.assertEqual(self.guest.post(f'/api/polls/{poll_id}/vote', json={'choice': 1}, headers=self.headers).status_code, 409) self.assertEqual(self.guest.get(f'/api/polls/{poll_id}').json()['results'], [0, 0]) self.assertEqual(self.owner.patch(f'/api/polls/{poll_id}', json={'closed': False}, headers=self.headers).status_code, 200) self.assertEqual(self.guest.post(f'/api/polls/{poll_id}/vote', json={'choice': 1}, headers=self.headers).status_code, 200) self.assertEqual(self.guest.get(f'/api/polls/{poll_id}').json()['results'], [0, 1]) def test_unlisted_and_legacy_polls_do_not_leak_into_browse(self): hidden_id = self.create(listed=False, question='Unlisted topic') legacy_id = self.create(question='Old link-only poll') legacy = self.module.state['polls'][legacy_id] del legacy['listed'] del legacy['expires_at'] self.assertEqual(self.guest.get('/api/browse?status=all').json()['total'], 0) self.assertEqual(self.guest.get('/api/browse?q=unlisted&status=all').json()['total'], 0) self.assertEqual(self.guest.get(f'/api/polls/{hidden_id}').status_code, 200) self.clock.return_value = self.at + timedelta(days=365) self.assertFalse(self.guest.get(f'/api/polls/{legacy_id}').json()['expired']) self.assertEqual(self.owner.patch(f'/api/polls/{legacy_id}', json={'listed': True}, headers=self.headers).status_code, 200) self.assertEqual(self.guest.get('/api/browse').json()['items'][0]['id'], legacy_id) self.assertEqual(self.owner.patch(f'/api/polls/{legacy_id}', json={'listed': False}, headers=self.headers).status_code, 200) self.assertEqual(self.guest.get('/api/browse').json()['total'], 0) def test_search_status_and_sorting(self): first = self.create(question='A tea break', description='Community choice', duration_hours=24) self.clock.return_value = self.at + timedelta(minutes=1) popular = self.create(question='Coffee time', duration_hours=72) self.clock.return_value = self.at + timedelta(minutes=2) newest = self.create(question='Lunch menu', duration_hours=1) self.guest.post(f'/api/polls/{popular}/vote', json={'choice': 0}, headers=self.headers) self.assertEqual(self.guest.get('/api/browse?sort=newest').json()['items'][0]['id'], newest) self.assertEqual(self.guest.get('/api/browse?sort=popular').json()['items'][0]['id'], popular) self.assertEqual(self.guest.get('/api/browse?sort=ending').json()['items'][0]['id'], newest) self.assertEqual(self.guest.get('/api/browse?q=COMMUNITY').json()['items'][0]['id'], first) self.assertEqual(self.guest.get('/api/browse?q=Tester').json()['total'], 3) self.owner.patch(f'/api/polls/{first}', json={'closed': True}, headers=self.headers) self.assertEqual(self.guest.get('/api/browse').json()['total'], 2) self.assertEqual(self.guest.get('/api/browse?status=closed').json()['items'][0]['id'], first) self.assertEqual(self.guest.get('/api/browse?status=all').json()['total'], 3) def test_pagination_and_empty_results(self): for index in range(14): self.clock.return_value = self.at + timedelta(seconds=index) self.create(question=f'Community poll {index}') first = self.guest.get('/api/browse').json() second = self.guest.get('/api/browse?page=2').json() self.assertEqual(first['total'], 14) self.assertEqual(len(first['items']), 12) self.assertEqual(len(second['items']), 2) self.assertTrue({p['id'] for p in first['items']}.isdisjoint(p['id'] for p in second['items'])) self.assertEqual(self.guest.get('/api/browse?page=999').json()['page'], 2) empty = self.guest.get('/api/browse?q=does-not-exist&page=2').json() self.assertEqual(empty['page'], 1) self.assertEqual(empty['items'], []) def test_invalid_requests_and_owner_permissions(self): for duration in [0, -1, 721]: response = self.owner.post('/api/polls', json={'question': 'Q', 'options': ['A', 'B'], 'duration_hours': duration}, headers=self.headers) self.assertEqual(response.status_code, 422) for query in ['status=invalid', 'sort=invalid', 'page=0', 'q=' + 'x' * 241]: self.assertEqual(self.guest.get('/api/browse?' + query).status_code, 422) poll_id = self.create() self.assertEqual(self.guest.patch(f'/api/polls/{poll_id}', json={'listed': False}, headers=self.headers).status_code, 401) with TestClient(self.module.app) as other: session = base64.b64encode(json.dumps({'user': {'id': 'other', 'name': 'Another user'}}).encode()) other.cookies.set('session', TimestampSigner(self.module.SECRET).sign(session).decode()) self.assertEqual(other.patch(f'/api/polls/{poll_id}', json={'listed': False}, headers=self.headers).status_code, 403) self.assertEqual(self.owner.patch(f'/api/polls/{poll_id}', json={}, headers=self.headers).status_code, 422) if __name__ == '__main__': unittest.main()