Event Listener

A salt events store for all daemons started by salt-factories

class saltfactories.plugins.event_listener.Event(*, daemon_id, tag, stamp, data, full_data, expire_seconds)

Bases: object

The Event class is a container for a salt event which will live on the EventListener store.

Parameters

• daemon_id (str) – The daemon ID which received this event.

• tag (str) – The event tag of the event.

• stamp (datetime) – When the event occurred

• data (dict) – The event payload, filtered of all of Salt’s private keys like _stamp which prevents proper assertions against it.

• full_data (dict) – The full event payload, as received by the daemon, including all of Salt’s private keys.

• expire_seconds (int,float) – The time, in seconds, after which the event should be considered as expired and removed from the store.

Return type

None

property expired

Property to identify if the event has expired, at which time it should be removed from the store.

class saltfactories.plugins.event_listener.MatchedEvents(*, matches, missed)

Bases: object

The MatchedEvents class is a container which is returned by wait_for_events().

Parameters

• matches (set) – A set of Event instances that matched.

• missed (set) – A set of Event instances that remained unmatched.

Return type

None

One can also easily iterate through all matched events of this class:

matched_events = MatchedEvents(..., ...)
for event in matched_events:
    print(event.tag)

property found_all_events

Return bool

True if all events were matched, or False otherwise.

class saltfactories.plugins.event_listener.EventListener(*, timeout=120)

Bases: object

The EventListener is a service started by salt-factories which receives all the events of all the salt masters that it starts. The service runs throughout the whole pytest session.

Parameters

timeout (int) – How long, in seconds, should a forwarded event stay in the store, after which, it will be deleted.

Return type

None

get_events(patterns, after_time=None)

Get events from the internal store.

Parameters

• pattern (Sequence) – An iterable of tuples in the form of ("<daemon-id>", "<event-tag-pattern>"), ie, which daemon ID we’re targeting and the event tag pattern which will be passed to fnmatch() to assert a match.

• after_time (datetime,float) – After which time to start matching events.

Return set

A set of matched events

wait_for_events(patterns, timeout=30, after_time=None)

Wait for a set of patterns to match or until timeout is reached.

Parameters

• pattern (Sequence) – An iterable of tuples in the form of ("<daemon-id>", "<event-tag-pattern>"), ie, which daemon ID we’re targeting and the event tag pattern which will be passed to fnmatch() to assert a match.

• timeout (int,float) – The amount of time to wait for the events, in seconds.

• after_time (datetime,float) – After which time to start matching events.

Returns

An instance of MatchedEvents.

Rtype ~saltfactories.plugins.event_listener.MatchedEvents

register_auth_event_handler(master_id, callback)

Register a callback to run for every authentication event, to accept or reject the minion authenticating.

Parameters

• master_id (str) – The master ID for which the callback should run

• callback (Callable) – The function while should be called

unregister_auth_event_handler(master_id)

Un-register the authentication event callback, if any, for the provided master ID

Parameters

master_id (str) – The master ID for which the callback is registered

saltfactories.plugins.event_listener.event_listener(request)

All started daemons will forward their events into an instance of EventListener.

This fixture can be used to wait for events:

def test_send(event_listener, salt_master, salt_minion, salt_call_cli):
    event_tag = random_string("salt/test/event/")
    data = {"event.fire": "just test it!!!!"}
    start_time = time.time()
    ret = salt_call_cli.run("event.send", event_tag, data=data)
    assert ret.exitcode == 0
    assert ret.json
    assert ret.json is True

    event_pattern = (salt_master.id, event_tag)
    matched_events = event_listener.wait_for_events(
        [event_pattern], after_time=start_time, timeout=30
    )
    assert matched_events.found_all_events
    # At this stage, we got all the events we were waiting for

And assert against those events events:

def test_send(event_listener, salt_master, salt_minion, salt_call_cli):
    # ... check the example above for the initial code ...
    assert matched_events.found_all_events
    # At this stage, we got all the events we were waiting for
    for event in matched_events:
        assert event.data["id"] == salt_minion.id
        assert event.data["cmd"] == "_minion_event"
        assert "event.fire" in event.data["data"]