import logging
import multiprocessing
import os
+import re
import sys
import types
import unittest
import click
import pytest
-import re
from click import unstyle
from click.testing import CliRunner
from pathspec import PathSpec
dump_to_stderr,
ff,
fs,
+ get_case_path,
read_data,
+ read_data_from_file,
)
THIS_FILE = Path(__file__)
)
def test_piping(self) -> None:
- source, expected = read_data("src/black/__init__", data=False)
+ source, expected = read_data_from_file(PROJECT_ROOT / "src/black/__init__.py")
result = BlackRunner().invoke(
black.main,
[
r"(STDIN|STDOUT)\t\d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d\.\d\d\d\d\d\d "
r"\+\d\d\d\d"
)
- source, _ = read_data("expression.py")
- expected, _ = read_data("expression.diff")
+ source, _ = read_data("simple_cases", "expression.py")
+ expected, _ = read_data("simple_cases", "expression.diff")
args = [
"-",
"--fast",
self.assertEqual(expected, actual)
def test_piping_diff_with_color(self) -> None:
- source, _ = read_data("expression.py")
+ source, _ = read_data("simple_cases", "expression.py")
args = [
"-",
"--fast",
@patch("black.dump_to_file", dump_to_stderr)
def _test_wip(self) -> None:
- source, expected = read_data("wip")
+ source, expected = read_data("miscellaneous", "wip")
sys.settrace(tracefunc)
mode = replace(
DEFAULT_MODE,
black.assert_stable(source, actual, black.FileMode())
def test_pep_572_version_detection(self) -> None:
- source, _ = read_data("pep_572")
+ source, _ = read_data("py_38", "pep_572")
root = black.lib2to3_parse(source)
features = black.get_features_used(root)
self.assertIn(black.Feature.ASSIGNMENT_EXPRESSIONS, features)
self.assertIn(black.TargetVersion.PY38, versions)
def test_expression_ff(self) -> None:
- source, expected = read_data("expression")
+ source, expected = read_data("simple_cases", "expression.py")
tmp_file = Path(black.dump_to_file(source))
try:
self.assertTrue(ff(tmp_file, write_back=black.WriteBack.YES))
black.assert_stable(source, actual, DEFAULT_MODE)
def test_expression_diff(self) -> None:
- source, _ = read_data("expression.py")
- expected, _ = read_data("expression.diff")
+ source, _ = read_data("simple_cases", "expression.py")
+ expected, _ = read_data("simple_cases", "expression.diff")
tmp_file = Path(black.dump_to_file(source))
diff_header = re.compile(
rf"{re.escape(str(tmp_file))}\t\d\d\d\d-\d\d-\d\d "
self.assertEqual(expected, actual, msg)
def test_expression_diff_with_color(self) -> None:
- source, _ = read_data("expression.py")
- expected, _ = read_data("expression.diff")
+ source, _ = read_data("simple_cases", "expression.py")
+ expected, _ = read_data("simple_cases", "expression.diff")
tmp_file = Path(black.dump_to_file(source))
try:
result = BlackRunner().invoke(
self.assertIn("\033[0m", actual)
def test_detect_pos_only_arguments(self) -> None:
- source, _ = read_data("pep_570")
+ source, _ = read_data("py_38", "pep_570")
root = black.lib2to3_parse(source)
features = black.get_features_used(root)
self.assertIn(black.Feature.POS_ONLY_ARGUMENTS, features)
versions = black.detect_target_versions(root)
self.assertIn(black.TargetVersion.PY38, versions)
+ def test_detect_debug_f_strings(self) -> None:
+ root = black.lib2to3_parse("""f"{x=}" """)
+ features = black.get_features_used(root)
+ self.assertIn(black.Feature.DEBUG_F_STRINGS, features)
+ versions = black.detect_target_versions(root)
+ self.assertIn(black.TargetVersion.PY38, versions)
+
+ root = black.lib2to3_parse(
+ """f"{x}"\nf'{"="}'\nf'{(x:=5)}'\nf'{f(a="3=")}'\nf'{x:=10}'\n"""
+ )
+ features = black.get_features_used(root)
+ self.assertNotIn(black.Feature.DEBUG_F_STRINGS, features)
+
+ # We don't yet support feature version detection in nested f-strings
+ root = black.lib2to3_parse(
+ """f"heard a rumour that { f'{1+1=}' } ... seems like it could be true" """
+ )
+ features = black.get_features_used(root)
+ self.assertNotIn(black.Feature.DEBUG_F_STRINGS, features)
+
@patch("black.dump_to_file", dump_to_stderr)
def test_string_quotes(self) -> None:
- source, expected = read_data("string_quotes")
+ source, expected = read_data("miscellaneous", "string_quotes")
mode = black.Mode(preview=True)
assert_format(source, expected, mode)
mode = replace(mode, string_normalization=False)
black.assert_equivalent(source, not_normalized)
black.assert_stable(source, not_normalized, mode=mode)
+ def test_skip_source_first_line(self) -> None:
+ source, _ = read_data("miscellaneous", "invalid_header")
+ tmp_file = Path(black.dump_to_file(source))
+ # Full source should fail (invalid syntax at header)
+ self.invokeBlack([str(tmp_file), "--diff", "--check"], exit_code=123)
+ # So, skipping the first line should work
+ result = BlackRunner().invoke(
+ black.main, [str(tmp_file), "-x", f"--config={EMPTY_CONFIG}"]
+ )
+ self.assertEqual(result.exit_code, 0)
+ with open(tmp_file, encoding="utf8") as f:
+ actual = f.read()
+ self.assertFormatEqual(source, actual)
+
+ def test_skip_source_first_line_when_mixing_newlines(self) -> None:
+ code_mixing_newlines = b"Header will be skipped\r\ni = [1,2,3]\nj = [1,2,3]\n"
+ expected = b"Header will be skipped\r\ni = [1, 2, 3]\nj = [1, 2, 3]\n"
+ with TemporaryDirectory() as workspace:
+ test_file = Path(workspace) / "skip_header.py"
+ test_file.write_bytes(code_mixing_newlines)
+ mode = replace(DEFAULT_MODE, skip_source_first_line=True)
+ ff(test_file, mode=mode, write_back=black.WriteBack.YES)
+ self.assertEqual(test_file.read_bytes(), expected)
+
def test_skip_magic_trailing_comma(self) -> None:
- source, _ = read_data("expression.py")
- expected, _ = read_data("expression_skip_magic_trailing_comma.diff")
+ source, _ = read_data("simple_cases", "expression")
+ expected, _ = read_data(
+ "miscellaneous", "expression_skip_magic_trailing_comma.diff"
+ )
tmp_file = Path(black.dump_to_file(source))
diff_header = re.compile(
rf"{re.escape(str(tmp_file))}\t\d\d\d\d-\d\d-\d\d "
@patch("black.dump_to_file", dump_to_stderr)
def test_async_as_identifier(self) -> None:
- source_path = (THIS_DIR / "data" / "async_as_identifier.py").resolve()
- source, expected = read_data("async_as_identifier")
+ source_path = get_case_path("miscellaneous", "async_as_identifier")
+ source, expected = read_data_from_file(source_path)
actual = fs(source)
self.assertFormatEqual(expected, actual)
major, minor = sys.version_info[:2]
@patch("black.dump_to_file", dump_to_stderr)
def test_python37(self) -> None:
- source_path = (THIS_DIR / "data" / "python37.py").resolve()
- source, expected = read_data("python37")
+ source_path = get_case_path("py_37", "python37")
+ source, expected = read_data_from_file(source_path)
actual = fs(source)
self.assertFormatEqual(expected, actual)
major, minor = sys.version_info[:2]
report.check = True
self.assertEqual(
unstyle(str(report)),
- "2 files would be reformatted, 3 files would be left unchanged, 2 files"
- " would fail to reformat.",
+ "2 files would be reformatted, 3 files would be left unchanged, 2"
+ " files would fail to reformat.",
)
report.check = False
report.diff = True
self.assertEqual(
unstyle(str(report)),
- "2 files would be reformatted, 3 files would be left unchanged, 2 files"
- " would fail to reformat.",
+ "2 files would be reformatted, 3 files would be left unchanged, 2"
+ " files would fail to reformat.",
)
def test_report_quiet(self) -> None:
report.check = True
self.assertEqual(
unstyle(str(report)),
- "2 files would be reformatted, 3 files would be left unchanged, 2 files"
- " would fail to reformat.",
+ "2 files would be reformatted, 3 files would be left unchanged, 2"
+ " files would fail to reformat.",
)
report.check = False
report.diff = True
self.assertEqual(
unstyle(str(report)),
- "2 files would be reformatted, 3 files would be left unchanged, 2 files"
- " would fail to reformat.",
+ "2 files would be reformatted, 3 files would be left unchanged, 2"
+ " files would fail to reformat.",
)
def test_report_normal(self) -> None:
report.check = True
self.assertEqual(
unstyle(str(report)),
- "2 files would be reformatted, 3 files would be left unchanged, 2 files"
- " would fail to reformat.",
+ "2 files would be reformatted, 3 files would be left unchanged, 2"
+ " files would fail to reformat.",
)
report.check = False
report.diff = True
self.assertEqual(
unstyle(str(report)),
- "2 files would be reformatted, 3 files would be left unchanged, 2 files"
- " would fail to reformat.",
+ "2 files would be reformatted, 3 files would be left unchanged, 2"
+ " files would fail to reformat.",
)
def test_lib2to3_parse(self) -> None:
# since this makes some test cases of test_get_features_used()
# fails if it fails, this is tested first so that a useful case
# is identified
- simples, relaxed = read_data("decorators")
+ simples, relaxed = read_data("miscellaneous", "decorators")
# skip explanation comments at the top of the file
for simple_test in simples.split("##")[1:]:
node = black.lib2to3_parse(simple_test)
self.assertEqual(black.get_features_used(node), {Feature.NUMERIC_UNDERSCORES})
node = black.lib2to3_parse("123456\n")
self.assertEqual(black.get_features_used(node), set())
- source, expected = read_data("function")
+ source, expected = read_data("simple_cases", "function")
node = black.lib2to3_parse(source)
expected_features = {
Feature.TRAILING_COMMA_IN_CALL,
self.assertEqual(black.get_features_used(node), expected_features)
node = black.lib2to3_parse(expected)
self.assertEqual(black.get_features_used(node), expected_features)
- source, expected = read_data("expression")
+ source, expected = read_data("simple_cases", "expression")
node = black.lib2to3_parse(source)
self.assertEqual(black.get_features_used(node), set())
node = black.lib2to3_parse(expected)
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})
+ node = black.lib2to3_parse("a[*b]")
+ self.assertEqual(black.get_features_used(node), {Feature.VARIADIC_GENERICS})
+ node = black.lib2to3_parse("a[x, *y(), z] = t")
+ self.assertEqual(black.get_features_used(node), {Feature.VARIADIC_GENERICS})
+ node = black.lib2to3_parse("def fn(*args: *T): pass")
+ self.assertEqual(black.get_features_used(node), {Feature.VARIADIC_GENERICS})
def test_get_features_used_for_future_flags(self) -> None:
for src, features in [
@pytest.mark.incompatible_with_mypyc
def test_debug_visitor(self) -> None:
- source, _ = read_data("debug_visitor.py")
- expected, _ = read_data("debug_visitor.out")
+ source, _ = read_data("miscellaneous", "debug_visitor")
+ expected, _ = read_data("miscellaneous", "debug_visitor.out")
out_lines = []
err_lines = []
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_check_diff_use_together(self) -> None:
with cache_dir():
# Files which will be reformatted.
- src1 = (THIS_DIR / "data" / "string_quotes.py").resolve()
+ src1 = get_case_path("miscellaneous", "string_quotes")
self.invokeBlack([str(src1), "--diff", "--check"], exit_code=1)
# Files which will not be reformatted.
- src2 = (THIS_DIR / "data" / "composition.py").resolve()
+ src2 = get_case_path("simple_cases", "composition")
self.invokeBlack([str(src2), "--diff", "--check"])
# Multi file command.
self.invokeBlack([str(src1), str(src2), "--diff", "--check"], exit_code=1)
def test_single_file_force_pyi(self) -> None:
pyi_mode = replace(DEFAULT_MODE, is_pyi=True)
- contents, expected = read_data("force_pyi")
+ contents, expected = read_data("miscellaneous", "force_pyi")
with cache_dir() as workspace:
path = (workspace / "file.py").resolve()
with open(path, "w") as fh:
def test_multi_file_force_pyi(self) -> None:
reg_mode = DEFAULT_MODE
pyi_mode = replace(DEFAULT_MODE, is_pyi=True)
- contents, expected = read_data("force_pyi")
+ contents, expected = read_data("miscellaneous", "force_pyi")
with cache_dir() as workspace:
paths = [
(workspace / "file1.py").resolve(),
self.assertNotIn(str(path), normal_cache)
def test_pipe_force_pyi(self) -> None:
- source, expected = read_data("force_pyi")
+ source, expected = read_data("miscellaneous", "force_pyi")
result = CliRunner().invoke(
black.main, ["-", "-q", "--pyi"], input=BytesIO(source.encode("utf8"))
)
def test_single_file_force_py36(self) -> None:
reg_mode = DEFAULT_MODE
py36_mode = replace(DEFAULT_MODE, target_versions=PY36_VERSIONS)
- source, expected = read_data("force_py36")
+ source, expected = read_data("miscellaneous", "force_py36")
with cache_dir() as workspace:
path = (workspace / "file.py").resolve()
with open(path, "w") as fh:
def test_multi_file_force_py36(self) -> None:
reg_mode = DEFAULT_MODE
py36_mode = replace(DEFAULT_MODE, target_versions=PY36_VERSIONS)
- source, expected = read_data("force_py36")
+ source, expected = read_data("miscellaneous", "force_py36")
with cache_dir() as workspace:
paths = [
(workspace / "file1.py").resolve(),
self.assertNotIn(str(path), normal_cache)
def test_pipe_force_py36(self) -> None:
- source, expected = read_data("force_py36")
+ source, expected = read_data("miscellaneous", "force_py36")
result = CliRunner().invoke(
black.main,
["-", "-q", "--target-version=py36"],
report = MagicMock()
# Even with an existing file, since we are forcing stdin, black
# should output to stdout and not modify the file inplace
- p = Path(str(THIS_DIR / "data/collections.py"))
+ p = THIS_DIR / "data" / "simple_cases" / "collections.py"
# Make sure is_file actually returns True
self.assertTrue(p.is_file())
path = Path(f"__BLACK_STDIN_FILENAME__{p}")
if nl == "\n":
self.assertNotIn(b"\r\n", output)
+ def test_normalize_line_endings(self) -> None:
+ with TemporaryDirectory() as workspace:
+ test_file = Path(workspace) / "test.py"
+ for data, expected in (
+ (b"c\r\nc\n ", b"c\r\nc\r\n"),
+ (b"l\nl\r\n ", b"l\nl\n"),
+ ):
+ test_file.write_bytes(data)
+ ff(test_file, write_back=black.WriteBack.YES)
+ self.assertEqual(test_file.read_bytes(), expected)
+
def test_assert_equivalent_different_asts(self) -> None:
with self.assertRaises(AssertionError):
black.assert_equivalent("{}", "None")
(src_dir.resolve(), "pyproject.toml"),
)
+ with change_directory(test_dir):
+ self.assertEqual(
+ black.find_project_root(("-",), stdin_filename="../src/a.py"),
+ (src_dir.resolve(), "pyproject.toml"),
+ )
+
@patch(
"black.files.find_user_pyproject_toml",
)
black.assert_stable(source, output, mode=DEFAULT_MODE)
def test_bpo_2142_workaround(self) -> None:
-
# https://bugs.python.org/issue2142
- source, _ = read_data("missing_final_newline.py")
+ source, _ = read_data("miscellaneous", "missing_final_newline")
# read_data adds a trailing newline
source = source.rstrip()
- expected, _ = read_data("missing_final_newline.diff")
+ expected, _ = read_data("miscellaneous", "missing_final_newline.diff")
tmp_file = Path(black.dump_to_file(source, ensure_final_newline=False))
diff_header = re.compile(
rf"{re.escape(str(tmp_file))}\t\d\d\d\d-\d\d-\d\d "
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:
src = (workspace / f"test{tag}.py").resolve()
with src.open("w") as fobj:
fobj.write("print('hello')")
- with patch("black.Manager", wraps=multiprocessing.Manager) as mgr:
+ with patch(
+ "black.concurrency.Manager", wraps=multiprocessing.Manager
+ ) as mgr:
cmd = ["--diff", str(workspace)]
if color:
cmd.append("--color")
str(cached): black.get_cache_info(cached),
str(cached_but_changed): (0.0, 0),
}
- todo, done = black.filter_cached(
+ todo, done = black.cache.filter_cached(
cache, {uncached, cached, cached_but_changed}
)
assert todo == {uncached, cached_but_changed}
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:
)
assert sorted(expected) == sorted(sources)
+ def test_nested_gitignore_directly_in_source_directory(self) -> None:
+ # https://github.com/psf/black/issues/2598
+ path = Path(DATA_DIR / "nested_gitignore_tests")
+ src = Path(path / "root" / "child")
+ expected = [src / "a.py", src / "c.py"]
+ assert_collected_sources([src], expected)
+
def test_invalid_gitignore(self) -> None:
path = THIS_DIR / "data" / "invalid_gitignore_tests"
empty_config = path / "pyproject.toml"
projects / etc / vim.git / blobdiff