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()