triggerflow/tests/ingestion-test/setup_redis.py at master · cloudbutton/triggerflow · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from uuid import uuid4
from platform import node

from triggerflow import Triggerflow, CloudEvent, DefaultActions, DefaultConditions
from triggerflow.eventsources import RedisEventSource

stream = 'ingestion'


def setup_triggers():
    red = RedisEventSource(host="127.0.0.1", port=6379, password="potato")
    red.set_stream(stream)
    tf = Triggerflow()
    tf.create_workspace(workspace_name='ingestion-test', event_source=red)

    for i in range(200):
        uuid = uuid4()
        cloudevent = (CloudEvent()
                      .SetSubject("join{}".format(i))
                      .SetEventType('event.triggerflow.test')
                      .SetEventID(uuid.hex)
                      .SetSource(f'urn:{node()}:{str(uuid)}'))

        tf.add_trigger(
            event=cloudevent,
            trigger_id="join{}".format(i),
            condition=DefaultConditions.JOIN,
            action=DefaultActions.TERMINATE,
            context={'join': 1000},
            transient=False
        )


if __name__ == '__main__':
    setup_triggers()