scripts: chromite: add function to deploy compressed ash

The CL introduces deployment code for an experimental model where
ash-chrome main binary is packaged into squashfs. Unlike Lacros, that
packages the whole /opt/google/chrome, this experiment packages the
binary only.

BUG=b:247397013
TEST=unittest
TEST=cros chrome-sdk ...
TEST=chrome_deploy ...

Change-Id: Ib6a49b2becb86c91996ad942c83b3beaa6ab6c9f
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/chromite/+/3899689
Commit-Queue: Daniil Lunev <dlunev@chromium.org>
Reviewed-by: Mike Frysinger <vapier@chromium.org>
Reviewed-by: Achuith Bhandarkar <achuith@chromium.org>
Tested-by: Daniil Lunev <dlunev@chromium.org>
diff --git a/scripts/deploy_chrome_unittest.py b/scripts/deploy_chrome_unittest.py
index 16f2550..f847089 100644
--- a/scripts/deploy_chrome_unittest.py
+++ b/scripts/deploy_chrome_unittest.py
@@ -202,7 +202,11 @@
     """Deploy Chrome Mock Class."""
 
     TARGET = "chromite.scripts.deploy_chrome.DeployChrome"
-    ATTRS = ("_KillAshChromeIfNeeded", "_DisableRootfsVerification")
+    ATTRS = (
+        "_ChromeFileInUse",
+        "_DisableRootfsVerification",
+        "_ShouldUseCompressedAsh",
+    )
 
     def __init__(self):
         partial_mock.PartialMock.__init__(self)
@@ -215,6 +219,10 @@
             deploy_chrome.LSOF_COMMAND_CHROME % (deploy_chrome._CHROME_DIR,), 1
         )
 
+        self.rsh_mock.AddCmdResult(
+            "status ui", stdout="ui start/running, process 123"
+        )
+
     def MockMountCmd(self, returnvalue):
         self.rsh_mock.AddCmdResult(deploy_chrome.MOUNT_RW_COMMAND, returnvalue)
 
@@ -230,9 +238,16 @@
         self.rsh_mock.stop()
         self.remote_device_mock.stop()
 
-    def _KillAshChromeIfNeeded(self, _inst):
-        # Fully stub out for now.
-        pass
+    def _ChromeFileInUse(self, _inst):
+        # Fully stub out for now. Can be replaced if further testing is added.
+        return False
+
+    def _ShouldUseCompressedAsh(self, inst):
+        with mock.patch.object(
+            remote_access.RemoteDevice, "IfFileExists"
+        ) as exists_mock:
+            exists_mock.return_value = False
+            self.backup["_ShouldUseCompressedAsh"](inst)
 
 
 class DeployTest(cros_test_lib.MockTempDirTestCase):
@@ -309,6 +324,29 @@
         self.assertFalse(self.deploy._root_dir_is_still_readonly.is_set())
 
 
+class TestDeployCompressedAsh(DeployTest):
+    """Testing deployments with --ash-compressed passed."""
+
+    def _GetDeployChrome(self, args):
+        args.append("--compressed-ash")
+        return super(TestDeployCompressedAsh, self)._GetDeployChrome(args)
+
+    def testUnmountSuccess(self):
+        """Test case for a successful 'umount' call."""
+        self.deploy._KillAshChromeIfNeeded()
+
+    def testUnmountFailure(self):
+        """Test case for a failed 'umount' call."""
+        self.deploy_mock.rsh_mock.AddCmdResult(
+            ["umount", deploy_chrome.RAW_ASH_PATH],
+            returncode=32,
+            stderr="umount failure",
+        )
+        self.assertRaises(
+            deploy_chrome.DeployFailure, self.deploy._KillAshChromeIfNeeded
+        )
+
+
 class TestMount(DeployTest):
     """Testing mount success and failure."""
 
@@ -476,6 +514,19 @@
             chrome_util._COPY_PATHS_CHROME,
         )
 
+    def testSloppyDeploySuccessLacros(self):
+        """Ensure the squashfs mechanism with --compressed-ash doesn't throw."""
+        options = _ParseCommandLine(
+            self.common_flags + ["--sloppy", "--compressed-ash"]
+        )
+        osutils.Touch(os.path.join(self.build_dir, "chrome"), makedirs=True)
+        deploy_chrome._PrepareStagingDir(
+            options,
+            self.tempdir,
+            self.staging_dir,
+            chrome_util._COPY_PATHS_CHROME,
+        )
+
     @cros_test_lib.pytestmark_network_test
     def testUploadStagingDir(self):
         """Upload staging directory."""
@@ -599,6 +650,9 @@
                 "--nostrip",
             ]
         )
+        self.remote_exists_mock = self.PatchObject(
+            remote_access.RemoteDevice, "IfFileExists", return_value=False
+        )
         self.deploy = deploy_chrome.DeployChrome(
             options, self.tempdir, os.path.join(self.tempdir, "staging")
         )