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