diff --git a/v3/README.md b/v3/README.md index cb82475..0ba97c7 100644 --- a/v3/README.md +++ b/v3/README.md @@ -216,6 +216,16 @@ present in the person database. Management of an existing election is restricted to its `owner_pid`. The `authz` field is reserved for a future delegation policy and currently grants no additional management access. +### Voter Lists + +From an election's management page, use **Manage voters** to view its eligible voters and the number of issues assigned to each. While the election is editable, paste ASF IDs separated by spaces, commas, or newlines and select the issues they may vote on. IDs must exactly match existing person-directory records; an unknown ID or invalid issue rejects the entire batch and keeps the input for correction. Repeated IDs and existing grants are accepted without duplicates, and adding voters preserves their other grants. + +Use **Edit eligibility** beside a voter to change their assigned issues, or **Remove voter** to remove their eligibility from every issue in this election. Removal leaves the global person record and other elections untouched. The web editor requires at least one selected issue when saving; removal is a separate explicit action. + +Eligibility belongs to individual issues, rather than an independent election-wide roster. Create issues before assigning voters. A newly added issue starts with no eligible voters and never automatically inherits another issue's list. Review its eligibility explicitly before opening. Once the election opens, the voter list and each person's issue assignments remain visible to the owner but cannot be changed, including after closure. These pages do not display participation or ballots. + +### Mutation and Storage Contracts + All `/do-*` mutations require POST and a session CSRF token. HTML forms send `csrf_token`; JSON requests send `X-CSRFToken`. Opening and closing elections via GET is not supported. State conflicts return HTTP 409, invalid ballots diff --git a/v3/TODO.md b/v3/TODO.md index 16edf10..f297a23 100644 --- a/v3/TODO.md +++ b/v3/TODO.md @@ -2,7 +2,7 @@ ## Community Self-Service -- Add voter roster management to the web UI, including ASF ID validation and a clear policy for issues added after a roster is configured. +- Add group-based voter selection once person-directory group membership and refresh semantics are defined. The web UI currently accepts explicit ASF IDs and issue assignments. - Add STV creation and editing for candidate labels, names, ASF IDs, and seats. The current web issue creation flow supports YNA only. - Define delegated management semantics for `authz` before granting access beyond the election owner. - Expose closed-election tallying through an owner-authorized workflow with execution status, failure reporting, stored results, and controlled downloads. Separate summary results from participation lists and anonymized raw ballots. diff --git a/v3/queries.yaml b/v3/queries.yaml index 301ca9c..7a168a4 100644 --- a/v3/queries.yaml +++ b/v3/queries.yaml @@ -168,6 +168,15 @@ election: WHERE i.eid = ? ORDER BY m.pid + q_voters: | + SELECT m.pid, p.name, COUNT(*) AS issue_count + FROM mayvote m + JOIN issue i ON m.iid = i.iid + JOIN person p ON m.pid = p.pid + WHERE i.eid = ? + GROUP BY m.pid, p.name + ORDER BY m.pid + q_closed_election_ids: SELECT eid FROM election WHERE closed = 1 q_open_election_ids: SELECT eid FROM election WHERE salt IS NOT NULL AND (closed IS NULL OR closed = 0) diff --git a/v3/server/pages.py b/v3/server/pages.py index d6d044a..bea774c 100644 --- a/v3/server/pages.py +++ b/v3/server/pages.py @@ -26,6 +26,7 @@ import logging import random import secrets +from urllib.parse import quote from contextlib import closing from easydict import EasyDict as edict @@ -58,6 +59,8 @@ T_BAD_EID = APP.load_template(TEMPLATES / 'e_bad_eid.ezt') T_BAD_IID = APP.load_template(TEMPLATES / 'e_bad_iid.ezt') T_BAD_PID = APP.load_template(TEMPLATES / 'e_bad_pid.ezt') +T_MANAGE_VOTERS = APP.load_template(TEMPLATES / 'manage-voters.ezt') +T_MANAGE_VOTER = APP.load_template(TEMPLATES / 'manage-voter.ezt') def rewrite_description(issue): @@ -258,6 +261,9 @@ async def loader(eid, *args, **kwargs): except steve.election.ElectionBadState: quart.abort(409, 'The election is not in the required state') + except steve.election.VoterNotFound: + quart.abort(404, 'This person is not on the election voter list') + return loader @@ -410,6 +416,134 @@ async def manage_page(election): return result +async def voter_management_data(election, *, pids='', selected=None, error=None): + result = await basic_info() + result.title = 'Manage Voters' + result.eid = election.eid + result.election = election.get_metadata() + result.issues = election.list_issues() + result.voters = election.list_voters() + for voter in result.voters: + voter.name = voter.name or voter.pid + voter.url_pid = quote(voter.pid, safe='') + result.voter_count = len(result.voters) + result.pids = pids + result.error = error + for issue in result.issues: + issue.selected = ezt.boolean(selected is None or issue.iid in selected) + return result + + +@APP.get('/manage-voters/') +@asfquart.auth.require({R.committer}) +@load_election +@require_election_owner +@APP.use_template(TEMPLATES / 'manage-voters.ezt') +async def manage_voters_page(election): + return await voter_management_data(election) + + +@APP.post('/do-add-voters/') +@asfquart.auth.require({R.committer}) +@load_election +@require_election_owner +async def do_add_voters_endpoint(election): + form = await quart.request.form + raw_pids = form.get('pids', '') + pids = list(dict.fromkeys(raw_pids.replace(',', ' ').split())) + iids = form.getlist('issue') + try: + election.add_voters(pids, iids) + except ( + ValueError, + steve.persondb.PersonNotFound, + steve.election.IssueNotFound, + ) as ex: + if isinstance(ex, steve.persondb.PersonNotFound): + error = f'ASF ID "{ex.pid}" is not in the voter directory. No voters were added.' + elif isinstance(ex, steve.election.IssueNotFound): + error = ( + 'An issue is no longer available. Review your selection and try again.' + ) + else: + error = str(ex) + data = await voter_management_data( + election, pids=raw_pids, selected=iids, error=error + ) + return quart.Response( + asfquart.utils.render(T_MANAGE_VOTERS, data), + status=400, + mimetype='text/html', + ) + await flash_success(f'Updated voter eligibility for {len(pids)} ASF IDs.') + return quart.redirect(f'/manage-voters/{election.eid}', code=303) + + +async def voter_editor_data(election, pid, *, selected=None, error=None): + eligible = election.get_voter_issues(pid) + result = await basic_info() + result.title = 'Edit Voter Eligibility' + result.eid = election.eid + result.election = election.get_metadata() + result.pid = pid + result.url_pid = quote(pid, safe='') + result.error = error + result.issues = election.list_issues() + for issue in result.issues: + issue.selected = ezt.boolean( + issue.iid in (eligible if selected is None else selected) + ) + return result + + +@APP.get('/manage-voter//') +@asfquart.auth.require({R.committer}) +@load_election +@require_election_owner +@APP.use_template(TEMPLATES / 'manage-voter.ezt') +async def manage_voter_page(election, pid): + return await voter_editor_data(election, pid) + + +@APP.post('/do-set-voter//') +@asfquart.auth.require({R.committer}) +@load_election +@require_election_owner +async def do_set_voter_endpoint(election, pid): + if not election.is_editable(): + quart.abort(409, 'The voter list is frozen') + form = await quart.request.form + iids = form.getlist('issue') + if not iids: + error = ( + 'Select at least one issue, or use Remove voter to remove all eligibility.' + ) + else: + try: + election.set_voter_issues(pid, iids) + except steve.election.IssueNotFound: + error = ( + 'An issue is no longer available. Review your selection and try again.' + ) + else: + await flash_success(f'Updated eligibility for {pid}.') + return quart.redirect(f'/manage-voters/{election.eid}', code=303) + data = await voter_editor_data(election, pid, selected=iids, error=error) + return quart.Response( + asfquart.utils.render(T_MANAGE_VOTER, data), status=400, mimetype='text/html' + ) + + +@APP.post('/do-remove-voter//') +@asfquart.auth.require({R.committer}) +@load_election +@require_election_owner +async def do_remove_voter_endpoint(election, pid): + election.set_voter_issues(pid, []) + await flash_success(f'Removed {pid} from this election.') + return quart.redirect(f'/manage-voters/{election.eid}', code=303) + + @APP.get('/manage-stv//') @asfquart.auth.require({R.committer}) ### need general solution @load_election_issue diff --git a/v3/server/templates/manage-voter.ezt b/v3/server/templates/manage-voter.ezt new file mode 100644 index 0000000..e914f35 --- /dev/null +++ b/v3/server/templates/manage-voter.ezt @@ -0,0 +1,40 @@ +[include "header.ezt"] +
+ Back to voter list +

