xref: /aosp_15_r20/external/toolchain-utils/rust_tools/rust_watch_test.py (revision 760c253c1ed00ce9abd48f8546f08516e57485fe) 
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3# Copyright 2020 The ChromiumOS Authors
4# Use of this source code is governed by a BSD-style license that can be
5# found in the LICENSE file.
6
7"""Tests for rust_watch.py."""
8
9import logging
10import pathlib
11import subprocess
12import time
13import unittest
14import unittest.mock
15
16from cros_utils import tiny_render
17import rust_watch
18
19
20class Test(unittest.TestCase):
21    """Tests."""
22
23    def _silence_logs(self):
24        """Silences all log output until the end of the current test."""
25
26        def should_log(_record):
27            return 0
28
29        logger = logging.root
30        logger.addFilter(should_log)
31        self.addCleanup(logger.removeFilter, should_log)
32
33    def test_release_version_parsing(self):
34        self.assertEqual(
35            rust_watch.RustReleaseVersion.from_string("1.2.3"),
36            rust_watch.RustReleaseVersion(1, 2, 3),
37        )
38
39    def test_release_version_json_round_trips(self):
40        ver = rust_watch.RustReleaseVersion(1, 2, 3)
41        self.assertEqual(
42            rust_watch.RustReleaseVersion.from_json(ver.to_json()), ver
43        )
44
45    def test_state_json_round_trips(self):
46        state = rust_watch.State(
47            last_seen_release=rust_watch.RustReleaseVersion(1, 2, 3),
48            last_gentoo_sha="abc123",
49        )
50
51        self.assertEqual(rust_watch.State.from_json(state.to_json()), state)
52
53    @unittest.mock.patch.object(subprocess, "run")
54    @unittest.mock.patch.object(time, "sleep")
55    def test_update_git_repo_tries_again_on_failure(self, sleep_mock, run_mock):
56        self._silence_logs()
57
58        oh_no_error = ValueError("oh no")
59
60        def check_returncode():
61            raise oh_no_error
62
63        run_call_count = 0
64
65        def run_sideeffect(*_args, **_kwargs):
66            nonlocal run_call_count
67            run_call_count += 1
68            result = unittest.mock.Mock()
69            result.returncode = 1
70            result.check_returncode = check_returncode
71            return result
72
73        run_mock.side_effect = run_sideeffect
74
75        with self.assertRaises(ValueError) as raised:
76            rust_watch.update_git_repo(pathlib.Path("/does/not/exist/at/all"))
77
78        self.assertIs(raised.exception, oh_no_error)
79        self.assertEqual(run_call_count, 5)
80
81        sleep_timings = [unittest.mock.call(60 * i) for i in range(1, 5)]
82        self.assertEqual(sleep_mock.mock_calls, sleep_timings)
83
84    @unittest.mock.patch.object(subprocess, "run")
85    def test_get_new_gentoo_commits_functions(self, run_mock):
86        returned = unittest.mock.Mock()
87        returned.returncode = 0
88        returned.stdout = "\n".join(
89            (
90                "abc123 newer commit",
91                "abcdef and an older commit",
92            )
93        )
94        run_mock.return_value = returned
95        results = rust_watch.get_new_gentoo_commits(
96            pathlib.Path("/does/not/exist/at/all"), "defabc"
97        )
98        self.assertEqual(
99            results,
100            [
101                rust_watch.GitCommit("abcdef", "and an older commit"),
102                rust_watch.GitCommit("abc123", "newer commit"),
103            ],
104        )
105
106    def test_compose_email_on_a_new_gentoo_commit(self):
107        sha_a = "a" * 40
108        new_commit = rust_watch.maybe_compose_email(
109            new_gentoo_commits=[
110                rust_watch.GitCommit(
111                    sha=sha_a,
112                    subject="summary_a",
113                ),
114            ],
115        )
116
117        self.assertEqual(
118            new_commit,
119            (
120                "[rust-watch] new rust ebuild commit detected",
121                [
122                    "commit:",
123                    tiny_render.UnorderedList(
124                        [
125                            [
126                                tiny_render.Link(
127                                    rust_watch.gentoo_sha_to_link(sha_a),
128                                    sha_a[:12],
129                                ),
130                                ": summary_a",
131                            ],
132                        ]
133                    ),
134                ],
135            ),
136        )
137
138    def test_compose_email_composes_nothing_when_no_new_updates_exist(self):
139        self.assertIsNone(rust_watch.maybe_compose_email(new_gentoo_commits=()))
140
141    def test_compose_bug_creates_bugs_on_new_versions(self):
142        bug_body_start = "A new Rust stable release has been detected;"
143        title, body = rust_watch.maybe_compose_bug(
144            old_state=rust_watch.State(
145                last_seen_release=rust_watch.RustReleaseVersion(1, 0, 0),
146                last_gentoo_sha="",
147            ),
148            newest_release=rust_watch.RustReleaseVersion(1, 0, 1),
149        )
150        self.assertEqual(title, "[Rust] Update to 1.0.1")
151        self.assertTrue(body.startswith(bug_body_start))
152
153        title, body = rust_watch.maybe_compose_bug(
154            old_state=rust_watch.State(
155                last_seen_release=rust_watch.RustReleaseVersion(1, 0, 0),
156                last_gentoo_sha="",
157            ),
158            newest_release=rust_watch.RustReleaseVersion(1, 1, 0),
159        )
160        self.assertEqual(title, "[Rust] Update to 1.1.0")
161        self.assertTrue(body.startswith(bug_body_start))
162
163        title, body = rust_watch.maybe_compose_bug(
164            old_state=rust_watch.State(
165                last_seen_release=rust_watch.RustReleaseVersion(1, 0, 0),
166                last_gentoo_sha="",
167            ),
168            newest_release=rust_watch.RustReleaseVersion(2, 0, 0),
169        )
170        self.assertEqual(title, "[Rust] Update to 2.0.0")
171        self.assertTrue(body.startswith(bug_body_start))
172
173    def test_compose_bug_does_nothing_when_no_new_updates_exist(self):
174        self.assertIsNone(
175            rust_watch.maybe_compose_bug(
176                old_state=rust_watch.State(
177                    last_seen_release=rust_watch.RustReleaseVersion(1, 0, 0),
178                    last_gentoo_sha="",
179                ),
180                newest_release=rust_watch.RustReleaseVersion(1, 0, 0),
181            )
182        )
183
184
185if __name__ == "__main__":
186    unittest.main()
187