Set exit code of b2luigi.process according to pipeline result, not scheduling result

[klieret]

I'm using pytest and want to also test my full luigi pipeline. Currently I'm facing two issues here:

• Issue 1: Calling process more than once: Once there is more than one test, b2luigi will make a fuzz about multiple process calls.
• Issue 2: process doesn't return exit code: Is there a clever way to figure out if the tasks ran successfully?

Currently I'm using two workarounds, neither of which are particularly pretty:

import b2luigi as luigi
import b2luigi.cli.process


def test_real_conversion_set(capsys):
    ...
    luigi.process(
        ...,
        batch=False,
        workers=1,
        ignore_additional_command_line_args=True,
    )
    captured = capsys.readouterr()
    assert "This progress looks :)" in captured.err
    # Deactivate the "run process only once" check, which might be useful IRL
    # but doesn't work with >= 2 tests
    luigi.cli.process.__has_run_already = False

This works fine, but eventually prettier solutions would be nice of course.

I also wanted to open this issue to document the workarounds. I could also open a PR for some documentation on that if you want.

[meliache]

Maybe you can take a look at B2LuigiTestCase and as an example of how to use it at our test_batch_process.py.

[klieret]

Ah, thanks for the suggestion, I didn't think of looking into your own unit tests. Your solution requires a bit more boiler plate (separate steering file, subprocess call and all), but it probably is the cleanest solution and doesn't require any modifications to the code. Let me close this :)

Edit: Okay, actually it does the same "This progress looks :)" hack as me :)

[meliache]

:)

[klieret]

Another thing where this comes up though is when chaining job completion to some other bash commands. Ideally I'd like to do ./submit_luigi_jobs.py && rsync .... And this currently isn't possible easily, right? As in: There's no easy way to set the exit code other than inspecting stdout/stderr.

[meliache]

Tbh I know that feeling, in the past I was quite annoyed about gbasf2 because it didn't set proper exit codes on failure. Haven't done bash chaining with b2luigi scripts yet, because for that I'd usually just use another b2luigi task. I haven't tested it yet, but shouldn't something like the following work?

main_task = MainTask()   # b2luigi wrapper task
b2luigi.process(main_task)
if not main_task.complete():  # tests if all outputs exist
    raise RuntimeError("Processing did not complete task") # or just: sys.exit(1)

[nils-braun]

So just a bit of background, why (b2)luigi does not give a non-zero error code: for luigi, the work is "successful" in the sense that it could correctly build the task DAG, execute it and finish up. The result was negative, but that is not luigi's concern (see e.g. here).

However, I like your suggestion, @meliache. Internally, luigi seems to produce a LuigiRunResult. I did not test, but maybe we can deduce the result from that? The call to luigi.build that b2luigi does eventually is basically just a wrapper around an internal luigi function, that exposes this result. That would maybe be a good entry point (haven't checked the result live though)