view#

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

Bases: object

name: str = 'TestResults'#
when: Literal['always', 'never', 'on_success', 'on_failure'] = '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(jobs: list[Job]) bool#
always_disabled() bool#
always_enabled() bool#
deferred_until_finish() bool#
class ViewManifestEntry(job_id: str, view_path: str, source: str, session: str, outcome: str, updated: str)#

Bases: object

job_id: str#
view_path: str#
source: str#
session: str#
outcome: str#
updated: str#
classmethod from_job(job: Job) ViewManifestEntry#
class ViewManifest(version: int = 1, settings: dict[str, typing.Any]=<factory>, entries: dict[str, _canary.view.ViewManifestEntry]=<factory>)#

Bases: object

version: int = 1#
settings: dict[str, Any]#
entries: dict[str, ViewManifestEntry]#
classmethod from_dict(data: dict[str, Any]) ViewManifest#
to_dict() dict[str, Any]#
class ResultsView(root: pathlib.Path, settings: _canary.view.ViewSettings)#

Bases: object

root: Path#
settings: ViewSettings#
static exists_at(p: Path) bool#
property dir: Path#
exists() bool#
make(exist_ok: bool = False) None#
update(jobs: list[Job]) bool#
add(job: Job) None#
property manifest_file: Path#
load_manifest() ViewManifest#
save_manifest(manifest: ViewManifest) None#
sync(job: Job, manifest: ViewManifest | None = None, *, save: bool = True) bool#

Synchronize this job’s latest result into the view.

This removes any previous entry for the job, then conditionally adds the current result depending on ViewSettings.include_job().

remove_entry(job_id: str, manifest: ViewManifest) bool#
remove_path(path: Path) None#
class ViewManager(workspace: Workspace, settings: ViewSettings, session: Session | None = None)#

Bases: object

Live manager for maintaining the session results view.

The view is updated incrementally as jobs finish. Updates are protected by a workspace-level file lock so multiple Canary processes can safely update the same view, e.g. re-entrant HPC batch runs.

workspace: Workspace#
settings: ViewSettings#
session: Session | None = None#
view: ResultsView | None = None#
enabled: bool = False#
started: bool = False#
finished: bool = False#
property lock_file: Path#
locked() Iterator[None]#
start() None#
finish() ResultsView | None#
sync(job: Job) None#
rebuild() ResultsView | None#

Rebuild the view from the latest results in the workspace.

The entire rebuild is protected by the same view lock used for live updates, so live syncs cannot interleave with a rebuild.