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