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, 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(
  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 class PrimerLibTests(unittest.TestCase):
  93     def test_analyze_results(self) -> None:
  94         fake_results = lib.Results(
  95             {
  96                 "disabled": 0,
  97                 "failed": 1,
  98                 "skipped_long_checkout": 0,
  99                 "success": 68,
 100                 "wrong_py_ver": 0,
 101             },
 102             {"black": CalledProcessError(69, ["black"], b"Black didn't work", b"")},
 103         )
 104         with capture_stdout(lib.analyze_results, 69, fake_results) as analyze_stdout:
 105             self.assertEqual(EXPECTED_ANALYSIS_OUTPUT, analyze_stdout)
 106
 107     @event_loop()
 108     def test_black_run(self) -> None:
 109         """Pretend to run Black to ensure we cater for all scenarios"""
 110         loop = asyncio.get_event_loop()
 111         project_name = "unittest"
 112         repo_path = Path(gettempdir())
 113         project_config = deepcopy(FAKE_PROJECT_CONFIG)
 114         results = lib.Results({"failed": 0, "success": 0}, {})
 115
 116         # Test a successful Black run
 117         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 118             loop.run_until_complete(
 119                 lib.black_run(project_name, repo_path, project_config, results)
 120             )
 121         self.assertEqual(1, results.stats["success"])
 122         self.assertFalse(results.failed_projects)
 123
 124         # Test a fail based on expecting formatting changes but not getting any
 125         project_config["expect_formatting_changes"] = True
 126         results = lib.Results({"failed": 0, "success": 0}, {})
 127         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 128             loop.run_until_complete(
 129                 lib.black_run(project_name, repo_path, project_config, results)
 130             )
 131         self.assertEqual(1, results.stats["failed"])
 132         self.assertTrue(results.failed_projects)
 133
 134         # Test a fail based on returning 1 and not expecting formatting changes
 135         project_config["expect_formatting_changes"] = False
 136         results = lib.Results({"failed": 0, "success": 0}, {})
 137         with patch("black_primer.lib._gen_check_output", raise_subprocess_error_1):
 138             loop.run_until_complete(
 139                 lib.black_run(project_name, repo_path, project_config, results)
 140             )
 141         self.assertEqual(1, results.stats["failed"])
 142         self.assertTrue(results.failed_projects)
 143
 144         # Test a formatting error based on returning 123
 145         with patch("black_primer.lib._gen_check_output", raise_subprocess_error_123):
 146             loop.run_until_complete(
 147                 lib.black_run(project_name, repo_path, project_config, results)
 148             )
 149         self.assertEqual(2, results.stats["failed"])
 150
 151     def test_flatten_cli_args(self) -> None:
 152         fake_long_args = ["--arg", ["really/", "|long", "|regex", "|splitup"], "--done"]
 153         expected = ["--arg", "really/|long|regex|splitup", "--done"]
 154         self.assertEqual(expected, lib._flatten_cli_args(fake_long_args))
 155
 156     @event_loop()
 157     def test_gen_check_output(self) -> None:
 158         loop = asyncio.get_event_loop()
 159         stdout, stderr = loop.run_until_complete(
 160             lib._gen_check_output([lib.BLACK_BINARY, "--help"])
 161         )
 162         self.assertTrue("The uncompromising code formatter" in stdout.decode("utf8"))
 163         self.assertEqual(None, stderr)
 164
 165         # TODO: Add a test to see failure works on Windows
 166         if lib.WINDOWS:
 167             return
 168
 169         false_bin = "/usr/bin/false" if system() == "Darwin" else "/bin/false"
 170         with self.assertRaises(CalledProcessError):
 171             loop.run_until_complete(lib._gen_check_output([false_bin]))
 172
 173         with self.assertRaises(asyncio.TimeoutError):
 174             loop.run_until_complete(
 175                 lib._gen_check_output(["/bin/sleep", "2"], timeout=0.1)
 176             )
 177
 178     @event_loop()
 179     def test_git_checkout_or_rebase(self) -> None:
 180         loop = asyncio.get_event_loop()
 181         project_config = deepcopy(FAKE_PROJECT_CONFIG)
 182         work_path = Path(gettempdir())
 183
 184         expected_repo_path = work_path / "black"
 185         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 186             returned_repo_path = loop.run_until_complete(
 187                 lib.git_checkout_or_rebase(work_path, project_config)
 188             )
 189         self.assertEqual(expected_repo_path, returned_repo_path)
 190
 191     @patch("sys.stdout", new_callable=StringIO)
 192     @event_loop()
 193     def test_process_queue(self, mock_stdout: Mock) -> None:
 194         """Test the process queue on primer itself
 195         - If you have non black conforming formatting in primer itself this can fail"""
 196         loop = asyncio.get_event_loop()
 197         config_path = Path(lib.__file__).parent / "primer.json"
 198         with patch("black_primer.lib.git_checkout_or_rebase", return_false):
 199             with TemporaryDirectory() as td:
 200                 return_val = loop.run_until_complete(
 201                     lib.process_queue(str(config_path), Path(td), 2)
 202                 )
 203                 self.assertEqual(0, return_val)
 204
 205
 206 class PrimerCLITests(unittest.TestCase):
 207     @event_loop()
 208     def test_async_main(self) -> None:
 209         loop = asyncio.get_event_loop()
 210         work_dir = Path(gettempdir()) / f"primer_ut_{getpid()}"
 211         args = {
 212             "config": "/config",
 213             "debug": False,
 214             "keep": False,
 215             "long_checkouts": False,
 216             "rebase": False,
 217             "workdir": str(work_dir),
 218             "workers": 69,
 219             "no_diff": False,
 220         }
 221         with patch("black_primer.cli.lib.process_queue", return_zero):
 222             return_val = loop.run_until_complete(cli.async_main(**args))  # type: ignore
 223             self.assertEqual(0, return_val)
 224
 225     def test_handle_debug(self) -> None:
 226         self.assertTrue(cli._handle_debug(None, None, True))
 227
 228     def test_help_output(self) -> None:
 229         runner = CliRunner()
 230         result = runner.invoke(cli.main, ["--help"])
 231         self.assertEqual(result.exit_code, 0)
 232
 233
 234 if __name__ == "__main__":
 235     unittest.main()