(
self,
config: pytest.Config,
specs: Sequence[execnet.XSpec | str] | None = None,
defaultchdir: str = "pyexecnetcache",
)
| 46 | DEFAULT_IGNORES = [".*", "*.pyc", "*.pyo", "*~"] |
| 47 | |
| 48 | def __init__( |
| 49 | self, |
| 50 | config: pytest.Config, |
| 51 | specs: Sequence[execnet.XSpec | str] | None = None, |
| 52 | defaultchdir: str = "pyexecnetcache", |
| 53 | ) -> None: |
| 54 | self.config = config |
| 55 | self.trace = self.config.trace.get("nodemanager") |
| 56 | self.testrunuid = self.config.getoption("testrunuid") |
| 57 | if self.testrunuid is None: |
| 58 | self.testrunuid = uuid.uuid4().hex |
| 59 | self.group = execnet.Group(execmodel="main_thread_only") |
| 60 | for proxy_spec in self._getpxspecs(): |
| 61 | # Proxy gateways do not run workers, and are meant to be passed with the `via` attribute |
| 62 | # to additional gateways. |
| 63 | # They are useful for running multiple workers on remote machines. |
| 64 | if getattr(proxy_spec, "id", None) is None: |
| 65 | raise pytest.UsageError( |
| 66 | f"Proxy gateway {proxy_spec} must include an id" |
| 67 | ) |
| 68 | self.group.makegateway(proxy_spec) |
| 69 | if specs is None: |
| 70 | specs = self._gettxspecs() |
| 71 | self.specs: list[execnet.XSpec] = [] |
| 72 | for spec in specs: |
| 73 | if not isinstance(spec, execnet.XSpec): |
| 74 | spec = execnet.XSpec(spec) |
| 75 | if getattr(spec, "execmodel", None) != "main_thread_only": |
| 76 | spec = execnet.XSpec(f"execmodel=main_thread_only//{spec}") |
| 77 | if not spec.chdir and not spec.popen: |
| 78 | spec.chdir = defaultchdir |
| 79 | self.group.allocate_id(spec) |
| 80 | self.specs.append(spec) |
| 81 | self.roots = self._getrsyncdirs() |
| 82 | self.rsyncoptions = self._getrsyncoptions() |
| 83 | self._rsynced_specs: set[tuple[Any, Any]] = set() |
| 84 | |
| 85 | def rsync_roots(self, gateway: execnet.Gateway) -> None: |
| 86 | """Rsync the set of roots to the node's gateway cwd.""" |
nothing calls this directly
no test coverage detected