Buckets:
| import os | |
| import pytest | |
| from filelock import FileLock | |
| from utils import * | |
| def configure_worker_port(request): | |
| worker_id = getattr(request.config, "workerinput", {}).get("workerid", "master") | |
| worker_num = 0 if worker_id == "master" else int(worker_id[2:]) | |
| port_base = int(os.environ.get("LLAMA_SERVER_TEST_PORT_BASE", "8080")) | |
| os.environ["PORT"] = str(port_base + worker_num * 10) | |
| # ref: https://stackoverflow.com/questions/22627659/run-code-before-and-after-each-test-in-py-test | |
| def stop_server_after_each_test(): | |
| # do nothing before each test | |
| yield | |
| # stop all servers after each test | |
| instances = set( | |
| server_instances | |
| ) # copy the set to prevent 'Set changed size during iteration' | |
| for server in instances: | |
| server.stop() | |
| def load_server_presets(configure_worker_port, tmp_path_factory): | |
| # this will be run once per test session, before any tests | |
| # serialize model downloads across parallel workers. | |
| root_tmp_dir = tmp_path_factory.getbasetemp().parent | |
| with FileLock(str(root_tmp_dir / "load_all.lock")): | |
| ServerPreset.load_all() | |
Xet Storage Details
- Size:
- 1.24 kB
- Xet hash:
- 2dc4581e5fea640ffeb14be375f320cb788443e04230a2482581fedf6f02ea16
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.