workspace#

Workspace management for Canary test execution.

This module defines the Workspace and Session classes which handle the lifecycle of test execution environments. A Workspace acts as the central repository for test specifications, results databases, and “views” (consolidated results). Sessions represent a specific execution run of a set of jobs.

Key functionalities include:

Creating and loading persistent workspaces on disk.
Managing a “View” of the latest test results via symlinks, hardlinks, or copies.
Coordinating the execution of Job objects within a Session.
Interfacing with the WorkspaceDatabase to store and retrieve job specifications.
Providing selection mechanisms to filter tests by tags, regex, or owners.

Example

>>> ws = Workspace.create(path=".")
>>> specs = ws.collect(scanpaths={"/src/tests": []})
>>> session = ws.run(specs=specs)

class ViewSettings(name: str = 'TestResults', when: Literal['always', 'on_success', 'on_failure', 'never'] = 'always', only: Literal['all', 'failed', 'not_pass', 'passed'] = 'all', mode: Literal['symlink', 'hardlink', 'copy'] = 'symlink')#

Bases: object

name: str = 'TestResults'#

when: Literal['always', 'on_success', 'on_failure', 'never'] = 'always'#

only: Literal['all', 'failed', 'not_pass', 'passed'] = 'all'#

mode: Literal['symlink', 'hardlink', 'copy'] = 'symlink'#

classmethod default() → ViewSettings#

include_job(job: Job) → bool#

is_enabled_for_status(status: int) → bool#

is_disabled() → bool#

class ResultsView(root: pathlib.Path, settings: _canary.workspace.ViewSettings)#

Bases: object

root: Path#

settings: ViewSettings#

property dir: Path#

exists() → bool#

make(exist_ok: bool = False) → None#

unlink(missing_ok: bool = False) → None#

update(jobs: list[Job]) → bool#

maybe_add(job: Job) → bool#

add(job: Job) → None#

class Session(name: str, jobs: list[_canary.job.Job], prefix: pathlib.Path)#

Bases: object

name: str#

jobs: list[Job]#

prefix: Path#

returncode: int = -1#

started_on: datetime = datetime.datetime(1, 1, 1, 0, 0)#

finished_on: datetime = datetime.datetime(1, 1, 1, 0, 0)#

run(workspace: Workspace) → None#

Executes the session’s jobs using a Runner.

Parameters:: workspace – The Workspace instance providing the environment.
Raises:: StopExecution – If no runnable jobs are found.

