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(*args: Any, **kwargs: Any) -> None:
  71     raise CalledProcessError(1, ["unittest", "error"], b"", b"")
  72
  73
  74 async def return_false(*args: Any, **kwargs: Any) -> bool:
  75     return False
  76
  77
  78 async def return_subproccess_output(*args: Any, **kwargs: Any) -> Tuple[bytes, bytes]:
  79     return (b"stdout", b"stderr")
  80
  81
  82 async def return_zero(*args: Any, **kwargs: Any) -> int:
  83     return 0
  84
  85
  86 class PrimerLibTests(unittest.TestCase):
  87     def test_analyze_results(self) -> None:
  88         fake_results = lib.Results(
  89             {
  90                 "disabled": 0,
  91                 "failed": 1,
  92                 "skipped_long_checkout": 0,
  93                 "success": 68,
  94                 "wrong_py_ver": 0,
  95             },
  96             {"black": CalledProcessError(69, ["black"], b"black didn't work", b"")},
  97         )
  98         with capture_stdout(lib.analyze_results, 69, fake_results) as analyze_stdout:
  99             self.assertEqual(EXPECTED_ANALYSIS_OUTPUT, analyze_stdout)
 100
 101     @event_loop()
 102     def test_black_run(self) -> None:
 103         """Pretend run black to ensure we cater for all scenarios"""
 104         loop = asyncio.get_event_loop()
 105         repo_path = Path(gettempdir())
 106         project_config = deepcopy(FAKE_PROJECT_CONFIG)
 107         results = lib.Results({"failed": 0, "success": 0}, {})
 108
 109         # Test a successful black run
 110         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 111             loop.run_until_complete(lib.black_run(repo_path, project_config, results))
 112         self.assertEqual(1, results.stats["success"])
 113         self.assertFalse(results.failed_projects)
 114
 115         # Test a fail based on expecting formatting changes but not getting any
 116         project_config["expect_formatting_changes"] = True
 117         results = lib.Results({"failed": 0, "success": 0}, {})
 118         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 119             loop.run_until_complete(lib.black_run(repo_path, project_config, results))
 120         self.assertEqual(1, results.stats["failed"])
 121         self.assertTrue(results.failed_projects)
 122
 123         # Test a fail based on returning 1 and not expecting formatting changes
 124         project_config["expect_formatting_changes"] = False
 125         results = lib.Results({"failed": 0, "success": 0}, {})
 126         with patch("black_primer.lib._gen_check_output", raise_subprocess_error):
 127             loop.run_until_complete(lib.black_run(repo_path, project_config, results))
 128         self.assertEqual(1, results.stats["failed"])
 129         self.assertTrue(results.failed_projects)
 130
 131     @event_loop()
 132     def test_gen_check_output(self) -> None:
 133         loop = asyncio.get_event_loop()
 134         stdout, stderr = loop.run_until_complete(
 135             lib._gen_check_output([lib.BLACK_BINARY, "--help"])
 136         )
 137         self.assertTrue("The uncompromising code formatter" in stdout.decode("utf8"))
 138         self.assertEqual(None, stderr)
 139
 140         # TODO: Add a test to see failure works on Windows
 141         if lib.WINDOWS:
 142             return
 143
 144         false_bin = "/usr/bin/false" if system() == "Darwin" else "/bin/false"
 145         with self.assertRaises(CalledProcessError):
 146             loop.run_until_complete(lib._gen_check_output([false_bin]))
 147
 148         with self.assertRaises(asyncio.TimeoutError):
 149             loop.run_until_complete(
 150                 lib._gen_check_output(["/bin/sleep", "2"], timeout=0.1)
 151             )
 152
 153     @event_loop()
 154     def test_git_checkout_or_rebase(self) -> None:
 155         loop = asyncio.get_event_loop()
 156         project_config = deepcopy(FAKE_PROJECT_CONFIG)
 157         work_path = Path(gettempdir())
 158
 159         expected_repo_path = work_path / "black"
 160         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 161             returned_repo_path = loop.run_until_complete(
 162                 lib.git_checkout_or_rebase(work_path, project_config)
 163             )
 164         self.assertEqual(expected_repo_path, returned_repo_path)
 165
 166     @patch("sys.stdout", new_callable=StringIO)
 167     @event_loop()
 168     def test_process_queue(self, mock_stdout: Mock) -> None:
 169         loop = asyncio.get_event_loop()
 170         config_path = Path(lib.__file__).parent / "primer.json"
 171         with patch("black_primer.lib.git_checkout_or_rebase", return_false):
 172             with TemporaryDirectory() as td:
 173                 return_val = loop.run_until_complete(
 174                     lib.process_queue(str(config_path), td, 2)
 175                 )
 176                 self.assertEqual(0, return_val)
 177
 178
 179 class PrimerCLITests(unittest.TestCase):
 180     @event_loop()
 181     def test_async_main(self) -> None:
 182         loop = asyncio.get_event_loop()
 183         work_dir = Path(gettempdir()) / f"primer_ut_{getpid()}"
 184         args = {
 185             "config": "/config",
 186             "debug": False,
 187             "keep": False,
 188             "long_checkouts": False,
 189             "rebase": False,
 190             "workdir": str(work_dir),
 191             "workers": 69,
 192         }
 193         with patch("black_primer.cli.lib.process_queue", return_zero):
 194             return_val = loop.run_until_complete(cli.async_main(**args))
 195             self.assertEqual(0, return_val)
 196
 197     def test_handle_debug(self) -> None:
 198         self.assertTrue(cli._handle_debug(None, None, True))
 199
 200     def test_help_output(self) -> None:
 201         runner = CliRunner()
 202         result = runner.invoke(cli.main, ["--help"])
 203         self.assertEqual(result.exit_code, 0)
 204
 205
 206 if __name__ == "__main__":
 207     unittest.main()