Source code for blissoda.demo.tests.confitest

import gc
import os

from ewoksjob.client.celery.utils import get_not_finished_futures

from ... import flint
from ...bliss_globals import current_session
from ...bliss_globals import setup_globals
from .. import BLISS_LOGS_DIR
from .. import testing




@testing.integration_fixture(autouse=True)
def cleanup_celery_jobs():
    yield
    futures = get_not_finished_futures()
    if futures:
        testing.print_message(
            f"{len(futures)} workflows are still running: abort them.", "warning"
        )
    for future in futures:
        _ = future.abort()
        try:
            _ = future.result(timeout=5)
        except Exception as ex:
            testing.print_message(f"Aborted workflow: {ex}", "warning")





@testing.integration_fixture
def gc_collect():
    while gc.collect():
        pass
    yield
    while gc.collect():
        pass





@testing.integration_fixture(autouse=True)
def debug_blissoda():
    setup_globals.debugon("blissoda.*")
    yield
    setup_globals.debugoff("blissoda.*")





@testing.integration_fixture
def debug_flint():
    keep = current_session.scan_display.extra_args

    log_file = os.path.join(BLISS_LOGS_DIR, "flint_itest.log")
    extra_args = ["--debug", "--log-file", log_file]

    try:
        flint.FlintClient().close()
    except RuntimeError:
        pass  # already closed

    current_session.scan_display.extra_args = extra_args

    yield

    current_session.scan_display.extra_args = keep

    try:
        flint.FlintClient().close()
    except RuntimeError:
        pass  # already closed