blob: 0faaf665c5c4434f0fe1f24ee20eff7033c0dbfa [file]
import sh
import unittest
from unittest.mock import MagicMock
from steps import upload_artifacts
class TestUploadArtifacts(unittest.TestCase):
def setUp(self):
self.step = upload_artifacts.UploadArtifacts()
self.step.gsutil_update_sstate_cache = MagicMock()
self.step.gsutil = MagicMock()
self.init_step()
def init_step(self):
self.step.firmware_machine = "machine0"
self.step.firmware_version = "x.y.z"
self.step.project_id = "test-id"
self.step.gcs_bucket_release = "bucket-release"
self.step.gcs_bucket = "bucket"
def test_gsutil_with_sstate_cache(self):
"""Tests run_script code flow with sstate_cache update enabled"""
self.step.update_cache = True
self.step.run_script()
artifacts_path = "build/machine0/tmp/deploy/images/machine0"
gcs_dir = "gs://bucket-release/machine0/x.y.z"
self.step.gsutil_update_sstate_cache.assert_called_with()
self.step.gsutil.assert_called_with("cp", artifacts_path, gcs_dir)
def test_gsutil_without_sstate_cache(self):
"""Tests run_script code flow with sstate_cache update disabled"""
self.setUp()
self.step.update_cache = False
self.step.run_script()
artifacts_path = "build/machine0/tmp/deploy/images/machine0"
gcs_dir = "gs://bucket-release/machine0/x.y.z"
self.step.gsutil_update_sstate_cache.assert_not_called()
self.step.gsutil.assert_called_with("cp", artifacts_path, gcs_dir)
def test_sh_with_sstate_cache(self):
"""Tests the arguments used to execute sh.sh with sstate-cache update enabled"""
self.step = upload_artifacts.UploadArtifacts()
self.init_step()
self.step.update_cache = True
sh.sh = MagicMock()
self.step.run_script()
actual_calls = sh.sh.call_args_list
sstate_cache_dir = "build/machine0/sstate-cache"
sstate_cache_bucket = "gs://bucket/cache-unzip/machine0"
call0 = f"gsutil -mq rsync -r -d {sstate_cache_dir} {sstate_cache_bucket}"
artifacts_path = "build/machine0/tmp/deploy/images/machine0"
gcs_dir = "gs://bucket-release/machine0/x.y.z"
call1 = f"gsutil -mq cp -r {artifacts_path} {gcs_dir}"
expected_calls = [(("-c", call0),), (("-c", call1),)]
self.assertEqual(actual_calls, expected_calls)
def test_sh_without_sstate_cache(self):
"""Tests the arguments used to execute sh.sh with sstate-cache update disabled"""
self.step = upload_artifacts.UploadArtifacts()
self.init_step()
self.step.update_cache = False
sh.sh = MagicMock()
self.step.run_script()
artifacts_path = "build/machine0/tmp/deploy/images/machine0"
gcs_dir = "gs://bucket-release/machine0/x.y.z"
actual_calls = sh.sh.call_args_list
call0 = f"gsutil -mq cp -r {artifacts_path} {gcs_dir}"
expected_calls = [(("-c", call0),)]
self.assertEqual(actual_calls, expected_calls)