Get current worker number and total worker count from pytest-xdist environment variables. Returns: WorkerInfo: Worker information or None values if not running in parallel
()
| 112 | |
| 113 | |
| 114 | def get_worker_info() -> WorkerInfo: |
| 115 | """ |
| 116 | Get current worker number and total worker count from pytest-xdist environment variables. |
| 117 | |
| 118 | Returns: |
| 119 | WorkerInfo: Worker information or None values if not running in parallel |
| 120 | """ |
| 121 | worker_number = os.getenv('PYTEST_XDIST_WORKER') |
| 122 | worker_count = os.getenv('PYTEST_XDIST_WORKER_COUNT') |
| 123 | |
| 124 | if worker_number and worker_count: |
| 125 | # Extract number from worker string like 'gw0', 'gw1', etc. |
| 126 | try: |
| 127 | worker_num = int(worker_number.replace('gw', '')) |
| 128 | total_workers = int(worker_count) |
| 129 | return WorkerInfo(worker_number=worker_num, total_workers=total_workers) |
| 130 | except (ValueError, AttributeError): |
| 131 | pass |
| 132 | |
| 133 | return WorkerInfo(worker_number=None, total_workers=None) |
| 134 | |
| 135 | |
| 136 | def get_wda_port() -> int: |
no test coverage detected