cfme.utils.events module¶
Library for event testing.
-
class
cfme.utils.events.
Event
(event_tool, *args)[source]¶ Bases:
object
represents either db event received by CFME and stored in event_streams or an expected event
-
add_attrs
(*attrs)[source]¶ event consists of attributes like event_type, etc. this method allows to add an attribute to event
-
-
class
cfme.utils.events.
EventAttr
(attr_type=None, cmp_func=None, **attrs)[source]¶ Bases:
object
contains one event attribute and the method for comparing it.
-
class
cfme.utils.events.
EventListener
(appliance)[source]¶ Bases:
threading.Thread
accepts “expected” events, listens to db events and compares showed up events with expected events. Runs callback function if expected events have it.
-
__call__
(*args, **kwargs)[source]¶ it is called by register_event fixture. bad idea, to replace register_event by object later
-
got_events
¶ returns dict with expected events and all the events matched to expected ones
-
listen_to
(*evts, **kwargs)[source]¶ accepts one or many events callback function will be called when event arrived in event_streams. callback will receive expected event and got event as params.
Parameters: - evts – list of events which EventListener should listen to
- callback – callback function that will be called if event is received
- first_event – EventListener waits for only first event of such type. it ignores such event in future if first matching event is found.
By default EventListener collects and receives all matching events.
-
new_event
(*attrs, **kwattrs)[source]¶ this method just simplifies “expected” event creation.
Usage:
listener = appliance.event_listener() evt = listener.new_event(target_type='VmOrTemplate', target_name='my_lovely_vm', event_type='vm_create') listener.listen_to(evt)
-
process_events
()[source]¶ processes all new db events and compares them with expected events. processed events are ignored next time
-
started
¶
-
-
class
cfme.utils.events.
EventTool
(appliance)[source]¶ Bases:
object
EventTool serves as a wrapper to getting the events from the database.
Variables: OBJECT_TABLE – Mapping of object types to tables and column names. -
OBJECT_TABLE
= {'Host': ('hosts', 'name', 'id'), 'Service': ('services', 'name', 'id'), 'VmOrTemplate': ('vms', 'name', 'id')}¶
-
ensure_event_happens
(*args, **kwds)[source]¶ Context manager usable for one-off checking of the events.
See also:
query_miq_events()
Parameters: - target_type – What kind of object is the target of the event (MiqServer, VmOrTemplate)
- target_id – What is the ID of the object (or name, see
process_id()
). - event_type – Type of the event. Ideally one of the
all_event_types()
but other kinds of events exist too.
-
event_streams
¶ event_streams
table.
-
event_streams_attributes
¶ event_streams
columns and python’s column types
-
miq_event_definitions
¶ miq_event_definitions
table.
-
process_id
(target_type, target_name)[source]¶ Resolves id, let it be a string or an id.
In case the
target_type
is defined in theOBJECT_TABLE
, you can pass a string with object’s name, otherwise a numeric id to the table is required.Parameters: - target_type – What kind of object is the target of the event (MiqServer, VmOrTemplate...)
- target_name – An id or a name of the object.
Returns: int
with id of the object in the database.
-
query_miq_events
(target_type=None, target_id=None, event_type=None, since=None, until=None, from_id=None)[source]¶ Checks whether an event occured.
Parameters: - target_type – What kind of object is the target of the event (MiqServer, VmOrTemplate)
- target_id – What is the ID of the object (or name, see
process_id()
). - event_type – Type of the event. Ideally one of the
all_event_types()
but other kinds of events exist too. - since – Since when you want to check it. UTC
- until – Until what time you want to check it.
-