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     def test_flatten_cli_args(self) -> None:
 150         fake_long_args = ["--arg", ["really/", "|long", "|regex", "|splitup"], "--done"]
 151         expected = ["--arg", "really/|long|regex|splitup", "--done"]
 152         self.assertEqual(expected, lib._flatten_cli_args(fake_long_args))
 153
 154     @event_loop()
 155     def test_gen_check_output(self) -> None:
 156         loop = asyncio.get_event_loop()
 157         stdout, stderr = loop.run_until_complete(
 158             lib._gen_check_output([lib.BLACK_BINARY, "--help"])
 159         )
 160         self.assertTrue("The uncompromising code formatter" in stdout.decode("utf8"))
 161         self.assertEqual(None, stderr)
 162
 163         # TODO: Add a test to see failure works on Windows
 164         if lib.WINDOWS:
 165             return
 166
 167         false_bin = "/usr/bin/false" if system() == "Darwin" else "/bin/false"
 168         with self.assertRaises(CalledProcessError):
 169             loop.run_until_complete(lib._gen_check_output([false_bin]))
 170
 171         with self.assertRaises(asyncio.TimeoutError):
 172             loop.run_until_complete(
 173                 lib._gen_check_output(["/bin/sleep", "2"], timeout=0.1)
 174             )
 175
 176     @event_loop()
 177     def test_git_checkout_or_rebase(self) -> None:
 178         loop = asyncio.get_event_loop()
 179         project_config = deepcopy(FAKE_PROJECT_CONFIG)
 180         work_path = Path(gettempdir())
 181
 182         expected_repo_path = work_path / "black"
 183         with patch("black_primer.lib._gen_check_output", return_subproccess_output):
 184             returned_repo_path = loop.run_until_complete(
 185                 lib.git_checkout_or_rebase(work_path, project_config)
 186             )
 187         self.assertEqual(expected_repo_path, returned_repo_path)
 188
 189     @patch("sys.stdout", new_callable=StringIO)
 190     @event_loop()
 191     def test_process_queue(self, mock_stdout: Mock) -> None:
 192         """Test the process queue on primer itself
 193         - If you have non black conforming formatting in primer itself this can fail"""
 194         loop = asyncio.get_event_loop()
 195         config_path = Path(lib.__file__).parent / "primer.json"
 196         with patch("black_primer.lib.git_checkout_or_rebase", return_false):
 197             with TemporaryDirectory() as td:
 198                 return_val = loop.run_until_complete(
 199                     lib.process_queue(str(config_path), Path(td), 2)
 200                 )
 201                 self.assertEqual(0, return_val)
 202
 203
 204 class PrimerCLITests(unittest.TestCase):
 205     @event_loop()
 206     def test_async_main(self) -> None:
 207         loop = asyncio.get_event_loop()
 208         work_dir = Path(gettempdir()) / f"primer_ut_{getpid()}"
 209         args = {
 210             "config": "/config",
 211             "debug": False,
 212             "keep": False,
 213             "long_checkouts": False,
 214             "rebase": False,
 215             "workdir": str(work_dir),
 216             "workers": 69,
 217             "no_diff": False,
 218         }
 219         with patch("black_primer.cli.lib.process_queue", return_zero):
 220             return_val = loop.run_until_complete(cli.async_main(**args))  # type: ignore
 221             self.assertEqual(0, return_val)
 222
 223     def test_handle_debug(self) -> None:
 224         self.assertTrue(cli._handle_debug(None, None, True))
 225
 226     def test_help_output(self) -> None:
 227         runner = CliRunner()
 228         result = runner.invoke(cli.main, ["--help"])
 229         self.assertEqual(result.exit_code, 0)
 230
 231
 232 if __name__ == "__main__":
 233     unittest.main()