| #!/usr/bin/env python3 |
| # Copyright 2024 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Tests for update_kernel_afdo.""" |
| |
| import datetime |
| from pathlib import Path |
| import shutil |
| import subprocess |
| import tempfile |
| import textwrap |
| import unittest |
| from unittest import mock |
| |
| import update_kernel_afdo |
| |
| |
| class Test(unittest.TestCase): |
| """Tests for update_kernel_afdo.""" |
| |
| def make_tempdir(self) -> Path: |
| x = Path(tempfile.mkdtemp(prefix="update_kernel_afdo_test_")) |
| self.addCleanup(shutil.rmtree, x) |
| return x |
| |
| def test_kernel_version_parsing(self): |
| self.assertEqual( |
| update_kernel_afdo.KernelVersion.parse("5.10"), |
| update_kernel_afdo.KernelVersion(major=5, minor=10), |
| ) |
| |
| with self.assertRaisesRegex(ValueError, ".*invalid kernel version.*"): |
| update_kernel_afdo.KernelVersion.parse("5") |
| |
| def test_kernel_version_formatting(self): |
| self.assertEqual( |
| str(update_kernel_afdo.KernelVersion(major=5, minor=10)), "5.10" |
| ) |
| |
| def test_channel_parsing(self): |
| with self.assertRaisesRegex(ValueError, "No such channel.*"): |
| update_kernel_afdo.Channel.parse("not a channel") |
| |
| # Ensure these round-trip. |
| for channel in update_kernel_afdo.Channel: |
| self.assertEqual( |
| channel, update_kernel_afdo.Channel.parse(channel.value) |
| ) |
| |
| @mock.patch.object(subprocess, "run") |
| def test_branch_autodetection(self, subprocess_run): |
| subprocess_run.return_value = subprocess.CompletedProcess( |
| args=[], |
| returncode=0, |
| stdout=textwrap.dedent( |
| """ |
| cros/not-a-release-branch |
| cros/release-R121-15699.B |
| cros/release-R122-15753.B |
| cros/release-R123-15786.B |
| cros/also-not-a-release-branch |
| m/main |
| """ |
| ), |
| ) |
| |
| branch_dict = update_kernel_afdo.autodetect_branches( |
| toolchain_utils=self.make_tempdir() |
| ) |
| |
| self.assertEqual( |
| branch_dict, |
| { |
| update_kernel_afdo.Channel.CANARY: update_kernel_afdo.GitBranch( |
| remote="cros", |
| release_number=124, |
| branch_name="main", |
| ), |
| update_kernel_afdo.Channel.BETA: update_kernel_afdo.GitBranch( |
| remote="cros", |
| release_number=123, |
| branch_name="release-R123-15786.B", |
| ), |
| update_kernel_afdo.Channel.STABLE: update_kernel_afdo.GitBranch( |
| remote="cros", |
| release_number=122, |
| branch_name="release-R122-15753.B", |
| ), |
| }, |
| ) |
| |
| def test_read_update_cfg_file(self): |
| valid_contents = textwrap.dedent( |
| """ |
| # some comment |
| # wow |
| AMD_KVERS="1.0 1.1" |
| ARM_KVERS="1.2" |
| AMD_METADATA_FILE="amd/file/path.json" # comment |
| ARM_METADATA_FILE="arm/file/path.json" |
| """ |
| ) |
| tmpdir = self.make_tempdir() |
| cfg_path = tmpdir / "test.cfg" |
| cfg_path.write_text(valid_contents, encoding="utf-8") |
| cfg = update_kernel_afdo.read_update_cfg_file(tmpdir, cfg_path) |
| expected_amd64 = update_kernel_afdo.ArchUpdateConfig( |
| versions_to_track=[ |
| update_kernel_afdo.KernelVersion(1, 0), |
| update_kernel_afdo.KernelVersion(1, 1), |
| ], |
| metadata_file=tmpdir / "amd/file/path.json", |
| ) |
| expected_arm = update_kernel_afdo.ArchUpdateConfig( |
| versions_to_track=[ |
| update_kernel_afdo.KernelVersion(1, 2), |
| ], |
| metadata_file=tmpdir / "arm/file/path.json", |
| ) |
| |
| self.assertEqual( |
| cfg, |
| { |
| update_kernel_afdo.Arch.AMD64: expected_amd64, |
| update_kernel_afdo.Arch.ARM: expected_arm, |
| }, |
| ) |
| |
| def test_parse_kernel_gs_profile(self): |
| timestamp = datetime.datetime.fromtimestamp(1234, datetime.timezone.utc) |
| profile = update_kernel_afdo.KernelGsProfile.from_file_name( |
| timestamp, |
| "R124-15808.0-1710149961.gcov.xz", |
| ) |
| self.assertEqual( |
| profile, |
| update_kernel_afdo.KernelGsProfile( |
| release_number=124, |
| chrome_build="15808.0", |
| cwp_timestamp=1710149961, |
| suffix=".gcov.xz", |
| gs_timestamp=timestamp, |
| ), |
| ) |
| |
| def test_kernel_gs_profile_file_name(self): |
| timestamp = datetime.datetime.fromtimestamp(1234, datetime.timezone.utc) |
| profile = update_kernel_afdo.KernelGsProfile.from_file_name( |
| timestamp, |
| "R124-15808.0-1710149961.gcov.xz", |
| ) |
| self.assertEqual(profile.file_name_no_suffix, "R124-15808.0-1710149961") |
| self.assertEqual(profile.file_name, "R124-15808.0-1710149961.gcov.xz") |
| |
| def test_gs_time_parsing(self): |
| self.assertEqual( |
| update_kernel_afdo.datetime_from_gs_time("2024-03-04T10:38:50Z"), |
| datetime.datetime( |
| year=2024, |
| month=3, |
| day=4, |
| hour=10, |
| minute=38, |
| second=50, |
| tzinfo=datetime.timezone.utc, |
| ), |
| ) |
| |
| @mock.patch.object(subprocess, "run") |
| def test_kernel_profile_fetcher_works(self, subprocess_run): |
| subprocess_run.return_value = subprocess.CompletedProcess( |
| args=[], |
| returncode=0, |
| # Don't use textwrap.dedent; linter complains about the line being |
| # too long in that case. |
| stdout=""" |
| 753112 2024-03-04T10:38:50Z gs://here/5.4/R124-15786.10-1709548729.gcov.xz |
| TOTAL: 2 objects, 1234 bytes (1.1KiB) |
| """, |
| ) |
| |
| fetcher = update_kernel_afdo.KernelProfileFetcher() |
| results = fetcher.fetch("gs://here/5.4") |
| |
| expected_results = [ |
| update_kernel_afdo.KernelGsProfile.from_file_name( |
| update_kernel_afdo.datetime_from_gs_time( |
| "2024-03-04T10:38:50Z" |
| ), |
| "R124-15786.10-1709548729.gcov.xz", |
| ), |
| ] |
| self.assertEqual(results, expected_results) |
| |
| @mock.patch.object(subprocess, "run") |
| def test_kernel_profile_fetcher_handles_no_profiles(self, subprocess_run): |
| subprocess_run.return_value = subprocess.CompletedProcess( |
| args=[], |
| returncode=1, |
| stderr="\nCommandException: One or more URLs matched no objects.\n", |
| ) |
| |
| fetcher = update_kernel_afdo.KernelProfileFetcher() |
| results = fetcher.fetch("gs://here/5.4") |
| self.assertEqual(results, []) |
| |
| @mock.patch.object(subprocess, "run") |
| def test_kernel_profile_fetcher_caches_urls(self, subprocess_run): |
| subprocess_run.return_value = subprocess.CompletedProcess( |
| args=[], |
| returncode=0, |
| # Don't use textwrap.dedent; linter complains about the line being |
| # too long in that case. |
| stdout=""" |
| 753112 2024-03-04T10:38:50Z gs://here/5.4/R124-15786.10-1709548729.gcov.xz |
| TOTAL: 2 objects, 1234 bytes (1.1KiB) |
| """, |
| ) |
| |
| fetcher = update_kernel_afdo.KernelProfileFetcher() |
| # Fetch these twice, and assert both that: |
| # - Only one fetch is performed. |
| # - Mutating the first list won't impact the later fetch. |
| result = fetcher.fetch("gs://here/5.4") |
| self.assertEqual(len(result), 1) |
| del result[:] |
| result = fetcher.fetch("gs://here/5.4") |
| self.assertEqual(len(result), 1) |
| subprocess_run.assert_called_once() |
| |
| @mock.patch.object(update_kernel_afdo.KernelProfileFetcher, "fetch") |
| def test_newest_afdo_artifact_finding_works(self, fetch): |
| late = update_kernel_afdo.KernelGsProfile.from_file_name( |
| datetime.datetime.fromtimestamp(1236, datetime.timezone.utc), |
| "R124-15786.10-1709548729.gcov.xz", |
| ) |
| early = update_kernel_afdo.KernelGsProfile.from_file_name( |
| datetime.datetime.fromtimestamp(1234, datetime.timezone.utc), |
| "R124-99999.99-9999999999.gcov.xz", |
| ) |
| fetch.return_value = [early, late] |
| |
| self.assertEqual( |
| update_kernel_afdo.find_newest_afdo_artifact( |
| update_kernel_afdo.KernelProfileFetcher(), |
| update_kernel_afdo.Arch.AMD64, |
| update_kernel_afdo.KernelVersion(5, 4), |
| release_number=124, |
| ), |
| late, |
| ) |
| |
| def test_afdo_descriptor_file_round_trips(self): |
| tmpdir = self.make_tempdir() |
| file_path = tmpdir / "desc-file.json" |
| |
| contents = { |
| update_kernel_afdo.KernelVersion(5, 10): "file1", |
| update_kernel_afdo.KernelVersion(5, 15): "file2", |
| } |
| self.assertTrue( |
| update_kernel_afdo.write_afdo_descriptor_file(file_path, contents) |
| ) |
| self.assertEqual( |
| update_kernel_afdo.read_afdo_descriptor_file(file_path), |
| contents, |
| ) |
| |
| def test_afdo_descriptor_file_refuses_to_rewrite_identical_contents(self): |
| tmpdir = self.make_tempdir() |
| file_path = tmpdir / "desc-file.json" |
| |
| contents = { |
| update_kernel_afdo.KernelVersion(5, 10): "file1", |
| update_kernel_afdo.KernelVersion(5, 15): "file2", |
| } |
| self.assertTrue( |
| update_kernel_afdo.write_afdo_descriptor_file(file_path, contents) |
| ) |
| self.assertFalse( |
| update_kernel_afdo.write_afdo_descriptor_file(file_path, contents) |
| ) |
| |
| def test_repo_autodetects_nothing_if_no_repo_dir(self): |
| self.assertIsNone( |
| update_kernel_afdo.find_chromeos_tree_root( |
| Path("/does/not/exist/nor/is/under/a/repo") |
| ) |
| ) |
| |
| def test_repo_autodetects_repo_dir_correctly(self): |
| tmpdir = self.make_tempdir() |
| test_subdir = tmpdir / "a/directory/and/another/one" |
| test_subdir.mkdir(parents=True) |
| (tmpdir / ".repo").mkdir() |
| self.assertEqual( |
| tmpdir, update_kernel_afdo.find_chromeos_tree_root(test_subdir) |
| ) |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |