updated: basic e2e update tests (#31742)

* e2e update test

* that too

* fix

* fix

* fix running in docker

* don't think GHA will work

* also test switching branches

* it's a test

* lets not delete that yet

* comment

* space
old-commit-hash: ac77129041

pyproject.toml

@@ -21,6 +21,7 @@ testpaths = [
   "selfdrive/thermald",
   "selfdrive/test/longitudinal_maneuvers",
   "selfdrive/test/process_replay/test_fuzzy.py",
+  "selfdrive/updated",
   "system/camerad",
   "system/hardware/tici",
   "system/loggerd",

selfdrive/updated/tests/test_updated.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+import os
+import pathlib
+import shutil
+import signal
+import subprocess
+import tempfile
+import time
+import unittest
+from unittest import mock
+
+import pytest
+from openpilot.selfdrive.manager.process import ManagerProcess
+
+
+from openpilot.selfdrive.test.helpers import processes_context
+from openpilot.common.params import Params
+
+
+def run(args, **kwargs):
+  return subprocess.run(args, **kwargs, check=True)
+
+
+def update_release(directory, name, version, release_notes):
+  with open(directory / "RELEASES.md", "w") as f:
+    f.write(release_notes)
+
+  (directory / "common").mkdir(exist_ok=True)
+
+  with open(directory / "common" / "version.h", "w") as f:
+    f.write(f'#define COMMA_VERSION "{version}"')
+
+  run(["git", "add", "."], cwd=directory)
+  run(["git", "commit", "-m", f"openpilot release {version}"], cwd=directory)
+
+
+@pytest.mark.slow # TODO: can we test overlayfs in GHA?
+class TestUpdateD(unittest.TestCase):
+  def setUp(self):
+    self.tmpdir = tempfile.mkdtemp()
+
+    run(["sudo", "mount", "-t", "tmpfs", "tmpfs", self.tmpdir]) # overlayfs doesn't work inside of docker unless this is a tmpfs
+
+    self.mock_update_path = pathlib.Path(self.tmpdir)
+
+    self.params = Params()
+
+    self.basedir = self.mock_update_path / "openpilot"
+    self.basedir.mkdir()
+
+    self.staging_root = self.mock_update_path / "safe_staging"
+    self.staging_root.mkdir()
+
+    self.remote_dir = self.mock_update_path / "remote"
+    self.remote_dir.mkdir()
+
+    mock.patch("openpilot.common.basedir.BASEDIR", self.basedir).start()
+
+    os.environ["UPDATER_STAGING_ROOT"] = str(self.staging_root)
+    os.environ["UPDATER_LOCK_FILE"] = str(self.mock_update_path / "safe_staging_overlay.lock")
+
+    self.MOCK_RELEASES = {
+      "release3": ("0.1.2", "0.1.2 release notes"),
+      "master": ("0.1.3", "0.1.3 release notes"),
+    }
+
+  def set_target_branch(self, branch):
+    self.params.put("UpdaterTargetBranch", branch)
+
+  def setup_basedir_release(self, release):
+    self.params = Params()
+    self.set_target_branch(release)
+    run(["git", "clone", "-b", release, self.remote_dir, self.basedir])
+
+  def update_remote_release(self, release):
+    update_release(self.remote_dir, release, *self.MOCK_RELEASES[release])
+
+  def setup_remote_release(self, release):
+    run(["git", "init"], cwd=self.remote_dir)
+    run(["git", "checkout", "-b", release], cwd=self.remote_dir)
+    self.update_remote_release(release)
+
+  def tearDown(self):
+    mock.patch.stopall()
+    run(["sudo", "umount", "-l", str(self.staging_root / "merged")])
+    run(["sudo", "umount", "-l", self.tmpdir])
+    shutil.rmtree(self.tmpdir)
+
+  def send_check_for_updates_signal(self, updated: ManagerProcess):
+    updated.signal(signal.SIGUSR1.value)
+
+  def send_download_signal(self, updated: ManagerProcess):
+    updated.signal(signal.SIGHUP.value)
+
+  def _test_params(self, branch, fetch_available, update_available):
+    self.assertEqual(self.params.get("UpdaterTargetBranch", encoding="utf-8"), branch)
+    self.assertEqual(self.params.get_bool("UpdaterFetchAvailable"), fetch_available)
+    self.assertEqual(self.params.get_bool("UpdateAvailable"), update_available)
+
+  def _test_update_params(self, branch, version, release_notes):
+    self.assertTrue(self.params.get("UpdaterNewDescription", encoding="utf-8").startswith(f"{version} / {branch}"))
+    self.assertEqual(self.params.get("UpdaterNewReleaseNotes", encoding="utf-8"), f"<p>{release_notes}</p>\n")
+
+  def wait_for_idle(self, timeout=5, min_wait_time=2):
+    start = time.monotonic()
+    time.sleep(min_wait_time)
+
+    while True:
+      waited = time.monotonic() - start
+      if self.params.get("UpdaterState", encoding="utf-8") == "idle":
+        print(f"waited {waited}s for idle")
+        break
+
+      if waited > timeout:
+        raise TimeoutError("timed out waiting for idle")
+
+      time.sleep(1)
+
+  def test_new_release(self):
+    # Start on release3, simulate a release3 commit, ensure we fetch that update properly
+    self.setup_remote_release("release3")
+    self.setup_basedir_release("release3")
+
+    with processes_context(["updated"]) as [updated]:
+      self._test_params("release3", False, False)
+      time.sleep(1)
+      self._test_params("release3", False, False)
+
+      self.MOCK_RELEASES["release3"] = ("0.1.3", "0.1.3 release notes")
+      self.update_remote_release("release3")
+
+      self.send_check_for_updates_signal(updated)
+
+      self.wait_for_idle()
+
+      self._test_params("release3", True, False)
+
+      self.send_download_signal(updated)
+
+      self.wait_for_idle()
+
+      self._test_params("release3", False, True)
+      self._test_update_params("release3", *self.MOCK_RELEASES["release3"])
+
+  def test_switch_branches(self):
+    # Start on release3, request to switch to master manually, ensure we switched
+    self.setup_remote_release("release3")
+    self.setup_remote_release("master")
+    self.setup_basedir_release("release3")
+
+    with processes_context(["updated"]) as [updated]:
+      self._test_params("release3", False, False)
+      self.wait_for_idle()
+      self._test_params("release3", False, False)
+
+      self.set_target_branch("master")
+      self.send_check_for_updates_signal(updated)
+
+      self.wait_for_idle()
+
+      self._test_params("master", True, False)
+
+      self.send_download_signal(updated)
+
+      self.wait_for_idle()
+
+      self._test_params("master", False, True)
+      self._test_update_params("master", *self.MOCK_RELEASES["master"])
+
+
+if __name__ == "__main__":
+  unittest.main()