fix(lua): add wait to socket connect

README.md

@@ -3,7 +3,7 @@
 [Neotest](https://github.com/rcarriga/neotest) adapter for python.
 Supports Pytest and unittest test files.
 
-Requires the treesitter parser for python.
+Requires [nvim-treesitter](https://github.com/nvim-treesitter/nvim-treesitter) and the parser for python.
 
 ```lua
 require("neotest").setup({

lua/neotest-python/base.lua

@@ -1,5 +1,6 @@
 local nio = require("nio")
 local lib = require("neotest.lib")
+local Path = require("plenary.path")
 
 local M = {}
 
@@ -7,7 +8,7 @@ function M.is_test_file(file_path)
   if not vim.endswith(file_path, ".py") then
     return false
   end
-  local elems = vim.split(file_path, lib.files.sep)
+  local elems = vim.split(file_path, Path.path.sep)
   local file_name = elems[#elems]
   return vim.startswith(file_name, "test_") or vim.endswith(file_name, "_test.py")
 end
@@ -34,14 +35,14 @@ function M.get_python_command(root)
   end
   -- Use activated virtualenv.
   if vim.env.VIRTUAL_ENV then
-    python_command_mem[root] = { vim.fs.joinpath(vim.env.VIRTUAL_ENV, venv_bin, "python") }
+    python_command_mem[root] = { Path:new(vim.env.VIRTUAL_ENV, venv_bin, "python").filename }
     return python_command_mem[root]
   end
 
   for _, pattern in ipairs({ "*", ".*" }) do
-    local match = nio.fn.glob(vim.fs.joinpath(root or nio.fn.getcwd(), pattern, "pyvenv.cfg"))
+    local match = nio.fn.glob(Path:new(root or nio.fn.getcwd(), pattern, "pyvenv.cfg").filename)
     if match ~= "" then
-      python_command_mem[root] = { vim.fs.joinpath(vim.fs.dirname(match), venv_bin, "python") }
+      python_command_mem[root] = { (Path:new(match):parent() / venv_bin / "python").filename }
       return python_command_mem[root]
     end
   end
@@ -51,7 +52,7 @@ function M.get_python_command(root)
     if success and exit_code == 0 then
       local venv = data.stdout:gsub("\r?\n", "")
       if venv then
-        python_command_mem[root] = { venv }
+        python_command_mem[root] = { Path:new(venv).filename }
         return python_command_mem[root]
       end
     end
@@ -66,7 +67,7 @@ function M.get_python_command(root)
     if success and exit_code == 0 then
       local venv = data.stdout:gsub("\r?\n", "")
       if venv then
-        python_command_mem[root] = { vim.fs.joinpath(venv, venv_bin, "python") }
+        python_command_mem[root] = { Path:new(venv, venv_bin, "python").filename }
         return python_command_mem[root]
       end
     end
@@ -79,7 +80,7 @@ function M.get_python_command(root)
       { stdout = true }
     )
     if success and exit_code == 0 then
-      python_command_mem[root] = { data }
+      python_command_mem[root] = { Path:new(data).filename }
       return python_command_mem[root]
     end
   end
@@ -109,20 +110,6 @@ end
 ---@return string
 local function scan_test_function_pattern(runner, config, python_command)
   local test_function_pattern = "^test"
-  if runner == "pytest" and config.pytest_discovery then
-    local cmd = vim
-      .iter({ python_command, M.get_script_path(), "--pytest-extract-test-name-template" })
-      :flatten()
-      :totable()
-    local _, data = lib.process.run(cmd, { stdout = true, stderr = true })
-
-    for line in vim.gsplit(data.stdout, "\n", true) do
-      if string.sub(line, 1, 1) == "{" and string.find(line, "python_functions") ~= nil then
-        local pytest_option = vim.json.decode(line)
-        test_function_pattern = pytest_option.python_functions
-      end
-    end
-  end
   return test_function_pattern
 end
 
@@ -198,7 +185,7 @@ function M.get_runner(python_path)
   then
     return vim_test_runner
   end
-  local runner = M.module_exists("pytest", python_path) and "pytest"
+  local runner = M.module_exists("pytest_", python_path) and "pytest"
     or M.module_exists("django", python_path) and "django"
     or "unittest"
   stored_runners[command_str] = runner

lua/neotest-python/pytest.lua

@@ -139,7 +139,7 @@ end
 ---@param positions neotest.Tree
 ---@param root string
 local function discover_params(python, script, path, positions, root)
-  local cmd = vim.iter({ python, script, "--pytest-collect", path }):flatten():totable()
+  local cmd = vim.tbl_flatten({ python, script, "--pytest-collect", path })
   logger.debug("Running test instance discovery:", cmd)
 
   local test_params = {}

neotest.py

@@ -17,4 +17,4 @@ with add_to_path():
     from neotest_python import main
 
 if __name__ == "__main__":
-    sys.exit(main(sys.argv[1:]))
+    main(sys.argv[1:])

neotest_python/__init__.py

@@ -50,18 +50,20 @@ parser.add_argument(
 parser.add_argument("args", nargs="*")
 
 
-def main(argv: List[str]) -> int:
+def main(argv: List[str]):
     if "--pytest-collect" in argv:
         argv.remove("--pytest-collect")
         from .pytest_ import collect
 
-        return collect(argv)
+        collect(argv)
+        return
 
     if "--pytest-extract-test-name-template" in argv:
         argv.remove("--pytest-extract-test-name-template")
         from .pytest_ import extract_test_name_template
 
-        return extract_test_name_template(argv)
+        extract_test_name_template(argv)
+        return
 
     args = parser.parse_args(argv)
     adapter = get_adapter(TestRunner(args.runner), args.emit_parameterized_ids)
@@ -72,9 +74,7 @@ def main(argv: List[str]) -> int:
             stream_file.write(json.dumps({"id": pos_id, "result": result}) + "\n")
             stream_file.flush()
 
-        results, exit_code = adapter.run(args.args, stream)
+        results = adapter.run(args.args, stream)
 
     with open(args.results_file, "w") as results_file:
         json.dump(results, results_file)
-
-    return exit_code

neotest_python/base.py

@@ -1,6 +1,6 @@
 import abc
 from enum import Enum
-from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Callable, Dict, List, Optional
 
 
 class NeotestResultStatus(str, Enum):
@@ -43,6 +43,6 @@ class NeotestAdapter(abc.ABC):
         }
 
     @abc.abstractmethod
-    def run(self, args: List[str], stream: Callable) -> Tuple[Dict, int]:
+    def run(self, args: List[str], stream: Callable):
         del args, stream
         raise NotImplementedError

neotest_python/django_unittest.py

@@ -1,7 +1,9 @@
 import inspect
 import os
+import subprocess
 import sys
 import traceback
+import unittest
 from argparse import ArgumentParser
 from pathlib import Path
 from types import TracebackType
@@ -31,31 +33,17 @@ class CaseUtilsMixin:
 
 
 class DjangoNeotestAdapter(CaseUtilsMixin, NeotestAdapter):
-    def get_django_root(self, path: str) -> Path:
-        """
-        Traverse the file system to locate the nearest manage.py parent
-        from the location of a given path.
-
-        This is the location of the django project
-        """
-        test_file_path = Path(path).resolve()
-        for parent in [test_file_path] + list(test_file_path.parents):
-            if (parent / "manage.py").exists():
-                return parent
-        raise FileNotFoundError("manage.py not found")
-
     def convert_args(self, case_id: str, args: List[str]) -> List[str]:
         """Converts a neotest ID into test specifier for unittest"""
         path, *child_ids = case_id.split("::")
         if not child_ids:
             child_ids = []
-        django_root = self.get_django_root(path)
+        relative_file = os.path.relpath(path, os.getcwd())
-        relative_file = os.path.relpath(path, django_root)
         relative_stem = os.path.splitext(relative_file)[0]
         relative_dotted = relative_stem.replace(os.sep, ".")
         return [*args, ".".join([relative_dotted, *child_ids])]
 
-    def run(self, args: List[str], _) -> Tuple[Dict, int]:
+    def run(self, args: List[str], _) -> Dict:
         errs: Dict[str, Tuple[Exception, Any, TracebackType]] = {}
         results = {}
 
@@ -129,16 +117,10 @@ class DjangoNeotestAdapter(CaseUtilsMixin, NeotestAdapter):
                     + len(suite_results.unexpectedSuccesses)
                 )
 
-        # Add the location of the django project to system path
+        # Make sure we can import relative to current path
-        # to ensure we have the same import paths as if the tests were ran
+        sys.path.insert(0, os.getcwd())
-        # by manage.py
-        case_id = args[-1]
-        path, *_ = case_id.split("::")
-        manage_py_location = self.get_django_root(path)
-        sys.path.insert(0, str(manage_py_location))
-
         # Prepend an executable name which is just used in output
-        argv = ["neotest-python"] + self.convert_args(case_id, args[:-1])
+        argv = ["neotest-python"] + self.convert_args(args[-1], args[:-1])
         # parse args
         parser = ArgumentParser()
         DjangoUnittestRunner.add_arguments(parser)
@@ -146,6 +128,5 @@ class DjangoNeotestAdapter(CaseUtilsMixin, NeotestAdapter):
         runner = DjangoUnittestRunner(
             **vars(parser.parse_args(argv[1:-1]))  # parse plugin config args
         )
-        failures = runner.run_tests(test_labels=[argv[-1]])  # pass test label
+        runner.run_tests(test_labels=[argv[-1]])  # pass test label
-        exit_code = 0 if failures == 0 else 1
+        return results
-        return results, exit_code

neotest_python/params_getter.py

@@ -1,5 +1,3 @@
-import logging
-from logging.handlers import WatchedFileHandler
 import asyncio
 import atexit
 from collections import deque
@@ -19,17 +17,9 @@ import argparse
 import inspect
 from pathlib import Path
 from typing import Any, Callable, Generator, Self, cast
+
 SOCKET_ROOT_DIR = Path("/tmp/neotest-python")
 
-python_socket_path = (
-    SOCKET_ROOT_DIR / hashlib.sha1(sys.executable.encode()).digest().hex()
-)
-python_socket_path.parent.mkdir(exist_ok=True,parents=True)
-
-logger = logging.getLogger('neotest-python')
-logger.addHandler(WatchedFileHandler(filename=python_socket_path.with_suffix('.log')))
-
-logger.setLevel(logging.DEBUG)
 
 class LineReceiver:
     def __init__(self, s: socket.socket) -> None:
@@ -66,7 +56,6 @@ class LineReceiver:
 
 
 def get_tests(paths: Iterable[str]) -> Generator[str, None, None]:
-    logger.info("loading tests from %s", paths)
     root = Path(os.curdir).absolute()
     if Path(os.curdir).absolute().as_posix() not in sys.path:
         sys.path.insert(0, Path(os.curdir).absolute().as_posix())
@@ -79,17 +68,16 @@ def get_tests(paths: Iterable[str]) -> Generator[str, None, None]:
         mod = importlib.util.module_from_spec(spec)
         spec.loader.exec_module(mod)
 
-        tests: Iterable[tuple[str, Callable[..., Any]]] = list(inspect.getmembers(
+        tests: Iterable[tuple[str, Callable[..., Any]]] = inspect.getmembers(
             mod,
             predicate=lambda member: inspect.isfunction(member)
             and member.__name__.startswith("test_"),
-        ))
+        )
         for _, test in tests:
             if not (marks := getattr(test, "pytestmark", None)):
                 yield (f"{path.relative_to(root).as_posix()}::{test.__name__}")
                 continue
 
-            logger.debug("path: %s - found parametrized test function %s", path, test.__name__)
             for mark in cast(Iterable[pytest.Mark], marks):
                 if mark.name == "parametrize":
                     ids = mark.kwargs.get("ids")
@@ -117,16 +105,6 @@ def get_tests(paths: Iterable[str]) -> Generator[str, None, None]:
                             test.__name__,  # pyright: ignore[reportArgumentType]
                         )
 
-                    try:
-                        id_maker = IdMaker(
-                            argnames,
-                            parametersets,
-                            idfn,
-                            ids_,
-                            None,
-                            nodeid=None,
-                        )
-                    except Exception:
                     id_maker = IdMaker(
                         argnames,
                         parametersets,
@@ -149,10 +127,12 @@ def _close_socket(path: Path) -> None:
 
 
 async def serve_socket():
-    global python_socket_path
     if not SOCKET_ROOT_DIR.exists():
         SOCKET_ROOT_DIR.mkdir()
 
+    python_socket_path = (
+        SOCKET_ROOT_DIR / hashlib.sha1(sys.executable.encode()).digest().hex()
+    )
 
     if python_socket_path.exists():
         print(python_socket_path)
@@ -165,14 +145,9 @@ async def serve_socket():
 
     async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
         path = await reader.readline()
-        try:
         tests = "\n".join(
-                [test for test in get_tests([path.decode().strip()]) if test]
+            [test for test in get_tests([cast(bytes, path).decode().strip()])]  # pyright: ignore[reportUnnecessaryCast]
         )
-            for test in tests.split('\n'):
-                logger.debug("found test: %s", test)
-        except Exception:
-            logger.exception("Failed to get tests for path: %s", path)
         writer.write(f"{tests}\n".encode())
         writer.close()
 
@@ -215,7 +190,6 @@ if __name__ == "__main__":
     )
     args = parser.parse_args()
     paths: list[str] = args.paths
-    try:
     main(
         paths=paths,
         quiet=args.quiet,
@@ -224,5 +198,3 @@ if __name__ == "__main__":
         socket_mode=args.socket_mode,
         no_fork=args.no_fork,
     )
-    except BaseException:
-        logger.exception("failed to run neotest-python pytest parser")

neotest_python/pytest_.py

@@ -1,20 +1,16 @@
+from io import StringIO
 import json
-import re
+from pathlib import Path
 from typing import Callable, Dict, List, Optional, Union
 from . import params_getter
-from io import StringIO
-from pathlib import Path
-from typing import Callable, Dict, Generator, List, Optional, Tuple, Union
 
 import pytest
 from _pytest._code.code import ExceptionRepr
-from _pytest.fixtures import FixtureLookupErrorRepr
 from _pytest.terminal import TerminalReporter
+from _pytest.fixtures import FixtureLookupErrorRepr
 
 from .base import NeotestAdapter, NeotestError, NeotestResult, NeotestResultStatus
 
-ANSI_ESCAPE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
-
 
 class PytestNeotestAdapter(NeotestAdapter):
     def __init__(self, emit_parameterized_ids: bool):
@@ -24,18 +20,18 @@ class PytestNeotestAdapter(NeotestAdapter):
         self,
         args: List[str],
         stream: Callable[[str, NeotestResult], None],
-    ) -> Tuple[Dict[str, NeotestResult], int]:
+    ) -> Dict[str, NeotestResult]:
         result_collector = NeotestResultCollector(
             self, stream=stream, emit_parameterized_ids=self.emit_parameterized_ids
         )
-        exit_code = pytest.main(
+        pytest.main(
             args=args,
             plugins=[
                 result_collector,
                 NeotestDebugpyPlugin(),
             ],
         )
-        return result_collector.results, int(exit_code)
+        return result_collector.results
 
 
 class NeotestResultCollector:
@@ -74,22 +70,10 @@ class NeotestResultCollector:
         buffer.seek(0)
         return buffer.read()
 
-    def pytest_configure(self, config: "pytest.Config"):
-        self.pytest_config = config
-
-    def _get_abs_path(self, file_path: Union[str, Path]):
-        try:
-            # rootpath is now the preferred way to access root
-            abs_path = str(self.pytest_config.rootpath / file_path)
-        except AttributeError:
-            # fallback to rootdir for older pytest versions
-            abs_path = str(Path(self.pytest_config.rootdir, file_path))
-        return abs_path
-
     def pytest_deselected(self, items: List["pytest.Item"]):
         for report in items:
             file_path, *name_path = report.nodeid.split("::")
-            abs_path = self._get_abs_path(file_path)
+            abs_path = str(Path(self.pytest_config.rootdir, file_path))
             *namespaces, test_name = name_path
             valid_test_name, *params = test_name.split("[")  # ]
             pos_id = "::".join([abs_path, *(namespaces), valid_test_name])
@@ -103,61 +87,44 @@ class NeotestResultCollector:
             )
             if not params:
                 self.stream(pos_id, result)
-
             self.results[pos_id] = result
 
+    def pytest_cmdline_main(self, config: "pytest.Config"):
+        self.pytest_config = config
+
     @pytest.hookimpl(hookwrapper=True)
     def pytest_runtest_makereport(
         self, item: "pytest.Item", call: "pytest.CallInfo"
-    ) -> Generator:
+    ) -> None:
+        # pytest generates the report.outcome field in its internal
+        # pytest_runtest_makereport implementation, so call it first.  (We don't
+        # implement pytest_runtest_logreport because it doesn't have access to
+        # call.excinfo.)
         outcome = yield
-        report: pytest.TestReport = outcome.get_result()
+        report = outcome.get_result()
 
-        if report.when not in {"call", "setup"} or report.outcome != "failed":
-            return
-
-        exc_repr = report.longrepr
-        if not isinstance(exc_repr, ExceptionRepr):
-            return
-
-        file_path, *_ = item.nodeid.split("::")
-        abs_path = str(Path(self.pytest_config.rootdir, file_path))
-        report.error_line = next(
-            (
-                traceback_entry.lineno
-                for traceback_entry in reversed(call.excinfo.traceback)
-                if str(traceback_entry.path) == abs_path
-            ),
-            None,
-        )
-
-    def pytest_runtest_logreport(self, report: "pytest.TestReport") -> None:
         if not (
             report.when == "call"
             or (report.when == "setup" and report.outcome in ("skipped", "failed"))
         ):
             return
 
-        file_path, *name_path = report.nodeid.split("::")
+        file_path, *name_path = item.nodeid.split("::")
-        abs_path = self._get_abs_path(file_path)
+        abs_path = str(Path(self.pytest_config.rootdir, file_path))
         *namespaces, test_name = name_path
         valid_test_name, *params = test_name.split("[")  # ]
-
         pos_id = "::".join([abs_path, *namespaces, valid_test_name])
 
         errors: List[NeotestError] = []
         short = self._get_short_output(self.pytest_config, report)
 
         msg_prefix = ""
-        param_id = None
+        if getattr(item, "callspec", None) is not None:
-        if "[" in test_name and test_name.endswith("]"):
+            # Parametrized test
-            param_id = test_name[len(valid_test_name) + 1 : -1]
-        if param_id:
             if self.emit_parameterized_ids:
-                pos_id += f"[{param_id}]"
+                pos_id += f"[{item.callspec.id}]"
             else:
-                msg_prefix = f"[{param_id}] "
+                msg_prefix = f"[{item.callspec.id}] "
-
         if report.outcome == "failed":
             exc_repr = report.longrepr
             # Test fails due to condition outside of test e.g. xfail
@@ -165,27 +132,27 @@ class NeotestResultCollector:
                 errors.append({"message": msg_prefix + exc_repr, "line": None})
             # Test failed internally
             elif isinstance(exc_repr, ExceptionRepr):
-                # Try to use reprcrash, but ensure the line is 0-based
+                error_message = exc_repr.reprcrash.message  # type: ignore
-                error_message = ANSI_ESCAPE.sub("", exc_repr.reprcrash.message)  # type: ignore
+                error_line = None
-                # error_line = report.error_line
+                for traceback_entry in reversed(call.excinfo.traceback):
-                error_line = getattr(report, "error_line", None)
+                    if str(traceback_entry.path) == abs_path:
+                        error_line = traceback_entry.lineno
                 errors.append(
                     {"message": msg_prefix + error_message, "line": error_line}
                 )
             elif isinstance(exc_repr, FixtureLookupErrorRepr):
-                line0 = getattr(exc_repr, "firstlineno", None)
-                if isinstance(line0, int):
-                    line0 = max(0, line0 - 1)  # 0-based
                 errors.append(
                     {
                         "message": msg_prefix + exc_repr.errorstring,
-                        "line": line0,
+                        "line": exc_repr.firstlineno,
                     }
                 )
             else:
-                # Preserve compatibility with previous behavior
+                # TODO: Figure out how these are returned and how to represent
-                errors.append({"message": msg_prefix + str(exc_repr), "line": None})
+                raise Exception(
-
+                    f"Unhandled error type ({type(exc_repr)}), please report to"
+                    " neotest-python repo"
+                )
         result: NeotestResult = self.adapter.update_result(
             self.results.get(pos_id),
             {
@@ -230,6 +197,7 @@ class NeotestDebugpyPlugin:
             # Do nothing if not running with a DAP debugger,
             # e.g. neotest was invoked with {strategy = dap}
             return
+
         thread = threading.current_thread()
         additional_info = py_db.set_additional_thread_info(thread)
         additional_info.is_tracing += 1

neotest_python/unittest.py

@@ -45,7 +45,7 @@ class UnittestNeotestAdapter(NeotestAdapter):
         return [*args, ".".join([relative_dotted, *child_ids])]
 
     # TODO: Stream results
-    def run(self, args: List[str], _) -> Tuple[Dict, int]:
+    def run(self, args: List[str], _) -> Dict:
         results = {}
 
         errs: Dict[str, Tuple[Exception, Any, TracebackType]] = {}
@@ -96,11 +96,11 @@ class UnittestNeotestAdapter(NeotestAdapter):
 
         # Prepend an executable name which is just used in output
         argv = ["neotest-python"] + self.convert_args(args[-1], args[:-1])
-        program = unittest.main(
+        unittest.main(
             module=None,
             argv=argv,
             testRunner=NeotestUnittestRunner(resultclass=NeotestTextTestResult),
             exit=False,
         )
-        exit_code = 0 if program.result.wasSuccessful() else 1
+
-        return results, exit_code
+        return results