* [launch_pytest] Fix issue when colcon --retest-until-fail flag is used

[ivanpauno]

Fixes #528 (comment).

test_module_scope.py / test_function_scope.py don't work when --count=N is passed, because now the test order is different.
I moved them to the example folder and I'm now running then with the pytester plugin to avoid the issue.

I'm also running the other examples we have in that folder, ignoring one that would introduce a dependency on launch_ros (we have to move that one somewhere else later).
I've also fixed some issues in the existing examples.

[Blast545]

LGTM after green CI

[Blast545]

Curious question, why was this assert removed?

[ivanpauno]

The function always returns None in this case, though I think that's confusing.

[ivanpauno]

See a9cc81c

[ivanpauno]

The function always returns None in this case, though I think that's confusing.

It was not only confusing, it was also blocking unnecessarely untile the timeout expired.

[hidmic]

@ivanpauno meta: we clearly need a less terse idiom to write predicates. If we were to assume the predicate passed if it returns None, that'd be practical yet surprising. If we were to simply return the result of evaluating the expression, we'd lose detailed assertion message (most of which are provided pytest rewriting logic). So how about splitting use cases? E.g.:

# Wait for incoming process output with a timeout
output = wait_for_output_sync(process, timeout=10)
output = await wait_for_output(process, timeout=10)

# Wait for incoming process output matching a predicate with a timeout
def predicate(output):
    return 'foo' in output
output = wait_for_output_sync(process, predicate, timeout=10)
output = await wait_for_output(process, predicate, timeout=10)

# Assert a statement on process output with a timeout 
def statement(output):
    assert output == 'foo'
assert_output_sync(process, statement, timeout=10)
await assert_output(process, statement, timeout=10)

In this case, it would read:

def hello_world_found():
     assert output.splitlines() == ['hello_world'], 'process never printed hello_world'
assert_output_sync(launch_context, hello_world_proc, hello_world_found, timeout=5)

[ivanpauno]

That sounds like the right approach to me.
I will address it in a follow-up PR so we solve the CI issue.

[hidmic]

Agreed, merge!

[ivanpauno]

CI using --retest-until-fail 2:

• Linux
• Linux-aarch64
• macOS
• Windows

[ivanpauno]

• Windows

[hidmic]

LGTM but for one meta comment

[hidmic]

@ivanpauno meta: we clearly need a less terse idiom to write predicates. If we were to assume the predicate passed if it returns None, that'd be practical yet surprising. If we were to simply return the result of evaluating the expression, we'd lose detailed assertion message (most of which are provided pytest rewriting logic). So how about splitting use cases? E.g.:

# Wait for incoming process output with a timeout
output = wait_for_output_sync(process, timeout=10)
output = await wait_for_output(process, timeout=10)

# Wait for incoming process output matching a predicate with a timeout
def predicate(output):
    return 'foo' in output
output = wait_for_output_sync(process, predicate, timeout=10)
output = await wait_for_output(process, predicate, timeout=10)

# Assert a statement on process output with a timeout 
def statement(output):
    assert output == 'foo'
assert_output_sync(process, statement, timeout=10)
await assert_output(process, statement, timeout=10)

In this case, it would read:

def hello_world_found():
     assert output.splitlines() == ['hello_world'], 'process never printed hello_world'
assert_output_sync(launch_context, hello_world_proc, hello_world_found, timeout=5)