Fix path tests.

.pylintrc

@@ -2,4 +2,3 @@
 disable=
     duplicate-code,
     import-outside-toplevel
-

bundled/tool/server.py

@@ -9,7 +9,7 @@
 import pathlib
 import sys
 import traceback
-from typing import Any, Dict, List, Sequence
+from typing import Any, Dict, List, Optional, Sequence
 
 
 # **********************************************************
@@ -90,6 +90,8 @@ def _linting_helper(document: workspace.Document) -> list[lsp.Diagnostic]:
     try:
         result = _run_tool_on_document(document, use_stdin=True)
         if result and result.stdout:
+            log_to_output(f"{document.uri} :\r\n{result.stdout}")
+
             # deep copy here to prevent accidentally updating global settings.
             settings = copy.deepcopy(_get_settings_by_document(document))
             return _parse_output(result.stdout, severity=settings["severity"])
@@ -273,12 +275,16 @@ def _get_settings_by_document(document: workspace.Document | None):
 def _run_tool_on_document(
     document: workspace.Document,
     use_stdin: bool = False,
+    extra_args: Optional[Sequence[str]] = None,
 ) -> utils.RunResult | None:
     """Runs tool on the given document.
 
     if use_stdin is true then contents of the document is passed to the
     tool via stdin.
     """
+    if extra_args is None:
+        extra_args = []
+
     if str(document.uri).startswith("vscode-notebook-cell"):
         log_warning(f"Skipping notebook cells [Not Supported]: {str(document.uri)}")
         return None
