(self)
| 266 | return [s for s in self.CMDS if isinstance(s, str)] |
| 267 | |
| 268 | def start_server(self): |
| 269 | process_name = multiprocessing.current_process().name |
| 270 | if SHARED_INTERPRETER_STATE.is_scan_process: |
| 271 | kwargs = dict(self.server_kwargs) |
| 272 | # if we're in tests, we use a single event loop to avoid weird race conditions |
| 273 | # this allows us to more easily mock http, etc. |
| 274 | if os.environ.get("BBOT_TESTING", "") == "True": |
| 275 | kwargs["_loop"] = get_event_loop() |
| 276 | kwargs["debug"] = self._engine_debug |
| 277 | self.process = CORE.create_process( |
| 278 | target=self.server_process, |
| 279 | args=( |
| 280 | self.SERVER_CLASS, |
| 281 | self.socket_path, |
| 282 | ), |
| 283 | kwargs=kwargs, |
| 284 | custom_name=f"BBOT {self.__class__.__name__}", |
| 285 | ) |
| 286 | self.process.start() |
| 287 | return self.process |
| 288 | else: |
| 289 | raise BBOTEngineError( |
| 290 | f"Tried to start server from process {process_name}. Did you forget \"if __name__ == '__main__'?\"" |
| 291 | ) |
| 292 | |
| 293 | @staticmethod |
| 294 | def server_process(server_class, socket_path, **kwargs): |
no test coverage detected