Fixes for ruff issues category PTH

pythonFiles/create_conda.py

@@ -48,17 +48,17 @@ def parse_args(argv: Sequence[str]) -> argparse.Namespace:
 
 
 def file_exists(path: Union[str, pathlib.PurePath]) -> bool:
-    return os.path.exists(path)
+    return pathlib.Path.exists(path)
 
 
 def conda_env_exists(name: Union[str, pathlib.PurePath]) -> bool:
-    return os.path.exists(CWD / name)
+    return pathlib.Path.exists(CWD / name)
 
 
 def run_process(args: Sequence[str], error_message: str) -> None:
     try:
         print("Running: " + " ".join(args))
-        subprocess.run(args, cwd=os.getcwd(), check=True)
+        subprocess.run(args, cwd=CWD, check=True)
     except subprocess.CalledProcessError:
         raise VenvError(error_message)
 
@@ -89,11 +89,10 @@ def install_packages(env_path: str) -> None:
 
 
 def add_gitignore(name: str) -> None:
-    git_ignore = os.fspath(CWD / name / ".gitignore")
+    git_ignore = CWD / name / ".gitignore"
     if not file_exists(git_ignore):
         print(f"Creating: {git_ignore}")
-        with open(git_ignore, "w") as f:
-            f.write("*")
+        git_ignore.write_text("*")
 
 
 def main(argv: Optional[Sequence[str]] = None) -> None:

pythonFiles/create_microvenv.py

@@ -20,7 +20,7 @@ class MicroVenvError(Exception):
 def run_process(args: Sequence[str], error_message: str) -> None:
     try:
         print("Running: " + " ".join(args))
-        subprocess.run(args, cwd=os.getcwd(), check=True)
+        subprocess.run(args, cwd=CWD, check=True)
     except subprocess.CalledProcessError:
         raise MicroVenvError(error_message)
 

pythonFiles/create_venv.py

@@ -71,17 +71,17 @@ def is_installed(module: str) -> bool:
 
 
 def file_exists(path: Union[str, pathlib.PurePath]) -> bool:
-    return os.path.exists(path)
+    return pathlib.Path.exists(path)
 
 
 def venv_exists(name: str) -> bool:
-    return os.path.exists(CWD / name) and file_exists(get_venv_path(name))
+    return pathlib.Path.exists(CWD / name) and file_exists(get_venv_path(name))
 
 
 def run_process(args: Sequence[str], error_message: str) -> None:
     try:
         print("Running: " + " ".join(args))
-        subprocess.run(args, cwd=os.getcwd(), check=True)
+        subprocess.run(args, cwd=CWD, check=True)
     except subprocess.CalledProcessError:
         raise VenvError(error_message)
 
@@ -129,9 +129,8 @@ def upgrade_pip(venv_path: str) -> None:
 def add_gitignore(name: str) -> None:
     git_ignore = CWD / name / ".gitignore"
     if not file_exists(git_ignore):
-        print("Creating: " + os.fspath(git_ignore))
-        with open(git_ignore, "w") as f:
-            f.write("*")
+        print(f"Creating: {git_ignore}")
+        git_ignore.write_text("*")
 
 
 def download_pip_pyz(name: str):
@@ -140,11 +139,9 @@ def download_pip_pyz(name: str):
 
     try:
         with url_lib.urlopen(url) as response:
-            pip_pyz_path = os.fspath(CWD / name / "pip.pyz")
-            with open(pip_pyz_path, "wb") as out_file:
-                data = response.read()
-                out_file.write(data)
-                out_file.flush()
+            pip_pyz_path = CWD / name / "pip.pyz"
+            pip_pyz_path.write_bytes(response.read())
+
     except Exception:
         raise VenvError("CREATE_VENV.DOWNLOAD_PIP_FAILED")
 

pythonFiles/download_get_pip.py

@@ -22,7 +22,7 @@ def _get_package_data():
 
 
 def _download_and_save(root, version):
