Unverified Commit 60add46f authored by Jose Borreguero's avatar Jose Borreguero Committed by GitHub
Browse files

integration test for auto_reduction_ready (#247) 

* integration test for autoreduction_ready
* substitute string_handler with fixture caplog
* remove unused imports
parent 11592802

scripts/reduce_CG1D.py

+8 −10
Original line number Diff line number Diff line
@@ -5,6 +5,7 @@ from imars3d.backend.dataio.config import save_config 
from imars3d.backend import auto_reduction_ready

from imars3d.backend import load_template_config

from imars3d.backend import extract_info_from_path

from imars3d.backend.autoredux import logger as logger_autoredux

from imars3d.backend.workflow.engine import WorkflowEngineAuto, WorkflowEngineError, WorkflowEngineExitCodes



# standard imports
@@ -14,11 +15,13 @@ from pathlib import Path 
from typing import Union

import sys





# declare the conda environment for this to run in

CONDA_ENV = "imars3d-dev"

ERROR_GENERAL = 1  # if more errors, they could be turned into enum.Enum

logger = logging.getLogger("reduce_CG1D")

WORKFLOW_SUCCESS: int = WorkflowEngineExitCodes.SUCCESS.value

WORKFLOW_ERROR: int = WorkflowEngineExitCodes.SUCCESS.value



logger = logger_autoredux.getChild("reduce_CG1D")





def _validate_inputs(inputfile: Path, outputdir: Path) -> int:
@@ -63,11 +66,6 @@ def main(inputfile: Union[str, Path], outputdir: Union[str, Path]) -> int: 


    # convert the parameters to make the rest easier

    inputfile = Path(inputfile)

    try:

        assert inputfile.exists()

    except AssertionError:

        logger.error(f"Input file {str(inputfile)} not found")

        return ERROR_GENERAL

    outputdir = Path(outputdir)

    logging.info("INPUT:", inputfile)  # TODO remove

    logging.info("OUTPUT:", outputdir)  # TODO remove
@@ -91,7 +89,7 @@ def main(inputfile: Union[str, Path], outputdir: Union[str, Path]) -> int: 
    config_dict = load_template_config(config_path)



    # step 2: extract info from inputfile

    update_dict = extract_info_from_path(inputfile)

    update_dict = extract_info_from_path(str(inputfile))



    # step 3: update the dict and save dict to disk

    config_dict.update(update_dict)
@@ -108,9 +106,9 @@ def main(inputfile: Union[str, Path], outputdir: Union[str, Path]) -> int: 
    workflow = WorkflowEngineAuto(config_dict)

    try:

        workflow.run()

        return WorkflowEngineExitCodes.SUCCESS.value

        return WORKFLOW_SUCCESS

    except WorkflowEngineError:

        return WorkflowEngineExitCodes.ERROR_GENERAL.value

        return WORKFLOW_ERROR





if __name__ == "__main__":

src/imars3d/backend/autoredux.py

+4 −2
Original line number Diff line number Diff line
@@ -8,6 +8,7 @@ import re 
from pathlib import Path

from typing import Union



# parent logger for all loggers instantiated in the auto-reduction scripts

logger = logging.getLogger(__name__)



facility_position = -6
@@ -33,14 +34,15 @@ def auto_reduction_ready(data_file: Union[str, Path]) -> bool: 
    """

    logger.info("Checking if scan has completed...")

    non_radiograph = {"dark-field": ["df"], "open-beam": ["ob"]}

    # /SNS/CG1D/IPTS-XXXXX/raw/_IMAGING_TYPE_/

    # canonical data_file path: /SNS/CG1D/IPTS-XXXXX/raw/_IMAGING_TYPE_/

    match = re.search(r"/raw/([-\w]+)/", str(data_file))

    if match:

        image_type = match.groups()[0]

        if image_type in non_radiograph["dark-field"] + non_radiograph["open-beam"]:

            logger.info("Scan is incomplete.")

            logger.info("The input image is not a radiograph. It's assumed the scan is incomplete.")

            return False

    else:

        logger.error("non-canonical data file path")

        return False



    # TODO: check the scan-compete signal from either the TIFF file or the Nexus file or some other file or PV

src/imars3d/backend/dataio/config.py

+2 −1
Original line number Diff line number Diff line
@@ -2,11 +2,12 @@ 
"""Configuration file handler for the imars3d."""

import json

from pathlib import Path

from typing import Union





def save_config(

    config_dict: dict,

    filepath: str,

    filepath: Union[str, Path],

):

    """

    Save a config dict to a file.

tests/integration/autoreduction/test_autoreduction_ready.py

0 → 100644
+55 −0
Original line number Diff line number Diff line 
# package imports

from reduce_CG1D import main as main_CG1D

from reduce_CG1D import ERROR_GENERAL



# third-party imports

import pytest



# standard imports

from pathlib import Path

from unittest import mock

from unittest.mock import MagicMock





@pytest.mark.datarepo

@mock.patch("reduce_CG1D._find_template_config")

def test_auto_reduction_ready(mock__find_template_config: MagicMock, caplog, IRON_MAN_DIR: Path, tmp_path: Path):

    # Force _find_template_config() to throw error in order to stop execution of main_CG1D()

    # right after auto_reduction_ready() is called

    mock__find_template_config.side_effect = FileNotFoundError("Mocked _find_template_config raises")

    raw_dir = IRON_MAN_DIR.parent.parent



    # non-canonical data file path

    bad_path_tiff = tmp_path / "20191030_OB_0070_0625.tiff"

    open(bad_path_tiff, "w").write("")

    assert main_CG1D(bad_path_tiff, tmp_path) == ERROR_GENERAL

    assert "non-canonical data file path" in caplog.text



    # checking an open beam

    caplog.clear()

    ob_tiff = raw_dir / "ob" / "Oct29_2019" / "20191030_OB_0070_0625.tiff"

    assert main_CG1D(ob_tiff, tmp_path) == ERROR_GENERAL

    assert "The input image is not a radiograph." in caplog.text



    # checking a dark field

    caplog.clear()

    df_tiff = raw_dir / "df" / "Oct29_2019" / "20191030_DF_0070_0632.tiff"

    assert main_CG1D(df_tiff, tmp_path) == ERROR_GENERAL

    assert "The input image is not a radiograph." in caplog.text



    # TODO: the first_tiff should return a non-complete scan

    caplog.clear()

    first_tiff = IRON_MAN_DIR / "20191029_ironman_small_0070_000_000_0002.tiff"

    assert main_CG1D(first_tiff, tmp_path) == ERROR_GENERAL

    assert "Mocked _find_template_config raises" in caplog.text

    assert "Scan is complete." in caplog.text  # should be "Scan is not complete."



    # TODO: last_tiff should return a complete scan but we're not picking the complete-scan signal

    caplog.clear()

    last_tiff = IRON_MAN_DIR / "20191030_ironman_small_0070_360_760_0624.tiff"

    assert main_CG1D(last_tiff, tmp_path) == ERROR_GENERAL

    assert "Scan is complete." in caplog.text





if __name__ == "__main__":

    pytest.main([__file__])