| # Copyright 2013 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit tests for xbuddy.py.""" |
| |
| import configparser |
| import os |
| import shutil |
| import tempfile |
| import time |
| from unittest import mock |
| |
| import pytest |
| |
| from chromite.lib import cros_test_lib |
| from chromite.lib import gs |
| from chromite.lib import path_util |
| from chromite.lib.xbuddy import xbuddy |
| |
| |
| pytestmark = cros_test_lib.pytestmark_inside_only |
| |
| |
| # pylint: disable=protected-access |
| # pylint: disable=no-value-for-parameter |
| |
| GS_ALTERNATE_DIR = "gs://chromeos-alternate-archive/" |
| |
| |
| class xBuddyTest(cros_test_lib.TestCase): |
| """Regression tests for xbuddy.""" |
| |
| def setUp(self) -> None: |
| self.static_image_dir = tempfile.mkdtemp("xbuddy_unittest_static") |
| |
| self.mock_xb = xbuddy.XBuddy(True, static_dir=self.static_image_dir) |
| self.images_dir = tempfile.mkdtemp("xbuddy_unittest_images") |
| self.mock_xb.images_dir = self.images_dir |
| |
| def tearDown(self) -> None: |
| """Removes testing files.""" |
| shutil.rmtree(self.static_image_dir) |
| shutil.rmtree(self.images_dir) |
| |
| def testParseBoolean(self) -> None: |
| """Check that some common True/False strings are handled.""" |
| self.assertEqual(xbuddy.XBuddy.ParseBoolean(None), False) |
| self.assertEqual(xbuddy.XBuddy.ParseBoolean("false"), False) |
| self.assertEqual(xbuddy.XBuddy.ParseBoolean("bs"), False) |
| self.assertEqual(xbuddy.XBuddy.ParseBoolean("true"), True) |
| self.assertEqual(xbuddy.XBuddy.ParseBoolean("y"), True) |
| |
| def testGetLatestVersionFromGsDir(self) -> None: |
| """Test that we can get the most recent version from gsutil calls.""" |
| mock_data1 = """gs://chromeos-releases/stable-channel/parrot/3701.96.0/ |
| gs://chromeos-releases/stable-channel/parrot/3701.98.0/ |
| gs://chromeos-releases/stable-channel/parrot/3912.100.0/ |
| gs://chromeos-releases/stable-channel/parrot/3912.101.0/ |
| gs://chromeos-releases/stable-channel/parrot/3912.79.0/ |
| gs://chromeos-releases/stable-channel/parrot/3912.79.1/""" |
| |
| mock_data2 = """\ |
| gs://chromeos-image-archive/parrot-release/R26-3912.101.0 |
| gs://chromeos-image-archive/parrot-release/R27-3912.101.0 |
| gs://chromeos-image-archive/parrot-release/R28-3912.101.0""" |
| |
| with mock.patch.object( |
| self.mock_xb, |
| "_LS", |
| side_effect=[mock_data1.splitlines(), mock_data2.splitlines()], |
| ) as ls_mock: |
| self.assertEqual( |
| self.mock_xb._GetLatestVersionFromGsDir( |
| "url1", with_release=False |
| ), |
| "3912.101.0", |
| ) |
| ls_mock.assert_called_with("url1", list_subdirectory=False) |
| self.assertEqual( |
| self.mock_xb._GetLatestVersionFromGsDir( |
| "url2", list_subdirectory=True, with_release=True |
| ), |
| "R28-3912.101.0", |
| ) |
| ls_mock.assert_called_with("url2", list_subdirectory=True) |
| |
| def testLookupOfficial(self) -> None: |
| """Basic test of _LookupOfficial. |
| |
| Checks that a given suffix is handled. |
| """ |
| with mock.patch.object( |
| gs.GSContext, "Cat", return_value="v" |
| ) as cat_mock: |
| self.assertEqual( |
| self.mock_xb._LookupOfficial("b", suffix="-s"), "b-s/v" |
| ) |
| cat_mock.assert_called_with( |
| "gs://chromeos-image-archive/b-s/LATEST-main", encoding="utf-8" |
| ) |
| |
| @mock.patch.object( |
| xbuddy.XBuddy, |
| "_GetLatestVersionFromGsDir", |
| side_effect=["4100.68.0", "R28-4100.68.0"], |
| ) |
| def testLookupChannel(self, version_mock) -> None: |
| """Basic test of _LookupChannel. |
| |
| Checks that a given suffix is handled. |
| """ |
| self.assertEqual( |
| self.mock_xb._LookupChannel("b", "-release"), |
| "b-release/R28-4100.68.0", |
| ) |
| version_mock.assert_called_with( |
| "gs://chromeos-image-archive/b-release/R*4100.68.0", |
| list_subdirectory=True, |
| ) |
| |
| def testLookupAliasPathRewrite(self) -> None: |
| """Tests LookupAlias of path rewrite, including keyword substitution.""" |
| path = "remote/BOARD/VERSION/test" |
| with mock.patch.object( |
| self.mock_xb.config, "get", side_effect=[configparser.Error, path] |
| ) as get_mock: |
| self.assertEqual( |
| ("remote/parrot/1.2.3/test", "-release"), |
| self.mock_xb.LookupAlias( |
| "foobar", board="parrot", version="1.2.3" |
| ), |
| ) |
| get_mock.assert_called_with("PATH_REWRITES", "foobar") |
| |
| def testLookupAliasSuffix(self) -> None: |
| """Tests LookupAlias of location suffix.""" |
| with mock.patch.object( |
| self.mock_xb.config, |
| "get", |
| side_effect=["-random", configparser.Error], |
| ) as get_mock: |
| self.assertEqual( |
| ("foobar", "-random"), |
| self.mock_xb.LookupAlias( |
| "foobar", board="parrot", version="1.2.3" |
| ), |
| ) |
| get_mock.assert_called_with("PATH_REWRITES", "foobar") |
| |
| def testLookupAliasPathRewriteAndSuffix(self) -> None: |
| """Tests LookupAlias with both path rewrite and suffix.""" |
| path = "remote/BOARD/VERSION/test" |
| with mock.patch.object( |
| self.mock_xb.config, "get", side_effect=["-random", path] |
| ) as get_mock: |
| self.assertEqual( |
| ("remote/parrot/1.2.3/test", "-random"), |
| self.mock_xb.LookupAlias( |
| "foobar", board="parrot", version="1.2.3" |
| ), |
| ) |
| get_mock.assert_called_with("PATH_REWRITES", "foobar") |
| |
| @mock.patch.object(xbuddy.XBuddy, "_LookupOfficial") |
| def testResolveVersionToBuildIdAndChannel_Official( |
| self, lookup_mock |
| ) -> None: |
| """Test _ResolveVersionToBuildIdAndChannel works for official build.""" |
| board = "chell" |
| suffix = "-s" |
| |
| version = "latest-official" |
| self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version) |
| lookup_mock.assert_called_with("chell", "-s", image_dir=None) |
| |
| self.mock_xb._ResolveVersionToBuildIdAndChannel( |
| board, suffix, version, image_dir=GS_ALTERNATE_DIR |
| ) |
| lookup_mock.assert_called_with( |
| "chell", "-s", image_dir="gs://chromeos-alternate-archive/" |
| ) |
| |
| version = "latest-official-paladin" |
| self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version) |
| lookup_mock.assert_called_with("chell", "paladin", image_dir=None) |
| |
| self.mock_xb._ResolveVersionToBuildIdAndChannel( |
| board, suffix, version, image_dir=GS_ALTERNATE_DIR |
| ) |
| lookup_mock.assert_called_with( |
| "chell", "paladin", image_dir="gs://chromeos-alternate-archive/" |
| ) |
| |
| @mock.patch.object(xbuddy.XBuddy, "_LookupChannel") |
| def testResolveVersionToBuildIdAndChannel_Channel( |
| self, lookup_mock |
| ) -> None: |
| """Check _ResolveVersionToBuildIdAndChannel support for channels.""" |
| board = "chell" |
| suffix = "-s" |
| |
| version = "latest" |
| self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version) |
| lookup_mock.assert_called_with("chell", "-s", image_dir=None) |
| |
| self.mock_xb._ResolveVersionToBuildIdAndChannel( |
| board, suffix, version, image_dir=GS_ALTERNATE_DIR |
| ) |
| lookup_mock.assert_called_with( |
| "chell", "-s", image_dir="gs://chromeos-alternate-archive/" |
| ) |
| |
| version = "latest-dev" |
| self.mock_xb._ResolveVersionToBuildIdAndChannel(board, suffix, version) |
| lookup_mock.assert_called_with( |
| "chell", "-s", channel="dev", image_dir=None |
| ) |
| |
| self.mock_xb._ResolveVersionToBuildIdAndChannel( |
| board, suffix, version, image_dir=GS_ALTERNATE_DIR |
| ) |
| lookup_mock.assert_called_with( |
| "chell", |
| "-s", |
| channel="dev", |
| image_dir="gs://chromeos-alternate-archive/", |
| ) |
| |
| # TODO(dgarrett): Re-enable when crbug.com/585914 is fixed. |
| # def testResolveVersionToBuildId_BaseVersion(self): |
| # """Check _ResolveVersionToBuildId handles a base version.""" |
| # board = 'b' |
| # suffix = '-s' |
| |
| # self.mox.StubOutWithMock(self.mock_xb, '_ResolveBuildVersion') |
| # self.mock_xb._ResolveBuildVersion(board, suffix, '1.2.3').AndReturn( |
| # 'R12-1.2.3') |
| # self.mox.StubOutWithMock(self.mock_xb, '_RemoteBuildId') |
| # self.mock_xb._RemoteBuildId(board, suffix, 'R12-1.2.3') |
| # self.mox.ReplayAll() |
| |
| # self.mock_xb._ResolveVersionToBuildId(board, suffix, '1.2.3') |
| # self.mox.VerifyAll() |
| |
| def testBasicInterpretPath(self) -> None: |
| """Basic checks for splitting a path""" |
| path = "parrot/R27-2455.0.0/test" |
| expected = ("test", "parrot", "R27-2455.0.0", True) |
| self.assertEqual(xbuddy.InterpretPath(path=path), expected) |
| |
| path = "parrot/R27-2455.0.0/full_payload" |
| expected = ("full_payload", "parrot", "R27-2455.0.0", True) |
| self.assertEqual(xbuddy.InterpretPath(path=path), expected) |
| |
| path = "parrot/R27-2455.0.0" |
| expected = ("ANY", "parrot", "R27-2455.0.0", True) |
| self.assertEqual(xbuddy.InterpretPath(path=path), expected) |
| |
| path = "remote/parrot/R27-2455.0.0" |
| expected = ("test", "parrot", "R27-2455.0.0", False) |
| self.assertEqual(xbuddy.InterpretPath(path=path), expected) |
| |
| path = "local/parrot/R27-2455.0.0" |
| expected = ("ANY", "parrot", "R27-2455.0.0", True) |
| self.assertEqual(xbuddy.InterpretPath(path=path), expected) |
| |
| path = "" |
| expected = ("ANY", None, "latest", True) |
| self.assertEqual(xbuddy.InterpretPath(path=path), expected) |
| |
| path = "local" |
| expected = ("ANY", None, "latest", True) |
| self.assertEqual(xbuddy.InterpretPath(path=path), expected) |
| |
| path = "local/parrot/latest/ANY" |
| expected = ("ANY", "parrot", "latest", True) |
| self.assertEqual(xbuddy.InterpretPath(path=path), expected) |
| |
| def testInterpretPathWithDefaults(self) -> None: |
| """Test path splitting with default board/version.""" |
| path = "" |
| expected = ("ANY", "parrot", "latest", True) |
| self.assertEqual( |
| expected, |
| xbuddy.InterpretPath(path=path, default_board="parrot"), |
| ) |
| |
| path = "" |
| expected = ("ANY", None, "1.2.3", True) |
| self.assertEqual( |
| expected, |
| xbuddy.InterpretPath(path=path, default_version="1.2.3"), |
| ) |
| |
| path = "" |
| expected = ("ANY", "parrot", "1.2.3", True) |
| self.assertEqual( |
| expected, |
| xbuddy.InterpretPath( |
| path=path, default_board="parrot", default_version="1.2.3" |
| ), |
| ) |
| |
| path = "1.2.3" |
| expected = ("ANY", None, "1.2.3", True) |
| self.assertEqual( |
| expected, |
| xbuddy.InterpretPath(path=path, default_version="1.2.3"), |
| ) |
| |
| path = "latest" |
| expected = ("ANY", None, "latest", True) |
| self.assertEqual( |
| expected, |
| xbuddy.InterpretPath(path=path, default_version="1.2.3"), |
| ) |
| |
| path = "1.2.3" |
| expected = ("ANY", "parrot", "1.2.3", True) |
| self.assertEqual( |
| expected, |
| xbuddy.InterpretPath( |
| path=path, default_board="parrot", default_version="1.2.3" |
| ), |
| ) |
| |
| path = "parrot" |
| expected = ("ANY", "parrot", "1.2.3", True) |
| self.assertEqual( |
| expected, |
| xbuddy.InterpretPath(path=path, default_version="1.2.3"), |
| ) |
| |
| path = "parrot" |
| expected = ("ANY", "parrot", "1.2.3", True) |
| self.assertEqual( |
| expected, |
| xbuddy.InterpretPath( |
| path=path, default_board="parrot", default_version="1.2.3" |
| ), |
| ) |
| |
| with mock.patch.object( |
| path_util, |
| "DetermineCheckout", |
| return_value=path_util.CheckoutInfo( |
| path_util.CheckoutType.GCLIENT, None, None |
| ), |
| ): |
| path = "" |
| expected = ("test", "parrot", "1.2.3", False) |
| self.assertEqual( |
| expected, |
| xbuddy.InterpretPath( |
| path=path, default_board="parrot", default_version="1.2.3" |
| ), |
| ) |
| |
| def testTimestampsAndList(self) -> None: |
| """Creation and listing of builds according to their timestamps.""" |
| # make 3 different timestamp files |
| b_id11 = "b1/v1" |
| b_id12 = "b1/v2" |
| b_id23 = "b2/v3" |
| xbuddy.update_timestamp(self.mock_xb._timestamp_folder, b_id11) |
| time.sleep(0.05) |
| xbuddy.update_timestamp(self.mock_xb._timestamp_folder, b_id12) |
| time.sleep(0.05) |
| xbuddy.update_timestamp(self.mock_xb._timestamp_folder, b_id23) |
| |
| # reference second one again |
| time.sleep(0.05) |
| xbuddy.update_timestamp(self.mock_xb._timestamp_folder, b_id12) |
| |
| # check that list returns the same 3 things, in last referenced order |
| result = self.mock_xb._ListBuildTimes() |
| self.assertEqual(result[0][0], b_id12) |
| self.assertEqual(result[1][0], b_id23) |
| self.assertEqual(result[2][0], b_id11) |
| |
| def testSyncRegistry(self) -> None: |
| # check that there are no builds initially |
| result = self.mock_xb._ListBuildTimes() |
| self.assertEqual(len(result), 0) |
| |
| # set up the stub build/images directory with images |
| boards = ["a", "b"] |
| versions = ["v1", "v2"] |
| for b in boards: |
| os.makedirs(os.path.join(self.mock_xb.images_dir, b)) |
| for v in versions: |
| os.makedirs(os.path.join(self.mock_xb.images_dir, b, v)) |
| |
| # Sync and check that they've been added to xBuddy's registry |
| self.mock_xb._SyncRegistryWithBuildImages() |
| result = self.mock_xb._ListBuildTimes() |
| self.assertEqual(len(result), 4) |
| |
| def testXBuddyCaching(self) -> None: |
| """Caching & replacement of timestamp files.""" |
| |
| def _ReleaseOnly(name): |
| # All non-release URLs are invalid so we can meet expectations. |
| if name.find("-release") == -1: |
| raise gs.GSContextException("bad URL") |
| return name |
| |
| with mock.patch.object( |
| gs.GSContext, "LS", side_effect=_ReleaseOnly |
| ) as ls_mock: |
| with mock.patch.object(self.mock_xb, "_Download") as download_mock: |
| version = "%s-release/R0" |
| |
| def _Download(image) -> None: |
| gs_image = "gs://chromeos-image-archive/" + version |
| self.mock_xb.Get(("remote", image, "R0", "test")) |
| ls_mock.assert_called_with(gs_image % image) |
| download_mock.assert_called_with( |
| gs_image % image, ["test_image"], version % image |
| ) |
| time.sleep(0.05) |
| |
| # Requires default capacity. |
| self.assertEqual(self.mock_xb._Capacity(), 5) |
| |
| # Get 6 different images: a,b,c,d,e,f. |
| images = ["a", "b", "c", "d", "e", "f"] |
| for image in images: |
| _Download(image) |
| |
| # Check that b,c,d,e,f are still stored. |
| result = self.mock_xb._ListBuildTimes() |
| self.assertEqual(len(result), 5) |
| |
| # Flip the list to get reverse chronological order |
| images.reverse() |
| for i in range(5): |
| self.assertEqual(result[i][0], version % images[i]) |
| |
| # Get b,a. |
| images = ["b", "a"] |
| for image in images: |
| _Download(image) |
| |
| # Check that d,e,f,b,a are still stored. |
| result = self.mock_xb._ListBuildTimes() |
| self.assertEqual(len(result), 5) |
| images_expected = ["a", "b", "f", "e", "d"] |
| for i in range(5): |
| self.assertEqual( |
| result[i][0], "%s-release/R0" % images_expected[i] |
| ) |
| |
| @mock.patch.object( |
| xbuddy.XBuddy, "_Download", return_value=[["/path/to/image.bin"]] |
| ) |
| @mock.patch.object( |
| xbuddy.XBuddy, |
| "_ResolveVersionToBuildIdAndChannel", |
| return_value=("reef/R1-1.2.3", "stable"), |
| ) |
| # pylint: disable=unused-argument |
| def testGet(self, resolve_mock, downloader_mock) -> None: |
| """Tests _GetArtifact method.""" |
| self.assertEqual( |
| self.mock_xb.Get(["remote", "reef", "R1-1.2.3", "test"]), |
| ("reef/R1-1.2.3", "/path/to/image.bin"), |
| ) |
| |
| self.assertEqual( |
| self.mock_xb.Get(["remote", "reef", "R1-1.2.3"]), |
| ("reef/R1-1.2.3", "/path/to/image.bin"), |
| ) |
| |
| |
| valid_full_uris = [ |
| ( |
| "xbuddy://strongbad/R109-14893.0.0/test", |
| xbuddy.XBuddyComponents("test", "strongbad", "R109-14893.0.0", True), |
| ), |
| ( |
| "xbuddy://betty/R106-14212.0.0/full_payload", |
| xbuddy.XBuddyComponents( |
| "full_payload", "betty", "R106-14212.0.0", True |
| ), |
| ), |
| ( |
| "xbuddy://boardname/R100-14000.0.0", |
| xbuddy.XBuddyComponents("ANY", "boardname", "R100-14000.0.0", True), |
| ), |
| ( |
| "xbuddy://remote/strongbad/R109-14893.0.0", |
| xbuddy.XBuddyComponents("test", "strongbad", "R109-14893.0.0", False), |
| ), |
| ( |
| "xbuddy://local/betty/R106-14212.0.0", |
| xbuddy.XBuddyComponents("ANY", "betty", "R106-14212.0.0", True), |
| ), |
| ("xbuddy://", xbuddy.XBuddyComponents("ANY", None, "latest", True)), |
| ( |
| "xbuddy://local", |
| xbuddy.XBuddyComponents("ANY", None, "latest", True), |
| ), |
| ( |
| "xbuddy://remote", |
| xbuddy.XBuddyComponents("test", None, "latest", False), |
| ), |
| ( |
| "xbuddy://local/boardname/latest/ANY", |
| xbuddy.XBuddyComponents("ANY", "boardname", "latest", True), |
| ), |
| ( |
| "xbuddy://remote/boardname/latest/ANY", |
| xbuddy.XBuddyComponents("ANY", "boardname", "latest", False), |
| ), |
| ] |
| |
| |
| @pytest.mark.parametrize("uri,expected", valid_full_uris) |
| def test_parse_valid_uri(uri, expected) -> None: |
| """Basic checks for parsing an XBuddy URI.""" |
| assert xbuddy.parse(uri) == expected |
| |
| |
| valid_uris_without_schema = [ |
| ( |
| "strongbad/R109-14893.0.0/test", |
| xbuddy.XBuddyComponents("test", "strongbad", "R109-14893.0.0", True), |
| ), |
| ("", xbuddy.XBuddyComponents("ANY", None, "latest", True)), |
| ] |
| |
| |
| @pytest.mark.parametrize("uri,expected", valid_uris_without_schema) |
| def test_parse_uri_with_empty_scheme_not_strict(uri, expected) -> None: |
| """Basic checks for parsing an XBuddy URI without a scheme.""" |
| assert xbuddy.parse(uri) == expected |
| |
| |
| @pytest.mark.parametrize("uri", [tc[0] for tc in valid_uris_without_schema]) |
| def test_parse_uri_with_empty_scheme_strict(uri) -> None: |
| """Fail strict scheme checking for URIs without scheme provided.""" |
| with pytest.raises(xbuddy.XBuddyInvalidSchemeException): |
| xbuddy.parse(uri, strict=True) |
| |
| |
| @pytest.mark.parametrize( |
| "uri", ["gs://test/bucket", "usb://", "ftp://path/to/image"] |
| ) |
| def test_parse_invalid_uri(uri) -> None: |
| """Test exception is raised when the scheme is specified and not xbuddy.""" |
| with pytest.raises(xbuddy.XBuddyInvalidSchemeException): |
| xbuddy.parse(uri) |
