Testbed

class catkit2.testbed.Testbed(port, is_simulated, config)

Bases: object

Manages services.

portinteger: The port on which the server should operate.
is_simulatedboolean: Whether the server should operate in simulated mode or not. This changes whether a simulated or hardware service is launched when a specific service is requested.
configdictionary: The full configuration as read in from the configuration files.

Methods Summary

`destroy_logging`()	Shut down all logging.
`do_heartbeats`()
`interrupt_service`(service_id)
`monitor_services`()
`on_get_info`(data)
`on_get_service_info`(data)
`on_interrupt_service`(data)
`on_register_service`(data)
`on_reload_config`(data)
`on_shut_down`(data)
`on_start_service`(data)
`on_stop_service`(data)
`on_terminate_service`(data)
`register_service_type`(service_type, path)	Register a service type.
`reload_config`(new_config)	Reload the testbed configuration.
`resolve_service_type`(service_type)	Resolve a service type into a path.
`run`()	Run the main loop of the server.
`setup_logging`()	Set up all logging.
`shut_down_all_services`()	Shut down all running services.
`start_log_distributor`()	Start the log distributor.
`start_service`(service_id)	Start a service.
`start_tracing_distributor`()	Start the tracing distributor.
`stop_log_distributor`()	Stop the log distributor.
`stop_service`(service_id)
`stop_tracing_distributor`()
`terminate_service`(service_id)

Methods Documentation

destroy_logging(): Shut down all logging.

do_heartbeats()

interrupt_service(service_id)

monitor_services()

on_get_info(data)

on_get_service_info(data)

on_interrupt_service(data)

on_register_service(data)

on_reload_config(data)

on_shut_down(data)

on_start_service(data)

on_stop_service(data)

on_terminate_service(data)

register_service_type(service_type, path)

Register a service type.

service_typestr: The service type.
pathstr: The path to the Python file to run for this service.

reload_config(new_config)

Reload the testbed configuration.

This updates the configuration without restarting the testbed. Services must be restarted to pick up their new configuration.

new_configdict: The new configuration dictionary.

RuntimeError: If the new configuration would produce an invalid dependency graph. In that case the new configuration is rejected and the testbed state is unchanged.

resolve_service_type(service_type)

Resolve a service type into a path.

service_typestring: The type of the service.

string: The path to the Python script of the service.

run(): Run the main loop of the server.

setup_logging(): Set up all logging.

shut_down_all_services()

Shut down all running services.

This sends a keyboard interrupt to all running services, and waits for each service to shut down. If a KeyboardInterrupt occurs during this shutdown process, all services that have not shut down already will be killed.

start_log_distributor(): Start the log distributor.

start_service(service_id)

Start a service.

service_idstring: The identifier of the service. This should correspond to an entry in the services section of the configuration of this testbed.

RuntimeError: If the service is not found in the configuration. If the service type path did not contain an executable or Python script.
ValueError: If the service type could not be found in the known services paths.

start_tracing_distributor(): Start the tracing distributor.

stop_log_distributor(): Stop the log distributor.

stop_service(service_id)

stop_tracing_distributor()

terminate_service(service_id)