libkirk package

Submodules

libkirk.com module

class libkirk.com.ComChannel[source]

Bases: Plugin

Communication channel. The objects implementing this class are usually using SSH, serial, shell, etc protocols. and they are used by the scheduler in order to execute commands or turning on/off the communication.

async active() → bool[source]

Returns:: Return True if communication is active. False otherwise.
Return type:: bool

async communicate(iobuffer: IOBuffer | None = None) → None[source]

Start communication.

Parameters:: iobuffer (IOBuffer) – Buffer used to write stdout.

async ensure_communicate(iobuffer: IOBuffer | None = None, retries: int = 10) → None[source]

Ensure that communicate is completed, retrying as many times we want in case of KirkException error. After each error, the communication is stopped and a new communication is performed.

Parameters:

iobuffer (IOBuffer) – Buffer used to write stdout.
retries (int) – Number of times we retry to communicate.

async fetch_file(target_path: str) → bytes[source]

Fetch file and return its content.

Parameters:: target_path (str) – Path of the file to download from target.
Returns:: Data contained in target_path.
Return type:: bytes

property parallel_execution: bool

Returns:: If True, communication supports commands parallel execution.
Return type:: bool

async ping() → float[source]

Send a ping request and verify how much reply takes in seconds.

Returns:: Time between ping and pong.
Return type:: float

async run_command(command: str, cwd: str | None = None, env: Dict[str, str] | None = None, iobuffer: IOBuffer | None = None) → Dict[str, Any] | None[source]

Run a command.

Parameters:

command (str) – Command to execute.
cwd (str) – Current working directory.
env (dict) – Environment variables.
iobuffer (IOBuffer) – Buffer used to write stdout.

Returns:

Dictionary containing information about the executed command.

{
    "command": <str>,
    "returncode": <int>,
    "stdout": <str>,
    "exec_time": <float>,
}

If None is returned, then callback has failed.

Return type:

dict

async stop(iobuffer: IOBuffer | None = None) → None[source]

Stop communication.

Parameters:: iobuffer (IOBuffer) – Buffer used to write stdout.

class libkirk.com.IOBuffer[source]

Bases: object

IO stdout buffer. The API is similar to IO types.

async write(data: str) → None[source]

Write data inside the buffer.

Parameters:: data (str) – Data to write.

libkirk.com.clone_channel(name: str, new_name: str) → Plugin[source]

Clone a channel implementation named name and rename it with new_name. The new plugin will be registered with the other plugins.

Parameters:

name (str) – Plugin name.
new_name (str) – New cloned plugin name.

Returns:

New plugin object.

Return type:

Plugin

libkirk.com.discover(path: str, extend: bool = True) → None[source]

Discover all ComChannel implementations inside path.

Parameters:

path (str) – Directory where searching for channel implementations.
extend – If True, it will add new discovered channels on top of the ones already found. If False, previous discovered channels will be cleared.

Return type:

bool

libkirk.com.get_channels() → List[ComChannel][source]

Returns:: List of loaded ComChannel implementations.
Return type:: list(ComChannel)

libkirk.data module

class libkirk.data.Suite(name: str, tests: List[Test])[source]

Bases: object

Testing suite definition class.

property name: str

Returns:: Name of the testing suite.
Return type:: str

property tests: List[Test]

Returns:: Tests definitions.
Return type:: list(Test)

class libkirk.data.Test(name: str, cmd: str, cwd: str | None = None, env: Dict[str, str] | None = None, args: List[str] | None = None, parallelizable: bool = False)[source]

Bases: object

Test definition.

property arguments: List[str]

Returns:: Arguments of the command.
Return type:: list(str)

property command: str

Returns:: Command to execute test.
Return type:: str

property cwd: str | None

Returns:: Current working directory.
Return type:: str | None

property env: Dict[str, str]

Returns:: Environment variables
Return type:: dict

force_parallel() → None[source]

Returns:: Force test to be parallelizable.

property full_command: str

Returns:: Return the full command, with arguments as well. For example, if command=”ls” and arguments=”-l -a”, full_command=”ls -l -a”.
Return type:: str

property name: str

Returns:: Name of the test.
Return type:: str

property parallelizable: bool

Returns:: If True, test can be run in parallel.
Return type:: bool

libkirk.errors module

exception libkirk.errors.CommunicationError[source]

Bases: KirkException

Raised when error occurs during channels communication.

exception libkirk.errors.ExporterError[source]

Bases: KirkException

Raised when an error occurs during Exporter operations.

exception libkirk.errors.FrameworkError[source]

Bases: KirkException

Raised when an error occurs inside a framework.

exception libkirk.errors.KernelPanicError[source]

Bases: KirkException

Raised during kernel panic.

exception libkirk.errors.KernelTaintedError[source]

Bases: KirkException

Raised when kernel is tainted.

exception libkirk.errors.KernelTimeoutError[source]

Bases: KirkException

Raised when kernel is not responding anymore.

exception libkirk.errors.KirkException[source]

Bases: Exception

The most generic exception that is raised by any module. All kirk errors derives from this class.

exception libkirk.errors.LTXError[source]

Bases: KirkException

Raised when an error occurs during LTX execution.

exception libkirk.errors.PluginError[source]

Bases: KirkException

Raised when plugins operations have failed.

exception libkirk.errors.SUTError[source]

Bases: KirkException

Raised when error occurs in SUT.

exception libkirk.errors.SchedulerError[source]

Bases: KirkException

Raised when an error occurs during Scheduler operations.

exception libkirk.errors.SessionError[source]

Bases: KirkException

Raised when an error occurs during Session operations.

libkirk.evt module

class libkirk.evt.Event(ordered: bool = False)[source]

Bases: object

An event to process.

create_tasks(*args: Any, **kwargs: Any) → List[Any][source]

Create tasks to run according to registered coroutines.

Parameters:

args (list) – Arguments to be passed to callback functions execution.
kwargs (dict) – Keyword arguments to be passed to callback functions execution.

Returns:

List of tasks to execute.

Return type:

list(asyncio.Task)

has_coros() → bool[source]

Check if there are still available registrations.

Returns:: True if there are registered coroutines.
Return type:: bool

register(coro: Callable) → None[source]

Register a new Callable.

Parameters:: coro (Callable) – Coroutine to register.

remove(coro: Callable) → None[source]

Remove a specific Callable associated to the event.

Parameters:: coro (Callable) – Callable to remove.

class libkirk.evt.EventsHandler[source]

Bases: object

This class implements event loop and events handling.

async fire(event_name: str, *args: Any, **kwargs: Any) → None[source]

Fire a specific event.

Parameters:

event_name (str) – Name of the event.
args (Any) – Arguments to be passed to callback functions execution.
kwargs (Any) – Keyword arguments to be passed to callback functions execution.

is_registered(event_name: str) → bool[source]

Returns True if event_name is registered.

Parameters:: event_name (str) – Name of the event.
Returns:: True if registered, False otherwise.
Return type:: bool

register(event_name: str, coro: Callable, ordered: bool = False) → None[source]

Register an event with event_name.

Parameters:

event_name (str) – Name of the event.
coro (Callable) – Callable associated with event_name.
ordered (bool) – If True, the event will raise coroutines in the order they arrive.

reset() → None[source]: Reset the entire events queue.

async start() → None[source]: Start the event loop.

async stop() → None[source]: Stop the event loop.

unregister(event_name: str, coro: Callable) → None[source]

Unregister a single event Callable with event_name. If coro is None, all coroutines registered will be removed.

Parameters:

event_name (str) – Name of the event.
coro (Callable) – Callable to unregister.

libkirk.export module

class libkirk.export.Exporter[source]

Bases: object

A class used to export Results into report file.

async save_file(results: List[SuiteResults], path: str) → None[source]

Save report into a file by taking information from SUT and testing results.

Parameters:

results (list(SuiteResults)) – List of suite results to export.
path (str) – Path of the file to save.

class libkirk.export.JSONExporter[source]

Bases: Exporter

Export testing results into a JSON file.

async save_file(results: List[SuiteResults], path: str) → None[source]

Save report into a file by taking information from SUT and testing results.

Parameters:

results (list(SuiteResults)) – List of suite results to export.
path (str) – Path of the file to save.

libkirk.framework module

class libkirk.framework.Framework[source]

Bases: object

Framework definition. Implement this class if you need to support more testing frameworks inside the application.

async find_command(channel: ComChannel, command: str) → Test[source]

Search for command inside Framework folder and, if it’s not found, search for command in the SUT. Then return a Test object which can be used to execute command.

Parameters:

channel (ComChannel) – Communication channel.
command (str) – Command to execute.

Returns:

Test object that has been found.

Return type:

Test

async find_suite(channel: ComChannel, name: str) → Suite[source]

Search for suite with given name inside the SUT.

Parameters:

channel (ComChannel) – Communication channel.
suite (str) – Name of the suite.

Returns:

Suite object that has been found.

Return type:

Suite

async get_suites(channel: ComChannel) → List[str][source]

Return the list of available suites.

Parameters:: channel (ComChannel) – Communication channel.
Returns:: List of suites names.
Return type:: list(str)

async read_result(test: Test, stdout: str, retcode: int, exec_t: float) → TestResults[source]

Return test results accoding with runner output and Test definition.

Parameters:

test (Test) – Test definition object.
stdout (str) – Test stdout.
retcode (int) – Test return code.
exec_t (float) – Test execution time in seconds.

Returns:

Test results.

Return type:

TestResults

libkirk.io module

class libkirk.io.AsyncFile(filename: str, mode: str = 'r')[source]

Bases: AsyncContextManager

Handle files in asynchronous way by running operations inside a separate thread.

async close() → None[source]: Close the file.

async open() → None[source]: Open the file according to the mode.

async read(size: int = -1) → str | bytes | None[source]

Asynchronous version of read().

Parameters:: size (int) – Amount of data to read. Default is -1, that means all data available.
Returns:: Data that has been read or None if file is not open.
Return type:: str | bytes | None

async readline() → str | bytes | None[source]

Asynchronous version of readline().

Returns:: Data that has been read or None if file is not open.
Return type:: str | bytes | None

async seek(pos: int) → None[source]

Asynchronous version of seek().

Parameters:: pos (int) – Position to search.

async tell() → int | None[source]

Asynchronous version of tell().

Returns:: Current file position or None if file is not open.
Return type:: int | None

async write(data: str | bytes) → None[source]

Asynchronous version of write().

Parameters:: data (str | bytes) – Data to write inside file.

libkirk.ltp module

class libkirk.ltp.LTPFramework(max_runtime: float = 0.0, timeout: float = 30.0)[source]

Bases: Framework

Linux Test Project framework definition.

PARALLEL_BLACKLIST: frozenset = frozenset({'format_device', 'max_runtime', 'mntpoint', 'mount_device', 'needs_device', 'needs_root', 'resource_file', 'save_restore'})

SUPPORTED_ENV: frozenset = frozenset({'CONNECTION_TOTAL', 'DOWNLOAD_BIGFILESIZE', 'DOWNLOAD_REGFILESIZE', 'FTP_DOWNLOAD_DIR', 'FTP_UPLOAD_DIR', 'FTP_UPLOAD_URLDIR', 'HTTP_DOWNLOAD_DIR', 'IF_UPDOWN_TIMES', 'IPSEC_MODE', 'IPSEC_PROTO', 'IPV4_LHOST', 'IPV4_RHOST', 'IPV6_LHOST', 'IPV6_RHOST', 'IP_TOTAL', 'IP_TOTAL_FOR_TCPIP', 'KCONFIG_PATH', 'KCONFIG_SKIP_CHECK', 'LHOST_IFACES', 'MCASTNUM_HEAVY', 'MCASTNUM_NORMAL', 'MTU_CHANGE_TIMES', 'NS_DURATION', 'NS_ICMPV4_SENDER_DATA_MAXSIZE', 'NS_ICMPV6_SENDER_DATA_MAXSIZE', 'NS_TIMES', 'PATH', 'PING_MAX', 'RHOST', 'RHOST_IFACES', 'ROUTE_CHANGE_IP', 'ROUTE_CHANGE_NETLINK', 'ROUTE_TOTAL', 'UPLOAD_BIGFILESIZE', 'UPLOAD_REGFILESIZE', 'VIRT_PERF_THRESHOLD'})

async find_command(channel: ComChannel, command: str) → Test[source]

Search for command inside Framework folder and, if it’s not found, search for command in the SUT. Then return a Test object which can be used to execute command.

Parameters:

channel (ComChannel) – Communication channel.
command (str) – Command to execute.

Returns:

Test object that has been found.

Return type:

Test

async find_suite(channel: ComChannel, name: str) → Suite[source]

Search for suite with given name inside the SUT.

Parameters:

channel (ComChannel) – Communication channel.
suite (str) – Name of the suite.

Returns:

Suite object that has been found.

Return type:

Suite

async get_suites(channel: ComChannel) → List[str][source]

Return the list of available suites.

Parameters:: channel (ComChannel) – Communication channel.
Returns:: List of suites names.
Return type:: list(str)

async read_result(test: Test, stdout: str, retcode: int, exec_t: float) → TestResults[source]

Return test results accoding with runner output and Test definition.

Parameters:

test (Test) – Test definition object.
stdout (str) – Test stdout.
retcode (int) – Test return code.
exec_t (float) – Test execution time in seconds.

Returns:

Test results.

Return type:

TestResults

libkirk.main module

libkirk.main.run(cmd_args: List[str] | None = None) → None[source]

Entry point of the application.

Parameters:: cmd_args (list(str) | None) – Command line arguments.

libkirk.monitor module

class libkirk.monitor.JSONFileMonitor(path: str)[source]

Bases: object

Monitor the current executor status and it redirects events to a file using JSON format.

async kernel_panic() → None[source]

async kernel_tainted(message: str) → None[source]

async run_cmd_start(cmd: str) → None[source]

async run_cmd_stop(command: str, stdout: str, returncode: int) → None[source]

async session_error(error: str) → None[source]

async session_restore(restore: str) → None[source]

async session_started(suites: list, tmpdir: str) → None[source]

async session_stopped() → None[source]

async session_warning(msg: str) → None[source]

async start() → None[source]: Attach to events and start writing data inside the monitor file.

async stop() → None[source]: Stop monitoring events.

async suite_completed(results: SuiteResults, exec_time: float) → None[source]

async suite_started(suite: Suite) → None[source]

async suite_timeout(suite: Suite, timeout: float) → None[source]

async sut_not_responding() → None[source]

async sut_restart(sut: str) → None[source]

async sut_start(sut: str) → None[source]

async sut_stdout(sut: str, data: str) → None[source]

async sut_stop(sut: str) → None[source]

async test_completed(results: TestResults) → None[source]

async test_started(test: Test) → None[source]

async test_stdout(test: Test, data: str) → None[source]

async test_timed_out(test: Test, timeout: int) → None[source]

libkirk.plugin module

class libkirk.plugin.Plugin[source]

Bases: object

Generic plugin definition.

clone(name: str) → _Self[source]

Copy plugin and return a new instance with a given name. Make sure that name is unique, so external modules can recognize it.

Parameters:: name (str) – Unique identifier string given to the plugin.
Returns:: Cloned plugin.
Return type:: Plugin

property config_help: Dict[str, str]

Associate each configuration option with a help message. This is used by the main menu application to generate –help message.

Returns:: Dictionary that associates options to help messages.
Return type:: dict

property name: str

Returns:: Unique name identifier of the plugin.
Return type:: str

setup(**kwargs: Dict[str, Any]) → None[source]

Initialize plugin using configuration dictionary.

Parameters:: kwargs (dict) – SUT configuration.

libkirk.plugin.discover(mytype: type, folder: str) → List[Plugin][source]

Discover mytype implementations inside a specific folder.

Parameters:

mtype (type) – Type of the object we are searching for.
folder (str) – Folder where to search for mtype.

Returns:

List of discovered plugins.

Return type:

list(Plugin)

libkirk.results module

class libkirk.results.ResultStatus[source]

Bases: object

Overall status of the test. This is a specific flag that is used to recognize final test status. For example, we might have 10 tests passing inside a single test binary, but the overall status of the test is fine, so we assign a PASS status.

BROK = 2: Test is broken.

CONF = 32: Test can’t run because of configuration error.

FAIL = 16: Test has failed.

PASS = 0: Test has passed.

WARN = 4: Test warnings received.

class libkirk.results.Results[source]

Bases: object

Base class for results.

property broken: int

Returns:: Number of broken tests.
Return type:: int

property exec_time: float

Returns:: Test execution time in seconds.
Return type:: float

property failed: int

Returns:: Number of failures.
Return type:: int

property passed: int

Returns:: Number of passed tests.
Return type:: int

property skipped: int

Returns:: Number of skipped tests.
Return type:: int

property warnings: int

Returns:: Number of warnings.
Return type:: int

Bases: Results

Testing suite results definition.

property arch: str | None

Returns:: Operating system architecture.
Return type:: str | None

property broken: int

Returns:: Number of broken tests.
Return type:: int

property cmdline: str | None

Returns:: /proc/cmdline.
Return type:: str | None

property cpu: str | None

Returns:: Current CPU type.
Return type:: str | None

property distro: str | None

Returns:: Linux distribution name.
Return type:: str | None

property distro_ver: str | None

Returns:: Linux distribution version.
Return type:: str | None

property exec_time: float

Returns:: Test execution time in seconds.
Return type:: float

property failed: int

Returns:: Number of failures.
Return type:: int

property kernel: str | None

Returns:: Kernel version.
Return type:: str | None

property passed: int

Returns:: Number of passed tests.
Return type:: int

property ram: str | None

Returns:: Current RAM occupation.
Return type:: str | None

property skipped: int

Returns:: Number of skipped tests.
Return type:: int

property suite: Suite

Returns:: Testing suite.
Return type:: Suite

property swap: str | None

Returns:: Current swap memory occupation.
Return type:: str | None

property tests_results: List[TestResults]

Returns:: list of all tests results.
Return type:: list(TestResults)

property warnings: int

Returns:: Number of warnings.
Return type:: int

class libkirk.results.TestResults(test: Test, failed: int = 0, passed: int = 0, broken: int = 0, skipped: int = 0, warnings: int = 0, exec_time: float = 0.0, status: int = 0, retcode: int = 0, stdout: str = '')[source]

Bases: Results

Test results definition.

property broken: int

Returns:: Number of broken tests.
Return type:: int

property exec_time: float

Returns:: Test execution time in seconds.
Return type:: float

property failed: int

Returns:: Number of failures.
Return type:: int

property passed: int

Returns:: Number of passed tests.
Return type:: int

property return_code: int

Returns:: Return code after execution.
Return type:: int

property skipped: int

Returns:: Number of skipped tests.
Return type:: int

property status: int

Returns:: Test result status.
Return type:: int

property stdout: str

Returns:: Test process stdout.
Return type:: str

property test: Test

Returns:: Test object.
Return type:: Test

property warnings: int

Returns:: Number of warnings.
Return type:: int

libkirk.scheduler module

class libkirk.scheduler.Scheduler[source]

Bases: object

Schedule jobs to run on target.

property results: List[Results]

Current results. It’s reset before every schedule call and it’s populated when a job completes the execution.

Returns:: List of results.
Return type:: list(Results)

async schedule(jobs: List[Any]) → None[source]

Schedule and execute a list of jobs.

Parameters:: jobs (list(object)) – Object containing jobs definition

async stop() → None[source]: Stop all running jobs.

property stopped: bool

Returns:: True when scheduler has been stopped. False otherwise.
Return type:: bool

class libkirk.scheduler.SuiteScheduler(sut: SUT, framework: Framework, suite_timeout: float = 0.0, exec_timeout: float = 0.0, max_workers: int = 1)[source]

Bases: Scheduler

The Scheduler class implementation for suites execution. This is a special scheduler that schedules suites tests, checking for kernel status and rebooting SUT if we have some issues with it (i.e. kernel panic).