@@ -311,7 +317,7 @@ def _run_tool_on_document(
         # process then run as module.
         argv = [TOOL_MODULE]
 
-    argv += TOOL_ARGS + settings["args"]
+    argv += TOOL_ARGS + settings["args"] + extra_args
 
     if use_stdin:
         argv += ["--from-stdin", document.path]
@@ -367,7 +373,6 @@ def _run_tool_on_document(
         if result.stderr:
             log_to_output(result.stderr)
 
-    log_to_output(f"{document.uri} :\r\n{result.stdout}")
     return result
 
 

src/test/python_tests/lsp_test_client/utils.py

@@ -26,12 +26,11 @@ def as_uri(path: str) -> str:
 
 
 @contextlib.contextmanager
-def python_file(contents: str, root: pathlib.Path):
+def python_file(contents: str, root: pathlib.Path, ext: str = ".py"):
     """Creates a temporary python file."""
     try:
         basename = (
-            "".join(random.choice("abcdefghijklmnopqrstuvwxyz") for _ in range(9))
-            + ".py"
+            "".join(random.choice("abcdefghijklmnopqrstuvwxyz") for _ in range(9)) + ext
         )
         fullpath = root / basename
         fullpath.write_text(contents)

src/test/python_tests/test_path_specialization.py

@@ -2,14 +2,16 @@
 Test for path and interpreter settings.
 """
 import copy
+from threading import Event
+from typing import Dict
 
 from hamcrest import assert_that, is_
 
 from .lsp_test_client import constants, defaults, session, utils
 
-FORMATTER = utils.get_server_info_defaults()
+TEST_FILE_PATH = constants.TEST_DATA / "sample1" / "sample.py"
+TEST_FILE_URI = utils.as_uri(str(TEST_FILE_PATH))
 TIMEOUT = 10  # 10 seconds
-TEST_FILE = constants.TEST_DATA / "sample1" / "sample.py"
 
 
 class CallbackObject:
@@ -22,10 +24,12 @@ def check_result(self):
         """returns Boolean result"""
         return self.result
 
-    def check_for_argv_duplication(self, argv):
+    def check_for_argv_duplication(self, argv: Dict[str, str]):
         """checks if argv duplication exists and sets result boolean"""
-        if argv["type"] == 4 and argv["message"].split().count("--from-stdin") > 1:
-            self.result = True
+        if argv["type"] == 4 and argv["message"].find("--from-stdin") >= 0:
+            parts = argv["message"].split()
+            count = len([x for x in parts if x.startswith("--from-stdin")])
+            self.result = count > 1
 
 
 def test_path():
@@ -35,43 +39,54 @@ def test_path():
     init_params["initializationOptions"]["settings"][0]["path"] = ["pylint"]
 
     argv_callback_object = CallbackObject()
-    contents = TEST_FILE.read_text()
-
-    actual = []
-    with utils.python_file(contents, TEST_FILE.parent) as file:
-        uri = utils.as_uri(str(file))
-
-        with session.LspSession() as ls_session:
-            ls_session.set_notification_callback(
-                session.WINDOW_LOG_MESSAGE,
-                argv_callback_object.check_for_argv_duplication,
-            )
-
-            ls_session.initialize(init_params)
-            ls_session.notify_did_open(
-                {
-                    "textDocument": {
-                        "uri": uri,
-                        "languageId": "python",
-                        "version": 1,
-                        "text": contents,
-                    }
+    contents = TEST_FILE_PATH.read_text()
+
+    actual = True
+    with session.LspSession() as ls_session:
+        ls_session.set_notification_callback(
+            session.WINDOW_LOG_MESSAGE,
+            argv_callback_object.check_for_argv_duplication,
+        )
+
+        done = Event()
+
+        def _handler(_params):
+            done.set()
+
+        ls_session.set_notification_callback(session.PUBLISH_DIAGNOSTICS, _handler)
+
+        ls_session.initialize(init_params)
+        ls_session.notify_did_open(
+            {
+                "textDocument": {
+                    "uri": TEST_FILE_URI,
+                    "languageId": "python",
+                    "version": 1,
+                    "text": contents,
                 }
-            )
-
-            # Call this second time to detect arg duplication.
-            ls_session.notify_did_open(
-                {
-                    "textDocument": {
-                        "uri": uri,
-                        "languageId": "python",
-                        "version": 1,
-                        "text": contents,
-                    }
+            }
+        )
+
+        # wait for some time to receive all notifications
+        done.wait(TIMEOUT)
+        done.clear()
+
+        # Call this second time to detect arg duplication.
+        ls_session.notify_did_open(
+            {
+                "textDocument": {
+                    "uri": TEST_FILE_URI,
+                    "languageId": "python",
+                    "version": 1,
+                    "text": contents,
                 }
-            )
+            }
+        )
 
-            actual = argv_callback_object.check_result()
+        # wait for some time to receive all notifications
+        done.wait(TIMEOUT)
+
+        actual = argv_callback_object.check_result()
 
     assert_that(actual, is_(False))
 
@@ -82,42 +97,53 @@ def test_interpreter():
     init_params["initializationOptions"]["settings"][0]["interpreter"] = ["python"]
 
     argv_callback_object = CallbackObject()
-    contents = TEST_FILE.read_text()
-
-    actual = []
-    with utils.python_file(contents, TEST_FILE.parent) as file:
-        uri = utils.as_uri(str(file))
-
-        with session.LspSession() as ls_session:
-            ls_session.set_notification_callback(
-                session.WINDOW_LOG_MESSAGE,
-                argv_callback_object.check_for_argv_duplication,
-            )
-
-            ls_session.initialize(init_params)
-            ls_session.notify_did_open(
-                {
-                    "textDocument": {
-                        "uri": uri,
-                        "languageId": "python",
-                        "version": 1,
-                        "text": contents,
-                    }
+    contents = TEST_FILE_PATH.read_text()
+
+    actual = True
+    with session.LspSession() as ls_session:
+        ls_session.set_notification_callback(
+            session.WINDOW_LOG_MESSAGE,
+            argv_callback_object.check_for_argv_duplication,
+        )
+
+        done = Event()
+
+        def _handler(_params):
+            done.set()
+
+        ls_session.set_notification_callback(session.PUBLISH_DIAGNOSTICS, _handler)
+
+        ls_session.initialize(init_params)
+        ls_session.notify_did_open(
+            {
+                "textDocument": {
+                    "uri": TEST_FILE_URI,
+                    "languageId": "python",
+                    "version": 1,
+                    "text": contents,
                 }
-            )
-
-            # Call this second time to detect arg duplication.
-            ls_session.notify_did_open(
-                {
-                    "textDocument": {
-                        "uri": uri,
-                        "languageId": "python",
-                        "version": 1,
-                        "text": contents,
-                    }
+            }
+        )
+
+        # wait for some time to receive all notifications
+        done.wait(TIMEOUT)
+        done.clear()
+
+        # Call this second time to detect arg duplication.
+        ls_session.notify_did_open(
+            {
+                "textDocument": {
+                    "uri": TEST_FILE_URI,
+                    "languageId": "python",
+                    "version": 1,
+                    "text": contents,
                 }
-            )
+            }
+        )
+
+        # wait for some time to receive all notifications
+        done.wait(TIMEOUT)
 
-            actual = argv_callback_object.check_result()
+        actual = argv_callback_object.check_result()
 
     assert_that(actual, is_(False))