Enable per-node API calls monitoring

[kmazurek]

In the current implementation of API assertions (#541) there's a single monitor being used for all API events. A serious drawback of this approach (as described here: golemfactory/yagna#1585) is the fact the order of the API calls may affect the result of the assertions.
A possible solution to this is introducing a Monitor object per Probe which is fed the same stream of API events.

[azawlocki]

@zakaprov Well, a single monitor for all API events is not that bad as it may seem.

It is true that the following test that uses a helper API step function wait_for_approve_agreement_response() defined in golemfactory/yagna#1585 may will, depending on the ordering of responses to approveAgreement made by two providers:

async with runner():
   ...
   wait_for_approve_agreement_response(provider_1)
   wait_for_approve_agreement_response(provider_2)

However that may merely mean that wait_for_approve_agreement_response if not especially good for the job you want to do here.

The following version is not sensitive to ordering of the two events in question:

async def agreement_approved(provider, agr_id, events):
    async for e in events:
        if is_approve_agreement(e, event_type=ApiResponse, node_name=provider.name, agr_id=agr_id):
            return True
    raise AssertionError(f"Provider {provider.name} did not approve the agreement {agr_id}")

async with runner():
    ...
    a1 = runner.add_api_assertion(partial(agreement_approved, agr_id_1, provider_1))
    a2 = runner.add_api_assertion(partial(agreement_approved, agr_id_2, provider_2))
    # This will succeed independent on the order in which `a1` and `a2` succeed:
    await a1.wait_for_result(timeout=10)
    await a2.wait_for_result(timeout=10)

The difference now is that a2.wait_for_result() does not require that a2 succeeds after a1, as it does not use the EventMonitor.wait_for_event() mechanism.

The drawback here is that the assertions a1 and a2 will only succeed if they are started before the actual events on which they wait (in contrast, EventMonitor.wait_for_event() could also examine past events) so you have to make sure a = runner.add_api_assertion(...) is not executed too late.

One solution to this problem would be to add API assertions even before the test is started, like this:

a = runner.add_api_assertion(some_assertion)
async with runner():
   ...
   await a.wait_for_result()

It has its own problems: assertion parameters such as probe names or agreement ids (as in case agreement_approved) may not be known before the test is started so they would have to be passed via global variables or as futures of some kind.

Another possibility is to create more declarative assertions. For example, instead of adding an assertion that states that the requestor will eventually accept an invoice with a particular id you could write an assertion that states that all received invoices are eventually accepted (that may be too strong a property, I don't know, but you get the idea):

async def all_invoices_accepted(events):
    not_accepted = set()
    async for e in events:
        if e is an event indicating that an invoice is received:
            not_accepted.add(invoice_id obtained from e)
        if e is an event indicating that an invoice is accepted:
            not_accepted.remove(invoice_id from e)
    if not_accepted:
        raise AssertionError('some invoices were not accepted')

runner.add_api_assertion(all_invoices_accepted)
async with runner(...):
   ...