[title]

+

[pid] · [election.title]

+ [include "flashes.ezt"] + [if-any error][end] + [is election.state "editable"] +
+ +
+ Eligible issues + [for issues] + + [end] +
+ +
+
+

Remove voter

+

This removes [pid] from every issue in this election. Their directory record and eligibility in other elections are kept.

+
+ + +
+
+ [else] +
This election has opened. Its voter list is frozen and cannot be changed.
+

Eligible issues

+
    + [for issues] + [if-any issues.selected]
  • [issues.title]
    • [end] + [end] +
+ [end] +
+[include "footer.ezt"] diff --git a/v3/server/templates/manage-voters.ezt b/v3/server/templates/manage-voters.ezt new file mode 100644 index 0000000..7646c89 --- /dev/null +++ b/v3/server/templates/manage-voters.ezt @@ -0,0 +1,68 @@ +[include "header.ezt"] +
+ Back to election +

[title]

+

[election.title]

+ [include "flashes.ezt"] + [if-any error][end] + [is election.state "editable"] +

Eligibility is assigned per issue. New issues start with no voters; adding an issue does not extend existing eligibility.

+ [else] +
This election has opened. Its voter list is frozen and cannot be changed.
+ [end] +
+
+

Eligible voters ([voter_count])

+ [if-any voters] +
+ + + + [for voters] + + + + + + + [end] + +
ASF IDNameEligible issues
[voters.pid][voters.name][voters.issue_count][is election.state "editable"]Edit eligibility[else]View eligibility[end]
+
+ [else] +

