Add tests for ./tools/cl

Makes a temporary copy of the git repo for each test so we can
modify the repo for testing purposes.
It's not the fastest test (~12s), but is only run when python files
have been modified. In contrast to other developer tooling, tools/cl
is not used by Luci, so it needs a dedicated test.

BUG=b:244185215
TEST=./tools/health-check

Change-Id: I06c90a580aa8ed0fa267a41ca40895710121767f
Reviewed-on: https://chromium-review.googlesource.com/c/crosvm/crosvm/+/3866692
Tested-by: Dennis Kempin <denniskempin@google.com>
Reviewed-by: Daniel Verkamp <dverkamp@chromium.org>
Commit-Queue: Dennis Kempin <denniskempin@google.com>

tools/cl

@@ -4,8 +4,9 @@
 # found in the LICENSE file.
 
 import functools
+from os import chdir
 from pathlib import Path
-from impl.common import GerritChange, confirm, run_commands, cmd
+from impl.common import CROSVM_ROOT, GerritChange, confirm, run_commands, cmd
 import sys
 
 USAGE = """\
@@ -184,18 +185,18 @@ def rebase():
     branch_name = git("branch --show-current").stdout()
     upstream_branch_name = branch_name + "-upstream"
 
-    print(f"Checking out '{upstream_branch_name}'")
-    rev = git("rev-parse", upstream_branch_name).stdout(check=False)
-    if rev:
-        print(f"Leaving behind previous revision of {upstream_branch_name}: {rev}")
-    git("fetch origin").fg()
-    git("checkout -B", upstream_branch_name, "origin/main").fg(quiet=True)
+    if git("rev-parse", upstream_branch_name).success():
+        print(f"Overwriting existing branch {upstream_branch_name}")
+        git("log -n1", upstream_branch_name).fg()
+
+    git("fetch -q origin main").fg()
+    git("checkout -B", upstream_branch_name, "origin/main").fg()
 
     print(f"Cherry-picking changes from {branch_name}")
     git(f"cherry-pick {branch_name}@{{u}}..{branch_name}").fg()
 
 
-def upload():
+def upload(dry_run: bool = False):
     """
     Uploads changes to the crosvm main branch.
     """
@@ -218,9 +219,10 @@ def upload():
             return
         print()
 
-    git("push origin HEAD:refs/for/main").fg()
+    git("push origin HEAD:refs/for/main").fg(dry_run=dry_run)
 
 
 if __name__ == "__main__":
+    chdir(CROSVM_ROOT)
     prerequisites()
     run_commands(upload, rebase, status, prune, usage=USAGE)

tools/dev_container

@@ -183,7 +183,12 @@ def main(
 
 
 class TestDevContainer(unittest.TestCase):
-    """Runs live tests using the docker service."""
+    """
+    Runs live tests using the docker service.
+
+    Note: This test is not run by health-check since it cannot be run inside the
+    container. It is run by infra/recipes/health_check.py before running health checks.
+    """
 
     docker: cmd
     docker_args = [

tools/health-check

@@ -11,13 +11,17 @@ from pathlib import Path
 from typing import List
 
 from impl.check_code_hygiene import has_crlf_line_endings
-from impl.common import CROSVM_ROOT, argh, chdir, cmd, cwd_context, parallel, run_main
+from impl.common import CROSVM_ROOT, TOOLS_ROOT, argh, chdir, cmd, cwd_context, parallel, run_main
 from impl.health_check import Check, CheckContext, run_checks
 
 
-def check_python_tests(context: CheckContext):
-    "Run all non-main python files to execute their unit tests."
-    parallel(*cmd("python3").foreach(context.all_files)).fg()
+def check_python_tests(_: CheckContext):
+    "No matter which python files have changed, run all available python tests."
+    PYTHON_TESTS = [
+        *TOOLS_ROOT.glob("tests/*.py"),
+        TOOLS_ROOT / "impl/common.py",
+    ]
+    parallel(*cmd("python3").foreach(PYTHON_TESTS)).fg()
 
 
 def check_python_types(context: CheckContext):
@@ -179,7 +183,8 @@ CHECKS: List[Check] = [
     ),
     Check(
         check_python_tests,
-        files=["tools/impl/common.py"],
+        files=["tools/**.py"],
+        python_tools=True,
     ),
     Check(
         check_python_types,

tools/impl/common.py

@@ -17,11 +17,6 @@ import json
 import sys
 import subprocess
 
-if sys.version_info.major != 3 or sys.version_info.minor < 8:
-    print("Python 3.8 or higher is required.")
-    print("Hint: Do not use crosvm tools inside cros_sdk.")
-    sys.exit(1)
-
 
 def ensure_package_exists(package: str):
     """Installs the specified package via pip if it does not exist."""
@@ -59,23 +54,47 @@ import re
 import shutil
 import traceback
 
-"Root directory of crosvm"
-CROSVM_ROOT = Path(__file__).parent.parent.parent.resolve()
+# File where to store http headers for gcloud authentication
+AUTH_HEADERS_FILE = Path(gettempdir()) / f"crosvm_gcloud_auth_headers_{getpass.getuser()}"
+
+PathLike = Union[Path, str]
+
+
+def find_crosvm_root():
+    "Walk up from CWD until we find the crosvm root dir."
+    path = Path("").resolve()
+    while True:
+        if (path / "tools/impl/common.py").is_file():
+            return path
+        if path.parent:
+            path = path.parent
+        else:
+            raise Exception("Cannot find crosvm root dir.")
+
+
+"Root directory of crosvm derived from CWD."
+CROSVM_ROOT = find_crosvm_root()
 
 "Cargo.toml file of crosvm"
 CROSVM_TOML = CROSVM_ROOT / "Cargo.toml"
 
+"""
+Root directory of crosvm devtools.
+
+May be different from `CROSVM_ROOT/tools`, which is allows you to run the crosvm dev
+tools from this directory on another crosvm repo.
+
+Use this if you want to call crosvm dev tools, which will use the scripts relative
+to this file.
+"""
+TOOLS_ROOT = Path(__file__).parent.parent.resolve()
+
 "Url of crosvm's gerrit review host"
 GERRIT_URL = "https://chromium-review.googlesource.com"
 
 # Ensure that we really found the crosvm root directory
 assert 'name = "crosvm"' in CROSVM_TOML.read_text()
 
-# File where to store http headers for gcloud authentication
-AUTH_HEADERS_FILE = Path(gettempdir()) / f"crosvm_gcloud_auth_headers_{getpass.getuser()}"
-
-PathLike = Union[Path, str]
-
 
 class CommandResult(NamedTuple):
     """Results of a command execution as returned by Command.run()"""
@@ -175,6 +194,7 @@ class Command(object):
         self,
         quiet: bool = False,
         check: bool = True,
+        dry_run: bool = False,
     ) -> int:
         """
         Runs a program in the foreground with output streamed to the user.
