How to manage two "then" whose order is unknown ?

[jp-pacotte]

I work on a legacy product that trigger events but the order is not stable, i can get an event A then an event B or event B then event A.
This events are collected via a SSH 'tail -f' command with paramiko-expect.

How can i write parallel 'then' because if event B is triggered before event A the log will be lost when the 'then' of event B will be started.

Feature: Foo
   dectection of 2 events not ordered

   Scenario: event A and event B detected
     Given the system is ready

     When i do action

     Then event A detected
     And event B detected

@then("event A detected")
def check_event_A(shh_interraction):
    shh_interraction.tail(stop_callback= lambda log: log_expected(log, 'event A'), timeout=10)

@then("event B detected")
def check_event_B(shh_interraction):
    shh_interraction.tail(stop_callback= lambda log: log_expected(log, 'event B'), timeout=10)

def log_expected(log, expected):
    return expected in log

thanks for your help.

[youtux]

pytest-bdd doesn't support that, you will have to come up with something that collects logs and that then checks the collected logs. Here is a suggestion of how the feature could look like:

Feature: Foo
   dectection of 2 events not ordered

   Scenario: event A and event B detected
     Given the system is ready
     And I am collecting logs
     
     When i do action

     Then event A detected
     And event B detected

and the "I collect logs" could be defined as

@pytest.fixture
def collected_logs():
    return []

@then("I collect logs")
def _(shh_interraction, collected_logs, request):
    shh_interraction.tail(callback=lambda log: collected_logs.append(log))
    # make sure to stop the ssh connection at the end of the test
    request.add_finalizer(lambda: ssh_interraction.disconnect())

@then("event A detected")
def check_event_A(shh_interraction):
    assert any(log_expected(log, 'event A') for log in logs)

@then("event B detected")
def check_event_B(captured_logs):
    assert any(log_expected(log, 'event B') for log in logs)

[jp-pacotte]

I forgot to update this discussion with my solution.
A LogExpectedManager that check if all events expected are found in a timing or not :

@then("event A and event B detected")
def check_eventA_eventB(shh_interraction):
    logs_expected_manager = LogsExpectedManager(['eventA', eventB'])
    shh_interraction.tail(stop_callback= lambda log: logs_expected_manager.process_log_expected(log), timeout=10)

from collections import namedtuple

class LogsExpectedManager:

    LogExpected = namedtuple('LogExpected',['log_expected','found'])

    def __init__(self, logs_expected):
        self.events = []

        for log_expected in logs_expected:
            self.events.append(self.LogExpected(log_expected, False))

    def process_log_expected(self, log):

        for i in range(len(self.events)):
            if self.events[i].log_expected in log:
                self.events[i] = self.events[i]._replace( found = True )

        return all([event.found for event in self.events])