Organize tests under line count limit

Author: cobycloud

pkgs/tigrcorn-certification/src/tigrcorn_certification/interop_runner/adapters.py

@@ -4,6 +4,7 @@
 from .models import *
 from .process import *
 from .helpers import *
+from .ports import _wait_for_server_ready
 
 class BasePeerAdapter:
     def inspect_version(self, spec: InteropProcessSpec, *, env: Mapping[str, str], cwd: Path | None) -> dict[str, Any]:

pkgs/tigrcorn-certification/src/tigrcorn_certification/interop_runner/environment.py

@@ -2,7 +2,10 @@
 
 from .imports import *
 from .models import *
+from .adapters import BasePeerAdapter, _ADAPTERS
 from .helpers import *
+from .helpers import _apply_template, _probe_command, _sha256_bytes, _sha256_path
+from tigrcorn_core.version import __version__
 
 def detect_source_revision(source_root: str | Path) -> str:
     env_commit = os.environ.get('TIGRCORN_COMMIT_HASH') or os.environ.get('GIT_COMMIT')
@@ -122,6 +125,14 @@ def _instantiate_adapter(name: str) -> BasePeerAdapter:
         return _ADAPTERS[name]()
     except KeyError as exc:
         raise InteropRunnerError(f'unknown interop adapter: {name}') from exc
+
+
+def _resolve_process_command(command: list[str]) -> list[str]:
+    if command and command[0] == '/opt/pyvenv/bin/python':
+        return [os.environ.get('TIGRCORN_INTEROP_PYTHON', os.sys.executable), *command[1:]]
+    return command
+
+
 def _materialize_process_spec(spec: InteropProcessSpec, context: Mapping[str, str]) -> InteropProcessSpec:
     return InteropProcessSpec(
         name=_apply_template(spec.name, context),

pkgs/tigrcorn-certification/src/tigrcorn_certification/interop_runner/process.py

@@ -10,6 +10,21 @@ def __init__(self, process: subprocess.Popen[Any], stdout_path: Path, stderr_pat
         self.stderr_path = stderr_path
 
     def stop(self, *, timeout: float = 5.0) -> int | None:
+        if os.name == 'nt' and self.process.poll() is None:
+            try:
+                subprocess.run(
+                    ['taskkill', '/PID', str(self.process.pid), '/T', '/F'],
+                    capture_output=True,
+                    text=True,
+                    timeout=timeout,
+                )
+            except Exception:
+                pass
+            try:
+                self.process.wait(timeout=timeout)
+            except subprocess.TimeoutExpired:
+                return None
+            return self.process.returncode
         if self.process.poll() is None:
             try:
                 self.process.terminate()

pkgs/tigrcorn-certification/src/tigrcorn_certification/interop_runner/proxies.py

@@ -2,6 +2,7 @@
 
 from .imports import *
 from .helpers import *
+from .ports import _normalize_sockaddr
 
 class _PacketTraceWriter:
     def __init__(self, path: Path) -> None:

pkgs/tigrcorn-certification/src/tigrcorn_certification/interop_runner/scenario.py

@@ -2,6 +2,7 @@
 from .imports import *
 from .models import *
 from .adapters import *
+from .assertions import *
 from .proxies import *
 from .environment import *
 from .matrix import *