]> git.madduck.net Git - etc/vim.git/blob - tests/test_primer.py

madduck's git repository

Every one of the projects in this repository is available at the canonical URL git://git.madduck.net/madduck/pub/<projectpath> — see each project's metadata for the exact URL.

All patches and comments are welcome. Please squash your changes to logical commits before using git-format-patch and git-send-email to patches@git.madduck.net. If you'd read over the Git project's submission guidelines and adhered to them, I'd be especially grateful.

SSH access, as well as push access can be individually arranged.

If you use my repositories frequently, consider adding the following snippet to ~/.gitconfig and using the third clone URL listed for each project:

[url "git://git.madduck.net/madduck/"]
  insteadOf = madduck:

Switch to Furo (#2793)
[etc/vim.git] / tests / test_primer.py
1 #!/usr/bin/env python3
2
3 import asyncio
4 import sys
5 import unittest
6 from contextlib import contextmanager
7 from copy import deepcopy
8 from io import StringIO
9 from os import getpid
10 from pathlib import Path
11 from platform import system
12 from pytest import LogCaptureFixture
13 from subprocess import CalledProcessError
14 from tempfile import TemporaryDirectory, gettempdir
15 from typing import Any, Callable, Iterator, List, Tuple, TypeVar
16 from unittest.mock import Mock, patch
17
18 from click.testing import CliRunner
19
20 from black_primer import cli, lib
21
22
23 EXPECTED_ANALYSIS_OUTPUT = """\
24
25 Failed projects:
26
27 ## black:
28  - Returned 69
29  - stdout:
30 Black didn't work
31
32 -- primer results 📊 --
33
34 68 / 69 succeeded (98.55%) ✅
35 1 / 69 FAILED (1.45%) 💩
36  - 0 projects disabled by config
37  - 0 projects skipped due to Python version
38  - 0 skipped due to long checkout
39
40 Failed projects: black
41
42 """
43 FAKE_PROJECT_CONFIG = {
44     "cli_arguments": ["--unittest"],
45     "expect_formatting_changes": False,
46     "git_clone_url": "https://github.com/psf/black.git",
47 }
48
49
50 @contextmanager
51 def capture_stdout(
52     command: Callable[..., Any], *args: Any, **kwargs: Any
53 ) -> Iterator[str]:
54     old_stdout, sys.stdout = sys.stdout, StringIO()
55     try:
56         command(*args, **kwargs)
57         sys.stdout.seek(0)
58         yield sys.stdout.read()
59     finally:
60         sys.stdout = old_stdout
61
62
63 @contextmanager
64 def event_loop() -> Iterator[None]:
65     policy = asyncio.get_event_loop_policy()
66     loop = policy.new_event_loop()
67     asyncio.set_event_loop(loop)
68     if sys.platform == "win32":
69         asyncio.set_event_loop(asyncio.ProactorEventLoop())
70     try:
71         yield
72     finally:
73         loop.close()
74
75
76 async def raise_subprocess_error_1(*args: Any, **kwargs: Any) -> None:
77     raise CalledProcessError(1, ["unittest", "error"], b"", b"")
78
79
80 async def raise_subprocess_error_123(*args: Any, **kwargs: Any) -> None:
81     raise CalledProcessError(123, ["unittest", "error"], b"", b"")
82
83
84 async def return_false(*args: Any, **kwargs: Any) -> bool:
85     return False
86
87
88 async def return_subproccess_output(*args: Any, **kwargs: Any) -> Tuple[bytes, bytes]:
89     return (b"stdout", b"stderr")
90
91
92 async def return_zero(*args: Any, **kwargs: Any) -> int:
93     return 0
94
95
96 if sys.version_info >= (3, 9):
97     T = TypeVar("T")
98     Q = asyncio.Queue[T]
99 else:
100     T = Any
101     Q = asyncio.Queue
102
103
104 def collect(queue: Q) -> List[T]:
105     ret = []
106     while True:
107         try:
108             item = queue.get_nowait()
109             ret.append(item)
110         except asyncio.QueueEmpty:
111             return ret
112
113
114 class PrimerLibTests(unittest.TestCase):
115     def test_analyze_results(self) -> None:
116         fake_results = lib.Results(
117             {
118                 "disabled": 0,
119                 "failed": 1,
120                 "skipped_long_checkout": 0,
121                 "success": 68,
122                 "wrong_py_ver": 0,
123             },
124             {"black": CalledProcessError(69, ["black"], b"Black didn't work", b"")},
125         )
126         with capture_stdout(lib.analyze_results, 69, fake_results) as analyze_stdout:
127             self.assertEqual(EXPECTED_ANALYSIS_OUTPUT, analyze_stdout)
128
129     @event_loop()
130     def test_black_run(self) -> None:
131         """Pretend to run Black to ensure we cater for all scenarios"""
132         loop = asyncio.get_event_loop()
133         project_name = "unittest"
134         repo_path = Path(gettempdir())
135         project_config = deepcopy(FAKE_PROJECT_CONFIG)
136         results = lib.Results({"failed": 0, "success": 0}, {})
137
138         # Test a successful Black run
139         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
140             loop.run_until_complete(
141                 lib.black_run(project_name, repo_path, project_config, results)
142             )
143         self.assertEqual(1, results.stats["success"])
144         self.assertFalse(results.failed_projects)
145
146         # Test a fail based on expecting formatting changes but not getting any
147         project_config["expect_formatting_changes"] = True
148         results = lib.Results({"failed": 0, "success": 0}, {})
149         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
150             loop.run_until_complete(
151                 lib.black_run(project_name, repo_path, project_config, results)
152             )
153         self.assertEqual(1, results.stats["failed"])
154         self.assertTrue(results.failed_projects)
155
156         # Test a fail based on returning 1 and not expecting formatting changes
157         project_config["expect_formatting_changes"] = False
158         results = lib.Results({"failed": 0, "success": 0}, {})
159         with patch("black_primer.lib._gen_check_output", raise_subprocess_error_1):
160             loop.run_until_complete(
161                 lib.black_run(project_name, repo_path, project_config, results)
162             )
163         self.assertEqual(1, results.stats["failed"])
164         self.assertTrue(results.failed_projects)
165
166         # Test a formatting error based on returning 123
167         with patch("black_primer.lib._gen_check_output", raise_subprocess_error_123):
168             loop.run_until_complete(
169                 lib.black_run(project_name, repo_path, project_config, results)
170             )
171         self.assertEqual(2, results.stats["failed"])
172
173     def test_flatten_cli_args(self) -> None:
174         fake_long_args = ["--arg", ["really/", "|long", "|regex", "|splitup"], "--done"]
175         expected = ["--arg", "really/|long|regex|splitup", "--done"]
176         self.assertEqual(expected, lib._flatten_cli_args(fake_long_args))
177
178     @event_loop()
179     def test_gen_check_output(self) -> None:
180         loop = asyncio.get_event_loop()
181         stdout, stderr = loop.run_until_complete(
182             lib._gen_check_output([lib.BLACK_BINARY, "--help"])
183         )
184         self.assertIn("The uncompromising code formatter", stdout.decode("utf8"))
185         self.assertEqual(None, stderr)
186
187         # TODO: Add a test to see failure works on Windows
188         if lib.WINDOWS:
189             return
190
191         false_bin = "/usr/bin/false" if system() == "Darwin" else "/bin/false"
192         with self.assertRaises(CalledProcessError):
193             loop.run_until_complete(lib._gen_check_output([false_bin]))
194
195         with self.assertRaises(asyncio.TimeoutError):
196             loop.run_until_complete(
197                 lib._gen_check_output(["/bin/sleep", "2"], timeout=0.1)
198             )
199
200     @event_loop()
201     def test_git_checkout_or_rebase(self) -> None:
202         loop = asyncio.get_event_loop()
203         project_config = deepcopy(FAKE_PROJECT_CONFIG)
204         work_path = Path(gettempdir())
205
206         expected_repo_path = work_path / "black"
207         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
208             returned_repo_path = loop.run_until_complete(
209                 lib.git_checkout_or_rebase(work_path, project_config)
210             )
211         self.assertEqual(expected_repo_path, returned_repo_path)
212
213     @patch("sys.stdout", new_callable=StringIO)
214     @event_loop()
215     def test_process_queue(self, mock_stdout: Mock) -> None:
216         """Test the process queue on primer itself
217         - If you have non black conforming formatting in primer itself this can fail"""
218         loop = asyncio.get_event_loop()
219         config_path = Path(lib.__file__).parent / "primer.json"
220         with patch("black_primer.lib.git_checkout_or_rebase", return_false):
221             with TemporaryDirectory() as td:
222                 return_val = loop.run_until_complete(
223                     lib.process_queue(
224                         str(config_path), Path(td), 2, ["django", "pyramid"]
225                     )
226                 )
227                 self.assertEqual(0, return_val)
228
229     @event_loop()
230     def test_load_projects_queue(self) -> None:
231         """Test the process queue on primer itself
232         - If you have non black conforming formatting in primer itself this can fail"""
233         loop = asyncio.get_event_loop()
234         config_path = Path(lib.__file__).parent / "primer.json"
235
236         config, projects_queue = loop.run_until_complete(
237             lib.load_projects_queue(config_path, ["django", "pyramid"])
238         )
239         projects = collect(projects_queue)
240         self.assertEqual(projects, ["django", "pyramid"])
241
242
243 class PrimerCLITests(unittest.TestCase):
244     @event_loop()
245     def test_async_main(self) -> None:
246         loop = asyncio.get_event_loop()
247         work_dir = Path(gettempdir()) / f"primer_ut_{getpid()}"
248         args = {
249             "config": "/config",
250             "debug": False,
251             "keep": False,
252             "long_checkouts": False,
253             "rebase": False,
254             "workdir": str(work_dir),
255             "workers": 69,
256             "no_diff": False,
257             "projects": "",
258         }
259         with patch("black_primer.cli.lib.process_queue", return_zero):
260             return_val = loop.run_until_complete(cli.async_main(**args))  # type: ignore
261             self.assertEqual(0, return_val)
262
263     def test_handle_debug(self) -> None:
264         self.assertTrue(cli._handle_debug(None, None, True))
265
266     def test_help_output(self) -> None:
267         runner = CliRunner()
268         result = runner.invoke(cli.main, ["--help"])
269         self.assertEqual(result.exit_code, 0)
270
271
272 def test_projects(caplog: LogCaptureFixture) -> None:
273     with event_loop():
274         runner = CliRunner()
275         result = runner.invoke(cli.main, ["--projects=STDIN,asdf"])
276         assert result.exit_code == 0
277         assert "1 / 1 succeeded" in result.output
278         assert "Projects not found: {'asdf'}" in caplog.text
279
280     caplog.clear()
281
282     with event_loop():
283         runner = CliRunner()
284         result = runner.invoke(cli.main, ["--projects=fdsa,STDIN"])
285         assert result.exit_code == 0
286         assert "1 / 1 succeeded" in result.output
287         assert "Projects not found: {'fdsa'}" in caplog.text
288
289
290 if __name__ == "__main__":
291     unittest.main()