05affd3d-48d7-4614-bb06-45190ab4242c

Committed 16 Jul 2024 03:22PM UTC coverage: 67.91% (-0.03%) from 67.937%

Build # 05affd3d-48d7-4614-bb06-45190ab4242c

Build Type

push

circleci

Committed by

VirginiaDooley

Commit Message

Filter by complete, incomplete and no resultset

This change adds resultset filter and separates the logic
from candidate results (winners elected)

Run Details

1561 of 2832 branches covered (55.12%)

Branch coverage included in aggregate %.

12 of 16 new or added lines in 2 files covered. (75.0%)

9 existing lines in 1 file now uncovered.

7488 of 10493 relevant lines covered (71.36%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.06

/ynr/apps/uk_results/models.py

from data_exports.models import MaterializedMemberships
from django.contrib.auth.models import User
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models, transaction
from django.db.models import JSONField, OuterRef, Subquery
from django_extensions.db.models import TimeStampedModel
from popolo.models import Membership


class ResultSet(TimeStampedModel):
    ballot = models.OneToOneField("candidates.Ballot", on_delete=models.CASCADE)

    num_turnout_reported = models.PositiveIntegerField(
        null=True, verbose_name="Reported Turnout"
    )
    turnout_percentage = models.IntegerField(
        validators=[MinValueValidator(0), MaxValueValidator(100)],
        blank=True,
        null=True,
    )
    num_spoilt_ballots = models.PositiveIntegerField(
        null=True, verbose_name="Spoilt Ballots"
    )
    total_electorate = models.PositiveIntegerField(
        verbose_name="Total Electorate", null=True, blank=True
    )
    source = models.TextField(null=True)

    user = models.ForeignKey(
        "auth.User",
        related_name="result_sets",
        null=True,
        on_delete=models.CASCADE,
    )
    versions = JSONField(default=list)

    ip_address = models.GenericIPAddressField(null=True)

    def __str__(self):
        return "Result for {}".format(self.ballot.ballot_paper_id)

    @property
    def resultset_complete(self):
        """
        Check if the result set has all the required fields filled in
        """
        if (
            all(
                [
                    self.num_turnout_reported,
                    self.total_electorate,
                    self.num_spoilt_ballots,
                    self.turnout_percentage,
                ]
            )
            and self.ballot.elected_count
        ):
            return True
        return False

    @property
    def rank(self):
        self.ballot.result_sets.order_by("-num_turnout_reported").index(self)

    def calculate_turnout_percentage(self):
        """
        Return turnout as a percentage, rounded to two decimal places
        """
        if not all([self.num_turnout_reported, self.total_electorate]):
            return

        percentage = (self.num_turnout_reported / self.total_electorate) * 100
        self.turnout_percentage = int(min(round(percentage, 2), 100))

    def as_dict(self):
        """
        A representation of the model and related CandidateResult models
        as JSON.

        Used for storing versions.

        # TODO use API serializer for this?
        """
        data = {
            "created": self.modified.isoformat(),
            "ballot_paper_id": self.ballot.ballot_paper_id,
            "turnout": self.num_turnout_reported,
            "spoilt_ballots": self.num_spoilt_ballots,
            "total_electorate": self.total_electorate,
            "source": self.source,
            "user": getattr(self.user, "username", None),
            "candidate_results": [],
        }

        for result in self.candidate_results.all():
            data["candidate_results"].append(
                {
                    "num_ballots": result.num_ballots,
                    "elected": result.membership.elected,
                    "person_id": result.membership.person_id,
                    "person_name": result.membership.person.name,
                }
            )
        return data

    def versions_equal(self, v1, v2):
        """
        Compare v1 and v2, ignoring the created key
        """
        ignore_keys = ["created"]
        comp1 = {k: v for k, v in v1.items() if k not in ignore_keys}
        comp2 = {k: v for k, v in v2.items() if k not in ignore_keys}
        return comp1 == comp2

    def record_version(self, force=False, save=True):
        existing = self.versions
        this_version = self.as_dict()
        changed = False
        if existing:
            latest = existing[0]
            if force or not self.versions_equal(latest, this_version):
                changed = True
                existing.insert(0, this_version)
        else:
            changed = True
            existing.insert(0, this_version)

        self.versions = existing
        if save:
            self.save()
        return (existing, changed)

    def save(self, **kwargs):
        self.calculate_turnout_percentage()
        return super().save(**kwargs)


class CandidateResult(TimeStampedModel):
    result_set = models.ForeignKey(
        "ResultSet", related_name="candidate_results", on_delete=models.CASCADE
    )

    membership = models.OneToOneField(
        "popolo.Membership", related_name="result", on_delete=models.CASCADE
    )

    num_ballots = models.PositiveIntegerField()
    tied_vote_winner = models.BooleanField(
        default=False,
        help_text="Did this person win after receiving same votes as another candidate, via coin toss, lots etc",
    )
    rank = models.PositiveIntegerField(null=True, verbose_name="Results Rank")

    class Meta:
        ordering = ("-num_ballots",)
        unique_together = (("result_set", "membership"),)

    def __unicode__(self):
        return "{} ({} votes)".format(self.membership.person, self.num_ballots)


# Useful when debugging
ALWAYS_REFRESH = False


class SuggestedWinner(TimeStampedModel):
    """
    Model that tracks when someone has marked a person as a winner
    by directly pressing a button, rather than entering votes cast.

    If the last `N_SUGGESTIONS_NEEDED` agree, then we mark the winner.

    """

    N_SUGGESTIONS_NEEDED = 2

    user = models.ForeignKey(User, on_delete=models.CASCADE)
    membership = models.ForeignKey(
        Membership, related_name="suggested_winners", on_delete=models.CASCADE
    )
    is_elected = models.BooleanField(default=True)

    def __str__(self):
        return f"{self.membership.party.ec_id}: {self.is_elected} ({self.user})"

    @classmethod
    def record_suggestion(
        cls, user: User, membership: Membership, is_elected: bool = True
    ):
        ballot = membership.ballot
        if ballot.winner_count > 1:
            raise ValueError(
                "Can't use mark winner system for multi-seat posts"
            )
        with transaction.atomic():
            # Create or update the vote
            suggestion, created = cls.objects.update_or_create(
                user=user,
                membership=membership,
                defaults={"is_elected": is_elected},
            )

            # If a different winner is already set on the ballot,
            # we shouldn't have allowed a user to make a suggestion.
            # We're here now though, so we still record the above, but
            # do nothing else
            other_winners = ballot.membership_set.filter(elected=True).exclude(
                pk=membership.pk
            )
            if other_winners.exists():
                return suggestion

            # Retrieve the last two votes for the membership
            subquery = (
                cls.objects.filter(user=OuterRef("user"), membership=membership)
                .order_by("-id")
                .values("id")[:1]
            )

            latest_votes = (
                cls.objects.filter(membership=membership)
                .annotate(latest_id=Subquery(subquery))
                .filter(id=Subquery(subquery))[:2]
            )

            elected_count = sum(v.is_elected for v in latest_votes)
            if elected_count == cls.N_SUGGESTIONS_NEEDED:
                membership.elected = True
                membership.save()
                if ALWAYS_REFRESH:
                    MaterializedMemberships.refresh_view()
                return suggestion
            if not is_elected:
                membership.elected = False
                membership.save()
                if ALWAYS_REFRESH:
                    MaterializedMemberships.refresh_view()
            return suggestion

1	from data_exports.models import MaterializedMemberships	1✔
2	from django.contrib.auth.models import User	1✔
3	from django.core.validators import MaxValueValidator, MinValueValidator	1✔
4	from django.db import models, transaction	1✔
5	from django.db.models import JSONField, OuterRef, Subquery	1✔
6	from django_extensions.db.models import TimeStampedModel	1✔
7	from popolo.models import Membership	1✔
8
9
10	class ResultSet(TimeStampedModel):	1✔
11	ballot = models.OneToOneField("candidates.Ballot", on_delete=models.CASCADE)	1✔
12
13	num_turnout_reported = models.PositiveIntegerField(	1✔
14	null=True, verbose_name="Reported Turnout"
15	)
16	turnout_percentage = models.IntegerField(	1✔
17	validators=[MinValueValidator(0), MaxValueValidator(100)],
18	blank=True,
19	null=True,
20	)
21	num_spoilt_ballots = models.PositiveIntegerField(	1✔
22	null=True, verbose_name="Spoilt Ballots"
23	)
24	total_electorate = models.PositiveIntegerField(	1✔
25	verbose_name="Total Electorate", null=True, blank=True
26	)
27	source = models.TextField(null=True)	1✔
28
29	user = models.ForeignKey(	1✔
30	"auth.User",
31	related_name="result_sets",
32	null=True,
33	on_delete=models.CASCADE,
34	)
35	versions = JSONField(default=list)	1✔
36
37	ip_address = models.GenericIPAddressField(null=True)	1✔
38
39	def __str__(self):	1✔
40	return "Result for {}".format(self.ballot.ballot_paper_id)	×
41
42	@property	1✔
43	def resultset_complete(self):	1✔
44	"""
45	Check if the result set has all the required fields filled in
46	"""
UNCOV 47	if (	×
48	all(
49	[
50	self.num_turnout_reported,
51	self.total_electorate,
52	self.num_spoilt_ballots,
53	self.turnout_percentage,
54	]
55	)
56	and self.ballot.elected_count
57	):
UNCOV 58	return True	×
UNCOV 59	return False	×
60
61	@property	1✔
62	def rank(self):	1✔
UNCOV 63	self.ballot.result_sets.order_by("-num_turnout_reported").index(self)	×
64
65	def calculate_turnout_percentage(self):	1✔
66	"""
67	Return turnout as a percentage, rounded to two decimal places
68	"""
69	if not all([self.num_turnout_reported, self.total_electorate]):	1✔
70	return	1✔
71
72	percentage = (self.num_turnout_reported / self.total_electorate) * 100	1✔
73	self.turnout_percentage = int(min(round(percentage, 2), 100))	1✔
74
75	def as_dict(self):	1✔
76	"""
77	A representation of the model and related CandidateResult models
78	as JSON.
79
80	Used for storing versions.
81
82	# TODO use API serializer for this?
83	"""
84	data = {	1✔
85	"created": self.modified.isoformat(),
86	"ballot_paper_id": self.ballot.ballot_paper_id,
87	"turnout": self.num_turnout_reported,
88	"spoilt_ballots": self.num_spoilt_ballots,
89	"total_electorate": self.total_electorate,
90	"source": self.source,
91	"user": getattr(self.user, "username", None),
92	"candidate_results": [],
93	}
94
95	for result in self.candidate_results.all():	1✔
96	data["candidate_results"].append(	1✔
97	{
98	"num_ballots": result.num_ballots,
99	"elected": result.membership.elected,
100	"person_id": result.membership.person_id,
101	"person_name": result.membership.person.name,
102	}
103	)
104	return data	1✔
105
106	def versions_equal(self, v1, v2):	1✔
107	"""
108	Compare v1 and v2, ignoring the created key
109	"""
110	ignore_keys = ["created"]	1✔
111	comp1 = {k: v for k, v in v1.items() if k not in ignore_keys}	1✔
112	comp2 = {k: v for k, v in v2.items() if k not in ignore_keys}	1✔
113	return comp1 == comp2	1✔
114
115	def record_version(self, force=False, save=True):	1✔
116	existing = self.versions	1✔
117	this_version = self.as_dict()	1✔
118	changed = False	1✔
119	if existing:	1✔
120	latest = existing[0]	1✔
121	if force or not self.versions_equal(latest, this_version):	1✔
122	changed = True	1✔
123	existing.insert(0, this_version)	1✔
124	else:
125	changed = True	1✔
126	existing.insert(0, this_version)	1✔
127
128	self.versions = existing	1✔
129	if save:	1!
130	self.save()	1✔
131	return (existing, changed)	1✔
132
133	def save(self, **kwargs):	1✔
134	self.calculate_turnout_percentage()	1✔
135	return super().save(**kwargs)	1✔
136
137
138	class CandidateResult(TimeStampedModel):	1✔
139	result_set = models.ForeignKey(	1✔
140	"ResultSet", related_name="candidate_results", on_delete=models.CASCADE
141	)
142
143	membership = models.OneToOneField(	1✔
144	"popolo.Membership", related_name="result", on_delete=models.CASCADE
145	)
146
147	num_ballots = models.PositiveIntegerField()	1✔
148	tied_vote_winner = models.BooleanField(	1✔
149	default=False,
150	help_text="Did this person win after receiving same votes as another candidate, via coin toss, lots etc",
151	)
152	rank = models.PositiveIntegerField(null=True, verbose_name="Results Rank")	1✔
153
154	class Meta:	1✔
155	ordering = ("-num_ballots",)	1✔
156	unique_together = (("result_set", "membership"),)	1✔
157
158	def __unicode__(self):	1✔
UNCOV 159	return "{} ({} votes)".format(self.membership.person, self.num_ballots)	×
160
161
162	# Useful when debugging
163	ALWAYS_REFRESH = False	1✔
164
165
166	class SuggestedWinner(TimeStampedModel):	1✔
167	"""
168	Model that tracks when someone has marked a person as a winner
169	by directly pressing a button, rather than entering votes cast.
170
171	If the last `N_SUGGESTIONS_NEEDED` agree, then we mark the winner.
172
173	"""
174
175	N_SUGGESTIONS_NEEDED = 2	1✔
176
177	user = models.ForeignKey(User, on_delete=models.CASCADE)	1✔
178	membership = models.ForeignKey(	1✔
179	Membership, related_name="suggested_winners", on_delete=models.CASCADE
180	)
181	is_elected = models.BooleanField(default=True)	1✔
182
183	def __str__(self):	1✔
UNCOV 184	return f"{self.membership.party.ec_id}: {self.is_elected} ({self.user})"	×
185
186	@classmethod	1✔
187	def record_suggestion(	1✔
188	cls, user: User, membership: Membership, is_elected: bool = True
189	):
190	ballot = membership.ballot	1✔
191	if ballot.winner_count > 1:	1!
UNCOV 192	raise ValueError(	×
193	"Can't use mark winner system for multi-seat posts"
194	)
195	with transaction.atomic():	1✔
196	# Create or update the vote
197	suggestion, created = cls.objects.update_or_create(	1✔
198	user=user,
199	membership=membership,
200	defaults={"is_elected": is_elected},
201	)
202
203	# If a different winner is already set on the ballot,
204	# we shouldn't have allowed a user to make a suggestion.
205	# We're here now though, so we still record the above, but
206	# do nothing else
207	other_winners = ballot.membership_set.filter(elected=True).exclude(	1✔
208	pk=membership.pk
209	)
210	if other_winners.exists():	1✔
211	return suggestion	1✔
212
213	# Retrieve the last two votes for the membership
214	subquery = (	1✔
215	cls.objects.filter(user=OuterRef("user"), membership=membership)
216	.order_by("-id")
217	.values("id")[:1]
218	)
219
220	latest_votes = (	1✔
221	cls.objects.filter(membership=membership)
222	.annotate(latest_id=Subquery(subquery))
223	.filter(id=Subquery(subquery))[:2]
224	)
225
226	elected_count = sum(v.is_elected for v in latest_votes)	1✔
227	if elected_count == cls.N_SUGGESTIONS_NEEDED:	1✔
228	membership.elected = True	1✔
229	membership.save()	1✔
230	if ALWAYS_REFRESH:	1!
UNCOV 231	MaterializedMemberships.refresh_view()	×
232	return suggestion	1✔
233	if not is_elected:	1✔
234	membership.elected = False	1✔
235	membership.save()	1✔
236	if ALWAYS_REFRESH:	1!
UNCOV 237	MaterializedMemberships.refresh_view()	×
238	return suggestion	1✔

DemocracyClub / yournextrepresentative / 05affd3d-48d7-4614-bb06-45190ab4242c

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous