How to write an extension in LISA

• Notifier

• Tool

• CustomScript

• Feature

  • Support an existing feature in a platform

  • Create a new feature

  • Use a feature

• Combinator

• Transformer

• Platform

• Hooks

  • Implement a hook

• Some notes

  • Extend schema

  • Which method must be implemented

LISA uses extensions to share code in test cases and makes it flexibly applicable to various situations. Before starting to extend, please make sure you understand the concepts of each extension.

The following content links to the code, which will be constructed using docstrings in the future.

Notifier

The base class is the Notifier in notifiers. All examples are in notifier.

class lisa.notifier.Notifier(runbook: TypedSchema)

_initialize(*args: Any, **kwargs: Any) → None

initialize is optional

_received_message(message: MessageBase) → None

Called by notifier, when a subscribed message happens.

_subscribed_message_type() → List[Type[MessageBase]]

Specify which message types want to be subscribed. Other types won’t be passed in.

finalize() → None

All test done. notifier should release resource, or do finalize work, like save to a file.

Even failed, this method will be called.

• console.py is the simplest example.

• html.py is a complete example.

If the notifier needs to be set up from the runbook, implement TypedSchema. Learn more from ConsoleSchema in console.py.

Note that the current implementation does not process messages in isolated threads, so if the implementation is slow, it may slow down the overall operation speed.

Tool

The base class is the Tool in executable. All examples are in tools.

class lisa.executable.Tool(node: Node, *args: Any, **kwargs: Any)

The base class, which wraps an executable, package, or scripts on a node. A tool can be installed, and execute on a node. When a tool is needed, call Tools[] to get one object. The Tools[] checks if it’s installed. If it’s not installed, then check if it can be installed, and then install or fail. After the tool instance returned, the run/Async of the tool will call execute/Async of node. So that the command passes to current node.

The must be implemented methods are marked with @abstractmethod, includes,

command: it’s the command name, like echo, ntttcp. it uses in run/Async to run it, and isInstalledInternal to check if it’s installed.

The should be implemented methods throws NotImplementedError, but not marked as abstract method, includes,

can_install: specify if a tool can be installed or not. If a tool is not builtin, it must implement this method.

_install: If a tool is not builtin, it must implement this method. This method needs to install a tool, and make sure it can be detected by isInstalledInternal.

The may be implemented methods is empty, includes,

initialize: It’s called when a tool is created, and before to call any other methods. It can be used to initialize variables or time-costing operations.

dependencies: All depended tools, they will be checked and installed before current tool installed. For example, ntttcp uses git to clone code and build. So it depends on Git tool.

See details on method descriptions.

_check_exists() → bool

Default implementation to check if a tool exists. This method is called by isInstalled, and cached result. Builtin tools can override it can return True directly to save time.

classmethod _freebsd_tool() → Type[Tool] | None

return a freebsd version tool class, if it’s needed

_initialize(*args: Any, **kwargs: Any) → None

Declare and initialize variables here, or some time costing initialization. This method is called before other methods, when initialing on a node.

_install() → bool

Execute installation process like build, install from packages. If other tools are depended, specify them in dependencies. Other tools can be used here, refer to ntttcp implementation.

classmethod _windows_tool() → Type[Tool] | None

return a windows version tool class, if it’s needed

property can_install: bool

Indicates if the tool supports installation or not. If it can return true, installInternal must be implemented.

property command: str

Return command string, which can be run in console. For example, echo. The command can be different under different conditions. For example, package management is ‘yum’ on CentOS, but ‘apt’ on Ubuntu.

command_exists(command: str) → Tuple[bool, bool]

classmethod create(node: Node, *args: Any, **kwargs: Any) → Tool

if there is a windows version tool, return the windows instance. override this method if richer creation factory is needed.

property dependencies: List[Type[Tool]]

Declare all dependencies here, it can be other tools, but prevent to be a circle dependency. The dependent tools are checked and installed firstly.

property exists: bool

Return if a tool installed. In most cases, overriding inInstalledInternal is enough. But if want to disable cached result and check tool every time, override this method. Notice, remote operations take times, that why caching is necessary.

get_tool_path(use_global: bool = False) → PurePath

compose a path, if the tool need to be installed

install() → bool

Default behavior of install a tool, including dependencies. It doesn’t need to be overridden.

property name: str

Unique name to a tool and used as path of tool. Don’t change it, or there may be unpredictable behavior.

property package_name: str

return package name, it may be different with command or different platform.

run(parameters: str = '', force_run: bool = False, shell: bool = False, sudo: bool = False, no_error_log: bool = False, no_info_log: bool = True, no_debug_log: bool = False, cwd: PurePath | None = None, update_envs: Dict[str, str] | None = None, encoding: str = '', timeout: int = 600, expected_exit_code: int | None = None, expected_exit_code_failure_message: str = '') → ExecutableResult

Run a process and wait for result.

run_async(parameters: str = '', force_run: bool = False, shell: bool = False, sudo: bool = False, no_error_log: bool = False, no_info_log: bool = True, no_debug_log: bool = False, cwd: pathlib.PurePath | None = None, update_envs: Dict[str, str] | None = None, node: 'Node' | None = None, encoding: str = '') → Process

Run a command async and return the Process. The process is used for async, or kill directly.

• cat.py is the simplest example.

• gcc.py supports installation.

• echo.py supports Windows.

• ntttcp.py shows how to specify dependencies between tools through the dependencies property.

• lsvmbus.py is a complex example, that handles different behaviors of Linux distributions and returns structured results to test cases.

In simple terms, the tool runs the command, returns the output, and parses it into a structure. When implementing tools, try to avoid returning original results to test cases, instead, parse the result and return a structured object, such as in lsvmbus.py. This code logic is preferred because it allows more coherence.

Learn more about how to use the tool from helloworld.py.

echo = node.tools[Echo]
...
result = echo.run(hello_world)
assert_that(result.stdout).is_equal_to(hello_world)

CustomScript

The CustomScript is like a lightweight tool, which is composited by one or more script files. However, please avoid using it unless there are serious performance concerns, compatible with existing test cases or other reasons, because it doesn’t leverage all advantages of LISA. For example, the script runs on nodes, the output may not be dumped into LISA log. The distro-agnostic modules of tools cannot be leveraged.

The base class is the CustomScript in executable.

class lisa.executable.CustomScript(name: str, node: Node, local_path: pathlib.Path, files: List[pathlib.PurePath], command: str | None = None, dependencies: List[Type[Tool]] | None = None)

_check_exists() → bool

Default implementation to check if a tool exists. This method is called by isInstalled, and cached result. Builtin tools can override it can return True directly to save time.

property can_install: bool

Indicates if the tool supports installation or not. If it can return true, installInternal must be implemented.

property command: str

Return command string, which can be run in console. For example, echo. The command can be different under different conditions. For example, package management is ‘yum’ on CentOS, but ‘apt’ on Ubuntu.

property dependencies: List[Type[Tool]]

Declare all dependencies here, it can be other tools, but prevent to be a circle dependency. The dependent tools are checked and installed firstly.

install() → bool

Default behavior of install a tool, including dependencies. It doesn’t need to be overridden.

property name: str

Unique name to a tool and used as path of tool. Don’t change it, or there may be unpredictable behavior.

run_async(parameters: str = '', force_run: bool = False, shell: bool = False, sudo: bool = False, no_error_log: bool = False, no_info_log: bool = True, no_debug_log: bool = False, cwd: pathlib.PurePath | None = None, update_envs: Dict[str, str] | None = None, node: 'Node' | None = None, encoding: str = '') → Process

Run a command async and return the Process. The process is used for async, or kill directly.

To use the scripts,

1. Define the scripts using CustomScriptBuilder.

   self._echo_script = CustomScriptBuilder(
       Path(__file__).parent.joinpath("scripts"), ["echo.sh"]
   )
   

2. Use it like a tool.

   script: CustomScript = node.tools[self._echo_script]
   result1 = script.run()
   

3. Learn more from withscript.py.

Feature

The base class is Feature in feature. All examples are in features and Azure’s features.py.

class lisa.feature.Feature(settings: FeatureSettings, node: Node, platform: Platform)

_initialize(*args: Any, **kwargs: Any) → None

override for initializing

classmethod can_disable() → bool

classmethod create_setting(*args: Any, **kwargs: Any) → FeatureSettings | None

It’s called in platform to check if a node support the feature or not. If it’s supported, create a setting.

enabled() → bool

classmethod get_feature_settings(feature: Type[Feature] | FeatureSettings | str) → FeatureSettings

classmethod name() → str

classmethod on_before_deployment(*args: Any, **kwargs: Any) → None

If a feature need to change something before deployment, it needs to implement this method. When this method is called, determined by the platform.

classmethod settings_type() → Type[FeatureSettings]

The following content takes SerialConsole as an example to introduce the feature.

Support an existing feature in a platform

1. Implement the feature, so that it can work normally. Learn more from the SerialConsole implementation in Azure’s features.py.

2. The platform should declare which features it supports, and where the implementations of features are.

   @classmethod
   def supported_features(cls) -> List[Type[Feature]]:
       return [features.StartStop, features.SerialConsole]
   

3. When preparing an environment, the platform should set the supported features on nodes.

   node_space.features = search_space.SetSpace[str](is_allow_set=True)
   node_space.features.update(
       [features.StartStop.name(), features.SerialConsole.name()]
   )
   

4. Learn more from Azure’s platform_.py.

Create a new feature

To create a new feature, you need to implement a base class that is called by the test cases, as to keep a common and shareable code logic. Learn more from SerialConsole in serial_console.py.

Use a feature

1. Declare in the metadata which features are required. If the environment does not support this feature, the test case will be skipped.

   requirement=simple_requirement(
       supported_features=[SerialConsole],
       ...
       )
   

2. Using features is like using tools.

   serial_console = node.features[SerialConsole]
   # if there is any panic, fail before partial pass
   serial_console.check_panic(saved_path=case_path, stage="reboot")
   

3. Learn more from provisioning.py.

Combinator

The base class is Combinator in combinator. All examples are in combinators.

class lisa.combinator.Combinator(runbook: Combinator)

Expand a couple of variables with multiple value to multiple runners. So LISA can run once to test different combinations of variables.

For example, v1: 1, 2 v2: 1, 2

With the grid combinations, there are 4 results: v1: 1, v2: 1 v1: 2, v2: 1 v1: 1, v2: 2 v1: 2, v2: 2

_initialize(*args: Any, **kwargs: Any) → None

if a combinator need long time initialization, it should be implemented here.

_next() → Dict[str, Any] | None

subclasses should implement this method to return a combination. Return None means no more.

_validate_entry(entry: Variable) → None

combinator reuse variable entry schema, but not allow the file type, and need the value to be a list.

fetch(current_variables: Dict[str, VariableEntry]) → Dict[str, VariableEntry] | None

Returns a combination each time. If there is no more, it returns None.

• grid_combinator.py supports a full matrix combination.

• batch_combinator.py supports a batch combination.

Transformer

The base class is Transformer in transformer. All examples are in transformers.

class lisa.transformer.Transformer(runbook: Transformer, runbook_builder: RunbookBuilder, *args: Any, **kwargs: Any)

_initialize(*args: Any, **kwargs: Any) → None

override for initialization logic. This mixin makes sure it’s called only once.

_internal_run() → Dict[str, Any]

The logic to transform

property _output_names: List[str]

List names of outputs, which are returned after run. It uses for pre-validation before the real run. It helps identifying variable name errors early.

run() → Dict[str, VariableEntry]

Call by the transformer flow, don’t override it in subclasses.

• to_list.py is the simplest example.

Platform

The base class is Platform in platform_.

class lisa.platform_.Platform(runbook: Platform)

_cleanup() → None

Called when the platform is being discarded. Perform any platform level cleanup work here.

_delete_environment(environment: Environment, log: Logger) → None

_deploy_environment(environment: Environment, log: Logger) → None

_get_environment_information(environment: Environment) → Dict[str, str]

_get_node_information(node: Node) → Dict[str, str]

_initialize(*args: Any, **kwargs: Any) → None

platform specified initialization

_initialize_guest_nodes(node: Node) → None

_prepare_environment(environment: Environment, log: Logger) → bool

Steps to prepare an environment.

1. check if platform can meet requirement of this environment.

2. if #1 is yes, specified platform context, so that the environment can be created in deploy phase with same spec as prepared.

3. set cost for environment priority.

return True, if environment can be deployed. False, if cannot.

cleanup() → None

delete_environment(environment: Environment) → None

deploy_environment(environment: Environment) → None

get_environment_information(environment: Environment) → Dict[str, str]

get_node_information(node: Node) → Dict[str, str]

prepare_environment(environment: Environment) → Environment

return prioritized environments.

user defined environment is higher priority than test cases, and then lower cost is prior to higher.

classmethod supported_features() → List[Type[Feature]]

Indicates which feature classes should be used to instance a feature.

For example, StartStop needs platform implementation, and LISA doesn’t know which type uses to start/stop for Azure. So Azure platform needs to return a type like azure.StartStop. The azure.StartStop use same feature string as lisa.features.StartStop. When test cases reference a feature by string, it can be instanced to azure.StartStop.

classmethod type_schema() → Type[TypedSchema]

• ready.py is the simplest example.

• platform_.py is a complete example of Azure.

If a platform needs to specify settings in runbook, it can be implemented in two places.

1. Platform schema. Learn more from AzurePlatformSchema in Azure’s platform_.py.

2. Node schema. Learn more from AzureNodeSchema in Azure’s common.py.

3. Use them in the platform code. Learn more from Azure’s platform_.py.

   azure_runbook: AzurePlatformSchema = self._runbook.get_extended_runbook(
       AzurePlatformSchema
   )
   azure_node_runbook = node_space.get_extended_runbook(
       AzureNodeSchema, type_name=AZURE
   )
   

Hooks

Hooks are imported by pluggy. The current list of hooks will expand due to new requirements. Take a look at A definitive example to quickly get started with pluggy.

Implement a hook

1. Create a hook specification namespace.

   class AzureHookSpec:
   
       @hookspec
       def azure_deploy_failed(self, error_message: str) -> None:
           ...
   

2. Define a hook and add some functions.

   class Platform(...):
   
       @hookimpl  # type: ignore
       def get_environment_information(self, environment: Environment) -> Dict[str, str]:
           ...
   

3. Add the spec to the manager and register the hook in place.

   plugin_manager.add_hookspecs(AzureHookSpec)
   plugin_manager.register(AzureHookSpecDefaultImpl())
   

4. Learn more from hooks in platform_.py.

Some notes

Extend schema

Extensions such as platforms and notifications support extended schema in runbook.

The runbook uses dataclass for definition, dataclass-json for deserialization, and marshmallow to validate the schema.

See more examples in schema.py, if you need to extend runbook schema.

Which method must be implemented

If a method in a parent class needs to be implemented in child class, it may raise a NotImplementedError inside the method body in the parent class and be annotated with @abstractmethod. Be careful with @abstractmethod to use use it only with NotImplementedError and nowhere else, because it is not support as a type in typing.

---

Back to how to write tests.