Fix propagating import strategy in multi-root case (microsoft#237)

bundled/tool/lsp_jsonrpc.py

@@ -6,12 +6,13 @@
 import atexit
 import io
 import json
+import os
 import pathlib
 import subprocess
 import threading
 import uuid
 from concurrent.futures import ThreadPoolExecutor
-from typing import BinaryIO, Dict, Sequence, Union
+from typing import BinaryIO, Dict, Optional, Sequence, Union
 
 CONTENT_LENGTH = "Content-Length: "
 RUNNER_SCRIPT = str(pathlib.Path(__file__).parent / "lsp_runner.py")
@@ -141,14 +142,24 @@ def stop_all_processes(self):
                 pass
         self._thread_pool.shutdown(wait=False)
 
-    def start_process(self, workspace: str, args: Sequence[str], cwd: str) -> None:
+    def start_process(
+        self,
+        workspace: str,
+        args: Sequence[str],
+        cwd: str,
+        env: Optional[Dict[str, str]],
+    ) -> None:
         """Starts a process and establishes JSON-RPC communication over stdio."""
+        new_env = os.environ.copy()
+        if env:
+            new_env.update(env)
         # pylint: disable=consider-using-with
         proc = subprocess.Popen(
             args,
             cwd=cwd,
             stdout=subprocess.PIPE,
             stdin=subprocess.PIPE,
+            env=new_env,
         )
         self._processes[workspace] = proc
         self._rpc[workspace] = create_json_rpc(proc.stdout, proc.stdin)
@@ -187,13 +198,16 @@ def _get_json_rpc(workspace: str) -> Union[JsonRpc, None]:
 
 
 def get_or_start_json_rpc(
-    workspace: str, interpreter: Sequence[str], cwd: str
+    workspace: str,
+    interpreter: Sequence[str],
+    cwd: str,
+    env: Optional[Dict[str, str]] = None,
 ) -> Union[JsonRpc, None]:
     """Gets an existing JSON-RPC connection or starts one and return it."""
     res = _get_json_rpc(workspace)
     if not res:
         args = [*interpreter, RUNNER_SCRIPT]
-        _process_manager.start_process(workspace, args, cwd)
+        _process_manager.start_process(workspace, args, cwd, env)
         res = _get_json_rpc(workspace)
     return res
 
@@ -216,10 +230,11 @@ def run_over_json_rpc(
     argv: Sequence[str],
     use_stdin: bool,
     cwd: str,
-    source: str = None,
+    source: Optional[str] = None,
+    env: Optional[Dict[str, str]] = None,
 ) -> RpcRunResult:
     """Uses JSON-RPC to execute a command."""
-    rpc: Union[JsonRpc, None] = get_or_start_json_rpc(workspace, interpreter, cwd)
+    rpc: Union[JsonRpc, None] = get_or_start_json_rpc(workspace, interpreter, cwd, env)
     if not rpc:
         raise Exception("Failed to run over JSON-RPC.")
 

bundled/tool/lsp_server.py

@@ -276,7 +276,7 @@ def _get_global_defaults():
 
 def _update_workspace_settings(settings):
     if not settings:
-        key = os.getcwd()
+        key = utils.normalize_path(os.getcwd())
         WORKSPACE_SETTINGS[key] = {
             "cwd": key,
             "workspaceFS": key,
@@ -286,7 +286,7 @@ def _update_workspace_settings(settings):
         return
 
     for setting in settings:
-        key = uris.to_fs_path(setting["workspace"])
+        key = utils.normalize_path(uris.to_fs_path(setting["workspace"]))
         WORKSPACE_SETTINGS[key] = {
             **setting,
             "workspaceFS": key,
@@ -297,7 +297,7 @@ def _get_settings_by_path(file_path: pathlib.Path):
     workspaces = {s["workspaceFS"] for s in WORKSPACE_SETTINGS.values()}
 
     while file_path != file_path.parent:
-        str_file_path = str(file_path)
+        str_file_path = utils.normalize_path(file_path)
         if str_file_path in workspaces:
             return WORKSPACE_SETTINGS[str_file_path]
         file_path = file_path.parent
@@ -313,8 +313,9 @@ def _get_document_key(document: workspace.Document):
 
         # Find workspace settings for the given file.
         while document_workspace != document_workspace.parent:
-            if str(document_workspace) in workspaces:
-                return str(document_workspace)
+            norm_path = utils.normalize_path(document_workspace)
+            if norm_path in workspaces:
+                return norm_path
             document_workspace = document_workspace.parent
 
     return None
@@ -327,7 +328,7 @@ def _get_settings_by_document(document: workspace.Document | None):
     key = _get_document_key(document)
     if key is None:
         # This is either a non-workspace file or there is no workspace.
-        key = os.fspath(pathlib.Path(document.path).parent)
+        key = utils.normalize_path(str(pathlib.Path(document.path).parent))
         return {
             "cwd": key,
             "workspaceFS": key,
@@ -415,6 +416,9 @@ def _run_tool_on_document(
             use_stdin=use_stdin,
             cwd=cwd,
             source=document.source,
+            env={
+                "LS_IMPORT_STRATEGY": settings["importStrategy"],
+            },
         )
         result = _to_run_result_with_logging(result)
     else:
@@ -485,6 +489,9 @@ def _run_tool(extra_args: Sequence[str], settings: Dict[str, Any]) -> utils.RunR
             argv=argv,
             use_stdin=True,
             cwd=cwd,
+            env={
+                "LS_IMPORT_STRATEGY": settings["importStrategy"],
+            },
         )
         result = _to_run_result_with_logging(result)
     else:

bundled/tool/lsp_utils.py

@@ -6,11 +6,12 @@
 import contextlib
 import io
 import os
-import os.path
+import pathlib
 import runpy
 import site
 import subprocess
 import sys
+import sysconfig
 import threading
 from typing import Any, Callable, List, Sequence, Tuple, Union
 
@@ -19,37 +20,42 @@
 CWD_LOCK = threading.Lock()
 
 
-def as_list(content: Union[Any, List[Any], Tuple[Any]]) -> Union[List[Any], Tuple[Any]]:
+def as_list(content: Union[Any, List[Any], Tuple[Any]]) -> List[Any]:
     """Ensures we always get a list"""
     if isinstance(content, (list, tuple)):
-        return content
+        return list(content)
     return [content]
 
 
-# pylint: disable-next=consider-using-generator
-_site_paths = tuple(
-    [
-        os.path.normcase(os.path.normpath(p))
-        for p in (as_list(site.getsitepackages()) + as_list(site.getusersitepackages()))
-    ]
+_site_paths = set(
+    str(pathlib.Path(p).resolve())
+    for p in (
+        as_list(site.getsitepackages())
+        + as_list(site.getusersitepackages())
+        + list(sysconfig.get_paths().values())
+    )
 )
 
 
-def is_same_path(file_path1, file_path2) -> bool:
+def is_same_path(file_path1: str, file_path2: str) -> bool:
     """Returns true if two paths are the same."""
-    return os.path.normcase(os.path.normpath(file_path1)) == os.path.normcase(
-        os.path.normpath(file_path2)
-    )
+    return pathlib.Path(file_path1) == pathlib.Path(file_path2)
+
+
+def normalize_path(file_path: str) -> str:
+    """Returns normalized path."""
+    return str(pathlib.Path(file_path).resolve())
 
 
 def is_current_interpreter(executable) -> bool:
     """Returns true if the executable path is same as the current interpreter."""
     return is_same_path(executable, sys.executable)
 
 
-def is_stdlib_file(file_path) -> bool:
-    """Return True if the file belongs to standard library."""
-    return os.path.normcase(os.path.normpath(file_path)).startswith(_site_paths)
+def is_stdlib_file(file_path: str) -> bool:
+    """Return True if the file belongs to the standard library."""
+    normalized_path = str(pathlib.Path(file_path).resolve())
+    return any(normalized_path.startswith(path) for path in _site_paths)
 
 
 # pylint: disable-next=too-few-public-methods

noxfile.py

@@ -139,18 +139,18 @@ def lint(session: nox.Session) -> None:
     )
     session.run("flake8", "noxfile.py")
 
+    # check import sorting using isort
+    session.install("isort")
+    session.run("isort", "--profile", "black", "--check", "./bundled/tool")
+    session.run("isort", "--profile", "black", "--check", "./src/test/python_tests")
+    session.run("isort", "--profile", "black", "--check", "noxfile.py")
+
     # check formatting using black
     session.install("black")
     session.run("black", "--check", "./bundled/tool")
     session.run("black", "--check", "./src/test/python_tests")
     session.run("black", "--check", "noxfile.py")
 
-    # check import sorting using isort
-    session.install("isort")
-    session.run("isort", "--check", "./bundled/tool")
-    session.run("isort", "--check", "./src/test/python_tests")
-    session.run("isort", "--check", "noxfile.py")
-
     # check typescript code
     session.run("npm", "run", "lint", external=True)
 

package.json

@@ -91,7 +91,7 @@
                         "%settings.importStrategy.fromEnvironment.description%",
                         "%settings.importStrategy.useBundled.description%"
                     ],
-                    "scope": "window",
+                    "scope": "resource",
                     "type": "string"
                 },
                 "black-formatter.interpreter": {

src/test/python_tests/test_formatting.py

@@ -3,6 +3,8 @@
 """
 Test for formatting over LSP.
 """
+import pathlib
+
 from hamcrest import assert_that, is_
 
 from .lsp_test_client import constants, session, utils
@@ -105,3 +107,34 @@ def test_formatting_cell():
     ]
 
     assert_that(actual, is_(expected))
+
+
+def test_skipping_site_packages_files():
+    """Test skipping formatting when the file is in site-packages"""
+
+    UNFORMATTED_TEST_FILE_PATH = constants.TEST_DATA / "sample1" / "sample.unformatted"
+    with session.LspSession() as ls_session:
+        # Use any stdlib path here
+        uri = utils.as_uri(pathlib.__file__)
+        ls_session.initialize()
+        ls_session.notify_did_open(
+            {
+                "textDocument": {
+                    "uri": uri,
+                    "languageId": "python",
+                    "version": 1,
+                    "text": UNFORMATTED_TEST_FILE_PATH.read_text(encoding="utf-8"),
+                }
+            }
+        )
+
+        actual = ls_session.text_document_formatting(
+            {
+                "textDocument": {"uri": uri},
+                # `options` is not used by black
+                "options": {"tabSize": 4, "insertSpaces": True},
+            }
+        )
+
+    expected = None
+    assert_that(actual, is_(expected))