]> 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:

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