Repository: Avaiga/taipy
File: tests/core/notification/test_events_published.py (Python)

Code before:
    task_config = Config.configure_task("the_task", identity, input=input_config, output=output_config)
    sc_config = Config.configure_scenario(
        "the_scenario", task_configs=[task_config], frequency=Frequency.DAILY, sequences={"the_seq": [task_config]}
    )
    scenario = tp.create_scenario(sc_config)
    register_id_0, register_queue_0 = Notifier.register()
    all_evts = RecordingConsumer(register_id_0, register_queue_0)
    all_evts.start()

    # Write input manually trigger 5 data node update events
    # for last_edit_date, editor_id, editor_expiration_date, edit_in_progress and edits
    scenario.the_input.write("test")
    snapshot = all_evts.capture()
    assert len(snapshot.collected_events) == 5
    assert snapshot.entity_type_collected.get(EventEntityType.DATA_NODE, 0) == 5
    assert snapshot.operation_collected.get(EventOperation.UPDATE, 0) == 5
    all_evts.stop()


def test_events_published_for_scenario_submission():
    _events_published_for_scenario_submission()

def test_events_published_for_scenario_submission_standalone():
    _events_published_for_scenario_submission(standalone=True)

def _events_published_for_scenario_submission(standalone=False):
    if standalone:
        Config.configure_job_executions(mode="standalone", max_nb_of_workers=2)
    input_config = Config.configure_data_node("the_input")
    output_config = Config.configure_data_node("the_output")
    task_config = Config.configure_task("the_task", identity, input=input_config, output=output_config)
    sc_config = Config.configure_scenario(
        "the_scenario", task_configs=[task_config], frequency=Frequency.DAILY, sequences={"the_seq": [task_config]}
    )
    scenario = tp.create_scenario(sc_config)
    scenario.the_input.write("test")
    register_id_0, register_queue_0 = Notifier.register()
    all_evts = RecordingConsumer(register_id_0, register_queue_0)
    all_evts.start()
    # Submit a scenario triggers:
    # 1 scenario submission event
    # 7 dn update events (for last_edit_date, editor_id(x2), editor_expiration_date(x2) and edit_in_progress(x2))
    # 1 job creation event
    # 3 job update events (for status: PENDING, RUNNING and COMPLETED)
    # 1 submission creation event
    # 1 submission update event for jobs
    # 3 submission update events (for status: PENDING, RUNNING and COMPLETED)
    # 1 submission update event for is_completed
    if standalone:
        Orchestrator().run()
        scenario.submit(wait=True)

Review comment (suggestion):
Using the `@pytest.mark.parametrize(mode, ["development", "standalone"])` here would be cleaner.

However, this seems ok as well.

Code after:
    input_config = Config.configure_data_node("the_input")
    output_config = Config.configure_data_node("the_output")
    task_config = Config.configure_task("the_task", identity, input=input_config, output=output_config)
    sc_config = Config.configure_scenario(
        "the_scenario", task_configs=[task_config], frequency=Frequency.DAILY, sequences={"the_seq": [task_config]}
    )
    scenario = tp.create_scenario(sc_config)
    register_id_0, register_queue_0 = Notifier.register()
    all_evts = RecordingConsumer(register_id_0, register_queue_0)
    all_evts.start()

    # Write input manually trigger 5 data node update events
    # for last_edit_date, editor_id, editor_expiration_date, edit_in_progress and edits
    scenario.the_input.write("test")
    snapshot = all_evts.capture()
    assert len(snapshot.collected_events) == 5
    assert snapshot.entity_type_collected.get(EventEntityType.DATA_NODE, 0) == 5
    assert snapshot.operation_collected.get(EventOperation.UPDATE, 0) == 5
    all_evts.stop()

@pytest.mark.parametrize("standalone", [False, True])
def test_events_published_for_scenario_submission(standalone):
    if standalone:
        Config.configure_job_executions(mode="standalone", max_nb_of_workers=2)
    input_config = Config.configure_data_node("the_input")
    output_config = Config.configure_data_node("the_output")
    task_config = Config.configure_task("the_task", identity, input=input_config, output=output_config)
    sc_config = Config.configure_scenario(
        "the_scenario", task_configs=[task_config], frequency=Frequency.DAILY, sequences={"the_seq": [task_config]}
    )
    scenario = tp.create_scenario(sc_config)
    scenario.the_input.write("test")
    register_id_0, register_queue_0 = Notifier.register()
    all_evts = RecordingConsumer(register_id_0, register_queue_0)
    all_evts.start()
    # Submit a scenario triggers:
    # 1 scenario submission event
    # 7 dn update events (for last_edit_date, editor_id(x2), editor_expiration_date(x2) and edit_in_progress(x2))
    # 1 job creation event
    # 3 job update events (for status: PENDING, RUNNING and COMPLETED)
    # 1 submission creation event
    # 1 submission update event for jobs
    # 3 submission update events (for status: PENDING, RUNNING and COMPLETED)
    # 1 submission update event for is_completed
    if standalone:
        Orchestrator().run()
        scenario.submit(wait=True)
    else:
        scenario.submit()
    snapshot = all_evts.capture()
    assert len(snapshot.collected_events) == 18