fix(lua): add wait to socket connect

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_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,26 +33,12 @@ 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])]
@@ -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)

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,16 +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()
-)
-
-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:
@@ -65,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())
@@ -78,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")
@@ -116,25 +105,15 @@ def get_tests(paths: Iterable[str]) -> Generator[str, None, None]:
                             test.__name__,  # pyright: ignore[reportArgumentType]
                         )
 
-                    try:
+                    id_maker = IdMaker(
-                        id_maker = IdMaker(
+                        argnames,
-                            argnames,
+                        parametersets,
-                            parametersets,
+                        idfn,
-                            idfn,
+                        ids_,
-                            ids_,
+                        None,
-                            None,
+                        nodeid=None,
-                            nodeid=None,
+                        func_name=test.__name__,
-                        )
+                    )
-                    except Exception:
-                        id_maker = IdMaker(
-                            argnames,
-                            parametersets,
-                            idfn,
-                            ids_,
-                            None,
-                            nodeid=None,
-                            func_name=test.__name__,
-                        )
                     yield from (
                         f"{path.relative_to(root).as_posix()}::{test.__name__}[{id_}]"
                         for id_ in id_maker.make_unique_parameterset_ids()
@@ -148,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)
@@ -164,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(
-            tests = "\n".join(
+            [test for test in get_tests([cast(bytes, path).decode().strip()])]  # pyright: ignore[reportUnnecessaryCast]
-                [test for test in get_tests([path.decode().strip()]) if test]
+        )
-            )
-            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()
 
@@ -214,14 +190,11 @@ if __name__ == "__main__":
     )
     args = parser.parse_args()
     paths: list[str] = args.paths
-    try:
+    main(
-        main(
+        paths=paths,
-            paths=paths,
+        quiet=args.quiet,
-            quiet=args.quiet,
+        collect_only=args.collect_only,
-            collect_only=args.collect_only,
+        verbosity=args.verbosity,
-            verbosity=args.verbosity,
+        socket_mode=args.socket_mode,
-            socket_mode=args.socket_mode,
+        no_fork=args.no_fork,
-            no_fork=args.no_fork,
+    )
-        )
-    except BaseException:
-        logger.exception("failed to run neotest-python pytest parser")

neotest_python/pytest_.py

@@ -1,7 +1,6 @@
 from io import StringIO
 import json
 from pathlib import Path
-import re
 from typing import Callable, Dict, List, Optional, Union
 from . import params_getter
 
@@ -12,7 +11,6 @@ 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):
@@ -134,7 +132,7 @@ class NeotestResultCollector:
                 errors.append({"message": msg_prefix + exc_repr, "line": None})
             # Test failed internally
             elif isinstance(exc_repr, ExceptionRepr):
-                error_message = ANSI_ESCAPE.sub('', exc_repr.reprcrash.message)  # type: ignore
+                error_message = exc_repr.reprcrash.message  # type: ignore
                 error_line = None
                 for traceback_entry in reversed(call.excinfo.traceback):
                     if str(traceback_entry.path) == abs_path: