Add additional tests and fix code issues

This adds a series of tests to validate that we can actually pull images
from our public repository.

Cleans up the tests to exclusively use pytest, and fixes a series of py3
issues.

Test: All pytests are green
Bug: 280634466

Author: Erwin Jansen

.gitignore

@@ -2,6 +2,7 @@
 __pycache__/
 *.py[cod]
 *$py.class
+bld/
 
 # C extensions
 *.so
@@ -98,7 +99,7 @@ celerybeat-schedule
 
 # Environments
 .env
-.venv
+.venv/
 env/
 venv/
 ENV/

aemu-container.code-workspace

@@ -8,6 +8,9 @@
 		}
 	],
 	"settings": {
-		"python.formatting.provider": "black"
+		"python.formatting.provider": "none",
+		"[python]": {
+			"editor.defaultFormatter": "ms-python.black-formatter"
+		}
 	}
 }

emu/android_release_zip.py

@@ -16,12 +16,17 @@
 import os
 import shutil
 import zipfile
+from typing import Dict, Set, Union
 
 from tqdm import tqdm
 
 from emu.utils import api_codename
 
 
+class NotAZipfile(Exception):
+    pass
+
+
 class AndroidReleaseZip(object):
     """Provides information of released android products.
 
@@ -30,74 +35,84 @@ class AndroidReleaseZip(object):
     about the contents of the zip.
     """
 
-    def __init__(self, file_name):
-        self.file_name = file_name
+    def __init__(self, file_name: str):
+        self.file_name: str = file_name
         if not zipfile.is_zipfile(file_name):
-            raise Exception("{} is not a zipfile!".format(file_name))
+            raise NotAZipfile(f"{file_name} is not a zipfile!")
+
         with zipfile.ZipFile(file_name, "r") as zip_file:
-            self.props = collections.defaultdict(set)
-            files = [x for x in zip_file.infolist() if "source.properties" in x.filename or "build.prop" in x.filename]
+            self.props: Dict[str, Set[str]] = collections.defaultdict(set)
+            files = [
+                x
+                for x in zip_file.infolist()
+                if "source.properties" in x.filename or "build.prop" in x.filename
+            ]
             for file in files:
                 for key, value in self._unpack_properties(zip_file, file).items():
                     self.props[key] = value
 
-    def _unpack_properties(self, zip_file, zip_info):
+    def _unpack_properties(
+        self, zip_file: zipfile.ZipFile, zip_info: zipfile.ZipInfo
+    ) -> Dict[str, str]:
         prop = zip_file.read(zip_info).decode("utf-8").splitlines()
         res = dict([a.split("=") for a in prop if "=" in a])
         return res
 
-    def __str__(self):
-        return "{}-{}".format(self.description(), self.revision())
+    def __str__(self) -> str:
+        return f"{self.description()}-{self.revision()}"
 
-    def description(self):
+    def description(self) -> Union[str, None]:
         """Descripton of this release."""
         return self.props.get("Pkg.Desc")
 
-    def revision(self):
+    def revision(self) -> Union[str, None]:
         """The revision of this release."""
         return self.props.get("Pkg.Revision")
 
-    def build_id(self):
+    def build_id(self) -> str:
         """The Pkg.BuildId or revision if build id is not available."""
         if "Pkg.BuildId" in self.props:
             return self.props.get("Pkg.BuildId")
         return self.revision()
 
-    def is_system_image(self):
+    def is_system_image(self) -> bool:
         """True if this zip file contains a system image."""
-        return "System Image" in self.description() or "Android SDK Platform" in self.description()
+        return (
+            "System Image" in self.description()
+            or "Android SDK Platform" in self.description()
+        )
 
-    def is_emulator(self):
+    def is_emulator(self) -> bool:
         """True if this zip file contains the android emulator."""
         return "Android Emulator" in self.description()
 
-    def copy(self, destination):
+    def copy(self, destination: str) -> str:
         """Copy the zipfile to the given destination.
 
         If the destination is the same as this zipfile the current path
         will be returned a no copy is made.
 
         Args:
-            destination ({string}): The destination to copy this zip to.
+            destination (str): The destination to copy this zip to.
 
         Returns:
-            {string}: The path where this zip file was copied to
+            str: The path where this zip file was copied to.
         """
         try:
             return shutil.copy2(self.file_name, destination)
         except shutil.SameFileError:
             logging.warning("Will not copy to itself, ignoring..")
             return self.file_name
 
-    def extract(self, destination):
+    def extract(self, destination: str) -> None:
         """Extract this release zip to the given destination
 
         Args:
-            destination ({string}): The destination to extract the zipfile to.
+            destination (str): The destination to extract the zipfile to.
         """
 
         zip_file = zipfile.ZipFile(self.file_name)
-        print("Extracting: {} -> {}".format(self.file_name, destination))
+        print(f"Extracting: {self.file_name} -> {destination}")
         for info in tqdm(iterable=zip_file.infolist(), total=len(zip_file.infolist())):
             filename = zip_file.extract(info, path=destination)
             mode = info.external_attr >> 16
