diff --git a/apt-deps.txt b/apt-deps.txt index fee47027..bf4587c3 100644 --- a/apt-deps.txt +++ b/apt-deps.txt @@ -27,6 +27,7 @@ python3-distutils-extra python3-flake8 python3-jsonschema python3-more-itertools +python3-mypy python3-nose python3-parameterized python3-pip diff --git a/scripts/run-mypy.py b/scripts/run-mypy.py new file mode 100755 index 00000000..9cf7cb15 --- /dev/null +++ b/scripts/run-mypy.py @@ -0,0 +1,93 @@ +#!/usr/bin/env python3 + +import argparse +import contextlib +import os +import subprocess +import tempfile +from typing import TextIO + + +def run_mypy(cwd: str | None, stdout: TextIO | None) -> None: + ''' Execute mypy and if requested, redirects the output to a file. ''' + mypy_cmd = [ + 'python3', '-m', 'mypy', + '--ignore-missing-imports', '--check-untyped-defs', + 'subiquity', 'subiquitycore', 'system_setup', 'console_conf', + 'scripts/replay-curtin-log.py', + ] + + subprocess.run(mypy_cmd, check=False, text=True, cwd=cwd, stdout=stdout) + + +@contextlib.contextmanager +def worktree(rev: str) -> str: + ''' When entered, deploy a git-worktree at the specified revision. Upon + exiting, remove the worktree. ''' + try: + with tempfile.TemporaryDirectory(suffix='.subiquity-worktree') as wt_dir: + subprocess.run([ + 'git', 'worktree', 'add', wt_dir, + '--detach', rev, + ], check=True) + subprocess.run([ + 'git', 'clone', 'probert', f'{wt_dir}/probert', + ], check=True) + subprocess.run([ + 'scripts/update-part.py', 'probert', + ], check=True, cwd=wt_dir) + subprocess.run([ + 'git', 'clone', 'curtin', f'{wt_dir}/curtin', + ], check=True) + subprocess.run([ + 'scripts/update-part.py', 'curtin', + ], check=True, cwd=wt_dir) + yield wt_dir + finally: + subprocess.run(['git', 'worktree', 'remove', '--force', wt_dir]) + + +def main() -> None: + parser = argparse.ArgumentParser() + + parser.add_argument('--diff-against', metavar='git-revision') + parser.add_argument('--checkout-head', action='store_true') + + args = parser.parse_args() + + if args.checkout_head: + cm_wd_head = worktree('HEAD') + else: + cm_wd_head = contextlib.nullcontext(enter_result=os.getcwd()) + + if args.diff_against is not None: + cm_wd_base = worktree(args.diff_against) + cm_stdout_head = tempfile.NamedTemporaryFile(suffix='.out') + cm_stdout_base = tempfile.NamedTemporaryFile(suffix='.out') + else: + cm_wd_base = contextlib.nullcontext() + cm_stdout_head = contextlib.nullcontext() + cm_stdout_base = contextlib.nullcontext() + + # Setup the output file(s) and worktree(s). + with ( + cm_wd_head as cwd_head, + cm_wd_base as cwd_base, + cm_stdout_head as stdout_head, + cm_stdout_base as stdout_base): + # Execute mypy on the head revision + run_mypy(stdout=stdout_head, cwd=cwd_head) + + if args.diff_against is None: + return + + run_mypy(stdout=stdout_base, cwd=cwd_base) + + subprocess.run([ + 'diff', '--color=always', '--unified=0', + '--', stdout_base.name, stdout_head.name, + ]) + + +if __name__ == '__main__': + main() diff --git a/subiquity/server/controllers/mirror.py b/subiquity/server/controllers/mirror.py index b68bc4bf..592d3188 100644 --- a/subiquity/server/controllers/mirror.py +++ b/subiquity/server/controllers/mirror.py @@ -316,6 +316,7 @@ class MirrorController(SubiquityController): # recommended. Clients should do a POST request to /mirror with # null as the body instead. await self.run_mirror_selection_or_fallback(self.context) + assert self.final_apt_configurer is not None await self.final_apt_configurer.apply_apt_config( self.context, final=True) @@ -326,6 +327,7 @@ class MirrorController(SubiquityController): # apply_apt_config and run_apt_config_check. Just make sure we still # use the original one. configurer = self.test_apt_configurer + assert configurer is not None await configurer.apply_apt_config( self.context, final=False) await configurer.run_apt_config_check(output) @@ -334,6 +336,7 @@ class MirrorController(SubiquityController): self.final_apt_configurer = get_apt_configurer( self.app, self.app.controllers.Source.get_handler(variation_name)) await self._promote_mirror() + assert self.final_apt_configurer is not None return self.final_apt_configurer async def GET(self) -> MirrorGet: