import types
import unittest
from concurrent.futures import ThreadPoolExecutor
-from contextlib import contextmanager
+from contextlib import contextmanager, redirect_stderr
from dataclasses import replace
from io import BytesIO
from pathlib import Path
self.assertEqual(
black.get_features_used(node), {Feature.ANN_ASSIGN_EXTENDED_RHS}
)
+ node = black.lib2to3_parse("try: pass\nexcept Something: pass")
+ self.assertEqual(black.get_features_used(node), set())
+ node = black.lib2to3_parse("try: pass\nexcept (*Something,): pass")
+ self.assertEqual(black.get_features_used(node), set())
+ node = black.lib2to3_parse("try: pass\nexcept *Group: pass")
+ self.assertEqual(black.get_features_used(node), {Feature.EXCEPT_STAR})
def test_get_features_used_for_future_flags(self) -> None:
for src, features in [
self.assertEqual("".join(err_lines), "")
@event_loop()
- @patch("black.ProcessPoolExecutor", MagicMock(side_effect=OSError))
+ @patch("concurrent.futures.ProcessPoolExecutor", MagicMock(side_effect=OSError))
def test_works_in_mono_process_only_environment(self) -> None:
with cache_dir() as workspace:
for f in [
def test_shhh_click(self) -> None:
try:
- from click import _unicodefun
- except ModuleNotFoundError:
+ from click import _unicodefun # type: ignore
+ except ImportError:
self.skipTest("Incompatible Click version")
- if not hasattr(_unicodefun, "_verify_python3_env"):
+
+ if not hasattr(_unicodefun, "_verify_python_env"):
self.skipTest("Incompatible Click version")
+
# First, let's see if Click is crashing with a preferred ASCII charset.
with patch("locale.getpreferredencoding") as gpe:
gpe.return_value = "ASCII"
with self.assertRaises(RuntimeError):
- _unicodefun._verify_python3_env() # type: ignore
+ _unicodefun._verify_python_env()
# Now, let's silence Click...
black.patch_click()
# ...and confirm it's silent.
with patch("locale.getpreferredencoding") as gpe:
gpe.return_value = "ASCII"
try:
- _unicodefun._verify_python3_env() # type: ignore
+ _unicodefun._verify_python_env()
except RuntimeError as re:
self.fail(f"`patch_click()` failed, exception still raised: {re}")
(src_dir.resolve(), "pyproject.toml"),
)
+ @patch(
+ "black.files.find_user_pyproject_toml",
+ )
+ def test_find_pyproject_toml(self, find_user_pyproject_toml: MagicMock) -> None:
+ find_user_pyproject_toml.side_effect = RuntimeError()
+
+ with redirect_stderr(io.StringIO()) as stderr:
+ result = black.files.find_pyproject_toml(
+ path_search_start=(str(Path.cwd().root),)
+ )
+
+ assert result is None
+ err = stderr.getvalue()
+ assert "Ignoring user configuration" in err
+
@patch(
"black.files.find_user_pyproject_toml",
black.files.find_user_pyproject_toml.__wrapped__,
normalized_path = black.normalize_path_maybe_ignore(path, root, report)
self.assertEqual(normalized_path, "workspace/project")
+ def test_normalize_path_ignore_windows_junctions_outside_of_root(self) -> None:
+ if system() != "Windows":
+ return
+
+ with TemporaryDirectory() as workspace:
+ root = Path(workspace)
+ junction_dir = root / "junction"
+ junction_target_outside_of_root = root / ".."
+ os.system(f"mklink /J {junction_dir} {junction_target_outside_of_root}")
+
+ report = black.Report(verbose=True)
+ normalized_path = black.normalize_path_maybe_ignore(
+ junction_dir, root, report
+ )
+ # Manually delete for Python < 3.8
+ os.system(f"rmdir {junction_dir}")
+
+ self.assertEqual(normalized_path, None)
+
def test_newline_comment_interaction(self) -> None:
source = "class A:\\\r\n# type: ignore\n pass\n"
output = black.format_str(source, mode=DEFAULT_MODE)
def test_cache_multiple_files(self) -> None:
mode = DEFAULT_MODE
with cache_dir() as workspace, patch(
- "black.ProcessPoolExecutor", new=ThreadPoolExecutor
+ "concurrent.futures.ProcessPoolExecutor", new=ThreadPoolExecutor
):
one = (workspace / "one.py").resolve()
with one.open("w") as fobj:
def test_failed_formatting_does_not_get_cached(self) -> None:
mode = DEFAULT_MODE
with cache_dir() as workspace, patch(
- "black.ProcessPoolExecutor", new=ThreadPoolExecutor
+ "concurrent.futures.ProcessPoolExecutor", new=ThreadPoolExecutor
):
failing = (workspace / "failing.py").resolve()
with failing.open("w") as fobj:
path.iterdir.return_value = [child]
child.resolve.return_value = Path("/a/b/c")
child.as_posix.return_value = "/a/b/c"
- child.is_symlink.return_value = True
try:
list(
black.gen_python_files(
pytest.fail(f"`get_python_files_in_dir()` failed: {ve}")
path.iterdir.assert_called_once()
child.resolve.assert_called_once()
- child.is_symlink.assert_called_once()
- # `child` should behave like a strange file which resolved path is clearly
- # outside of the `root` directory.
- child.is_symlink.return_value = False
- with pytest.raises(ValueError):
- list(
- black.gen_python_files(
- path.iterdir(),
- root,
- include,
- exclude,
- None,
- None,
- report,
- gitignore,
- verbose=False,
- quiet=False,
- )
- )
- path.iterdir.assert_called()
- assert path.iterdir.call_count == 2
- child.resolve.assert_called()
- assert child.resolve.call_count == 2
- child.is_symlink.assert_called()
- assert child.is_symlink.call_count == 2
@patch("black.find_project_root", lambda *args: (THIS_DIR.resolve(), None))
def test_get_sources_with_stdin(self) -> None:
projects / etc / vim.git / blobdiff