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.
6 from contextlib import contextmanager
7 from copy import deepcopy
8 from io import StringIO
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
18 from click.testing import CliRunner
20 from black_primer import cli, lib
23 EXPECTED_ANALYSIS_OUTPUT = """\
32 -- primer results 📊 --
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
40 Failed projects: black
43 FAKE_PROJECT_CONFIG = {
44 "cli_arguments": ["--unittest"],
45 "expect_formatting_changes": False,
46 "git_clone_url": "https://github.com/psf/black.git",
52 command: Callable[..., Any], *args: Any, **kwargs: Any
54 old_stdout, sys.stdout = sys.stdout, StringIO()
56 command(*args, **kwargs)
58 yield sys.stdout.read()
60 sys.stdout = old_stdout
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())
76 async def raise_subprocess_error_1(*args: Any, **kwargs: Any) -> None:
77 raise CalledProcessError(1, ["unittest", "error"], b"", b"")
80 async def raise_subprocess_error_123(*args: Any, **kwargs: Any) -> None:
81 raise CalledProcessError(123, ["unittest", "error"], b"", b"")
84 async def return_false(*args: Any, **kwargs: Any) -> bool:
88 async def return_subproccess_output(*args: Any, **kwargs: Any) -> Tuple[bytes, bytes]:
89 return (b"stdout", b"stderr")
92 async def return_zero(*args: Any, **kwargs: Any) -> int:
96 if sys.version_info >= (3, 9):
104 def collect(queue: Q) -> List[T]:
108 item = queue.get_nowait()
110 except asyncio.QueueEmpty:
114 class PrimerLibTests(unittest.TestCase):
115 def test_analyze_results(self) -> None:
116 fake_results = lib.Results(
120 "skipped_long_checkout": 0,
124 {"black": CalledProcessError(69, ["black"], b"Black didn't work", b"")},
126 with capture_stdout(lib.analyze_results, 69, fake_results) as analyze_stdout:
127 self.assertEqual(EXPECTED_ANALYSIS_OUTPUT, analyze_stdout)
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}, {})
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)
143 self.assertEqual(1, results.stats["success"])
144 self.assertFalse(results.failed_projects)
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)
153 self.assertEqual(1, results.stats["failed"])
154 self.assertTrue(results.failed_projects)
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)
163 self.assertEqual(1, results.stats["failed"])
164 self.assertTrue(results.failed_projects)
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)
171 self.assertEqual(2, results.stats["failed"])
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))
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"])
184 self.assertTrue("The uncompromising code formatter" in stdout.decode("utf8"))
185 self.assertEqual(None, stderr)
187 # TODO: Add a test to see failure works on Windows
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]))
195 with self.assertRaises(asyncio.TimeoutError):
196 loop.run_until_complete(
197 lib._gen_check_output(["/bin/sleep", "2"], timeout=0.1)
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())
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)
211 self.assertEqual(expected_repo_path, returned_repo_path)
213 @patch("sys.stdout", new_callable=StringIO)
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(
224 str(config_path), Path(td), 2, ["django", "pyramid"]
227 self.assertEqual(0, return_val)
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"
236 config, projects_queue = loop.run_until_complete(
237 lib.load_projects_queue(config_path, ["django", "pyramid"])
239 projects = collect(projects_queue)
240 self.assertEqual(projects, ["django", "pyramid"])
243 class PrimerCLITests(unittest.TestCase):
245 def test_async_main(self) -> None:
246 loop = asyncio.get_event_loop()
247 work_dir = Path(gettempdir()) / f"primer_ut_{getpid()}"
252 "long_checkouts": False,
254 "workdir": str(work_dir),
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)
263 def test_handle_debug(self) -> None:
264 self.assertTrue(cli._handle_debug(None, None, True))
266 def test_help_output(self) -> None:
268 result = runner.invoke(cli.main, ["--help"])
269 self.assertEqual(result.exit_code, 0)
272 def test_projects(caplog: LogCaptureFixture) -> None:
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
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
290 if __name__ == "__main__":