scheduler.py File Reference

Go to the source code of this file.

Classes
class	oc_chat.testing.scheduler.SchedulerUnavailable
class	oc_chat.testing.scheduler.WorkerPlan
class	oc_chat.testing.scheduler.ScheduleUnit
class	oc_chat.testing.scheduler.ChildRequest
class	oc_chat.testing.scheduler.UnitOutcome

Namespaces
namespace	oc_chat
namespace	oc_chat.testing
namespace	oc_chat.testing.scheduler

Functions
Any	oc_chat.testing.scheduler.load_parallel_module (Path|None workspace_root=None)
Path	oc_chat.testing.scheduler.manifest_path_for (Path root, Sequence[str] testpaths, Path|None workspace_root=None)
tuple[dict[str, str], tuple[str,...]]	oc_chat.testing.scheduler.load_groups (Path manifest_path, Iterable[str] relpaths, Path|None workspace_root=None)
WorkerPlan	oc_chat.testing.scheduler.resolve_worker_plan (int|str requested, Path workspace_root, str package, *, int|None cpu_count=None)
dict	oc_chat.testing.scheduler.clamp_error (WorkerPlan plan)
tuple[ScheduleUnit,...]	oc_chat.testing.scheduler.build_schedule (Mapping[str, str] groups, Mapping[str, Sequence[str]] nodeids_by_file, *, Iterable[str] whole_file=(), int|None max_free_unit_tests=None)
None	oc_chat.testing.scheduler.child_entry (ChildRequest request, str result_path)
Any	oc_chat.testing.scheduler._spawn (ScheduleUnit unit, *, Path root, Sequence[str] pytest_args, int|None timeout, Path result_path)
UnitOutcome	oc_chat.testing.scheduler._read_result (ScheduleUnit unit, Any process, Path result_path, float started, float finished)
tuple[list[UnitOutcome], bool]	oc_chat.testing.scheduler.run_units (Sequence[ScheduleUnit] units, *, int workers, Path root, Sequence[str] pytest_args, int|None timeout, float origin)
dict	oc_chat.testing.scheduler.crash_violation (UnitOutcome outcome)
dict	oc_chat.testing.scheduler.crash_error (UnitOutcome outcome)
dict	oc_chat.testing.scheduler.unclassified_violation (Path root, Sequence[str] missing)

Variables
int	oc_chat.testing.scheduler.DEFAULT_WORKER_CAP = 24
str	oc_chat.testing.scheduler.GROUP_UNCLASSIFIED = "unclassified-serial"
str	oc_chat.testing.scheduler.GROUP_FREE = "free"
str	oc_chat.testing.scheduler.WHOLE_FILE_MARKER = "whole_file"
int	oc_chat.testing.scheduler.MAX_FREE_UNIT_TESTS = 100
str	oc_chat.testing.scheduler.DIST_MODE_PARALLEL = "oct_det_loadscope"
str	oc_chat.testing.scheduler.DIST_MODE_SERIAL = "serial"
str	oc_chat.testing.scheduler.CODE_WORKERS_CLAMPED = "9007"
float	oc_chat.testing.scheduler._POLL_SECONDS = 0.02