(self)
| 46 | self.events = Queue() # type: ignore[var-annotated] |
| 47 | |
| 48 | def setup(self) -> None: |
| 49 | self.pytester.chdir() |
| 50 | # import os ; os.environ['EXECNET_DEBUG'] = "2" |
| 51 | self.gateway = execnet.makegateway("execmodel=main_thread_only//popen") |
| 52 | self.config = config = self.pytester.parseconfigure() |
| 53 | putevent = self.events.put if self.use_callback else None |
| 54 | |
| 55 | class DummyMananger: |
| 56 | testrunuid = uuid.uuid4().hex |
| 57 | specs = [0, 1] |
| 58 | |
| 59 | nodemanager = cast(NodeManager, DummyMananger) |
| 60 | |
| 61 | self.slp = WorkerController( |
| 62 | nodemanager=nodemanager, |
| 63 | gateway=self.gateway, |
| 64 | config=config, |
| 65 | putevent=putevent, # type: ignore[arg-type] |
| 66 | ) |
| 67 | self.request.addfinalizer(self.slp.ensure_teardown) |
| 68 | self.slp.setup() |
| 69 | |
| 70 | def popevent(self, name: str | None = None) -> EventCall: |
| 71 | while 1: |
no test coverage detected