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         project_name = "unittest"
 110         repo_path = Path(gettempdir())
 111         project_config = deepcopy(FAKE_PROJECT_CONFIG)
 112         results = lib.Results({"failed": 0, "success": 0}, {})
 113
 114         # Test a successful Black run
 115         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 116             loop.run_until_complete(
 117                 lib.black_run(project_name, repo_path, project_config, results)
 118             )
 119         self.assertEqual(1, results.stats["success"])
 120         self.assertFalse(results.failed_projects)
 121
 122         # Test a fail based on expecting formatting changes but not getting any
 123         project_config["expect_formatting_changes"] = True
 124         results = lib.Results({"failed": 0, "success": 0}, {})
 125         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 126             loop.run_until_complete(
 127                 lib.black_run(project_name, repo_path, project_config, results)
 128             )
 129         self.assertEqual(1, results.stats["failed"])
 130         self.assertTrue(results.failed_projects)
 131
 132         # Test a fail based on returning 1 and not expecting formatting changes
 133         project_config["expect_formatting_changes"] = False
 134         results = lib.Results({"failed": 0, "success": 0}, {})
 135         with patch("black_primer.lib._gen_check_output", raise_subprocess_error_1):
 136             loop.run_until_complete(
 137                 lib.black_run(project_name, repo_path, project_config, results)
 138             )
 139         self.assertEqual(1, results.stats["failed"])
 140         self.assertTrue(results.failed_projects)
 141
 142         # Test a formatting error based on returning 123
 143         with patch("black_primer.lib._gen_check_output", raise_subprocess_error_123):
 144             loop.run_until_complete(
 145                 lib.black_run(project_name, repo_path, project_config, results)
 146             )
 147         self.assertEqual(2, results.stats["failed"])
 148
 149     @event_loop()
 150     def test_gen_check_output(self) -> None:
 151         loop = asyncio.get_event_loop()
 152         stdout, stderr = loop.run_until_complete(
 153             lib._gen_check_output([lib.BLACK_BINARY, "--help"])
 154         )
 155         self.assertTrue("The uncompromising code formatter" in stdout.decode("utf8"))
 156         self.assertEqual(None, stderr)
 157
 158         # TODO: Add a test to see failure works on Windows
 159         if lib.WINDOWS:
 160             return
 161
 162         false_bin = "/usr/bin/false" if system() == "Darwin" else "/bin/false"
 163         with self.assertRaises(CalledProcessError):
 164             loop.run_until_complete(lib._gen_check_output([false_bin]))
 165
 166         with self.assertRaises(asyncio.TimeoutError):
 167             loop.run_until_complete(
 168                 lib._gen_check_output(["/bin/sleep", "2"], timeout=0.1)
 169             )
 170
 171     @event_loop()
 172     def test_git_checkout_or_rebase(self) -> None:
 173         loop = asyncio.get_event_loop()
 174         project_config = deepcopy(FAKE_PROJECT_CONFIG)
 175         work_path = Path(gettempdir())
 176
 177         expected_repo_path = work_path / "black"
 178         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 179             returned_repo_path = loop.run_until_complete(
 180                 lib.git_checkout_or_rebase(work_path, project_config)
 181             )
 182         self.assertEqual(expected_repo_path, returned_repo_path)
 183
 184     @patch("sys.stdout", new_callable=StringIO)
 185     @event_loop()
 186     def test_process_queue(self, mock_stdout: Mock) -> None:
 187         loop = asyncio.get_event_loop()
 188         config_path = Path(lib.__file__).parent / "primer.json"
 189         with patch("black_primer.lib.git_checkout_or_rebase", return_false):
 190             with TemporaryDirectory() as td:
 191                 return_val = loop.run_until_complete(
 192                     lib.process_queue(str(config_path), td, 2)
 193                 )
 194                 self.assertEqual(0, return_val)
 195
 196
 197 class PrimerCLITests(unittest.TestCase):
 198     @event_loop()
 199     def test_async_main(self) -> None:
 200         loop = asyncio.get_event_loop()
 201         work_dir = Path(gettempdir()) / f"primer_ut_{getpid()}"
 202         args = {
 203             "config": "/config",
 204             "debug": False,
 205             "keep": False,
 206             "long_checkouts": False,
 207             "rebase": False,
 208             "workdir": str(work_dir),
 209             "workers": 69,
 210             "no_diff": False,
 211         }
 212         with patch("black_primer.cli.lib.process_queue", return_zero):
 213             return_val = loop.run_until_complete(cli.async_main(**args))
 214             self.assertEqual(0, return_val)
 215
 216     def test_handle_debug(self) -> None:
 217         self.assertTrue(cli._handle_debug(None, None, True))
 218
 219     def test_help_output(self) -> None:
 220         runner = CliRunner()
 221         result = runner.invoke(cli.main, ["--help"])
 222         self.assertEqual(result.exit_code, 0)
 223
 224
 225 if __name__ == "__main__":
 226     unittest.main()