property results: List[Results]

Current results. It’s reset before every schedule call and it’s populated when a job completes the execution.

Returns:: List of results.
Return type:: list(Results)

async schedule(jobs: List[Any]) → None[source]

Schedule and execute a list of jobs.

Parameters:: jobs (list(object)) – Object containing jobs definition

async stop() → None[source]: Stop all running jobs.

property stopped: bool

Returns:: True when scheduler has been stopped. False otherwise.
Return type:: bool

class libkirk.scheduler.TestScheduler(sut: SUT, framework: Framework, timeout: float = 0.0, max_workers: int = 1)[source]

Bases: Scheduler

Schedule and run tests, taking into account status of the kernel during their execution, as well as tests timeout.

property results: List[Results]

Current results. It’s reset before every schedule call and it’s populated when a job completes the execution.

Returns:: List of results.
Return type:: list(Results)

async schedule(jobs: List[Any]) → None[source]

Schedule and execute a list of jobs.

Parameters:: jobs (list(object)) – Object containing jobs definition

async stop() → None[source]: Stop all running jobs.

property stopped: bool

Returns:: True when scheduler has been stopped. False otherwise.
Return type:: bool

class libkirk.scheduler.TestStatus(*values)[source]

Bases: IntEnum

Status codes returned by test execution in the scheduler.

KERNEL_PANIC = 2

KERNEL_TAINTED = 3

KERNEL_TIMEOUT = 4

OK = 0

TEST_TIMEOUT = 1

libkirk.session module

class libkirk.session.Session(tmpdir: TempDir, sut: SUT, exec_timeout: float = 3600.0, suite_timeout: float = 3600.0, workers: int = 1, force_parallel: bool = False)[source]

Bases: object

The session runner.

async run(command: str | None = None, suites: List[str] | None = None, pattern: str | None = None, skip_tests: str | None = None, report_path: str | None = None, restore_path: str | None = None, suite_iterate: int = 1, randomize: bool = False, runtime: float = 0, fault_prob: int = 0, fault_interval: int = 1) → None[source]

Run a new session and store results inside a JSON file.

Parameters:

command (str) – Single command to run before suites.
suites (list) – List of suites to execute.
pattern (str) – Regex pattern to include tests.
skip_tests (str) – Regex for tests to skip.
report_path (str) – JSON report path.
restore_path (str) – Temporary directory generated by a previous session.
suite_iterate (int) – Execute all suites multiple times.
randomize (bool) – Randomize all tests if True.
runtime (float) – For how long we want to run the session.
fault_prob (int) – Fault injection probability.
fault_interval (int) – Fault injection interval.

async stop() → None[source]: Stop the current session.

libkirk.sut module

class libkirk.sut.RedirectSUTStdout(sut: SUT, is_cmd: bool = False)[source]

Bases: IOBuffer

Redirect SUT stdout data to UI events.

async write(data: str) → None[source]

Write data inside the buffer.

Parameters:: data (str) – Data to write.

class libkirk.sut.RedirectTestStdout(test: Test)[source]

Bases: IOBuffer

Redirect test stdout data to UI events and save it.

async write(data: str) → None[source]

Write data inside the buffer.

Parameters:: data (str) – Data to write.

class libkirk.sut.SUT[source]

Bases: Plugin

SUT abstraction class. It could be a remote host, a local host, a virtual machine instance, or any complex system we want to test.

FAULT_INJECTION_FILES = ['fail_io_timeout', 'fail_make_request', 'fail_page_alloc', 'failslab']

TAINTED_MSG = ['proprietary module was loaded', 'module was force loaded', 'kernel running on an out of specification system', 'module was force unloaded', 'processor reported a Machine Check Exception (MCE)', 'bad page referenced or some unexpected page flags', 'taint requested by userspace application', 'kernel died recently, i.e. there was an OOPS or BUG', 'ACPI table overridden by user', 'kernel issued warning', 'staging driver was loaded', 'workaround for bug in platform firmware applied', 'externally-built (“out-of-tree”) module was loaded', 'unsigned module was loaded', 'soft lockup occurred', 'kernel has been live patched', 'auxiliary taint, defined for and used by distros', 'kernel was built with the struct randomization plugin']