-    root = os.getcwd() if root is None or root == "." else root
+    root = pathlib.Path.cwd() if root is None or root == "." else root
     url = f"https://raw.githubusercontent.com/pypa/get-pip/{version}/public/get-pip.py"
     print(url)
     with url_lib.urlopen(url) as response:

pythonFiles/printEnvVariablesToFile.py

@@ -2,11 +2,12 @@
 # Licensed under the MIT License.
 
 import os
+import pathlib
 import sys
 
 # Last argument is the target file into which we'll write the env variables line by line.
-output_file = sys.argv[-1]
+output_file = pathlib.Path(sys.argv[-1])
 
-with open(output_file, "w") as outfile:
+with output_file.open("w") as outfile:
     for key, val in os.environ.items():
         outfile.write(f"{key}={val}\n")

pythonFiles/run-jedi-language-server.py

@@ -1,9 +1,14 @@
-import sys
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
 import os
+import sys
+from pathlib import Path
+
 
 # Add the lib path to our sys path so jedi_language_server can find its references
-EXTENSION_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-sys.path.insert(0, os.path.join(EXTENSION_ROOT, "pythonFiles", "lib", "jedilsp"))
+EXTENSION_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, os.fspath(EXTENSION_ROOT / "pythonFiles" / "lib" / "jedilsp"))
 
 
 from jedi_language_server.cli import cli

pythonFiles/shell_exec.py

@@ -3,6 +3,7 @@
 
 import subprocess
 import sys
+from pathlib import Path
 
 # This is a simple solution to waiting for completion of commands sent to terminal.
 # 1. Intercept commands send to a terminal
@@ -11,12 +12,12 @@
 # 4. Calling code monitors the contents of the file to determine state of execution.
 
 # Last argument is a file that's used for synchronizing the actions in the terminal with the calling code in extension.
-lock_file = sys.argv[-1]
+lock_file = Path(sys.argv[-1])
 shell_args = sys.argv[1:-1]
 
 print("Executing command in shell >> " + " ".join(shell_args))
 
-with open(lock_file, "w") as fp:
+with lock_file.open("w") as fp:
     try:
         # Signal start of execution.
         fp.write("START\n")
@@ -36,7 +37,8 @@
         fp.flush()
         try:
             # ALso log the error for use from the other side.
-            with open(lock_file + ".error", "w") as fpError:
+            error_path = lock_file.with_suffix(".error")
+            with error_path.open("w") as fpError:
                 fpError.write(traceback.format_exc())
         except Exception:
             pass

pythonFiles/testing_tools/run_adapter.py

@@ -2,17 +2,11 @@
 # Licensed under the MIT License.
 
 # Replace the "." entry.
-import os.path
+import os
 import sys
+from pathlib import Path
 
-sys.path.insert(
-    1,
-    os.path.dirname(  # pythonFiles
-        os.path.dirname(  # pythonFiles/testing_tools
-            os.path.abspath(__file__)  # this file
-        )
-    ),
-)
+sys.path.insert(1, os.fspath(Path(__file__).resolve().parent.parent))
 
 from testing_tools.adapter.__main__ import parse_args, main
 

pythonFiles/tests/__init__.py

@@ -1,11 +1,12 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
-import os.path
+import os
+from pathlib import Path
 
-TEST_ROOT = os.path.dirname(__file__)
-SRC_ROOT = os.path.dirname(TEST_ROOT)
-PROJECT_ROOT = os.path.dirname(SRC_ROOT)
-TESTING_TOOLS_ROOT = os.path.join(SRC_ROOT, "testing_tools")
-DEBUG_ADAPTER_ROOT = os.path.join(SRC_ROOT, "debug_adapter")
+TEST_ROOT = os.fspath(Path(__file__).resolve().parent)
+SRC_ROOT = os.fspath(Path(TEST_ROOT).parent)
+PROJECT_ROOT = os.fspath(Path(SRC_ROOT).parent)
+TESTING_TOOLS_ROOT = os.fspath(Path(SRC_ROOT) / "testing_tools")
+DEBUG_ADAPTER_ROOT = os.fspath(Path(SRC_ROOT) / "debug_adapter")
 
