(self, config: pytest.Config, log: Producer | None = None)
| 91 | """ |
| 92 | |
| 93 | def __init__(self, config: pytest.Config, log: Producer | None = None) -> None: |
| 94 | self.numnodes = len(parse_tx_spec_config(config)) |
| 95 | self.collection: list[str] | None = None |
| 96 | |
| 97 | self.workqueue: OrderedDict[str, dict[str, bool]] = OrderedDict() |
| 98 | self.assigned_work: dict[WorkerController, dict[str, dict[str, bool]]] = {} |
| 99 | self.registered_collections: dict[WorkerController, list[str]] = {} |
| 100 | |
| 101 | if log is None: |
| 102 | self.log = Producer("loadscopesched") |
| 103 | else: |
| 104 | self.log = log.loadscopesched |
| 105 | |
| 106 | self.config = config |
| 107 | |
| 108 | @property |
| 109 | def nodes(self) -> list[WorkerController]: |
nothing calls this directly
no test coverage detected