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_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.
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_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.

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)