get_channel() → ComChannel[source]

Returns:: Main channel to communicated with SUT.
Return type:: ComChannel

async get_info() → Dict[str, str][source]

Return SUT information.

Returns:

Dictionary containing the SUT information in the form of:

{
    "distro": str,
    "distro_ver": str,
    "kernel": str,
    "cmdline": str,
    "arch": str,
    "cpu" : str,
    "swap" : str,
    "ram" : str,
}

Return type:

dict

async get_tainted_info() → Tuple[int, List[str]][source]

Return information about kernel if tainted.

Returns:: A tuple containing tainted information. First element is the tainted code and the second element is the tainted message.
Return type:: (int, list[str])

async is_fault_injection_enabled() → bool[source]

Returns:: True if fault injection is enabled in the kernel. False otherwise.
Return type:: bool

async is_running() → bool[source]

Returns:: True if system under test is up and running. False otherwise.
Return type:: bool

async logged_as_root() → bool[source]

Returns:: True if we are logged as root inside the SUT. False otherwise.
Return type:: bool

property optimize: bool

Optimize the commands execution by applying parallelization when it’s available.

Returns:: True if SUT will optimize commands execution on SUT.

async restart(iobuffer: IOBuffer | None = None) → None[source]

Restart the SUT.

Parameters:: iobuffer (IOBuffer) – IO channel where to write stdout.

async setup_fault_injection(prob: int, interval: int = 1) → None[source]

Configure kernel fault injection. When prob is zero, the fault injection is set to default values.

Parameters:

prob (int) – Fault probability in between 0-100.
interval (int) – Fault interval.

async start(iobuffer: IOBuffer | None = None) → None[source]

Start the SUT.

Parameters:: iobuffer (IOBuffer) – IO channel where to write stdout.

async stop(iobuffer: IOBuffer | None = None) → None[source]

Stop the SUT.

Parameters:: iobuffer (IOBuffer) – IO channel where to write stdout.

libkirk.sut.discover(path: str, extend: bool = True) → None[source]

Discover all SUT implementations inside path.

Parameters:

path (str) – Directory where searching for SUT implementations.
extend – If True, it will add new discovered SUT on top of the ones already found. If False, previous discovered SUT will be cleared.

libkirk.sut.get_suts() → List[SUT][source]

Returns:: List of loaded SUT implementations.
Return type:: list(SUT)

libkirk.sut_base module

class libkirk.sut_base.GenericSUT[source]

Bases: SUT

Generic SUT which is defining a structure to start implementing it.

This class can be inherited in order to provide all its features also to the new implementations. Used by itself, it’s just a SUT named ‘default’ for the kirk application and it’s recognized by the plugin system.

property config_help: Dict[str, str]

Associate each configuration option with a help message. This is used by the main menu application to generate –help message.

Returns:: Dictionary that associates options to help messages.
Return type:: dict

get_channel() → ComChannel[source]

Returns:: Main channel to communicated with SUT.
Return type:: ComChannel

async is_running() → bool[source]

Returns:: True if system under test is up and running. False otherwise.
Return type:: bool

property name: str

Returns:: Unique name identifier of the plugin.
Return type:: str

async restart(iobuffer: IOBuffer | None = None) → None[source]

Restart the SUT.

Parameters:: iobuffer (IOBuffer) – IO channel where to write stdout.

setup(**kwargs: Dict[str, Any]) → None[source]

Initialize plugin using configuration dictionary.

Parameters:: kwargs (dict) – SUT configuration.

async start(iobuffer: IOBuffer | None = None) → None[source]

Start the SUT.

Parameters:: iobuffer (IOBuffer) – IO channel where to write stdout.

async stop(iobuffer: IOBuffer | None = None) → None[source]

Stop the SUT.

Parameters:: iobuffer (IOBuffer) – IO channel where to write stdout.

libkirk.tempfile module

class libkirk.tempfile.TempDir(root: str | None, max_rotate: int = 5)[source]

Bases: object

