Merge pull request #1683 from ogayot/mypy-diff-add-local-script

scripts: add script to run mypy and compare with output in another revision

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

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()

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: