Skip to content

Commit

Permalink
ruff 0.8.0 fixes (#24488)
Browse files Browse the repository at this point in the history
  • Loading branch information
joar authored Nov 25, 2024
1 parent 2fed954 commit 98fea6e
Show file tree
Hide file tree
Showing 19 changed files with 90 additions and 88 deletions.
3 changes: 2 additions & 1 deletion python_files/create_conda.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@
import pathlib
import subprocess
import sys
from typing import Optional, Sequence, Union
from collections.abc import Sequence
from typing import Optional, Union

CONDA_ENV_NAME = ".conda"
CWD = pathlib.Path.cwd()
Expand Down
3 changes: 2 additions & 1 deletion python_files/create_microvenv.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@
import pathlib
import subprocess
import sys
from typing import Optional, Sequence
from collections.abc import Sequence
from typing import Optional

VENV_NAME = ".venv"
LIB_ROOT = pathlib.Path(__file__).parent / "lib" / "python"
Expand Down
9 changes: 5 additions & 4 deletions python_files/create_venv.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@
import subprocess
import sys
import urllib.request as url_lib
from typing import List, Optional, Sequence, Union
from collections.abc import Sequence
from typing import Optional, Union

VENV_NAME = ".venv"
CWD = pathlib.Path.cwd()
Expand Down Expand Up @@ -107,7 +108,7 @@ def get_venv_path(name: str) -> str:
return os.fspath(CWD / name / "bin" / "python")


def install_requirements(venv_path: str, requirements: List[str]) -> None:
def install_requirements(venv_path: str, requirements: list[str]) -> None:
if not requirements:
return

Expand All @@ -120,7 +121,7 @@ def install_requirements(venv_path: str, requirements: List[str]) -> None:
print("CREATE_VENV.PIP_INSTALLED_REQUIREMENTS")


def install_toml(venv_path: str, extras: List[str]) -> None:
def install_toml(venv_path: str, extras: list[str]) -> None:
args = "." if len(extras) == 0 else f".[{','.join(extras)}]"
run_process(
[venv_path, "-m", "pip", "install", "-e", args],
Expand Down Expand Up @@ -171,7 +172,7 @@ def install_pip(name: str):
)


def get_requirements_from_args(args: argparse.Namespace) -> List[str]:
def get_requirements_from_args(args: argparse.Namespace) -> list[str]:
requirements = []
if args.stdin:
data = json.loads(sys.stdin.read())
Expand Down
11 changes: 6 additions & 5 deletions python_files/installed_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@
import os
import pathlib
import sys
from typing import Dict, List, Optional, Sequence, Tuple, Union
from collections.abc import Sequence
from typing import Optional, Union

LIB_ROOT = pathlib.Path(__file__).parent / "lib" / "python"
sys.path.insert(0, os.fspath(LIB_ROOT))
Expand Down Expand Up @@ -43,7 +44,7 @@ def parse_requirements(line: str) -> Optional[Requirement]:
return None


def process_requirements(req_file: pathlib.Path) -> List[Dict[str, Union[str, int]]]:
def process_requirements(req_file: pathlib.Path) -> list[dict[str, Union[str, int]]]:
diagnostics = []
for n, line in enumerate(req_file.read_text(encoding="utf-8").splitlines()):
if line.startswith(("#", "-", " ")) or line == "":
Expand All @@ -69,15 +70,15 @@ def process_requirements(req_file: pathlib.Path) -> List[Dict[str, Union[str, in
return diagnostics


def get_pos(lines: List[str], text: str) -> Tuple[int, int, int, int]:
def get_pos(lines: list[str], text: str) -> tuple[int, int, int, int]:
for n, line in enumerate(lines):
index = line.find(text)
if index >= 0:
return n, index, n, index + len(text)
return (0, 0, 0, 0)


def process_pyproject(req_file: pathlib.Path) -> List[Dict[str, Union[str, int]]]:
def process_pyproject(req_file: pathlib.Path) -> list[dict[str, Union[str, int]]]:
diagnostics = []
try:
raw_text = req_file.read_text(encoding="utf-8")
Expand Down Expand Up @@ -109,7 +110,7 @@ def process_pyproject(req_file: pathlib.Path) -> List[Dict[str, Union[str, int]]
return diagnostics


def get_diagnostics(req_file: pathlib.Path) -> List[Dict[str, Union[str, int]]]:
def get_diagnostics(req_file: pathlib.Path) -> list[dict[str, Union[str, int]]]:
diagnostics = []
if not req_file.exists():
return diagnostics
Expand Down
2 changes: 1 addition & 1 deletion python_files/normalizeSelection.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import re
import sys
import textwrap
from typing import Iterable
from collections.abc import Iterable

attach_bracket_paste = sys.version_info >= (3, 13)

Expand Down
4 changes: 2 additions & 2 deletions python_files/python_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import sys
import traceback
import uuid
from typing import Dict, List, Optional, Union
from typing import Optional, Union

STDIN = sys.stdin
STDOUT = sys.stdout
Expand Down Expand Up @@ -38,7 +38,7 @@ def send_response(
)


def send_request(params: Optional[Union[List, Dict]] = None):
def send_request(params: Optional[Union[list, dict]] = None):
request_id = uuid.uuid4().hex
if params is None:
send_message(id=request_id, method="input")
Expand Down
3 changes: 1 addition & 2 deletions python_files/testing_tools/process_json_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,11 @@
# Licensed under the MIT License.
import io
import json
from typing import Dict, List

CONTENT_LENGTH: str = "Content-Length:"


def process_rpc_json(data: str) -> Dict[str, List[str]]:
def process_rpc_json(data: str) -> dict[str, list[str]]:
"""Process the JSON data which comes from the server."""
str_stream: io.StringIO = io.StringIO(data)

Expand Down
24 changes: 12 additions & 12 deletions python_files/tests/pytestadapter/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
import tempfile
import threading
import uuid
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Optional

if sys.platform == "win32":
from namedpipe import NPopen
Expand Down Expand Up @@ -63,7 +63,7 @@ def create_symlink(root: pathlib.Path, target_ext: str, destination_ext: str):
print("destination unlinked", destination)


def process_data_received(data: str) -> List[Dict[str, Any]]:
def process_data_received(data: str) -> list[dict[str, Any]]:
"""Process the all JSON data which comes from the server.
After listen is finished, this function will be called.
Expand All @@ -87,7 +87,7 @@ def process_data_received(data: str) -> List[Dict[str, Any]]:
return json_messages # return the list of json messages


def parse_rpc_message(data: str) -> Tuple[Dict[str, str], str]:
def parse_rpc_message(data: str) -> tuple[dict[str, str], str]:
"""Process the JSON data which comes from the server.
A single rpc payload is in the format:
Expand Down Expand Up @@ -128,7 +128,7 @@ def parse_rpc_message(data: str) -> Tuple[Dict[str, str], str]:
print("json decode error")


def _listen_on_fifo(pipe_name: str, result: List[str], completed: threading.Event):
def _listen_on_fifo(pipe_name: str, result: list[str], completed: threading.Event):
# Open the FIFO for reading
fifo_path = pathlib.Path(pipe_name)
with fifo_path.open() as fifo:
Expand All @@ -144,7 +144,7 @@ def _listen_on_fifo(pipe_name: str, result: List[str], completed: threading.Even
result.append(data)


def _listen_on_pipe_new(listener, result: List[str], completed: threading.Event):
def _listen_on_pipe_new(listener, result: list[str], completed: threading.Event):
"""Listen on the named pipe or Unix domain socket for JSON data from the server.
Created as a separate function for clarity in threading context.
Expand Down Expand Up @@ -197,24 +197,24 @@ def _listen_on_pipe_new(listener, result: List[str], completed: threading.Event)
result.append("".join(all_data))


def _run_test_code(proc_args: List[str], proc_env, proc_cwd: str, completed: threading.Event):
def _run_test_code(proc_args: list[str], proc_env, proc_cwd: str, completed: threading.Event):
result = subprocess.run(proc_args, env=proc_env, cwd=proc_cwd)
completed.set()
return result


def runner(args: List[str]) -> Optional[List[Dict[str, Any]]]:
def runner(args: list[str]) -> Optional[list[dict[str, Any]]]:
"""Run a subprocess and a named-pipe to listen for messages at the same time with threading."""
print("\n Running python test subprocess with cwd set to: ", TEST_DATA_PATH)
return runner_with_cwd(args, TEST_DATA_PATH)


def runner_with_cwd(args: List[str], path: pathlib.Path) -> Optional[List[Dict[str, Any]]]:
def runner_with_cwd(args: list[str], path: pathlib.Path) -> Optional[list[dict[str, Any]]]:
"""Run a subprocess and a named-pipe to listen for messages at the same time with threading."""
return runner_with_cwd_env(args, path, {})


def split_array_at_item(arr: List[str], item: str) -> Tuple[List[str], List[str]]:
def split_array_at_item(arr: list[str], item: str) -> tuple[list[str], list[str]]:
"""
Splits an array into two subarrays at the specified item.
Expand All @@ -235,14 +235,14 @@ def split_array_at_item(arr: List[str], item: str) -> Tuple[List[str], List[str]


def runner_with_cwd_env(
args: List[str], path: pathlib.Path, env_add: Dict[str, str]
) -> Optional[List[Dict[str, Any]]]:
args: list[str], path: pathlib.Path, env_add: dict[str, str]
) -> Optional[list[dict[str, Any]]]:
"""
Run a subprocess and a named-pipe to listen for messages at the same time with threading.
Includes environment variables to add to the test environment.
"""
process_args: List[str]
process_args: list[str]
pipe_name: str
if "MANAGE_PY_PATH" in env_add:
# If we are running Django, generate a unittest-specific pipe name.
Expand Down
20 changes: 10 additions & 10 deletions python_files/tests/pytestadapter/test_discovery.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import json
import os
import sys
from typing import Any, Dict, List, Optional
from typing import Any, Optional

import pytest

Expand All @@ -25,10 +25,10 @@ def test_import_error():
"""
file_path = helpers.TEST_DATA_PATH / "error_pytest_import.txt"
with helpers.text_to_python_file(file_path) as p:
actual: Optional[List[Dict[str, Any]]] = helpers.runner(["--collect-only", os.fspath(p)])
actual: Optional[list[dict[str, Any]]] = helpers.runner(["--collect-only", os.fspath(p)])

assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
if actual_list is not None:
for actual_item in actual_list:
assert all(item in actual_item for item in ("status", "cwd", "error"))
Expand Down Expand Up @@ -64,7 +64,7 @@ def test_syntax_error(tmp_path): # noqa: ARG001
actual = helpers.runner(["--collect-only", os.fspath(p)])

assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
if actual_list is not None:
for actual_item in actual_list:
assert all(item in actual_item for item in ("status", "cwd", "error"))
Expand All @@ -89,7 +89,7 @@ def test_parameterized_error_collect():
file_path_str = "error_parametrize_discovery.py"
actual = helpers.runner(["--collect-only", file_path_str])
assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
if actual_list is not None:
for actual_item in actual_list:
assert all(item in actual_item for item in ("status", "cwd", "error"))
Expand Down Expand Up @@ -191,7 +191,7 @@ def test_pytest_collect(file, expected_const):
)

assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
if actual_list is not None:
actual_item = actual_list.pop(0)
assert all(item in actual_item for item in ("status", "cwd", "error"))
Expand Down Expand Up @@ -227,7 +227,7 @@ def test_symlink_root_dir():
)
expected = expected_discovery_test_output.symlink_expected_discovery_output
assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
if actual_list is not None:
actual_item = actual_list.pop(0)
try:
Expand Down Expand Up @@ -260,7 +260,7 @@ def test_pytest_root_dir():
helpers.TEST_DATA_PATH / "root",
)
assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
if actual_list is not None:
actual_item = actual_list.pop(0)

Expand All @@ -287,7 +287,7 @@ def test_pytest_config_file():
helpers.TEST_DATA_PATH / "root",
)
assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
if actual_list is not None:
actual_item = actual_list.pop(0)

Expand Down Expand Up @@ -319,7 +319,7 @@ def test_config_sub_folder():
)

assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
if actual_list is not None:
actual_item = actual_list.pop(0)
assert all(item in actual_item for item in ("status", "cwd", "error"))
Expand Down
10 changes: 5 additions & 5 deletions python_files/tests/pytestadapter/test_execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import os
import pathlib
import sys
from typing import Any, Dict, List
from typing import Any

import pytest

Expand Down Expand Up @@ -33,7 +33,7 @@ def test_config_file():
actual = runner_with_cwd(args, new_cwd)
expected_const = expected_execution_test_output.config_file_pytest_expected_execution_output
assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
assert len(actual_list) == len(expected_const)
actual_result_dict = {}
if actual_list is not None:
Expand All @@ -53,7 +53,7 @@ def test_rootdir_specified():
actual = runner_with_cwd(args, new_cwd)
expected_const = expected_execution_test_output.config_file_pytest_expected_execution_output
assert actual
actual_list: List[Dict[str, Dict[str, Any]]] = actual
actual_list: list[dict[str, dict[str, Any]]] = actual
assert len(actual_list) == len(expected_const)
actual_result_dict = {}
if actual_list is not None:
Expand Down Expand Up @@ -207,7 +207,7 @@ def test_pytest_execution(test_ids, expected_const):
args = test_ids
actual = runner(args)
assert actual
actual_list: List[Dict[str, Dict[str, Any]]] = actual
actual_list: list[dict[str, dict[str, Any]]] = actual
assert len(actual_list) == len(expected_const)
actual_result_dict = {}
if actual_list is not None:
Expand Down Expand Up @@ -248,7 +248,7 @@ def test_symlink_run():

expected_const = expected_execution_test_output.symlink_run_expected_execution_output
assert actual
actual_list: List[Dict[str, Any]] = actual
actual_list: list[dict[str, Any]] = actual
if actual_list is not None:
actual_item = actual_list.pop(0)
try:
Expand Down
4 changes: 2 additions & 2 deletions python_files/tests/test_installed_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import pathlib
import subprocess
import sys
from typing import Dict, List, Optional, Union
from typing import Optional, Union

import pytest

Expand All @@ -31,7 +31,7 @@ def generate_file(base_file: pathlib.Path):

def run_on_file(
file_path: pathlib.Path, severity: Optional[str] = None
) -> List[Dict[str, Union[str, int]]]:
) -> list[dict[str, Union[str, int]]]:
env = os.environ.copy()
if severity:
env["VSCODE_MISSING_PGK_SEVERITY"] = severity
Expand Down
Loading

0 comments on commit 98fea6e

Please sign in to comment.