conductor#

class CanaryHPCConductor(*, backend: str)#

Bases: object

register(pluginmanager: CanaryPluginManager) None#
run(args: Namespace) int#
property slots_per_resource_type: Counter[str]#
canary_resource_pool_count(type: str) int#
canary_resource_pool_count_per_node(type: str) int#
canary_resource_pool_types() list[str]#
canary_resource_pool_accommodates(case: TestCase) Outcome#
backend_accommodates(case: TestCase) Outcome#

determine if the resources for this test are available

canary_runtests(runner: Runner) bool#

Run each test case in cases.

Args: cases: test cases to run

Returns: The session returncode (0 for success)

static setup_parser(parser: Parser | LegacyParserAdapter | _ArgumentGroup) None#

Exists to accomodate canary hpc run and canary run -b ...

static setup_legacy_parser(parser: Parser) None#
class LegacyParserAdapter(parser: Parser)#

Bases: object

add_argument(flag: str, *args, **kwargs)#
parse_args(args: Sequence[str] | None = None) Namespace#
exception KeyboardQuit#

Bases: Exception

class BatchExecutor#

Bases: object

Class for running ResourceQueue.