Add mypy / flake8 checks for python files

Author: danielfullmer

.flake8

@@ -0,0 +1,2 @@
+[flake8]
+max-line-length = 120

check.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+mypy --exclude apks/chromium .
+flake8 --exclude apks/chromium .

flake.nix

@@ -27,8 +27,6 @@
     nixosModule = import ./nixos; # Contains all robotnix nixos modules
     nixosModules.attestation-server = import ./nixos/attestation-server/module.nix;
 
-    checks.x86_64-linux = {};
-
     packages.x86_64-linux = {
       manual = (import ./docs { inherit pkgs; }).manual;
     };

flavors/lineageos/update-device-dirs.py

@@ -2,11 +2,11 @@
 # SPDX-FileCopyrightText: 2020 Daniel Fullmer and robotnix contributors
 # SPDX-License-Identifier: MIT
 
+from typing import Any, Dict, List, Tuple
 import argparse
 import json
 import os
 import subprocess
-import urllib.request
 
 # A full run took approximately 12 minutes total. Needed to set TMPDIR=/tmp
 #
@@ -18,45 +18,57 @@
 VENDOR_REPO_BASE = "https://github.com/TheMuppets"
 ROBOTNIX_GIT_MIRRORS = os.environ.get('ROBOTNIX_GIT_MIRRORS', '')
 if ROBOTNIX_GIT_MIRRORS:
-    MIRRORS = dict(mirror.split("=") for mirror in ROBOTNIX_GIT_MIRRORS).split('|'))
+    MIRRORS: Dict[str, str] = dict(
+            (mirror.split("=")[0], mirror.split("=")[1])
+            for mirror in ROBOTNIX_GIT_MIRRORS.split('|')
+            )
 else:
     MIRRORS = {}
-REMOTE_REFS = {} # url: { ref: rev }
+REMOTE_REFS: Dict[str, Dict[str, str]] = {}  # url: { ref: rev }
 
-def save(filename, data):
+
+def save(filename: str, data: str) -> None:
     open(filename, 'w').write(json.dumps(data, sort_keys=True, indent=2, separators=(',', ': ')))
 
-def get_mirrored_url(url):
+
+def get_mirrored_url(url: str) -> str:
     for mirror_url, mirror_path in MIRRORS.items():
         if url.startswith(mirror_url):
             url = url.replace(mirror_url, mirror_path)
     return url
 
-def ls_remote(url):
+
+def ls_remote(url: str) -> Dict[str, str]:
     if url in REMOTE_REFS:
         return REMOTE_REFS[url]
 
     orig_url = url
     url = get_mirrored_url(url)
 
-    remote_info = subprocess.check_output([ "git", "ls-remote", url ]).decode()
-    REMOTE_REFS[orig_url] = dict(reversed(line.split('\t')) for line in remote_info.split('\n') if line)
+    remote_info = subprocess.check_output(["git", "ls-remote", url]).decode()
+    REMOTE_REFS[orig_url] = {}
+    for line in remote_info.split('\n'):
+        if line:
+            ref, rev = reversed(line.split('\t'))
+            REMOTE_REFS[orig_url][ref] = rev
     return REMOTE_REFS[orig_url]
 
-def checkout_git(url, rev):
+
+def checkout_git(url: str, rev: str) -> Any:
     print("Checking out %s %s" % (url, rev))
-    json_text = subprocess.check_output([ "nix-prefetch-git", "--url", url, "--rev", rev]).decode()
+    json_text = subprocess.check_output(["nix-prefetch-git", "--url", url, "--rev", rev]).decode()
     return json.loads(json_text)
 
-def fetch_relpath(dirs, relpath, url, branch):
+
+def fetch_relpath(dirs: Dict[str, Any], relpath: str, url: str, branch: str) -> Any:
     orig_url = url
     url = get_mirrored_url(url)
 
     current_rev = dirs.get(relpath, {}).get('rev', None)
     refs = ls_remote(url)
     ref = f'refs/heads/{branch}'
     if ref not in refs:
