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.assertTrue("The uncompromising code formatter" in 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()