@@ -108,61 +123,71 @@ def extract(self, destination):
 class SystemImageReleaseZip(AndroidReleaseZip):
     """An Android Release Zipfile containing an emulator system image."""
 
-    ABI_CPU_MAP = {"armeabi-v7a": "arm", "arm64-v8a": "arm64", "x86_64": "x86_64", "x86": "x86"}
-    SHORT_MAP = {"armeabi-v7a": "a32", "arm64-v8a": "a64", "x86_64": "x64", "x86": "x86"}
-    SHORT_TAG = {
+    ABI_CPU_MAP: Dict[str, str] = {
+        "armeabi-v7a": "arm",
+        "arm64-v8a": "arm64",
+        "x86_64": "x86_64",
+        "x86": "x86",
+    }
+    SHORT_MAP: Dict[str, str] = {
+        "armeabi-v7a": "a32",
+        "arm64-v8a": "a64",
+        "x86_64": "x64",
+        "x86": "x86",
+    }
+    SHORT_TAG: Dict[str, str] = {
         "android": "aosp",
         "google_apis": "google",
         "google_apis_playstore": "playstore",
         "google_ndk_playstore": "ndk_playstore",
         "android-tv": "tv",
     }
 
-    def __init__(self, file_name):
+    def __init__(self, file_name: str):
         super().__init__(file_name)
         if not self.is_system_image():
-            raise Exception("{} is not a zip file with a system image".format(file_name))
+            raise NotAZipfile(f"{file_name} is not a zip file with a system image")
 
         self.props["qemu.cpu"] = self.qemu_cpu()
         self.props["qemu.tag"] = self.tag()
         self.props["qemu.short_tag"] = self.short_tag()
         self.props["qemu.short_abi"] = self.short_abi()
 
-    def api(self):
+    def api(self) -> str:
         """The api level, if any."""
         return self.props.get("AndroidVersion.ApiLevel", "")
 
-    def codename(self):
+    def codename(self) -> str:
         """First letter of the desert, if any."""
         if "AndroidVersion.CodeName" in self.props:
             return self.props["AndroidVersion.CodeName"]
         return api_codename(self.api())
 
-    def abi(self):
+    def abi(self) -> str:
         """The abi if any."""
         return self.props.get("SystemImage.Abi", "")
 
-    def short_abi(self):
+    def short_abi(self) -> str:
         """Shortened version of the ABI string."""
         if self.abi() not in self.SHORT_MAP:
             logging.error("%s not in short map", self)
         return self.SHORT_MAP[self.abi()]
 
-    def qemu_cpu(self):
+    def qemu_cpu(self) -> str:
         """Returns the cpu architecture, derived from the abi."""
         return self.ABI_CPU_MAP.get(self.abi(), "None")
 
-    def gpu(self):
+    def gpu(self) -> str:
         """Returns whether or not the system has gpu support."""
         return self.props.get("SystemImage.GpuSupport")
 
-    def tag(self):
+    def tag(self) -> str:
         """The tag associated with this release."""
         tag = self.props.get("SystemImage.TagId", "")
         if tag == "default" or tag.strip() == "":
             tag = "android"
         return tag
 
-    def short_tag(self):
+    def short_tag(self) -> str:
         """A shorthand tag."""
         return self.SHORT_TAG[self.tag()]

emu/cloud_build.py

@@ -99,26 +99,25 @@ def cloud_build(args):
                 emulators.add(emulator_container.props["emu_build_id"])
                 steps.append(create_build_step(emulator_container, args.dest))
                 images.append(emulator_container.full_name())
-                images.append(emulator_container.latest_name())
                 emulator_images.append(emulator_container.full_name())
-                emulator_images.append(emulator_container.latest_name())
 
     cloudbuild = {"steps": steps, "images": images, "timeout": "21600s"}
     logging.info("Writing cloud yaml [%s] in %s", yaml, args.dest)
-    with open(os.path.join(args.dest, "cloudbuild.yaml"), "w") as ymlfile:
+    with open(os.path.join(args.dest, "cloudbuild.yaml"), "w", encoding="utf-8") as ymlfile:
         yaml.dump(cloudbuild, ymlfile)
 
     writer = TemplateWriter(args.dest)
     writer.write_template(
         "cloudbuild.README.MD",
-        {"emu_version": ", ".join(emulators), "emu_images": "\n".join(["* {}".format(x) for x in emulator_images])},
+        {"emu_version": ", ".join(emulators),
+         "emu_images": "\n".join([f"* {x}" for x in emulator_images])},
         rename_as="README.MD",
     )
     writer.write_template(
         "registry.README.MD",
         {
             "emu_version": ", ".join(emulators),
-            "emu_images": "\n".join(["* {}".format(x) for x in images]),
+            "emu_images": "\n".join([f"* {x}" for x in images]),
             "first_image": next(iter(images), None),
         },
         rename_as="REGISTRY.MD",