Temporary directory handler.

FOLDER_PREFIX = 'kirk.'

SYMLINK_NAME = 'latest'

property abspath: str

Returns:: Absolute path of the temporary directory.
Return type:: str

mkdir(path: str) → None[source]

Create a directory inside temporary directory.

Parameters:: path (str) – Path of the directory.
Returns:: Folder path.

mkfile(path: str, content: str) → None[source]

Create a file inside temporary directory.

Parameters:

path (str) – Path of the file.
content (str) – File content.

property root: str

Returns:: The root folder. For example, if temporary folder is “/tmp/kirk.acer/tmpf547ftxv” the method will return “/tmp”. If root folder has not been given during object creation, this method returns an empty string.
Return type:: str

libkirk.types module

libkirk.types.dict_item(data: Dict[str, Any], key: str, cls: Type, default: Any | None = None) → Any[source]

Extract a value from a dictionary according to the key, ensuring that correct type is returned.

Parameters:

data (dict) – Dictionary from where we want to extract data.
key (str) – Key we are searching for.
cls (Type) – Type we want to extract.
default (Any | None) – Default value.

Returns:

Type of the default value.

Return type:

Any

libkirk.ui module

class libkirk.ui.ConsoleUserInterface(no_colors: bool = False)[source]

Bases: object

Console based user interface.

CYAN = '\x1b[1;36m'

GREEN = '\x1b[1;32m'

RED = '\x1b[1;31m'

RESET_COLOR = '\x1b[0m'

RESET_SCREEN = '\x1b[2J'

WHITE = '\x1b[1;37m'

YELLOW = '\x1b[1;33m'

async internal_error(exc: BaseException, func_name: str) → None[source]

async print_message(msg: str, end: str = '\n', flush: bool = True) → None[source]: Print a message in console, avoiding any I/O blocking operation done by the print built-in function, using asyncio.to_thread().

async run_cmd_start(cmd: str) → None[source]

async run_cmd_stdout(data: str) → None[source]

async run_cmd_stop(command: str, stdout: str, returncode: int) → None[source]

async session_completed(results: List[SuiteResults]) → None[source]

async session_error(error: str) → None[source]

async session_restore(restore: str) → None[source]

async session_started(suites: list, tmpdir: str) → None[source]

async session_stopped() → None[source]

async session_warning(msg: str) → None[source]

async suite_completed(results: SuiteResults, exec_time: float) → None[source]

async suite_started(suite: Suite) → None[source]

async suite_timeout(suite: Suite, timeout: float) → None[source]

async sut_restart(sut: str) → None[source]

async sut_start(sut: str) → None[source]

async sut_stop(sut: str) → None[source]

class libkirk.ui.ParallelUserInterface(no_colors: bool = False)[source]

Bases: ConsoleUserInterface

Console based user interface for parallel execution of the tests.

async kernel_panic() → None[source]

async kernel_tainted(message: str) → None[source]

async print_parallel(suite: Suite) → None[source]

async sut_not_responding() → None[source]

async test_completed(results: TestResults) → None[source]

async test_timed_out(_: Test, timeout: int) → None[source]

class libkirk.ui.SimpleUserInterface(no_colors: bool = False)[source]

Bases: ConsoleUserInterface

Console based user interface without many fancy stuff.

async kernel_panic() → None[source]

async kernel_tainted(message: str) → None[source]

async sut_not_responding() → None[source]

async test_completed(results: TestResults) → None[source]

async test_started(test: Test) → None[source]

async test_timed_out(_: Test, timeout: int) → None[source]

class libkirk.ui.VerboseUserInterface(no_colors: bool = False)[source]

Bases: ConsoleUserInterface

Verbose console based user interface.

async kernel_tainted(message: str) → None[source]

async sut_stdout(_: str, data: str) → None[source]

async test_completed(results: TestResults) → None[source]

async test_started(test: Test) → None[source]

async test_stdout(_: Test, data: str) → None[source]

async test_timed_out(_: Test, timeout: int) → None[source]

Module contents

libkirk.get_event_loop() → AbstractEventLoop[source]: Return the current asyncio event loop.