Spaces:
Sleeping
Sleeping
| 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): | |
| 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') | |
| 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() | |