class Workspace(anchor: str | Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/canary-wm/checkouts/latest/docs/source'))#

Bases: object

version_info = (1, 0, 0)#

root: Path#

refs_dir: Path#

sessions_dir: Path#

cache_dir: Path#

tmp_dir: Path#

logs_dir: Path#

db: WorkspaceDatabase#

canary_level: int#

initialize_properties(*, anchor: Path) → None#

Sets up the internal directory structure paths based on the anchor.

Parameters:: anchor – The base directory where the .canary folder resides.

static remove(start: str | Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/canary-wm/checkouts/latest/docs/source')) → Path | None#

Deletes a workspace and its associated view.

Parameters:: start – Path to start searching for the workspace.
Returns:: The path to the removed workspace, or None if no workspace was found.

rmf() → None#: Dangerously removes the workspace and view directories from disk.

latest_view() → ResultsView | None#

register_view(view: ResultsView) → None#

static find_anchor(start: str | Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/canary-wm/checkouts/latest/docs/source')) → Path | None#

Searches upwards from start to find the directory containing the workspace.

Parameters:: start – The directory to start the search from.
Returns:: The anchor Path if found, otherwise None.

static find_workspace(start: str | Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/canary-wm/checkouts/latest/docs/source')) → Path | None#

Locates the .canary workspace directory.

Parameters:: start – The directory to start the search from.
Returns:: The path to the workspace directory if found, otherwise None.

classmethod create(path: str | Path = PosixPath('/home/docs/checkouts/readthedocs.org/user_builds/canary-wm/checkouts/latest/docs/source'), force: bool = False) → Workspace#

Creates a new Canary workspace at the specified path.

Parameters:

path – The anchor directory for the workspace.
force – If True, remove existing workspace at path before creating.

Returns:

The newly created Workspace instance.

classmethod load(start: str | Path | None = None) → Workspace#

Loads an existing workspace from the filesystem.

Parameters:: start – The directory to start searching for the workspace.
Returns:: A loaded Workspace instance.
Raises:: NotAWorkspaceError – If no workspace is found.

run(specs: list[JobSpec], session: str | None = None, inplace: bool = False, view_t: ViewSettings | None = None, only: str = 'not_pass') → Session#

Executes a set of job specifications in a new or existing session.

Parameters:

specs – List of job specs to run.
session – Optional existing session name to reuse.
inplace – If True, run jobs in their existing result directories.
view_t – View settings.
only – Rerun strategy (e.g., ‘not_pass’).

Returns:

The resulting Session object.

add_session_results(session: Session, view_t: ViewSettings | None = None) → None#: Update latest results, view, and refs with results from session

rebuild_view(view_t: ViewSettings | None = None) → None#: Keep only the latest results.

relative_to_view(path: str | PathLike[str]) → str | None#

If path is inside TestResults, return the relative path (which may include glob characters). Otherwise return None.

Examples

/ws/TestResults/foo/bar/test.py -> foo/bar/test.py

is_session_dir(path: str | PathLike[str]) → bool#

Checks if a path is located within the workspace’s sessions directory.

Parameters:: path – The path to check.
Returns:: True if the path is inside the sessions directory, False otherwise.

info() → dict[str, Any]#

Returns summary information about the workspace.

Returns:: A dictionary containing root, session count, latest session, and version.

collect(scanpaths: dict[str, list[str]], on_options: list[str] | None = None) → list[JobSpec]#

Find test job generators in scan_paths and add them to this workspace.

Parameters:

scanpaths – Dictionary of root paths to scan.
on_options – Options used to filter tests by option.

Returns:

A list of resolved JobSpecs.

store_specs(specs: list[JobSpec]) → None#

Caches the provided job specifications into the workspace database.

Parameters:: specs – The resolved job specifications to store.

Selects job specifications from the database using filters and saves as a tag.

Parameters:

tag – The name of the selection tag to create.
prefixes – Filter by path prefixes.
keyword_exprs – Filter by keywords.
parameter_expr – Filter by parameter expressions.
owners – Filter by owners.
regex – Filter by regular expression.

Returns:

The list of selected JobSpecs.

Filters a list of JobSpecs using the provided rules.

Parameters:

resolved – The list of specs to filter.
prefixes – Filter by path prefixes.
keyword_exprs – Filter by keywords.
parameter_expr – Filter by parameter expressions.
owners – Filter by owners.
regex – Filter by regular expression.

Returns:

The filtered list of JobSpecs.

Collects generators from paths and creates a tagged selection.

Parameters:

tag – Tag name (randomly generated if None).
scanpaths – Paths to scan for generators.
on_options – Options to filter tests by.
keyword_exprs – Filter by keywords.
parameter_expr – Filter by parameters.
owners – Filter by owners.
regex – Filter by regex.

Returns:

The created selection of JobSpecs.

Filters the provided specs in-place using selection rules.

Parameters:

specs – The list of specs to filter.
keyword_exprs – Filter by keywords.
parameter_expr – Filter by parameters.
owners – Filter by owners.
regex – Filter by regex.
ids – Filter by specific IDs.

load_jobs(ids: list[str] | None = None) → list[Job]#

Loads jobs from the database, reconstructing them with their latest results.

Parameters:: ids – Optional list of specific job IDs to load.
Returns:: A list of Job objects in static dependency order.

select_from_view(path: Path) → list[JobSpec]#

Identifies JobSpecs based on ‘testcase.lock’ files found in the view.

Parameters:: path – The directory path to scan for lock files.
Returns:: A list of corresponding JobSpecs.

remove_tag(tag: str) → bool#

Deletes a selection tag from the database.

Parameters:: tag – The tag name to remove.
Returns:: True if the tag was removed, False if it didn’t exist.

is_tag(tag: str) → bool#

Checks if a given string is a valid selection tag.

Parameters:: tag – The string to check.
Returns:: True if it is a selection tag, False otherwise.

generate_jobspecs(generators: list[AbstractTestGenerator], on_options: list[str] | None = None) → list[JobSpec]#

Generate resolved test specs.

Parameters:

generators – List of test generators.
on_options – Used to filter tests by option.

Returns:

A list of resolved JobSpecs.

construct_jobs(specs: list[JobSpec], session: Path) → list[Job]#

Creates Job objects from JobSpecs, attempting to link latest results.

Parameters:

specs – The specifications to turn into jobs.
session – The directory for the current session.

Returns:

A list of constructed Job objects.

get_selection(tag: str | None) → list[JobSpec]#

Retrieves a list of JobSpecs associated with a tag.

Parameters:: tag – The tag name, or None/:all: for all specs.
Returns:: A list of JobSpecs.

gc(dryrun: bool = False) → None#

Garbage collects old result directories, keeping only the latest per job.

Parameters:: dryrun – If True, only log what would be removed without actually deleting.

find(*, job: str | None = None, spec: str | None = None) → Any#

Locates a Job or JobSpec in the workspace.

Parameters:

job – Identifier to find a Job.
spec – Identifier to find a JobSpec.

Returns:

The found Job or JobSpec.

find_job(root: str) → Job#

Locates a Job by ID or matching pattern.

Parameters:: root – The ID or pattern to match.
Returns:: The matching Job object.
Raises:: ValueError – If no matching job is found.

find_jobspec(root: str) → JobSpec#

Locates a JobSpec by ID or matching pattern.

Parameters:: root – The ID or pattern to match.
Returns:: The matching JobSpec object.
Raises:: ValueError – If no matching spec is found.

find_specids(ids: list[str]) → list[str | None]#

Resolves a list of potentially partial IDs or names to full spec IDs.

Parameters:: ids – List of strings to resolve.
Returns:: A list of resolved spec IDs, or None if not found.

testcase_done_callback(event: EventTypes, *args: Any) → None#

Callback to queue job results for database persistence.

Parameters:

event – The event type.
*args – Event arguments, expected to contain the finished job.

exception WorkspaceExistsError#

Bases: Exception

Raised when attempting to create a workspace in a directory that already exists.

exception NotAWorkspaceError#

Bases: Exception

Raised when a directory is not recognized as a Canary workspace.

exception SpecNotFoundError#

Bases: Exception

Raised when a requested JobSpec cannot be located in the workspace.