Source code for ewoksid13.tests.test_script_integration

import os

import pytest
import yaml

from ewoksid13.scripts.integration import _main_pyfai_integration_from_template
from ewoksid13.scripts.resources import TEMPLATE_INTEGRATION

pytestmark = pytest.mark.skip(reason="Waiting for PR ewoksdata: !173")




@pytest.mark.parametrize(
    "submit_parameters",
    [
        {"submit": False},
        {"submit": True, "local_execution": True},
        {"submit": True, "local_execution": False},
    ],
    ids=["dry_run", "local_execution", "celery"],
)
def test_script_integration_dataset_template(
    bliss_proposal, pyfai_config, tmp_path, submit_parameters
):
    submit_parameters = dict(submit_parameters)

    # Celery execution requires a live worker: beacon_host and queue must be
    # provided through the environment, otherwise the test is skipped.
    if submit_parameters.get("submit") and not submit_parameters.get(
        "local_execution", False
    ):
        queue = os.environ.get("EWOKS_CELERY_QUEUE")
        if not queue:
            pytest.skip(
                "Celery execution requires the BEACON_HOST and EWOKS_CELERY_QUEUE "
                "environment variables to be set."
            )
        submit_parameters["queue"] = queue

    bliss_filename = bliss_proposal["filenames_datasets"][0]

    with open(TEMPLATE_INTEGRATION, "r") as f:
        template = yaml.safe_load(f)

    output_root_folder = tmp_path / "PROCESSED_DATA"
    template["bliss_filenames"] = [bliss_filename]
    template["pyfai_configurations"] = [{"filename": pyfai_config}]
    template["output_root_folder"] = str(output_root_folder)
    template["submit_parameters"].update(submit_parameters)

    template_filename = tmp_path / "integration.yaml"
    with open(template_filename, "w") as f:
        yaml.safe_dump(template, f)

    _main_pyfai_integration_from_template(filename=str(template_filename))

    ewoks_files = list(output_root_folder.rglob("*_ewoks.json"))
    assert ewoks_files, "No ewoks workflow file was generated"