(
self, worker: WorkerSetup, unserialize_report: UnserializerReport
)
| 300 | assert "workeroutput" in ev.kwargs |
| 301 | |
| 302 | def test_steal_empty_queue( |
| 303 | self, worker: WorkerSetup, unserialize_report: UnserializerReport |
| 304 | ) -> None: |
| 305 | worker.pytester.makepyfile( |
| 306 | """ |
| 307 | def test_func(): pass |
| 308 | def test_func2(): pass |
| 309 | """ |
| 310 | ) |
| 311 | worker.setup() |
| 312 | ev = worker.popevent("collectionfinish") |
| 313 | ids = ev.kwargs["ids"] |
| 314 | assert len(ids) == 2 |
| 315 | worker.sendcommand("runtests_all") |
| 316 | |
| 317 | for when in ["setup", "call", "teardown"]: |
| 318 | ev = worker.popevent("testreport") |
| 319 | rep = unserialize_report(ev.kwargs["data"]) |
| 320 | assert rep.nodeid.endswith("::test_func") |
| 321 | assert rep.when == when |
| 322 | |
| 323 | worker.sendcommand("steal", indices=[0, 1]) |
| 324 | ev = worker.popevent("unscheduled") |
| 325 | assert ev.kwargs["indices"] == [] |
| 326 | |
| 327 | worker.sendcommand("shutdown") |
| 328 | |
| 329 | for when in ["setup", "call", "teardown"]: |
| 330 | ev = worker.popevent("testreport") |
| 331 | rep = unserialize_report(ev.kwargs["data"]) |
| 332 | assert rep.nodeid.endswith("::test_func2") |
| 333 | assert rep.when == when |
| 334 | |
| 335 | ev = worker.popevent("workerfinished") |
| 336 | assert "workeroutput" in ev.kwargs |
| 337 | |
| 338 | |
| 339 | def test_remote_env_vars(pytester: pytest.Pytester) -> None: |
nothing calls this directly
no test coverage detected