No voters are eligible for this election yet.

+ [end] +
+ [is election.state "editable"] +
+
+
+

Add voters

+ [if-any issues] +
+ + + +

Separate IDs with spaces, commas, or new lines. IDs must match the voter directory exactly. Unknown IDs reject the whole batch.

+
+ Eligible issues + [for issues] + + [end] +
+

Existing eligibility is preserved. Repeating an ID does not create a duplicate.

+ +
+ [else] +

Add an issue on the election page before assigning voters.

+ [end] +
+
+
+ [end] +
+
+[include "footer.ezt"] diff --git a/v3/server/templates/manage.ezt b/v3/server/templates/manage.ezt index 50ead2a..8a0fe34 100644 --- a/v3/server/templates/manage.ezt +++ b/v3/server/templates/manage.ezt @@ -7,6 +7,13 @@

[e_title]

+
+ Manage voters + [is e_state "editable"] +

New issues start with no eligible voters. Assign their voters before opening the election.

+ [end] +
+ diff --git a/v3/steve/election.py b/v3/steve/election.py index 9240f19..822d56c 100644 --- a/v3/steve/election.py +++ b/v3/steve/election.py @@ -26,6 +26,7 @@ from . import crypto from . import vtypes +from .persondb import PersonNotFound _LOGGER = logging.getLogger(__name__) @@ -279,6 +280,63 @@ def add_voter(self, pid: str, iid: str | None = None) -> None: else: self.c_add_mayvote_all.perform(pid, self.eid) + def list_voters(self): + """List this election's eligible people without disclosing participation.""" + self.q_voters.perform(self.eid) + return list(self.q_voters.fetchall()) + + def get_voter_issues(self, pid: str) -> set[str]: + """Return eligible issue IDs, or raise VoterNotFound for a nonmember.""" + self.q_find_issues.perform(pid, self.eid) + iids = {row.iid for row in self.q_find_issues.fetchall()} + if not iids: + raise VoterNotFound(pid) + return iids + + def add_voters(self, pids: list[str], iids: list[str]) -> None: + """Atomically grant existing people access to selected current issues. + + Existing grants are preserved. Future issues never inherit these grants. + """ + with self.transaction(): + self._all_metadata(self.S_EDITABLE) + if not pids or not iids: + raise ValueError('Select at least one voter and one issue') + pids = list(dict.fromkeys(pids)) + iids = list(dict.fromkeys(iids)) + for iid in iids: + self.get_issue(iid) + for pid in pids: + if ( + self.db.conn.execute( + 'SELECT 1 FROM person WHERE pid = ?', (pid,) + ).fetchone() + is None + ): + raise PersonNotFound(pid) + self.db.conn.executemany( + 'INSERT INTO mayvote (pid, iid) VALUES (?, ?) ' + 'ON CONFLICT (pid, iid) DO NOTHING', + ((pid, iid) for pid in pids for iid in iids), + ) + + def set_voter_issues(self, pid: str, iids: list[str]) -> None: + """Replace one voter's eligibility; an empty selection removes the voter.""" + with self.transaction(): + self._all_metadata(self.S_EDITABLE) + current = self.get_voter_issues(pid) + selected = set(iids) + for iid in selected: + self.get_issue(iid) + self.db.conn.executemany( + 'DELETE FROM mayvote WHERE pid = ? AND iid = ?', + ((pid, iid) for iid in current - selected), + ) + self.db.conn.executemany( + 'INSERT INTO mayvote (pid, iid) VALUES (?, ?)', + ((pid, iid) for iid in selected - current), + ) + def validate_vote(self, pid: str, iid: str, votestring: str): """Validate eligibility and ballot syntax without recording a vote.""" if not isinstance(votestring, str): @@ -606,3 +664,9 @@ def __init__(self, iid): def __str__(self): return f'Issue[I:{self.iid}] not found' + + +class VoterNotFound(Exception): + def __init__(self, pid): + self.pid = pid + super().__init__(f'Voter[P:{pid}] not found in this election') diff --git a/v3/tests/test_pages.py b/v3/tests/test_pages.py index 7ae2952..dc6d2e7 100644 --- a/v3/tests/test_pages.py +++ b/v3/tests/test_pages.py @@ -22,8 +22,10 @@ import unittest from html.parser import HTMLParser from unittest.mock import patch +from urllib.parse import urlencode import asfquart +from werkzeug.datastructures import MultiDict from support import DatabaseTestCase, ROOT @@ -33,19 +35,31 @@ def __init__(self, html): super().__init__() self.forms = {} self.current = None + self.textarea = None self.feed(html) def handle_starttag(self, tag, attrs): attrs = dict(attrs) if tag == 'form': - self.current = {'method': attrs.get('method', 'GET'), 'fields': {}} + self.current = {'method': attrs.get('method', 'GET'), 'fields': MultiDict()} self.forms[attrs.get('action', '')] = self.current elif tag == 'input' and self.current is not None and attrs.get('name'): - self.current['fields'][attrs['name']] = attrs.get('value', '') + if attrs.get('type') != 'checkbox' or 'checked' in attrs: + self.current['fields'].add(attrs['name'], attrs.get('value', '')) + elif tag == 'textarea' and self.current is not None: + self.textarea = attrs.get('name') + if self.textarea: + self.current['fields'][self.textarea] = '' + + def handle_data(self, data): + if self.textarea and self.current is not None: + self.current['fields'][self.textarea] += data def handle_endtag(self, tag): if tag == 'form': self.current = None + elif tag == 'textarea': + self.textarea = None class PageTests(DatabaseTestCase, unittest.IsolatedAsyncioTestCase): @@ -54,7 +68,11 @@ def setUpClass(cls): cls.app_directory = tempfile.TemporaryDirectory(prefix='steve-http-test-') cls.addClassCleanup(cls.app_directory.cleanup) cls.app = asfquart.construct( - 'steve-tests', app_dir=cls.app_directory.name, token_file=None, oauth=False + 'steve-tests', + app_dir=cls.app_directory.name, + token_file=None, + oauth=False, + static_folder=None, ) cls.app.cfg.db = 'unused.db' cls.app.config['TESTING'] = True @@ -94,8 +112,18 @@ async def post(self, path, *, form=None, json=None): return await self.client.post( path, json=json, headers={'X-CSRFToken': self.csrf}, scheme='https' ) - return await self.client.post( - path, form={'csrf_token': self.csrf, **(form or {})}, scheme='https' + data = MultiDict(form or {}) + data.setdefault('csrf_token', self.csrf) + return await self.submit(path, {'method': 'POST', 'fields': data}) + + async def submit(self, path, form): + # Quart's form= helper flattens repeated keys; encode browser checkboxes. + return await self.client.open( + path, + method=form['method'], + data=urlencode(list(form['fields'].items(multi=True))), + headers={'Content-Type': 'application/x-www-form-urlencoded'}, + scheme='https', ) async def test_create_form_and_owner_lifecycle_work_with_csrf(self): @@ -267,3 +295,177 @@ async def test_invalid_stv_submission_is_rejected_before_any_issue_is_recorded( ) self.assertEqual(response.status_code, 400) self.assertFalse(any(election.has_voted_upon('alice').values())) + + async def test_voter_forms_cover_bulk_add_edit_remove_and_voting(self): + election = self.create_election() + first = election.add_issue('Approve plan', '', 'yna', None) + second = election.add_issue('Approve budget', '', 'yna', None) + response = await self.client.get(f'/manage/{election.eid}', scheme='https') + self.assertIn( + f'/manage-voters/{election.eid}', await response.get_data(as_text=True) + ) + response = await self.client.get( + f'/manage-voters/{election.eid}', scheme='https' + ) + action = f'/do-add-voters/{election.eid}' + addition = Forms(await response.get_data(as_text=True)).forms[action] + self.assertEqual(set(addition['fields'].getlist('issue')), {first, second}) + addition['fields']['pids'] = 'bob, carlos\nbob' + response = await self.submit(action, addition) + self.assertEqual(response.status_code, 303) + response = await self.client.get( + f'/manage-voter/{election.eid}/bob', scheme='https' + ) + forms = Forms(await response.get_data(as_text=True)) + action = f'/do-set-voter/{election.eid}/bob' + editing = forms.forms[action] + self.assertEqual(set(editing['fields'].getlist('issue')), {first, second}) + editing['fields'].setlist('issue', [first]) + response = await self.submit(action, editing) + self.assertEqual(response.status_code, 303) + response = await self.client.get( + f'/manage-voter/{election.eid}/carlos', scheme='https' + ) + action = f'/do-remove-voter/{election.eid}/carlos' + removal = Forms(await response.get_data(as_text=True)).forms[action] + response = await self.submit(action, removal) + self.assertEqual(response.status_code, 303) + self.assertEqual([v.pid for v in election.list_voters()], ['bob']) + response = await self.post(f'/do-open/{election.eid}') + self.assertEqual(response.status_code, 303) + await self.login('bob') + response = await self.client.get(f'/vote-on/{election.eid}', scheme='https') + html = await response.get_data(as_text=True) + self.assertIn('Approve plan', html) + self.assertNotIn('Approve budget', html) + response = await self.post( + f'/do-vote/{election.eid}', form={f'vote-{first}': 'y'} + ) + self.assertEqual(response.status_code, 303) + election.close() + self.assertFalse(election.is_tampered(self.pdb)) + self.assertEqual(election.tally_issue(first)[0][1]['y'], 1) + + async def test_voter_management_rejects_nonowners_csrf_and_get_mutations(self): + election = self.create_election() + iid = election.add_issue('Approval', '', 'yna', None) + election.add_voter('carlos', iid) + paths = ( + f'/do-add-voters/{election.eid}', + f'/do-set-voter/{election.eid}/carlos', + f'/do-remove-voter/{election.eid}/carlos', + ) + for path in paths: + response = await self.client.get(path, scheme='https') + self.assertEqual(response.status_code, 405) + response = await self.client.post( + path, form={'pids': 'bob', 'issue': iid}, scheme='https' + ) + self.assertEqual(response.status_code, 400) + await self.login('bob') + for path in ( + f'/manage-voters/{election.eid}', + f'/manage-voter/{election.eid}/carlos', + ): + response = await self.client.get(path, scheme='https') + self.assertEqual(response.status_code, 403) + for path in paths: + response = await self.post(path, form={'pids': 'bob', 'issue': iid}) + self.assertEqual(response.status_code, 403) + self.assertEqual([v.pid for v in election.list_voters()], ['carlos']) + + async def test_invalid_voter_batch_keeps_input_and_has_no_partial_effect(self): + election = self.create_election() + first = election.add_issue('First', '', 'yna', None) + second = election.add_issue('Second', '', 'yna', None) + action = f'/do-add-voters/{election.eid}' + raw_pids = 'bob\n' + response = await self.post(action, form={'pids': raw_pids, 'issue': second}) + self.assertEqual(response.status_code, 400) + html = await response.get_data(as_text=True) + self.assertNotIn('', html) + form = Forms(html).forms[action]['fields'] + self.assertEqual(form['pids'], raw_pids) + self.assertEqual(form.getlist('issue'), [second]) + self.assertEqual(election.list_voters(), []) + form['pids'] = 'bob' + response = await self.post(action, form=form) + self.assertEqual(response.status_code, 303) + self.assertEqual(election.get_voter_issues('bob'), {second}) + for fields in ({'pids': 'carlos'}, {'pids': '', 'issue': first}): + response = await self.post(action, form=fields) + self.assertEqual(response.status_code, 400) + self.assertEqual([v.pid for v in election.list_voters()], ['bob']) + + async def test_voter_scope_checks_and_stale_forms_do_not_change_other_elections( + self, + ): + election = self.create_election() + iid = election.add_issue('Owned', '', 'yna', None) + other = self.create_election('bob') + foreign = other.add_issue('Foreign', '', 'yna', None) + other.add_voter('carlos', foreign) + for path in ( + f'/manage-voter/{election.eid}/carlos', + f'/manage-voter/{election.eid}/missing', + ): + response = await self.client.get(path, scheme='https') + self.assertEqual(response.status_code, 404) + response = await self.post( + f'/do-set-voter/{election.eid}/carlos', form={'issue': iid} + ) + self.assertEqual(response.status_code, 404) + response = await self.post(f'/do-remove-voter/{election.eid}/carlos') + self.assertEqual(response.status_code, 404) + response = await self.post( + f'/do-add-voters/{election.eid}', form={'pids': 'carlos', 'issue': foreign} + ) + self.assertEqual(response.status_code, 400) + election.add_voter('carlos', iid) + response = await self.post( + f'/do-set-voter/{election.eid}/carlos', form={'issue': foreign} + ) + self.assertEqual(response.status_code, 400) + self.assertEqual(election.get_voter_issues('carlos'), {iid}) + response = await self.post(f'/do-set-voter/{election.eid}/carlos') + self.assertEqual(response.status_code, 400) + self.assertEqual(election.get_voter_issues('carlos'), {iid}) + election.delete_issue(iid) + response = await self.post( + f'/do-add-voters/{election.eid}', form={'pids': 'bob', 'issue': iid} + ) + self.assertEqual(response.status_code, 400) + self.assertEqual(other.get_voter_issues('carlos'), {foreign}) + + async def test_open_and_closed_rosters_are_read_only(self): + election = self.create_election() + iid = election.add_issue('Approval', '', 'yna', None) + election.add_voter('bob', iid) + election.open(self.pdb) + for state in ('open', 'closed'): + if state == 'closed': + election.close() + for path in ( + f'/manage-voters/{election.eid}', + f'/manage-voter/{election.eid}/bob', + ): + response = await self.client.get(path, scheme='https') + self.assertEqual(response.status_code, 200) + self.assertEqual(Forms(await response.get_data(as_text=True)).forms, {}) + for path in ( + f'/do-add-voters/{election.eid}', + f'/do-set-voter/{election.eid}/bob', + f'/do-remove-voter/{election.eid}/bob', + ): + response = await self.post(path, form={'pids': 'carlos', 'issue': iid}) + self.assertEqual(response.status_code, 409) + self.assertEqual(election.get_voter_issues('bob'), {iid}) + self.assertFalse(election.is_tampered(self.pdb)) + + async def test_roster_without_issues_guides_owner_to_create_an_issue(self): + election = self.create_election() + response = await self.client.get( + f'/manage-voters/{election.eid}', scheme='https' + ) + self.assertEqual(response.status_code, 200) + self.assertEqual(Forms(await response.get_data(as_text=True)).forms, {}) diff --git a/v3/tests/test_voters.py b/v3/tests/test_voters.py new file mode 100644 index 0000000..ae350b1 --- /dev/null +++ b/v3/tests/test_voters.py @@ -0,0 +1,118 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import sqlite3 + +from steve.election import ElectionBadState, IssueNotFound, VoterNotFound +from steve.persondb import PersonNotFound +from support import DatabaseTestCase + + +class VoterTests(DatabaseTestCase): + def setUp(self): + super().setUp() + self.election = self.create_election() + self.first = self.election.add_issue('First issue', '', 'yna', None) + self.second = self.election.add_issue('Second issue', '', 'yna', None) + + def test_batch_add_preserves_existing_grants_and_deduplicates(self): + self.election.add_voter('bob', self.first) + self.election.add_voters(['bob', 'carlos', 'bob'], [self.second, self.second]) + self.election.add_voters(['bob', 'carlos'], [self.second]) + self.assertEqual( + self.election.get_voter_issues('bob'), {self.first, self.second} + ) + self.assertEqual(self.election.get_voter_issues('carlos'), {self.second}) + self.assertEqual( + [(v.pid, v.issue_count) for v in self.election.list_voters()], + [('bob', 2), ('carlos', 1)], + ) + self.assertEqual(self.election.get_voters_for_email()[0].pid, 'bob') + + def test_batch_rejects_unknown_people_and_foreign_issues_without_partial_grants( + self, + ): + other = self.create_election('bob') + foreign = other.add_issue('Other election', '', 'yna', None) + for pids, iids, error in ( + (['bob', 'missing'], [self.first], PersonNotFound), + (['Bob'], [self.first], PersonNotFound), + (['bob'], [self.first, foreign], IssueNotFound), + (['bob'], [], ValueError), + ([], [self.first], ValueError), + ): + with self.subTest(pids=pids, iids=iids), self.assertRaises(error): + self.election.add_voters(pids, iids) + self.assertEqual(self.election.list_voters(), []) + + def test_batch_database_failure_rolls_back_inside_caller_transaction(self): + self.election.db.conn.executescript(""" + CREATE TRIGGER reject_second_grant BEFORE INSERT ON mayvote + WHEN (SELECT COUNT(*) FROM mayvote) > 0 + BEGIN SELECT RAISE(ABORT, 'storage failure'); END; + """) + with self.election.transaction(): + with self.assertRaises(sqlite3.IntegrityError): + self.election.add_voters(['bob'], [self.first, self.second]) + self.assertEqual(self.election.list_voters(), []) + + def test_replacement_and_removal_are_scoped_to_one_voter_and_election(self): + other = self.create_election('bob') + foreign = other.add_issue('Other election', '', 'yna', None) + other.add_voter('bob', foreign) + self.election.add_voters(['bob', 'carlos'], [self.first]) + with self.assertRaises(IssueNotFound): + self.election.set_voter_issues('bob', [self.second, foreign]) + self.assertEqual(self.election.get_voter_issues('bob'), {self.first}) + self.election.set_voter_issues('bob', [self.second]) + self.assertEqual(self.election.get_voter_issues('bob'), {self.second}) + self.election.set_voter_issues('bob', []) + with self.assertRaises(VoterNotFound): + self.election.set_voter_issues('bob', [self.first]) + self.assertEqual(self.election.get_voter_issues('carlos'), {self.first}) + self.assertEqual(other.get_voter_issues('bob'), {foreign}) + self.assertEqual(self.pdb.get_person('bob')[0], 'Bob') + + def test_replacement_storage_failure_restores_previous_grants(self): + self.election.add_voter('bob', self.first) + self.election.db.conn.executescript(""" + CREATE TRIGGER reject_grant BEFORE INSERT ON mayvote + BEGIN SELECT RAISE(ABORT, 'storage failure'); END; + """) + with self.assertRaises(sqlite3.IntegrityError): + self.election.set_voter_issues('bob', [self.second]) + self.assertEqual(self.election.get_voter_issues('bob'), {self.first}) + + def test_new_issues_start_empty_and_opening_freezes_all_grants(self): + self.election.add_voters(['bob'], [self.first, self.second]) + new = self.election.add_issue('Later issue', '', 'yna', None) + self.assertNotIn(new, self.election.get_voter_issues('bob')) + self.election.open(self.pdb) + for state in ('open', 'closed'): + if state == 'closed': + self.election.close() + for change in ( + lambda: self.election.add_voters(['carlos'], [new]), + lambda: self.election.set_voter_issues('bob', [new]), + lambda: self.election.set_voter_issues('bob', []), + ): + with self.subTest(state=state), self.assertRaises(ElectionBadState): + change() + self.assertEqual( + self.election.get_voter_issues('bob'), {self.first, self.second} + ) + self.assertFalse(self.election.is_tampered(self.pdb))