-PYTHONFILES = os.path.join(SRC_ROOT, "lib", "python")
+PYTHONFILES = os.fspath(Path(SRC_ROOT) / "lib" / "python")

pythonFiles/tests/pytestadapter/expected_discovery_test_output.py

@@ -1,4 +1,8 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
 import os
+from pathlib import Path
 
 
 from .helpers import TEST_DATA_PATH, find_test_line_number, get_absolute_test_id
@@ -845,10 +849,10 @@
             "children": [
                 {
                     "name": "test_a_function",
-                    "path": os.fspath(os.path.join(tests_path, "test_a.py")),
+                    "path": os.fspath(tests_path / "test_a.py"),
                     "lineno": find_test_line_number(
                         "test_a_function",
-                        os.path.join(tests_path, "test_a.py"),
+                        os.fspath(tests_path / "test_a.py"),
                     ),
                     "type_": "test",
                     "id_": get_absolute_test_id(
@@ -868,10 +872,10 @@
             "children": [
                 {
                     "name": "test_b_function",
-                    "path": os.fspath(os.path.join(tests_path, "test_b.py")),
+                    "path": os.fspath(tests_path / "test_b.py"),
                     "lineno": find_test_line_number(
                         "test_b_function",
-                        os.path.join(tests_path, "test_b.py"),
+                        os.fspath(tests_path / "test_b.py"),
                     ),
                     "type_": "test",
                     "id_": get_absolute_test_id(

pythonFiles/tests/pytestadapter/helpers.py

@@ -10,7 +10,7 @@
 import sys
 import threading
 import uuid
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 script_dir = pathlib.Path(__file__).parent.parent.parent
 sys.path.append(os.fspath(script_dir))
@@ -195,7 +195,9 @@ def _run_test_code(
     return result
 
 
-def find_test_line_number(test_name: str, test_file_path) -> str:
+def find_test_line_number(
+    test_name: str, test_file_path: Union[str, pathlib.Path]
+) -> str:
     """Function which finds the correct line number for a test by looking for the "test_marker--[test_name]" string.
 
     The test_name is split on the "[" character to remove the parameterization information.
@@ -205,9 +207,9 @@ def find_test_line_number(test_name: str, test_file_path) -> str:
     test_file_path: The path to the test file where the test is located.
     """
     test_file_unique_id: str = "test_marker--" + test_name.split("[")[0]
-    with open(test_file_path) as f:
-        for i, line in enumerate(f):
-            if test_file_unique_id in line:
-                return str(i + 1)
+    lines = pathlib.Path(test_file_path).read_text().splitlines()
+    for i, line in enumerate(lines):
+        if test_file_unique_id in line:
+            return str(i + 1)
     error_str: str = f"Test {test_name!r} not found on any line in {test_file_path}"
     raise ValueError(error_str)

pythonFiles/tests/run_all.py

@@ -2,10 +2,11 @@
 # Licensed under the MIT License.
 
 # Replace the "." entry.
-import os.path
+import os
 import sys
+from pathlib import Path
 
-sys.path[0] = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path[0] = os.fspath(Path(__file__).resolve().parent.parent)
 
 from tests.__main__ import main, parse_args
 

pythonFiles/tests/test_installed_check.py

@@ -20,13 +20,12 @@
 def generate_file(base_file: pathlib.Path):
     basename = "pyproject.toml" if "pyproject" in base_file.name else "requirements.txt"
     fullpath = base_file.parent / basename
-    if fullpath.exists():
-        os.unlink(os.fspath(fullpath))
+    fullpath.unlink(missing_ok=True)
     fullpath.write_text(base_file.read_text(encoding="utf-8"))
     try:
         yield fullpath
     finally:
-        os.unlink(str(fullpath))
+        fullpath.unlink(missing_ok=True)
 
 
 def run_on_file(

pythonFiles/unittestadapter/discovery.py

@@ -76,9 +76,9 @@ def discover_tests(
         "status": "error",
     }
     """
-    cwd = os.path.abspath(start_dir)
+    cwd = os.fspath(pathlib.Path(start_dir).resolve())
     if "/" in start_dir:  #  is a subdir
-        parent_dir = os.path.dirname(start_dir)
+        parent_dir = os.fspath(pathlib.Path(start_dir).parent)
         sys.path.insert(0, parent_dir)
     else:
         sys.path.insert(0, cwd)
@@ -95,7 +95,7 @@ def discover_tests(
             top_level_dir = start_dir
 
         # Get abspath of top level directory for build_test_tree.
-        top_level_dir = os.path.abspath(top_level_dir)
+        top_level_dir = os.fspath(pathlib.Path(top_level_dir).resolve())
 
         tests, error = build_test_tree(
             suite, top_level_dir

pythonFiles/unittestadapter/execution.py

@@ -171,7 +171,7 @@ def run_tests(
     failfast: Optional[bool],
     locals: Optional[bool] = None,
 ) -> PayloadDict:
-    cwd = os.path.abspath(start_dir)
+    cwd = os.fspath(pathlib.Path(start_dir).resolve())
     status = TestExecutionStatus.error
     error = None
     payload: PayloadDict = {"cwd": cwd, "status": status, "result": None}
@@ -180,8 +180,8 @@ def run_tests(
         # If it's a file, split path and file name.
         start_dir = cwd
         if cwd.endswith(".py"):
-            start_dir = os.path.dirname(cwd)
-            pattern = os.path.basename(cwd)
+            start_dir = pathlib.Path(cwd).parent
+            pattern = pathlib.Path(cwd).name
 
         # Discover tests at path with the file name as a pattern (if any).
         loader = unittest.TestLoader()
@@ -232,7 +232,7 @@ def run_tests(
 
 def send_run_data(raw_data, port, uuid):
     status = raw_data["outcome"]
-    cwd = os.path.abspath(START_DIR)
+    cwd = os.fspath(pathlib.Path(START_DIR).resolve())
     if raw_data["subtest"]:
         test_id = raw_data["subtest"]
     else:
@@ -349,7 +349,7 @@ def post_response(
             locals,
         )
     else:
-        cwd = os.path.abspath(start_dir)
+        cwd = os.fspath(pathlib.Path(start_dir).resolve())
         status = TestExecutionStatus.error
         payload: PayloadDict = {
             "cwd": cwd,

pythonFiles/unittestadapter/pvsc_utils.py

@@ -166,7 +166,7 @@ def build_test_tree(
             components = test_id.split(".")
             class_name = f"{components[-1]}.py"
             # Find/build class node.
-            file_path = os.fsdecode(os.path.join(directory_path, class_name))
+            file_path = os.fsdecode(directory_path / class_name)
             current_node = get_child_node(
                 class_name, file_path, TestNodeTypeEnum.file, root
             )

pythonFiles/vscode_datascience_helpers/tests/logParser.py

@@ -44,8 +44,8 @@ def printTestOutput(testlog):
 
 def splitByPid(testlog):
     # Split testlog into prefixed logs based on pid
-    baseFile = os.path.splitext(testlog[0])[0]
     p = Path(testlog[0])
+    baseFile = p.parent / p.stem
     pids = set()
     logs = {}
     pid = None
@@ -64,7 +64,7 @@ def splitByPid(testlog):
                     # See if we've created a log for this pid or not
                     if pid not in pids:
                         pids.add(pid)
-                        logFile = "{}_{}.log".format(baseFile, pid)
+                        logFile = f"{os.fspath(baseFile)}_{pid}.log"
                         print("Writing to new log: " + logFile)
                         logs[pid] = Path(logFile).open(mode="w")