(idx: int)
| 1223 | errors: list[BaseException] = [] |
| 1224 | |
| 1225 | def writer(idx: int) -> None: |
| 1226 | try: |
| 1227 | barrier.wait() |
| 1228 | r._write_to_disk( |
| 1229 | { |
| 1230 | "access_token": f"a-{idx}", |
| 1231 | "refresh_token": f"r-{idx}", |
| 1232 | "expires_at": 1_700_000_000 + idx, |
| 1233 | "issuer": DEFAULT_ISSUER, |
| 1234 | "client_id": "openshell-cli", |
| 1235 | } |
| 1236 | ) |
| 1237 | except BaseException as e: |
| 1238 | errors.append(e) |
| 1239 | |
| 1240 | threads = [threading.Thread(target=writer, args=(i,)) for i in range(N)] |
| 1241 | for t in threads: |
nothing calls this directly
no test coverage detected