095e3f47-e82d-4d0d-80ac-cf1093df78bb

Build # 095e3f47-e82d-4d0d-80ac-cf1093df78bb

Build Type

push

circleci

Committed by Will Roper

Commit Message

Support multiple charismatic election dates

Run Details

16 of 16 new or added lines in 3 files covered. (100.0%)

3134 of 4320 relevant lines covered (72.55%)

0.73 hits per line

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

86.5

/polling_stations/apps/data_finder/helpers/every_election.py

from datetime import datetime
from typing import Optional
from typing import List

import requests
from django.conf import settings
from django.core.cache import cache
from uk_geo_utils.helpers import Postcode

session = requests.session()


class EveryElectionWrapper:
    def __init__(self, postcode=None, point=None, council_id=None):
        if not any((postcode, point, council_id)):
            raise ValueError("Expected either a point, postcode or council_id")
        try:
            self.request_success = False
            if postcode:
                self.elections = self.get_data_by_postcode(
                    Postcode(postcode).with_space
                )
                self.request_success = True
            if point:
                self.elections = self.get_data_by_point(point)
                self.request_success = True
            if council_id:
                self.elections = self.get_election_intersecting_local_authority(
                    council_id
                )
                self.request_success = True
            self.ballots = self.get_ballots_for_next_date()
            self.cancelled_ballots = self.get_cancelled_ballots()
        except requests.exceptions.RequestException:
            self.request_success = False

    def get_data_by_postcode(self, postcode):
        query_url = (
            "%sapi/elections.json?postcode=%s&future=1&current=1&identifier_type=ballot"
            % (
                settings.EE_BASE,
                postcode,
            )
        )
        return self.get_data(query_url)

    def get_data_by_point(self, point):
        query_url = (
            "%sapi/elections.json?coords=%s,%s&future=1&current=1&identifier_type=ballot"
            % (
                settings.EE_BASE,
                point.y,
                point.x,
            )
        )
        return self.get_data(query_url)

    def get_election_intersecting_local_authority(self, council_id):
        query_url = (
            "%sapi/elections.json?election_intersects_local_authority=%s&future=1&identifier_type=ballot"
            % (
                settings.EE_BASE,
                council_id,
            )
        )
        # Only used by council users atm so seems safe to cache for a whole day
        return self.get_data(query_url, cache_hours=24)

    def get_data(self, query_url, cache_hours=0):
        res_json = None
        if cache_hours:
            res_json = cache.get(query_url)
        if not res_json:
            headers = {}
            if hasattr(settings, "CUSTOM_UA"):
                headers["User-Agent"] = settings.CUSTOM_UA

            res = session.get(query_url, timeout=10, headers=headers)

            if res.status_code != 200:
                res.raise_for_status()

            res_json = res.json()
            if cache_hours:
                cache.set(query_url, res_json, 60 * 60 * cache_hours)

        if "results" in res_json:
            return res_json["results"]
        return res_json

    def get_all_ballots(self):
        if not self.request_success:
            return []
        ballots = [e for e in self.elections if e["group_type"] is None]
        ballots = [
            e for e in ballots if e["election_id"] not in settings.ELECTION_BLACKLIST
        ]
        return sorted(ballots, key=lambda k: k["poll_open_date"])

    def get_future_election_dates(self):
        if not self.request_success:
            return []
        dates = set([e["poll_open_date"] for e in self.elections if not e["cancelled"]])
        return sorted(list(dates))

    def _get_next_election_date(self):
        ballots = self.get_all_ballots()
        # if no ballots, return early
        if len(ballots) == 0:
            return None

        next_charismatic_election_dates = getattr(
            settings, "NEXT_CHARISMATIC_ELECTION_DATES", []
        )
        next_charismatic_election_dates.sort()
        dates = [datetime.strptime(b["poll_open_date"], "%Y-%m-%d") for b in ballots]
        dates.sort()

        if next_charismatic_election_dates:
            # If we have some dates return the first one that is in NEXT_CHARISMATIC_ELECTION_DATES
            for date in dates:
                if date in next_charismatic_election_dates:
                    return date
            # If none of them are in NEXT_CHARISMATIC_ELECTION_DATES,
            # return the earliest charismatic election date
            return next_charismatic_election_dates[0]

        # If we haven't set NEXT_CHARISMATIC_ELECTION_DATES,
        # return the earliest election
        return dates[0].strftime("%Y-%m-%d")

    def get_ballots_for_next_date(self):
        if not self.request_success:
            return []
        ballots = self.get_all_ballots()
        if len(ballots) == 0:
            return ballots
        next_election_date = self._get_next_election_date()
        ballots = [e for e in ballots if e["poll_open_date"] == next_election_date]
        return ballots

    def get_cancelled_ballots(self):
        return [b for b in self.ballots if b["cancelled"]]

    @property
    def all_ballots_cancelled(self):
        return len(self.cancelled_ballots) > 0 and len(self.ballots) == len(
            self.cancelled_ballots
        )

    def get_cancelled_election_info(self):
        rec = {
            "cancelled": None,
            "name": None,
            "rescheduled_date": None,
            "metadata": None,
        }

        # bypass the rest of this if we're not checking EE
        # or we failed to contact EE
        if not settings.EVERY_ELECTION["CHECK"] or not self.request_success:
            rec["cancelled"] = False
            return rec

        rec["cancelled"] = self.all_ballots_cancelled
        # What we care about here is if _all_
        # applicable ballot objects are cancelled.

        # i.e: If the user has a local election and a mayoral election
        # and the local one is cancelled but the mayoral one is going ahead
        # we just want to tell them where the polling station is.

        # For the purposes of WhereDIV we can abstract
        # the complexity that they will receive a different
        # number of ballots than expected when they get there.
        if not rec["cancelled"]:
            return rec

        cancelled_ballot = self.cancelled_ballots[0]
        if len(self.cancelled_ballots) == 1:
            rec["name"] = cancelled_ballot["election_title"]
        rec["metadata"] = cancelled_ballot["metadata"]

        if cancelled_ballot["replaced_by"]:
            try:
                query_url = "%sapi/elections/%s.json" % (
                    settings.EE_BASE,
                    cancelled_ballot["replaced_by"],
                )
                new_ballot = self.get_data(query_url)
                rec["rescheduled_date"] = datetime.strptime(
                    new_ballot["poll_open_date"], "%Y-%m-%d"
                ).strftime("%-d %B %Y")
            except requests.exceptions.RequestException:
                rec["rescheduled_date"] = None

        return rec

    def has_election(self):
        if not settings.EVERY_ELECTION["CHECK"]:
            return settings.EVERY_ELECTION["HAS_ELECTION"]

        if not self.request_success:
            # if the request was unsuccessful for some reason,
            # assume there *is* an upcoming election
            return True

        if len(self.ballots) > 0 and not self.all_ballots_cancelled:
            return True
        return False

    def get_explanations(self):
        explanations = []
        if not self.request_success:
            # if the request was unsuccessful for some reason,
            # return no explanations
            return explanations

        if len(self.elections) > 0:
            for election in self.elections:
                if (
                    "explanation" in election
                    and election["explanation"]
                    and election["poll_open_date"] == self._get_next_election_date()
                ):
                    explanations.append(
                        {
                            "title": election["election_title"],
                            "explanation": election["explanation"],
                        }
                    )
        return explanations

    def get_metadata(self):
        cancelled = self.get_cancelled_election_info()
        if cancelled["cancelled"]:
            return {"cancelled_election": cancelled["metadata"]}

        return None

    def get_metadata_by_key(self, key):
        if not settings.EVERY_ELECTION["CHECK"] or not self.request_success:
            return None

        for b in self.ballots:
            if "metadata" in b and b["metadata"] and key in b["metadata"]:
                return b["metadata"][key]
        return None

    def get_voter_id_status(self) -> Optional[str]:
        """
        For a given election, determine whether any ballots require photo ID
        If yes, return the stub value (e.g. EA-2022)
        If no, return None
        """
        ballot_with_id = next(
            (
                ballot
                for ballot in self.get_all_ballots()
                if ballot.get("requires_voter_id") and not ballot.get("cancelled")
            ),
            {},
        )
        return ballot_with_id.get("requires_voter_id")

    @property
    def multiple_elections(self):
        if self.has_election and self.request_success:
            uncancelled_ballots = [b for b in self.ballots if not b["cancelled"]]
            return len(uncancelled_ballots) > 1
        else:
            return False


class EmptyEveryElectionWrapper:
    """
    There are times when we know that we don't want to query EE, for example
    when we are going to show an address picker.

    This class allows us to swap out the EveryElectionWrapper while keeping
    exisitng code the same.
    """

    @staticmethod
    def has_election() -> bool:
        return False

    @staticmethod
    def get_metadata() -> None:
        return None

    @staticmethod
    def get_ballots_for_next_date() -> List:
        return []

    @staticmethod
    def get_all_ballots() -> List:
        return []

1	from datetime import datetime	1✔
2	from typing import Optional	1✔
3	from typing import List	1✔
4
5	import requests	1✔
6	from django.conf import settings	1✔
7	from django.core.cache import cache	1✔
8	from uk_geo_utils.helpers import Postcode	1✔
9
10	session = requests.session()	1✔
11
12
13	class EveryElectionWrapper:	1✔
14	def __init__(self, postcode=None, point=None, council_id=None):	1✔
15	if not any((postcode, point, council_id)):	1✔
16	raise ValueError("Expected either a point, postcode or council_id")	×
17	try:	1✔
18	self.request_success = False	1✔
19	if postcode:	1✔
20	self.elections = self.get_data_by_postcode(	1✔
21	Postcode(postcode).with_space
22	)
23	self.request_success = True	1✔
24	if point:	1✔
25	self.elections = self.get_data_by_point(point)	1✔
26	self.request_success = True	1✔
27	if council_id:	1✔
28	self.elections = self.get_election_intersecting_local_authority(	1✔
29	council_id
30	)
31	self.request_success = True	×
32	self.ballots = self.get_ballots_for_next_date()	1✔
33	self.cancelled_ballots = self.get_cancelled_ballots()	1✔
34	except requests.exceptions.RequestException:	1✔
35	self.request_success = False	1✔
36
37	def get_data_by_postcode(self, postcode):	1✔
38	query_url = (	1✔
39	"%sapi/elections.json?postcode=%s&future=1&current=1&identifier_type=ballot"
40	% (
41	settings.EE_BASE,
42	postcode,
43	)
44	)
45	return self.get_data(query_url)	1✔
46
47	def get_data_by_point(self, point):	1✔
48	query_url = (	1✔
49	"%sapi/elections.json?coords=%s,%s&future=1&current=1&identifier_type=ballot"
50	% (
51	settings.EE_BASE,
52	point.y,
53	point.x,
54	)
55	)
56	return self.get_data(query_url)	1✔
57
58	def get_election_intersecting_local_authority(self, council_id):	1✔
59	query_url = (	1✔
60	"%sapi/elections.json?election_intersects_local_authority=%s&future=1&identifier_type=ballot"
61	% (
62	settings.EE_BASE,
63	council_id,
64	)
65	)
66	# Only used by council users atm so seems safe to cache for a whole day
67	return self.get_data(query_url, cache_hours=24)	1✔
68
69	def get_data(self, query_url, cache_hours=0):	1✔
70	res_json = None	1✔
71	if cache_hours:	1✔
72	res_json = cache.get(query_url)	1✔
73	if not res_json:	1✔
74	headers = {}	1✔
75	if hasattr(settings, "CUSTOM_UA"):	1✔
76	headers["User-Agent"] = settings.CUSTOM_UA	×
77
78	res = session.get(query_url, timeout=10, headers=headers)	1✔
79
80	if res.status_code != 200:	1✔
81	res.raise_for_status()	1✔
82
83	res_json = res.json()	1✔
84	if cache_hours:	1✔
85	cache.set(query_url, res_json, 60 * 60 * cache_hours)	×
86
87	if "results" in res_json:	1✔
88	return res_json["results"]	1✔
89	return res_json	×
90
91	def get_all_ballots(self):	1✔
92	if not self.request_success:	1✔
93	return []	1✔
94	ballots = [e for e in self.elections if e["group_type"] is None]	1✔
95	ballots = [	1✔
96	e for e in ballots if e["election_id"] not in settings.ELECTION_BLACKLIST
97	]
98	return sorted(ballots, key=lambda k: k["poll_open_date"])	1✔
99
100	def get_future_election_dates(self):	1✔
101	if not self.request_success:	1✔
102	return []	1✔
103	dates = set([e["poll_open_date"] for e in self.elections if not e["cancelled"]])	×
104	return sorted(list(dates))	×
105
106	def _get_next_election_date(self):	1✔
107	ballots = self.get_all_ballots()	1✔
108	# if no ballots, return early
109	if len(ballots) == 0:	1✔
110	return None	×
111
112	next_charismatic_election_dates = getattr(	1✔
113	settings, "NEXT_CHARISMATIC_ELECTION_DATES", []
114	)
115	next_charismatic_election_dates.sort()	1✔
116	dates = [datetime.strptime(b["poll_open_date"], "%Y-%m-%d") for b in ballots]	1✔
117	dates.sort()	1✔
118
119	if next_charismatic_election_dates:	1✔
120	# If we have some dates return the first one that is in NEXT_CHARISMATIC_ELECTION_DATES
121	for date in dates:	1✔
122	if date in next_charismatic_election_dates:	1✔
123	return date	×
124	# If none of them are in NEXT_CHARISMATIC_ELECTION_DATES,
125	# return the earliest charismatic election date
126	return next_charismatic_election_dates[0]	1✔
127
128	# If we haven't set NEXT_CHARISMATIC_ELECTION_DATES,
129	# return the earliest election
130	return dates[0].strftime("%Y-%m-%d")	1✔
131
132	def get_ballots_for_next_date(self):	1✔
133	if not self.request_success:	1✔
134	return []	×
135	ballots = self.get_all_ballots()	1✔
136	if len(ballots) == 0:	1✔
137	return ballots	1✔
138	next_election_date = self._get_next_election_date()	1✔
139	ballots = [e for e in ballots if e["poll_open_date"] == next_election_date]	1✔
140	return ballots	1✔
141
142	def get_cancelled_ballots(self):	1✔
143	return [b for b in self.ballots if b["cancelled"]]	1✔
144
145	@property	1✔
146	def all_ballots_cancelled(self):	1✔
147	return len(self.cancelled_ballots) > 0 and len(self.ballots) == len(	1✔
148	self.cancelled_ballots
149	)
150
151	def get_cancelled_election_info(self):	1✔
152	rec = {	1✔
153	"cancelled": None,
154	"name": None,
155	"rescheduled_date": None,
156	"metadata": None,
157	}
158
159	# bypass the rest of this if we're not checking EE
160	# or we failed to contact EE
161	if not settings.EVERY_ELECTION["CHECK"] or not self.request_success:	1✔
162	rec["cancelled"] = False	1✔
163	return rec	1✔
164
165	rec["cancelled"] = self.all_ballots_cancelled	1✔
166	# What we care about here is if _all_
167	# applicable ballot objects are cancelled.
168
169	# i.e: If the user has a local election and a mayoral election
170	# and the local one is cancelled but the mayoral one is going ahead
171	# we just want to tell them where the polling station is.
172
173	# For the purposes of WhereDIV we can abstract
174	# the complexity that they will receive a different
175	# number of ballots than expected when they get there.
176	if not rec["cancelled"]:	1✔
177	return rec	1✔
178
179	cancelled_ballot = self.cancelled_ballots[0]	1✔
180	if len(self.cancelled_ballots) == 1:	1✔
181	rec["name"] = cancelled_ballot["election_title"]	1✔
182	rec["metadata"] = cancelled_ballot["metadata"]	1✔
183
184	if cancelled_ballot["replaced_by"]:	1✔
185	try:	1✔
186	query_url = "%sapi/elections/%s.json" % (	1✔
187	settings.EE_BASE,
188	cancelled_ballot["replaced_by"],
189	)
190	new_ballot = self.get_data(query_url)	1✔
191	rec["rescheduled_date"] = datetime.strptime(	1✔
192	new_ballot["poll_open_date"], "%Y-%m-%d"
193	).strftime("%-d %B %Y")
194	except requests.exceptions.RequestException:	×
195	rec["rescheduled_date"] = None	×
196
197	return rec	1✔
198
199	def has_election(self):	1✔
200	if not settings.EVERY_ELECTION["CHECK"]:	1✔
201	return settings.EVERY_ELECTION["HAS_ELECTION"]	1✔
202
203	if not self.request_success:	1✔
204	# if the request was unsuccessful for some reason,
205	# assume there is an upcoming election
206	return True	1✔
207
208	if len(self.ballots) > 0 and not self.all_ballots_cancelled:	1✔
209	return True	1✔
210	return False	1✔
211
212	def get_explanations(self):	1✔
213	explanations = []	1✔
214	if not self.request_success:	1✔
215	# if the request was unsuccessful for some reason,
216	# return no explanations
217	return explanations	1✔
218
219	if len(self.elections) > 0:	1✔
220	for election in self.elections:	1✔
221	if (	1✔
222	"explanation" in election
223	and election["explanation"]
224	and election["poll_open_date"] == self._get_next_election_date()
225	):
226	explanations.append(	1✔
227	{
228	"title": election["election_title"],
229	"explanation": election["explanation"],
230	}
231	)
232	return explanations	1✔
233
234	def get_metadata(self):	1✔
235	cancelled = self.get_cancelled_election_info()	1✔
236	if cancelled["cancelled"]:	1✔
237	return {"cancelled_election": cancelled["metadata"]}	1✔
238
239	return None	1✔
240
241	def get_metadata_by_key(self, key):	1✔
242	if not settings.EVERY_ELECTION["CHECK"] or not self.request_success:	×
243	return None	×
244
245	for b in self.ballots:	×
246	if "metadata" in b and b["metadata"] and key in b["metadata"]:	×
247	return b["metadata"][key]	×
248	return None	×
249
250	def get_voter_id_status(self) -> Optional[str]:	1✔
251	"""
252	For a given election, determine whether any ballots require photo ID
253	If yes, return the stub value (e.g. EA-2022)
254	If no, return None
255	"""
256	ballot_with_id = next(	1✔
257	(
258	ballot
259	for ballot in self.get_all_ballots()
260	if ballot.get("requires_voter_id") and not ballot.get("cancelled")
261	),
262	{},
263	)
264	return ballot_with_id.get("requires_voter_id")	1✔
265
266	@property	1✔
267	def multiple_elections(self):	1✔
268	if self.has_election and self.request_success:	1✔
269	uncancelled_ballots = [b for b in self.ballots if not b["cancelled"]]	1✔
270	return len(uncancelled_ballots) > 1	1✔
271	else:
272	return False	1✔
273
274
275	class EmptyEveryElectionWrapper:	1✔
276	"""
277	There are times when we know that we don't want to query EE, for example
278	when we are going to show an address picker.
279
280	This class allows us to swap out the EveryElectionWrapper while keeping
281	exisitng code the same.
282	"""
283
284	@staticmethod	1✔
285	def has_election() -> bool:	1✔
286	return False	×
287
288	@staticmethod	1✔
289	def get_metadata() -> None:	1✔
290	return None	×
291
292	@staticmethod	1✔
293	def get_ballots_for_next_date() -> List:	1✔
294	return []	×
295
296	@staticmethod	1✔
297	def get_all_ballots() -> List:	1✔
298	return []	×

DemocracyClub / UK-Polling-Stations / 095e3f47-e82d-4d0d-80ac-cf1093df78bb

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