-        raise Exception(f'{url} is missing {ref}')
+        raise ValueError(f'{url} is missing {ref}')
     newest_rev = refs[ref]
     if (current_rev != newest_rev
             or ('path' not in dirs[relpath])
@@ -70,14 +82,14 @@ def fetch_relpath(dirs, relpath, url, branch):
 
 
 # Fetch device source trees for devices in metadata and save their information into filename
-def fetch_device_dirs(metadata, filename, branch):
+def fetch_device_dirs(metadata: Any, filename: str, branch: str) -> Tuple[Any, Dict[str, List[str]]]:
     if os.path.exists(filename):
         dirs = json.load(open(filename))
     else:
         dirs = {}
 
-    dirs_to_fetch = set() # Pairs of (relpath, url)
-    dirs_fetched = set() # Just strings of relpath
+    dirs_to_fetch = set()  # Pairs of (relpath, url)
+    dirs_fetched = set()  # Just strings of relpath
     for device, data in metadata.items():
         vendor = data['vendor']
         url = f'{LINEAGE_REPO_BASE}/android_device_{vendor}_{device}'
@@ -88,12 +100,12 @@ def fetch_device_dirs(metadata, filename, branch):
         else:
             print(f'SKIP: {branch} branch does not exist for {device}')
 
-    dir_dependencies = {} # key -> [ values ]. 
+    dir_dependencies: Dict[str, List[str]] = {}  # key -> [ values ].
     while len(dirs_to_fetch) > 0:
         relpath, url = dirs_to_fetch.pop()
         try:
             dir_info = fetch_relpath(dirs, relpath, url, branch)
-        except:
+        except ValueError:
             continue
 
         # Also grab any dirs that this one depends on
@@ -105,16 +117,17 @@ def fetch_device_dirs(metadata, filename, branch):
                 if dep['target_path'] not in dirs_fetched:
                     dirs_to_fetch.add((dep['target_path'], f"{LINEAGE_REPO_BASE}/{dep['repository']}"))
 
-            dir_info['deps'] = [ dep['target_path'] for dep in lineage_dependencies ]
+            dir_info['deps'] = [dep['target_path'] for dep in lineage_dependencies]
         else:
             dir_info['deps'] = []
 
-        save(filename, dirs) # Save after every step, for resuming
+        save(filename, dirs)  # Save after every step, for resuming
         dirs_fetched.add(relpath)
 
     return dirs, dir_dependencies
 
-def fetch_vendor_dirs(metadata, filename, branch):
+
+def fetch_vendor_dirs(metadata: Any, filename: str, branch: str) -> Any:
     required_vendor = set()
     for device, data in metadata.items():
         if 'vendor' in data:
@@ -151,11 +164,13 @@ def fetch_vendor_dirs(metadata, filename, branch):
 
     return dirs
 
-def main():
+
+def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument('--branch', help="lineageos version")
     parser.add_argument('product', nargs='*',
-                        help='product to fetch directory metadata for, specified by <vendor>_<device> (example: google_crosshatch) '
+                        help='product to fetch directory metadata for, specified by <vendor>_<device> '
+                        '(example: google_crosshatch) '
                         'If no products are specified, all products in device-metadata.json will be updated')
     args = parser.parse_args()
 
@@ -165,10 +180,11 @@ def main():
         metadata = {}
         for product in args.product:
             vendor, device = product.split('_', 1)
-            metadata[device] = { 'vendor': vendor }
+            metadata[device] = {'vendor': vendor}
+
+    fetch_device_dirs(metadata, os.path.join(args.branch, 'device-dirs.json'), args.branch)
+    fetch_vendor_dirs(metadata, os.path.join(args.branch, 'vendor-dirs.json'), args.branch)
 
-    device_dirs, dir_dependencies = fetch_device_dirs(metadata, os.path.join(args.branch, 'device-dirs.json'), args.branch)
-    vendor_dirs = fetch_vendor_dirs(metadata, os.path.join(args.branch, 'vendor-dirs.json'), args.branch)
 
 if __name__ == '__main__':
     main()

flavors/lineageos/update-device-metadata.py

@@ -2,17 +2,21 @@
 # SPDX-FileCopyrightText: 2020 Daniel Fullmer and robotnix contributors
 # SPDX-License-Identifier: MIT
 
+from typing import Any
 import json
-import os
 import urllib.request
 
-def save(filename, data):
+
+def save(filename: str, data: str) -> None:
     open(filename, 'w').write(json.dumps(data, sort_keys=True, indent=2, separators=(',', ': ')))
 
-def fetch_metadata():
+
+def fetch_metadata() -> Any:
     metadata = {}
 
-    lineage_build_targets_str = urllib.request.urlopen("https://github.com/LineageOS/hudson/raw/master/lineage-build-targets").read().decode()
+    lineage_build_targets_str = urllib.request.urlopen(
+            "https://github.com/LineageOS/hudson/raw/master/lineage-build-targets"
+            ).read().decode()
     for line in lineage_build_targets_str.split("\n"):
         line = line.strip()
         if line == "":
@@ -54,6 +58,7 @@ def fetch_metadata():
 
     return metadata
 
+
 if __name__ == '__main__':
     metadata = fetch_metadata()
     save('device-metadata.json', metadata)

mk-repo-file.py

@@ -2,7 +2,7 @@
 # SPDX-FileCopyrightText: 2020 Daniel Fullmer and robotnix contributors
 # SPDX-License-Identifier: MIT
 
-from typing import Optional, Dict, List, Tuple
+from typing import Any, Optional, Dict, List, Tuple
 from enum import Enum
 
 import argparse
@@ -20,45 +20,56 @@
     "--depth=1",
 ]
 