@@ -200,6 +220,10 @@ class Command(object):
         Returns: The return code of the program.
         """
         self.__debug_print()
+        if dry_run:
+            print(f"Not running: {self}")
+            return 0
+
         if quiet:
             result = subprocess.run(
                 self.args,

tools/tests/cl_tests.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+# Copyright 2022 The ChromiumOS Authors.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+import os
+from pathlib import Path
+import shutil
+import sys
+import tempfile
+import unittest
+
+sys.path.append(os.path.dirname(sys.path[0]))
+
+from impl.common import CROSVM_ROOT, cmd, quoted, TOOLS_ROOT
+
+git = cmd("git")
+cl = cmd(f"{TOOLS_ROOT}/cl")
+
+
+class ClTests(unittest.TestCase):
+    test_root: Path
+
+    def setUp(self):
+        self.test_root = Path(tempfile.mkdtemp())
+        os.chdir(self.test_root)
+        git("clone", CROSVM_ROOT, ".").fg(quiet=True)
+
+        # Set up user name (it's not set up in Luci)
+        git("config user.name Nobody").fg(quiet=True)
+        git("config user.email nobody@chromium.org").fg(quiet=True)
+
+        # Check out a detached head and delete all branches.
+        git("checkout -d HEAD").fg(quiet=True)
+        branch_list = git("branch").lines()
+        for branch in branch_list:
+            if not branch.startswith("*"):
+                git("branch -D", branch).fg(quiet=True)
+
+        # Set up the origin for testing uploads and rebases.
+        git("remote set-url origin https://chromium.googlesource.com/crosvm/crosvm").fg(quiet=True)
+        git("fetch -q origin main").fg(quiet=True)
+        git("fetch -q origin chromeos").fg(quiet=True)
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.test_root)
+
+    def create_test_commit(self, message: str, branch: str, upstream: str = "origin/main"):
+        git("checkout -b", branch, "--track", upstream).fg(quiet=True)
+        with Path("Cargo.toml").open("a") as file:
+            file.write("# Foo")
+        git("commit -a -m", quoted(message)).fg(quiet=True)
+        return git("rev-parse HEAD").stdout()
+
+    def test_cl_upload(self):
+        sha = self.create_test_commit("Test Commit", "foo")
+        expected = f"""\
+Uploading to origin/main:
+  {sha} Test Commit
+
+Not running: git push origin HEAD:refs/for/main"""
+
+        self.assertEqual(cl("upload --dry-run").stdout(), expected)
+
+    def test_cl_status(self):
+        self.create_test_commit("Test Commit", "foo")
+        expected = """\
+Branch foo tracking origin/main
+  NOT_UPLOADED Test Commit"""
+
+        self.assertEqual(cl("status").stdout(), expected)
+
+    def test_cl_rebase(self):
+        self.create_test_commit("Test Commit", "foo", "origin/chromeos")
+        cl("rebase").fg()
+
+        # Expect foo-upstream to be tracking `main` and have the same commit
+        self.assertEqual(git("rev-parse --abbrev-ref foo-upstream@{u}").stdout(), "origin/main")
+        self.assertEqual(
+            git("log -1 --format=%s foo").stdout(),
+            git("log -1 --format=%s foo-upstream").stdout(),
+        )
+
+    def test_cl_rebase_with_existing_branch(self):
+        previous_sha = self.create_test_commit("Previous commit", "foo-upstream ")
+        self.create_test_commit("Test Commit", "foo", "origin/chromeos")
+        message = cl("rebase").stdout()
+
+        # `cl rebase` will overwrite the branch, but we should print the previous sha in case
+        # the user needs to recover it.
+        self.assertIn(previous_sha, message)
+
+        # Expect foo-upstream to be tracking `main` and have the same commit. The previous commit
+        # would be dropped.
+        self.assertEqual(git("rev-parse --abbrev-ref foo-upstream@{u}").stdout(), "origin/main")
+        self.assertEqual(
+            git("log -1 --format=%s foo").stdout(),
+            git("log -1 --format=%s foo-upstream").stdout(),
+        )
+
+    def test_prune(self):
+        self.create_test_commit("Test Commit", "foo")
+        git("branch foo-no-commit origin/main").fg()
+        cl("prune --force").fg()
+
+        # `foo` has unsubmitted commits, it should still be there.
+        self.assertTrue(git("rev-parse foo").success())
+
+        # `foo-no-commit` has no commits, it should have been pruned.
+        self.assertFalse(git("rev-parse foo-no-commit").success())
+
+
+if __name__ == "__main__":
+    unittest.main()