• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3# Copyright 2020 The Chromium OS Authors. All rights reserved.
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
9# pylint: disable=cros-logging-import
10
11import logging
12import pathlib
13import subprocess
14import time
15import unittest
16import unittest.mock
17
18import rust_watch
19from cros_utils import tiny_render
20
21
22class Test(unittest.TestCase):
23  """Tests."""
24
25  def _silence_logs(self):
26    """Silences all log output until the end of the current test."""
27
28    def should_log(_record):
29      return 0
30
31    logger = logging.root
32    logger.addFilter(should_log)
33    self.addCleanup(logger.removeFilter, should_log)
34
35  def test_release_version_parsing(self):
36    self.assertEqual(
37        rust_watch.RustReleaseVersion.from_string('1.2.3'),
38        rust_watch.RustReleaseVersion(1, 2, 3),
39    )
40
41  def test_release_version_json_round_trips(self):
42    ver = rust_watch.RustReleaseVersion(1, 2, 3)
43    self.assertEqual(
44        rust_watch.RustReleaseVersion.from_json(ver.to_json()), ver)
45
46  def test_state_json_round_trips(self):
47    state = rust_watch.State(
48        last_seen_release=rust_watch.RustReleaseVersion(1, 2, 3),
49        last_gentoo_sha='abc123',
50    )
51
52    self.assertEqual(rust_watch.State.from_json(state.to_json()), state)
53
54  @unittest.mock.patch.object(subprocess, 'run')
55  @unittest.mock.patch.object(time, 'sleep')
56  def test_update_git_repo_tries_again_on_failure(self, sleep_mock, run_mock):
57    self._silence_logs()
58
59    oh_no_error = ValueError('oh no')
60
61    def check_returncode():
62      raise oh_no_error
63
64    run_call_count = 0
65
66    def run_sideeffect(*_args, **_kwargs):
67      nonlocal run_call_count
68      run_call_count += 1
69      result = unittest.mock.Mock()
70      result.returncode = 1
71      result.check_returncode = check_returncode
72      return result
73
74    run_mock.side_effect = run_sideeffect
75
76    with self.assertRaises(ValueError) as raised:
77      rust_watch.update_git_repo(pathlib.Path('/does/not/exist/at/all'))
78
79    self.assertIs(raised.exception, oh_no_error)
80    self.assertEqual(run_call_count, 5)
81
82    sleep_timings = [unittest.mock.call(60 * i) for i in range(1, 5)]
83    self.assertEqual(sleep_mock.mock_calls, sleep_timings)
84
85  @unittest.mock.patch.object(subprocess, 'run')
86  def test_get_new_gentoo_commits_functions(self, run_mock):
87    returned = unittest.mock.Mock()
88    returned.returncode = 0
89    returned.stdout = '\n'.join((
90        'abc123 newer commit',
91        'abcdef and an older commit',
92    ))
93    run_mock.return_value = returned
94    results = rust_watch.get_new_gentoo_commits(
95        pathlib.Path('/does/not/exist/at/all'), 'defabc')
96    self.assertEqual(results, [
97        rust_watch.GitCommit('abcdef', 'and an older commit'),
98        rust_watch.GitCommit('abc123', 'newer commit'),
99    ])
100
101  def test_compose_email_on_a_new_release(self):
102    new_release = rust_watch.maybe_compose_email(
103        old_state=rust_watch.State(
104            last_seen_release=rust_watch.RustReleaseVersion(1, 0, 0),
105            last_gentoo_sha='',
106        ),
107        newest_release=rust_watch.RustReleaseVersion(1, 1, 0),
108        new_gentoo_commits=[],
109    )
110
111    self.assertEqual(new_release, ('[rust-watch] new rustc release detected',
112                                   ['Rustc tag for v1.1.0 was found.']))
113
114  def test_compose_email_on_a_new_gentoo_commit(self):
115    sha_a = 'a' * 40
116    new_commit = rust_watch.maybe_compose_email(
117        old_state=rust_watch.State(
118            last_seen_release=rust_watch.RustReleaseVersion(1, 0, 0),
119            last_gentoo_sha='',
120        ),
121        newest_release=rust_watch.RustReleaseVersion(1, 0, 0),
122        new_gentoo_commits=[
123            rust_watch.GitCommit(
124                sha=sha_a,
125                subject='summary_a',
126            ),
127        ],
128    )
129
130    self.assertEqual(new_commit,
131                     ('[rust-watch] new rust ebuild commit detected', [
132                         'commit:',
133                         tiny_render.UnorderedList([
134                             [
135                                 tiny_render.Link(
136                                     rust_watch.gentoo_sha_to_link(sha_a),
137                                     sha_a[:12],
138                                 ),
139                                 ': summary_a',
140                             ],
141                         ])
142                     ]))
143
144  def test_compose_email_on_multiple_events(self):
145    sha_a = 'a' * 40
146    new_commit_and_release = rust_watch.maybe_compose_email(
147        old_state=rust_watch.State(
148            last_seen_release=rust_watch.RustReleaseVersion(1, 0, 0),
149            last_gentoo_sha='',
150        ),
151        newest_release=rust_watch.RustReleaseVersion(1, 1, 0),
152        new_gentoo_commits=[
153            rust_watch.GitCommit(
154                sha=sha_a,
155                subject='summary_a',
156            ),
157        ],
158    )
159
160    self.assertEqual(
161        new_commit_and_release,
162        ('[rust-watch] new rustc release detected; new rust ebuild commit '
163         'detected', [
164             'Rustc tag for v1.1.0 was found.',
165             tiny_render.line_break,
166             tiny_render.line_break,
167             'commit:',
168             tiny_render.UnorderedList([
169                 [
170                     tiny_render.Link(
171                         rust_watch.gentoo_sha_to_link(sha_a),
172                         sha_a[:12],
173                     ),
174                     ': summary_a',
175                 ],
176             ]),
177         ]))
178
179  def test_compose_email_composes_nothing_when_no_new_updates_exist(self):
180    self.assertIsNone(
181        rust_watch.maybe_compose_email(
182            old_state=rust_watch.State(
183                last_seen_release=rust_watch.RustReleaseVersion(1, 0, 0),
184                last_gentoo_sha='',
185            ),
186            newest_release=rust_watch.RustReleaseVersion(1, 0, 0),
187            new_gentoo_commits=[],
188        ))
189
190    self.assertIsNone(
191        rust_watch.maybe_compose_email(
192            old_state=rust_watch.State(
193                last_seen_release=rust_watch.RustReleaseVersion(1, 1, 0),
194                last_gentoo_sha='',
195            ),
196            newest_release=rust_watch.RustReleaseVersion(1, 0, 0),
197            new_gentoo_commits=[],
198        ))
199
200
201if __name__ == '__main__':
202  unittest.main()
203