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

Add compatible configuration files. (psf#1789) (#1792)
[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, Generator, Iterator, Tuple
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 -- primer results 📊 --
24
25 68 / 69 succeeded (98.55%) ✅
26 1 / 69 FAILED (1.45%) 💩
27  - 0 projects disabled by config
28  - 0 projects skipped due to Python version
29  - 0 skipped due to long checkout
30
31 Failed projects:
32
33 ## black:
34  - Returned 69
35  - stdout:
36 Black didn't work
37
38 """
39 FAKE_PROJECT_CONFIG = {
40     "cli_arguments": ["--unittest"],
41     "expect_formatting_changes": False,
42     "git_clone_url": "https://github.com/psf/black.git",
43 }
44
45
46 @contextmanager
47 def capture_stdout(command: Callable, *args: Any, **kwargs: Any) -> Generator:
48     old_stdout, sys.stdout = sys.stdout, StringIO()
49     try:
50         command(*args, **kwargs)
51         sys.stdout.seek(0)
52         yield sys.stdout.read()
53     finally:
54         sys.stdout = old_stdout
55
56
57 @contextmanager
58 def event_loop() -> Iterator[None]:
59     policy = asyncio.get_event_loop_policy()
60     loop = policy.new_event_loop()
61     asyncio.set_event_loop(loop)
62     if sys.platform == "win32":
63         asyncio.set_event_loop(asyncio.ProactorEventLoop())
64     try:
65         yield
66     finally:
67         loop.close()
68
69
70 async def raise_subprocess_error_1(*args: Any, **kwargs: Any) -> None:
71     raise CalledProcessError(1, ["unittest", "error"], b"", b"")
72
73
74 async def raise_subprocess_error_123(*args: Any, **kwargs: Any) -> None:
75     raise CalledProcessError(123, ["unittest", "error"], b"", b"")
76
77
78 async def return_false(*args: Any, **kwargs: Any) -> bool:
79     return False
80
81
82 async def return_subproccess_output(*args: Any, **kwargs: Any) -> Tuple[bytes, bytes]:
83     return (b"stdout", b"stderr")
84
85
86 async def return_zero(*args: Any, **kwargs: Any) -> int:
87     return 0
88
89
90 class PrimerLibTests(unittest.TestCase):
91     def test_analyze_results(self) -> None:
92         fake_results = lib.Results(
93             {
94                 "disabled": 0,
95                 "failed": 1,
96                 "skipped_long_checkout": 0,
97                 "success": 68,
98                 "wrong_py_ver": 0,
99             },
100             {"black": CalledProcessError(69, ["black"], b"Black didn't work", b"")},
101         )
102         with capture_stdout(lib.analyze_results, 69, fake_results) as analyze_stdout:
103             self.assertEqual(EXPECTED_ANALYSIS_OUTPUT, analyze_stdout)
104
105     @event_loop()
106     def test_black_run(self) -> None:
107         """Pretend to run Black to ensure we cater for all scenarios"""
108         loop = asyncio.get_event_loop()
109         repo_path = Path(gettempdir())
110         project_config = deepcopy(FAKE_PROJECT_CONFIG)
111         results = lib.Results({"failed": 0, "success": 0}, {})
112
113         # Test a successful Black run
114         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
115             loop.run_until_complete(lib.black_run(repo_path, project_config, results))
116         self.assertEqual(1, results.stats["success"])
117         self.assertFalse(results.failed_projects)
118
119         # Test a fail based on expecting formatting changes but not getting any
120         project_config["expect_formatting_changes"] = True
121         results = lib.Results({"failed": 0, "success": 0}, {})
122         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
123             loop.run_until_complete(lib.black_run(repo_path, project_config, results))
124         self.assertEqual(1, results.stats["failed"])
125         self.assertTrue(results.failed_projects)
126
127         # Test a fail based on returning 1 and not expecting formatting changes
128         project_config["expect_formatting_changes"] = False
129         results = lib.Results({"failed": 0, "success": 0}, {})
130         with patch("black_primer.lib._gen_check_output", raise_subprocess_error_1):
131             loop.run_until_complete(lib.black_run(repo_path, project_config, results))
132         self.assertEqual(1, results.stats["failed"])
133         self.assertTrue(results.failed_projects)
134
135         # Test a formatting error based on returning 123
136         with patch("black_primer.lib._gen_check_output", raise_subprocess_error_123):
137             loop.run_until_complete(lib.black_run(repo_path, project_config, results))
138         self.assertEqual(2, results.stats["failed"])
139
140     @event_loop()
141     def test_gen_check_output(self) -> None:
142         loop = asyncio.get_event_loop()
143         stdout, stderr = loop.run_until_complete(
144             lib._gen_check_output([lib.BLACK_BINARY, "--help"])
145         )
146         self.assertTrue("The uncompromising code formatter" in stdout.decode("utf8"))
147         self.assertEqual(None, stderr)
148
149         # TODO: Add a test to see failure works on Windows
150         if lib.WINDOWS:
151             return
152
153         false_bin = "/usr/bin/false" if system() == "Darwin" else "/bin/false"
154         with self.assertRaises(CalledProcessError):
155             loop.run_until_complete(lib._gen_check_output([false_bin]))
156
157         with self.assertRaises(asyncio.TimeoutError):
158             loop.run_until_complete(
159                 lib._gen_check_output(["/bin/sleep", "2"], timeout=0.1)
160             )
161
162     @event_loop()
163     def test_git_checkout_or_rebase(self) -> None:
164         loop = asyncio.get_event_loop()
165         project_config = deepcopy(FAKE_PROJECT_CONFIG)
166         work_path = Path(gettempdir())
167
168         expected_repo_path = work_path / "black"
169         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
170             returned_repo_path = loop.run_until_complete(
171                 lib.git_checkout_or_rebase(work_path, project_config)
172             )
173         self.assertEqual(expected_repo_path, returned_repo_path)
174
175     @patch("sys.stdout", new_callable=StringIO)
176     @event_loop()
177     def test_process_queue(self, mock_stdout: Mock) -> None:
178         loop = asyncio.get_event_loop()
179         config_path = Path(lib.__file__).parent / "primer.json"
180         with patch("black_primer.lib.git_checkout_or_rebase", return_false):
181             with TemporaryDirectory() as td:
182                 return_val = loop.run_until_complete(
183                     lib.process_queue(str(config_path), td, 2)
184                 )
185                 self.assertEqual(0, return_val)
186
187
188 class PrimerCLITests(unittest.TestCase):
189     @event_loop()
190     def test_async_main(self) -> None:
191         loop = asyncio.get_event_loop()
192         work_dir = Path(gettempdir()) / f"primer_ut_{getpid()}"
193         args = {
194             "config": "/config",
195             "debug": False,
196             "keep": False,
197             "long_checkouts": False,
198             "rebase": False,
199             "workdir": str(work_dir),
200             "workers": 69,
201         }
202         with patch("black_primer.cli.lib.process_queue", return_zero):
203             return_val = loop.run_until_complete(cli.async_main(**args))
204             self.assertEqual(0, return_val)
205
206     def test_handle_debug(self) -> None:
207         self.assertTrue(cli._handle_debug(None, None, True))
208
209     def test_help_output(self) -> None:
210         runner = CliRunner()
211         result = runner.invoke(cli.main, ["--help"])
212         self.assertEqual(result.exit_code, 0)
213
214
215 if __name__ == "__main__":
216     unittest.main()