diff --git a/README.md b/README.md index 02c1238..dadc1f8 100644 --- a/README.md +++ b/README.md @@ -5,20 +5,17 @@ A Python 3.11 module for votings on the c-base space-station. ## Usage ```python -from voting.voting import QuorumKind, Quorum, Vote, Voting +from voting.voting import Quorum, QuorumKind, Vote, Voting, Result # initializing a new voting voting = Voting( - title="TITLE", - quorum=Quorum(), + title="EXAMPLE", + quorum=Quorum(), # equals Quorum(kind=QuorumKind.NONE, value=None) #quorum=Quorum(kind=QuorumKind.ABSOLUTE, value=42), #quorum=Quorum(kind=QuorumKind.PERCENT, value=42.0), voters=['alice', 'bob'], ) -# starting the voting -voting.start() - # placing a vote voting.vote('alice', Vote.NO) @@ -26,6 +23,6 @@ voting.vote('alice', Vote.NO) voting.vote('alice', Vote.YES) # obtaining the result -res = voting.result() -print(res) +res = Result(voting) +print(res.result()) ``` \ No newline at end of file diff --git a/TODO.md b/TODO.md index 7763316..ce9d2f9 100644 --- a/TODO.md +++ b/TODO.md @@ -1,8 +1,5 @@ # ToDos -- refactor `Voting.result()` to `Voting.state()` that returns a data dictionary -- refactor `Result` to be an abstract class that can handle `Voting.state` allowing users to provide their own custom `Result` -- add `DefaultResult` abstraction of `Result` - add support for pickling/unpickling `Voting` - add `click` and provide a user-friendly CLI - add `fastapi` and implement a builtin JSON/REST API server diff --git a/test/test_quorum.py b/test/test_quorum.py new file mode 100644 index 0000000..43cd8c8 --- /dev/null +++ b/test/test_quorum.py @@ -0,0 +1,27 @@ +import unittest + +from voting import QuorumKind, Quorum + +voters = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j'] + +class TestQuorumSanity(unittest.TestCase): + def test_invalid_absolute_values(self): + self.assertRaises(ValueError, Quorum, kind=QuorumKind.ABSOLUTE, value=0) + self.assertRaises(ValueError, Quorum, kind=QuorumKind.ABSOLUTE, value=-1) + + def test_valid_absolute_values(self): + Quorum(kind=QuorumKind.ABSOLUTE, value=1) + + def test_invalid_percent_values(self): + self.assertRaises(ValueError, Quorum, kind=QuorumKind.PERCENT, value=0) + self.assertRaises(ValueError, Quorum, kind=QuorumKind.PERCENT, value=-1) + self.assertRaises(ValueError, Quorum, kind=QuorumKind.PERCENT, value=100.1) + + def test_valid_percent_values(self): + Quorum(kind=QuorumKind.PERCENT, value=0.1) + Quorum(kind=QuorumKind.PERCENT, value=100) + + +if __name__ == '__main__': + unittest.main() + diff --git a/test/test_voting.py b/test/test_voting.py new file mode 100644 index 0000000..0162cc1 --- /dev/null +++ b/test/test_voting.py @@ -0,0 +1,72 @@ +import unittest + +from arrow import get as arrow_get, Arrow +from voting import Quorum, Vote, Voting, Result + +voters = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j'] + +class TestVoter(unittest.TestCase): + def setUp(self) -> None: + self.voting = Voting(title="Test", quorum=Quorum(), voters=voters) + + def test_fail_with_no_voters(self): + self.assertRaises(RuntimeError, Voting, title="Test", quorum=Quorum(), voters=[],) + + def test_one_voter(self): + Voting(title="Test", quorum=Quorum(), voters=['a']) + + def test_vote_before_end(self): + self.voting.vote(voter='a', vote=Vote.YES) + + def test_vote_after_end(self): + self.voting.days = 0 + self.assertRaises(RuntimeError, self.voting.vote, voter='a', vote=Vote.YES) + + def test_unknown_voter(self): + self.assertRaises(ValueError, self.voting.vote, voter='z', vote=Vote.YES) + + def test_known_voter(self): + self.voting.vote('a', Vote.YES) + + +class TestResult(unittest.TestCase): + def setUp(self) -> None: + self.timestamp = str(Arrow.utcnow()) + self.voting = Voting(title="Test", quorum=Quorum(), voters=voters) + self.voting.start = self.timestamp + self.result = Result(self.voting) + + def test_result_data(self): + res = self.result.result() + assert res['state'] == "LAUFEND" + assert res['voting']['title'] == 'Test' + assert res['voting']['quorum']['kind'] == 'NONE' + assert res['voting']['quorum']['value'] == 0 + assert res['voting']['voters'] == voters + assert res['voting']['start'] == self.timestamp + assert res['voting']['days'] == 7 + assert res['voting']['votes'] == {} + assert res['votes'] == {'YES': 0, 'NO': 0, 'ABSTENTION': 0} + assert res['quorum_reached'] == True + + def test_result_data_with_votes(self): + self.voting.vote(voter='a', vote=Vote.YES) + res = self.result.result() + assert res['voting']['votes'] == {'a': 'YES'} + assert res['votes'] == {'YES': 1, 'NO': 0, 'ABSTENTION': 0} + + def test_result_data_with_outcome_accept(self): + self.voting.vote(voter='a', vote=Vote.YES) + self.voting.days = 0 + res = self.result.result() + assert res['state'] == 'ANGENOMMEN' + + def test_result_data_with_outcome_declined(self): + self.voting.vote(voter='a', vote=Vote.NO) + self.voting.days = 0 + res = self.result.result() + assert res['state'] == 'ABGELEHNT' + + +if __name__ == '__main__': + unittest.main() diff --git a/voting/__init__.py b/voting/__init__.py new file mode 100644 index 0000000..740ac76 --- /dev/null +++ b/voting/__init__.py @@ -0,0 +1,2 @@ +from .quorum import Quorum, QuorumKind +from .voting import Vote, Voting, Result diff --git a/voting/quorum.py b/voting/quorum.py new file mode 100644 index 0000000..00a5408 --- /dev/null +++ b/voting/quorum.py @@ -0,0 +1,54 @@ +from dataclasses import dataclass +from enum import Enum +from typing import Dict, Union + + +class QuorumKind(Enum): + """QuorumKind defines the kind of quorum. + + Kinds: + - NONE - there ain't no quorum + - ABSOLUTE - absolute number of minimum voters is required + - PERCENT - minimum percentage of valid voters must have voted + """ + NONE = 0 + ABSOLUTE = 1 + PERCENT = 2 + + +@dataclass +class Quorum: + """Quorm defines the required quorum of a Voting. + + Absolute quorums must be greater than 0. + Percent quorums must be greather than 0.0 and less then or equal to 100.0. + """ + + kind: QuorumKind + value: Union[int, float] + + def __init__(self, kind: QuorumKind = None, value: Union[float, int, None] = None): + super(Quorum) + if kind is None: + self.kind = QuorumKind.NONE + elif isinstance(kind, str): + self.kind = QuorumKind[kind] + else: + self.kind = kind + + if self.kind == QuorumKind.NONE: + self.value = 0 + elif self.kind == QuorumKind.ABSOLUTE: + self.value = int(value) + if self.value <= 0: + raise ValueError(f"Value can not be less than or equal to zero: {self.value}") + elif self.kind == QuorumKind.PERCENT: + self.value = float(value) + if self.value <= 0.0 or self.value > 100.0: + raise ValueError(f"Value must be greater than 0.0 and less then or equal to 100.0: {self.value}") + + def as_dict(self) -> Dict: + return { + "kind": self.kind.name, + "value": self.value, + } diff --git a/voting/test_voting.py b/voting/test_voting.py deleted file mode 100644 index aefdf62..0000000 --- a/voting/test_voting.py +++ /dev/null @@ -1,107 +0,0 @@ -import unittest - -import voting as V -from voting import QuorumKind, Quorum, Vote, Voting - -voters = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j'] - -class TestAtLeastOneVoter(unittest.TestCase): - def test_no_voters(self): - self.assertRaises(RuntimeError, Voting, - title="Test", - quorum=Quorum(), - voters=[], - ) - - def test_one_voter(self): - Voting(title="Test", quorum=Quorum(), voters=['a']) - - -class TestNoVotesBeforeStart(unittest.TestCase): - def setUp(self) -> None: - self.voting = Voting(title="Test", quorum=Quorum(), voters=voters) - - def test_vote_before_started(self): - self.assertRaises(RuntimeError, self.voting.vote, voter='a', vote=Vote.YES) - - def test_vote_after_started(self): - self.voting.start() - self.voting.vote(voter='a', vote=Vote.YES) - - -class TestQuorumSanity(unittest.TestCase): - def test_invalid_absolute_values(self): - self.assertRaises(ValueError, Quorum, kind=QuorumKind.ABSOLUTE, value=0) - self.assertRaises(ValueError, Quorum, kind=QuorumKind.ABSOLUTE, value=-1) - - def test_valid_absolute_values(self): - Quorum(kind=QuorumKind.ABSOLUTE, value=1) - - def test_invalid_percent_values(self): - self.assertRaises(ValueError, Quorum, kind=QuorumKind.PERCENT, value=0) - self.assertRaises(ValueError, Quorum, kind=QuorumKind.PERCENT, value=-1) - self.assertRaises(ValueError, Quorum, kind=QuorumKind.PERCENT, value=100.1) - - def test_valid_percent_values(self): - Quorum(kind=QuorumKind.PERCENT, value=0.1) - Quorum(kind=QuorumKind.PERCENT, value=100) - - -class TestVoteMustBeLegal(unittest.TestCase): - def setUp(self) -> None: - self.voting = Voting(title="Test", quorum=Quorum(), voters=voters) - self.voting.start() - - def test_unknown_voter(self): - self.assertRaises(ValueError, self.voting.vote, voter='z', vote=Vote.YES) - - def test_known_voter(self): - self.voting.vote('a', Vote.YES) - - -class TestResultAbsoluteQuorum(unittest.TestCase): - def setUp(self) -> None: - def new_voting(): - v = Voting( - title="Test", - quorum=Quorum(kind=QuorumKind.ABSOLUTE, value=1), - voters=voters, - ) - v.start() - return v - self.new_voting = new_voting - - def test_quorum(self): - v = self.new_voting() - assert not v.result().quorum_reached - - v.vote(voter='a', vote=Vote.YES) - assert v.result().quorum_reached - - -class TestResultPercentQuorum(unittest.TestCase): - def setUp(self) -> None: - def new_voting(): - v = Voting( - title="Test", - quorum=Quorum(kind=QuorumKind.PERCENT, value=15), - voters=voters, - ) - v.start() - return v - self.new_voting = new_voting - - def test_quorum(self): - v = self.new_voting() - assert not v.result().quorum_reached - - v.vote('a', Vote.YES) - assert not v.result().quorum_reached - - v.vote('b', Vote.YES) - assert v.result().quorum_reached - - - -if __name__ == '__main__': - unittest.main() diff --git a/voting/voting.py b/voting/voting.py index 465948b..667812f 100644 --- a/voting/voting.py +++ b/voting/voting.py @@ -3,57 +3,15 @@ __AUTHOR__ = 'Brian Wiborg ' __LICENSE__ = 'MIT' -from arrow import Arrow +from .quorum import Quorum, QuorumKind + +from arrow import Arrow, get as arrow_get +from dataclasses import dataclass from enum import Enum -from typing import Dict, List, Union +from typing import Dict, List -TIMEFRAME_IN_DAYS: int = 7 - - -def timeframe_is_over(started: Arrow) -> bool: - """Check if the current time is less than the given timestamp plus TIMEFRAME_IN_DAYS.""" - return Arrow.utcnow() > started.shift(days=TIMEFRAME_IN_DAYS) - - -class QuorumKind(Enum): - """QuorumKind defines the kind of quorum. - - Kinds: - - NONE - there ain't no quorum - - ABSOLUTE - absolute number of minimum voters is required - - PERCENT - minimum percentage of valid voters must have voted - """ - NONE = 0 - ABSOLUTE = 1 - PERCENT = 2 - - -class Quorum(object): - """Quorm defines the required quorum of a Voting. - - Absolute quorums must be greater than 0. - Percent quorums must be greather than 0.0 and less then or equal to 100.0. - """ - def __init__(self, kind: QuorumKind = QuorumKind.NONE, value: Union[float, int, None] = None): - self.__kind: QuorumKind = kind - self.__value: Union[float, int, None] = None - - if kind == QuorumKind.ABSOLUTE: - self.__value = int(value) - if self.__value <= 0: - raise ValueError(f"Value can not be less than or equal to zero: {self.__value}") - elif kind == QuorumKind.PERCENT: - self.__value = float(value) - if self.__value <= 0.0 or self.__value > 100.0: - raise ValueError(f"Value must be greater than 0.0 and less then or equal to 100.0: {self.__value}") - - @property - def kind(self) -> bool: - return self.__kind - - @property - def value(self) -> Union[float, int]: - return self.__value +import abc +import json.tool class Vote(Enum): @@ -63,115 +21,101 @@ class Vote(Enum): ABSTENTION = 2 -class Result(object): - """Result represents a voting result.""" - def __init__(self, started: Arrow, quorum: Quorum, voters: int, yes: int, no: int, absent: int): - self.__started: Arrow = started - self.__quroum: Quorum = quorum - self.__voters: int = voters - self.__votes_yes: int = yes - self.__votes_no: int = no - self.__votes_absent: int = absent - - def __repr__(self) -> str: - data = { - 'STIMMEN': self.__voters, - 'JA': self.__votes_yes, - 'NEIN': self.__votes_no, - 'ENTHALTUNGEN': self.__votes_absent, - } - - out = ' | '.join([f"{e}: {data[e]}" for e in data]) - if self.status == 'ready': - out += f"\nDamit is der Antrag: {self.outcome}" - return out - - @property - def status(self) -> str: - """Returns the human-readable status of the result.""" - is_running = self.__started is not None - is_ready = is_running and timeframe_is_over(self.__started) - out = "not started" - if is_ready: - out = "ready" - elif is_running: - out = "running" - return out - - @property - def quorum_reached(self) -> bool: - """Checks if the quorum has been reached.""" - total_votes = sum([self.__votes_yes, self.__votes_no, self.__votes_absent]) - if self.__quroum.kind == QuorumKind.NONE: - return True - elif self.__quroum.kind == QuorumKind.ABSOLUTE: - return total_votes >= self.__quroum.value - elif self.__quroum.kind == QuorumKind.PERCENT: - return total_votes * 100 / self.__voters >= self.__quroum.value - else: - raise RuntimeError("BUG! this code should never be reached!") - - @property - def outcome(self) -> str: - """Returns the human-readable outcome.""" - return 'ANGENOMMEN' if self.quorum_reached and self.__votes_yes > self.__votes_no else 'ABGELEHNT' - - class Voting(object): - """Voting represents a voting.""" - def __init__(self, title: str, quorum: Quorum, voters: List[str]): + title: str + quorum: Quorum + voters: List[str] + start: str + days: int + votes: Dict[str, Vote] + + def __init__(self, + title: str, + quorum: Quorum, + voters: List[str], + start: str = "", + days: int = 7, + votes: Dict[str, Vote] = None + ): if len(voters) == 0: - raise RuntimeError(f"Can not start a voting with 0 voters: {voters}") - self.__title: str = title - self.__quorum: Quorum = quorum - self.__voters: List[str] = voters - self.__start: Union[Arrow, None] = None - self.__votes: Dict[str, Vote] = {} + raise RuntimeError(f"A voting must have at least one voter: {voters}") - @property - def title(self) -> str: - return self.__title + self.title: str = title + self.quorum: Quorum = quorum if isinstance(quorum, Quorum) else Quorum(**quorum) + self.voters: List[str] = voters + self.start: str = start if start else str(Arrow.utcnow()) + self.days: int = days + self.votes: Dict[str, Vote] = {} - @property - def voters(self) -> List[str]: - return self.__voters - - def start(self) -> None: - """Starts the voting.""" - if self.__start is not None: - raise RuntimeWarning(f"Voting already running since: {self.__start}") - self.__start = Arrow.utcnow() - - def is_running(self) -> bool: - """Checks if the voting is currently running.""" - return self.__start is not None and not timeframe_is_over(self.__start) + if votes: + for voter in votes: + self.votes[voter] = self.votes[voter] if isinstance(votes[voter], Vote) else Vote[votes[voter]] def is_over(self) -> bool: - """Checks if the voting is over.""" - return self.is_running() and timeframe_is_over(self.__start) + return Arrow.utcnow() >= arrow_get(self.start).shift(days=self.days) def vote(self, voter: str, vote: Vote) -> None: - """Makes a vote. - - The voter must be in the list of legal voters. - If a voter votes multiple times, the latest vote wins. - """ - if not self.is_running() or self.is_over(): - raise RuntimeError(f"Voting has ended or hasn't even been started! (started: {self.__start})") - if voter not in self.__voters: - raise ValueError(f"{voter} not in {self.__voters}") - self.__votes[voter] = vote + if voter not in self.voters: + raise ValueError(f"{voter} not in {self.voters}") + if self.is_over(): + raise RuntimeError(f"Voting has ended! ({self.start} - {arrow_get(self.start).shift(days=self.days)})") + self.votes[voter] = vote - def result(self) -> Result: - """Obtains the current result for this voting.""" - if self.__start is None: - return RuntimeError("Voting hasn't even been started, yet!") + def as_dict(self) -> Dict: + return { + "title": self.title, + "quorum": self.quorum.as_dict(), + "voters": self.voters, + "start": str(self.start), + "days": self.days, + "votes": {k: v.name for (k, v) in self.votes.items()} + } - return Result( - quorum=self.__quorum, - started=self.__start, - voters=len(self.__voters), - yes=len([e for e in self.__votes if self.__votes[e] == Vote.YES]), - no=len([e for e in self.__votes if self.__votes[e] == Vote.NO]), - absent=len([e for e in self.__votes if self.__votes[e] == Vote.ABSTENTION]), - ) + def dumps(self) -> str: + return json.dumps(self.as_dict()) + + @classmethod + def loads(cls, data: Dict): + return cls(**json.loads(data)) + + +@dataclass +class Result: + voting: Voting + + @abc.abstractmethod + def result(self) -> Dict: + return { + "state": self.outcome(), + "voting": self.voting.as_dict(), + "votes": self.votes(), + "quorum_reached": self.quorum_reached(), + } + + @abc.abstractmethod + def outcome(self) -> str: + if not self.has_ended(): + return "LAUFEND" + else: + return "ANGENOMMEN" if self.votes()[Vote.YES.name] > self.votes()[Vote.NO.name] and self.quorum_reached() else "ABGELEHNT" + + def has_ended(self) -> bool: + end_at = arrow_get(self.voting.start).shift(days=self.voting.days) + return Arrow.utcnow() >= end_at + + def votes(self) -> Dict: + return { + Vote.YES.name: len([vote for vote in self.voting.votes if self.voting.votes[vote] == Vote.YES]), + Vote.NO.name: len([vote for vote in self.voting.votes if self.voting.votes[vote] == Vote.NO]), + Vote.ABSTENTION.name: len([vote for vote in self.voting.votes if self.voting.votes[vote] == Vote.ABSTENTION]), + } + + def quorum_reached(self) -> bool: + if self.voting.quorum.kind == QuorumKind.NONE: + return True + elif self.voting.quorum.kind == QuorumKind.ABSOLUTE: + return sum(self.votes().values()) >= self.voting.quorum.value + elif self.voting.quorum.kind == QuorumKind.PERCENT: + return sum(self.votes().values()) * 100 / len(self.voting.voters) >= self.voting.quorum.value + else: + raise RuntimeError("BUG: This code should never be reached!")