+
 # The kind of remote a "commitish" refers to.
 # These are used for the --ref-type CLI arg.
 class ManifestRefType(Enum):
     BRANCH = "heads"
     TAG = "tags"
 
+
 revHashes: Dict[Tuple[str, bool], str] = {}  # (rev, fetch_submodules) -> sha256hash
 revTrees: Dict[str, str] = {}           # rev -> treeHash
-treeHashes: Dict[Tuple[str, bool], str] = {} # (treeHash, fetch_submodules) -> sha256hash
+treeHashes: Dict[Tuple[str, bool], str] = {}  # (treeHash, fetch_submodules) -> sha256hash
 
-def save(filename, data):
+
+def save(filename: str, data: Any) -> None:
     open(filename, 'w').write(json.dumps(data, sort_keys=True, indent=2, separators=(',', ': ')))
 
-def checkout_git(url, rev, fetch_submodules=False):
+
+def checkout_git(url: str, rev: str, fetch_submodules: bool = False) -> Any:
     print("Checking out %s %s" % (url, rev))
-    args = [ "nix-prefetch-git", "--url", url, "--rev", rev ]
+    args = ["nix-prefetch-git", "--url", url, "--rev", rev]
     if fetch_submodules:
         args.append("--fetch-submodules")
     json_text = subprocess.check_output(args).decode()
     return json.loads(json_text)
 
-def ls_remote(url, rev):
-    remote_info = subprocess.check_output([ "git", "ls-remote", url, rev ]).decode()
+
+def ls_remote(url: str, rev: str) -> str:
+    remote_info = subprocess.check_output(["git", "ls-remote", url, rev]).decode()
     remote_rev = remote_info.split('\t')[0]
     assert remote_rev != ""
     return remote_rev
 
+
 def make_repo_file(url: str, ref: str, filename: str, ref_type: ManifestRefType,
                    override_project_revs: Dict[str, str], resume: bool,
                    mirrors: Dict[str, str], project_fetch_submodules: List[str],
                    override_tag: Optional[str], include_prefix: List[str],
-                   exclude_path: List[str]):
+                   exclude_path: List[str]) -> None:
     if resume and os.path.exists(filename):
         data = json.load(open(filename))
     else:
         print("Fetching information for %s %s" % (url, ref))
         with tempfile.TemporaryDirectory() as tmpdir:
-            subprocess.check_call(['repo', 'init', f'--manifest-url={url}', f'--manifest-branch=refs/{ref_type.value}/{ref}', *REPO_FLAGS], cwd=tmpdir)
-            json_text = subprocess.check_output(['repo', 'dumpjson'] + (["--local-only"] if override_project_revs else []), cwd=tmpdir).decode()
+            subprocess.check_call([
+                'repo', 'init', f'--manifest-url={url}', f'--manifest-branch=refs/{ref_type.value}/{ref}', *REPO_FLAGS
+                ], cwd=tmpdir)
+            json_text = subprocess.check_output(
+                    ['repo', 'dumpjson']
+                    + (["--local-only"] if override_project_revs else []),
+                    cwd=tmpdir).decode()
             data = json.loads(json_text)
 
             save(filename, data)
@@ -109,7 +120,9 @@ def make_repo_file(url: str, ref: str, filename: str, ref_type: ManifestRefType,
             for mirror_url, mirror_path in mirrors.items():
                 if p['url'].startswith(mirror_url):
                     p_url = p['url'].replace(mirror_url, mirror_path)
-                    p['tree'] = subprocess.check_output(['git', 'log','-1', '--pretty=%T', p['rev']], cwd=p_url+'.git').decode().strip()
+                    p['tree'] = subprocess.check_output(
+                        ['git', 'log', '-1', '--pretty=%T', p['rev']],
+                        cwd=p_url+'.git').decode().strip()
                     if (p['tree'], fetch_submodules) in treeHashes:
                         p['sha256'] = treeHashes[p['tree'], fetch_submodules]
                         found_treehash = True
@@ -135,16 +148,19 @@ def make_repo_file(url: str, ref: str, filename: str, ref_type: ManifestRefType,
     # Save at the end as well!
     save(filename, data)
 
-def main():
+
+def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument('--out', default=None, help="path to output file, defaults to repo-{rev}.json")
     parser.add_argument('--ref-type', help="the kind of ref that is to be fetched",
                         choices=[t.name.lower() for t in ManifestRefType], default=ManifestRefType.TAG.name.lower())
     parser.add_argument('--resume', help="resume a previous download", action='store_true')
     parser.add_argument('--repo-prop', help="repo.prop file to use as source for project git revisions")
     parser.add_argument('--override-tag', help="tag to fetch for subrepos, ignoring revisions from manifest")
-    parser.add_argument('--project-fetch-submodules', action="append", default=[], help="fetch submodules for the specified project path")
-    parser.add_argument('--include-prefix', action="append", default=[], help="only include paths if they start with the specified prefix")
+    parser.add_argument('--project-fetch-submodules', action="append", default=[],
+                        help="fetch submodules for the specified project path")
+    parser.add_argument('--include-prefix', action="append", default=[],
+                        help="only include paths if they start with the specified prefix")
     parser.add_argument('--exclude-path', action="append", default=[], help="paths to exclude from fetching")
     parser.add_argument('url', help="manifest URL")
     parser.add_argument('ref', help="manifest ref")
@@ -153,7 +169,10 @@ def main():
 
     ROBOTNIX_GIT_MIRRORS = os.environ.get('ROBOTNIX_GIT_MIRRORS', '')
     if ROBOTNIX_GIT_MIRRORS:
-        mirrors = dict(mirror.split("=") for mirror in ROBOTNIX_GIT_MIRRORS.split('|'))
+        mirrors: Dict[str, str] = dict(
+                (mirror.split("=")[0], mirror.split("=")[1])
+                for mirror in ROBOTNIX_GIT_MIRRORS.split('|')
+                )
     else:
         mirrors = {}
 
@@ -192,5 +211,6 @@ def main():
                    exclude_path=args.exclude_path,
                    )
 
+
 if __name__ == "__main__":
     main()

mypy.ini

@@ -0,0 +1,2 @@
+[mypy]
+strict = true

shell.nix

@@ -3,7 +3,7 @@ pkgs.mkShell {
   name = "robotnix-scripts";
   nativeBuildInputs = with pkgs; [
     # For android updater scripts
-    python3
+    (python3.withPackages (p: with p; [ mypy flake8 pytest ]))
     gitRepo nix-prefetch-git
     curl go-pup jq
 
@@ -12,4 +12,5 @@ pkgs.mkShell {
 
     cachix
   ];
+  PYTHONPATH=".";
 }