28b2cb5e-29d5-4b2a-bba5-65e9410e6204

Committed 08 Mar 2024 11:40AM UTC coverage: 71.569% (-0.4%) from 71.995%

Build # 28b2cb5e-29d5-4b2a-bba5-65e9410e6204

Build Type

Pull #6452

circleci

Committed by

awdem

Commit Message

Import script for City of Lincoln (2024-05-02) (closes #6451)

Pull Request Pull Request #6452: Import script for City of Lincoln (2024-05-02) (closes #6451)

Run Details

3411 of 4766 relevant lines covered (71.57%)

0.72 hits per line

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

89.36

/polling_stations/apps/data_importers/management/commands/run_new_imports.py

import re
import subprocess
from pathlib import Path

import boto3
import botocore
from django.apps import apps
from django.core.management import BaseCommand, call_command


def get_paths_changed(from_sha, to_sha):
    args = ["git", "diff", "--name-only", from_sha, to_sha]
    output = subprocess.check_output(args)
    return output.decode().splitlines()


def git_rev_parse(rev):
    return subprocess.check_output(["git", "rev-parse", f"{rev}"]).decode().strip()


def get_changed_scripts(changed):
    if any_import_scripts(changed):
        return [p for p in changed if is_import_script(p)]
    return []


def is_import_script(path):
    if re.search(
        r"polling_stations/apps/data_importers/management/commands/import_(?!eoni).+\.py",
        path,
    ):
        return True
    return False


def any_import_scripts(changed):
    return any(is_import_script(path) for path in changed)


def any_non_import_scripts(changed):
    return any(not is_import_script(path) for path in changed)


def get_last_import_sha_from_ssm():
    ssm_client = boto3.client("ssm")
    response = ssm_client.get_parameter(Name="LAST_IMPORT_SHA")
    return response["Parameter"]["Value"]


class Command(BaseCommand):
    help = """
        Checks the files changed between two GIT commits
        and decides if import scripts should be run
    """

    # Turn off auto system check for all apps
    # We will manually run system checks only for the
    # 'data_importers' and 'pollingstations' apps
    requires_system_checks = []

    summary = []

    def add_arguments(self, parser):
        parser.add_argument(
            "-f",
            "--from-sha",
            help="<Optional> The earliest commit hash against which to search for changed scripts. "
            "Defaults to value of LAST_IMPORT_SHA in ssm parameter store",
        )
        parser.add_argument(
            "-t",
            "--to-sha",
            help="<Optional> The latest commit hash against which to search for changed scripts. Defaults to HEAD",
            default=git_rev_parse("HEAD"),
        )
        parser.add_argument("--post-deploy", action="store_true", default=False)

    def run_scripts(self, changed_paths, opts):
        for script in changed_paths:
            script_path = Path(script)
            try:
                call_command(script_path.stem, **opts)
                self.summary.append(
                    (
                        "INFO",
                        f"Ran import script: {script_path.name}",
                    )
                )
            except Exception as e:
                # usually we want to handle a specific exception, but in this situation
                # if there is any issue (at all) trying to run the command,
                # we just want to log it and move on to the next script
                self.summary.append(
                    ("WARNING", f"{script_path.name} could not be run. Due to {e}")
                )
                continue

    def run_misc_fixes(self):
        call_command("misc_fixes")

    def output_summary(self):
        for line in self.summary:
            if line[0] == "INFO":
                self.stdout.write(line[1])
            elif line[0] == "WARNING":
                self.stdout.write(self.style.ERROR(line[1]))
            else:
                self.stdout.write(line[1])

    def update_last_import_sha_on_ssm(self, to_sha):
        try:
            ssm_client = boto3.client("ssm")
            ssm_client.put_parameter(
                Name="LAST_IMPORT_SHA", Value=to_sha, Overwrite=True
            )
        except botocore.exceptions.ClientError as e:
            self.summary.append(
                (
                    "WARNING",
                    f"LAST_IMPORT_SHA not updated in parameter store, due to:\n{e}",
                )
            )
        except botocore.exceptions.NoCredentialsError as e:
            self.summary.append(
                (
                    "WARNING",
                    f"LAST_IMPORT_SHA not updated in parameter store, due to:\n{e}",
                )
            )

    def handle(self, *args, **options):
        self.check(
            [
                apps.get_app_config("data_importers"),
                apps.get_app_config("pollingstations"),
            ]
        )

        if not (from_sha := options.get("from_sha")):
            from_sha = get_last_import_sha_from_ssm()

        to_sha = options.get("to_sha")
        is_post_deploy = options.get("post_deploy")

        changed_paths = get_paths_changed(from_sha, to_sha)
        changed_scripts = get_changed_scripts(changed_paths)
        has_imports = any_import_scripts(changed_paths)
        has_application = any_non_import_scripts(changed_paths)

        cmd_opts = {
            "nochecks": True,
            "verbosity": 1,
            "use_postcode_centroids": False,
            "include_past_elections": False,
        }

        self.stdout.write(
            f"Comparing repo between {from_sha} and {to_sha}\n"
            f"\tFrom: https://github.com/DemocracyClub/UK-Polling-Stations/commit/{git_rev_parse(from_sha)}\n"
            f"\tTo: https://github.com/DemocracyClub/UK-Polling-Stations/commit/{git_rev_parse(to_sha)}\n"
        )

        if has_imports and not has_application:
            self.stdout.write("Only import scripts have changed\n")
            self.stdout.write("Running import scripts\n")
            self.run_scripts(changed_scripts, cmd_opts)
            self.run_misc_fixes()
            self.stdout.write("updating LAST_IMPORT_SHA on ssm")
            self.update_last_import_sha_on_ssm(to_sha)
        elif has_imports and has_application and is_post_deploy:
            self.stdout.write("App has deployed. OK to run import scripts")
            self.run_scripts(changed_scripts, cmd_opts)
            self.run_misc_fixes()
            self.update_last_import_sha_on_ssm(to_sha)
        elif has_imports and has_application and not is_post_deploy:
            self.stdout.write("Need to deploy before running import scripts\n")
        elif not has_imports:
            self.stdout.write(
                "No import scripts have changed. So nothing new to import.\n"
            )
            self.update_last_import_sha_on_ssm(to_sha)
        else:
            self.stdout.write("Not running import scripts")

        self.output_summary()

1	import re	1✔
2	import subprocess	1✔
3	from pathlib import Path	1✔
4
5	import boto3	1✔
6	import botocore	1✔
7	from django.apps import apps	1✔
8	from django.core.management import BaseCommand, call_command	1✔
9
10
11	def get_paths_changed(from_sha, to_sha):	1✔
12	args = ["git", "diff", "--name-only", from_sha, to_sha]	1✔
13	output = subprocess.check_output(args)	1✔
14	return output.decode().splitlines()	1✔
15
16
17	def git_rev_parse(rev):	1✔
18	return subprocess.check_output(["git", "rev-parse", f"{rev}"]).decode().strip()	1✔
19
20
21	def get_changed_scripts(changed):	1✔
22	if any_import_scripts(changed):	1✔
23	return [p for p in changed if is_import_script(p)]	1✔
24	return []	1✔
25
26
27	def is_import_script(path):	1✔
28	if re.search(	1✔
29	r"polling_stations/apps/data_importers/management/commands/import_(?!eoni).+\.py",
30	path,
31	):
32	return True	1✔
33	return False	1✔
34
35
36	def any_import_scripts(changed):	1✔
37	return any(is_import_script(path) for path in changed)	1✔
38
39
40	def any_non_import_scripts(changed):	1✔
41	return any(not is_import_script(path) for path in changed)	1✔
42
43
44	def get_last_import_sha_from_ssm():	1✔
45	ssm_client = boto3.client("ssm")	×
46	response = ssm_client.get_parameter(Name="LAST_IMPORT_SHA")	×
47	return response["Parameter"]["Value"]	×
48
49
50	class Command(BaseCommand):	1✔
51	help = """	1✔
52	Checks the files changed between two GIT commits
53	and decides if import scripts should be run
54	"""
55
56	# Turn off auto system check for all apps
57	# We will manually run system checks only for the
58	# 'data_importers' and 'pollingstations' apps
59	requires_system_checks = []	1✔
60
61	summary = []	1✔
62
63	def add_arguments(self, parser):	1✔
64	parser.add_argument(	1✔
65	"-f",
66	"--from-sha",
67	help="<Optional> The earliest commit hash against which to search for changed scripts. "
68	"Defaults to value of LAST_IMPORT_SHA in ssm parameter store",
69	)
70	parser.add_argument(	1✔
71	"-t",
72	"--to-sha",
73	help="<Optional> The latest commit hash against which to search for changed scripts. Defaults to HEAD",
74	default=git_rev_parse("HEAD"),
75	)
76	parser.add_argument("--post-deploy", action="store_true", default=False)	1✔
77
78	def run_scripts(self, changed_paths, opts):	1✔
79	for script in changed_paths:	1✔
80	script_path = Path(script)	1✔
81	try:	1✔
82	call_command(script_path.stem, **opts)	1✔
83	self.summary.append(	×
84	(
85	"INFO",
86	f"Ran import script: {script_path.name}",
87	)
88	)
89	except Exception as e:	1✔
90	# usually we want to handle a specific exception, but in this situation
91	# if there is any issue (at all) trying to run the command,
92	# we just want to log it and move on to the next script
93	self.summary.append(	1✔
94	("WARNING", f"{script_path.name} could not be run. Due to {e}")
95	)
96	continue	1✔
97
98	def run_misc_fixes(self):	1✔
99	call_command("misc_fixes")	1✔
100
101	def output_summary(self):	1✔
102	for line in self.summary:	1✔
103	if line[0] == "INFO":	1✔
104	self.stdout.write(line[1])	×
105	elif line[0] == "WARNING":	1✔
106	self.stdout.write(self.style.ERROR(line[1]))	1✔
107	else:
108	self.stdout.write(line[1])	×
109
110	def update_last_import_sha_on_ssm(self, to_sha):	1✔
111	try:	1✔
112	ssm_client = boto3.client("ssm")	1✔
113	ssm_client.put_parameter(	1✔
114	Name="LAST_IMPORT_SHA", Value=to_sha, Overwrite=True
115	)
116	except botocore.exceptions.ClientError as e:	1✔
117	self.summary.append(	1✔
118	(
119	"WARNING",
120	f"LAST_IMPORT_SHA not updated in parameter store, due to:\n{e}",
121	)
122	)
123	except botocore.exceptions.NoCredentialsError as e:	×
124	self.summary.append(	×
125	(
126	"WARNING",
127	f"LAST_IMPORT_SHA not updated in parameter store, due to:\n{e}",
128	)
129	)
130
131	def handle(self, args, *options):	1✔
132	self.check(	1✔
133	[
134	apps.get_app_config("data_importers"),
135	apps.get_app_config("pollingstations"),
136	]
137	)
138
139	if not (from_sha := options.get("from_sha")):	1✔
140	from_sha = get_last_import_sha_from_ssm()	×
141
142	to_sha = options.get("to_sha")	1✔
143	is_post_deploy = options.get("post_deploy")	1✔
144
145	changed_paths = get_paths_changed(from_sha, to_sha)	1✔
146	changed_scripts = get_changed_scripts(changed_paths)	1✔
147	has_imports = any_import_scripts(changed_paths)	1✔
148	has_application = any_non_import_scripts(changed_paths)	1✔
149
150	cmd_opts = {	1✔
151	"nochecks": True,
152	"verbosity": 1,
153	"use_postcode_centroids": False,
154	"include_past_elections": False,
155	}
156
157	self.stdout.write(	1✔
158	f"Comparing repo between {from_sha} and {to_sha}\n"
159	f"\tFrom: https://github.com/DemocracyClub/UK-Polling-Stations/commit/{git_rev_parse(from_sha)}\n"
160	f"\tTo: https://github.com/DemocracyClub/UK-Polling-Stations/commit/{git_rev_parse(to_sha)}\n"
161	)
162
163	if has_imports and not has_application:	1✔
164	self.stdout.write("Only import scripts have changed\n")	1✔
165	self.stdout.write("Running import scripts\n")	1✔
166	self.run_scripts(changed_scripts, cmd_opts)	1✔
167	self.run_misc_fixes()	1✔
168	self.stdout.write("updating LAST_IMPORT_SHA on ssm")	1✔
169	self.update_last_import_sha_on_ssm(to_sha)	1✔
170	elif has_imports and has_application and is_post_deploy:	1✔
171	self.stdout.write("App has deployed. OK to run import scripts")	1✔
172	self.run_scripts(changed_scripts, cmd_opts)	1✔
173	self.run_misc_fixes()	1✔
174	self.update_last_import_sha_on_ssm(to_sha)	1✔
175	elif has_imports and has_application and not is_post_deploy:	1✔
176	self.stdout.write("Need to deploy before running import scripts\n")	1✔
177	elif not has_imports:	1✔
178	self.stdout.write(	1✔
179	"No import scripts have changed. So nothing new to import.\n"
180	)
181	self.update_last_import_sha_on_ssm(to_sha)	1✔
182	else:
183	self.stdout.write("Not running import scripts")	×
184
185	self.output_summary()	1✔

DemocracyClub / UK-Polling-Stations / 28b2cb5e-29d5-4b2a-bba5-65e9410e6204

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