conftest.py
1 import logging 2 import pathlib 3 import subprocess 4 5 import pytest 6 import requests 7 8 from sirocco import pretty_print 9 from sirocco.core import _tasks as core_tasks 10 from sirocco.core import workflow 11 from sirocco.parsing import yaml_data_models as models 12 13 pytest_plugins = ["aiida.tools.pytest_fixtures"] 14 15 LOGGER = logging.getLogger(__name__) 16 17 18 class DownloadError(RuntimeError): 19 def __init__(self, url: str, response: requests.Response): 20 super().__init__(f"Failed downloading file {url} , exited with response {response}") 21 22 23 def download_file(url: str, file_path: pathlib.Path): 24 response = requests.get(url) 25 if not response.ok: 26 raise DownloadError(url, response) 27 28 file_path.write_bytes(response.content) 29 30 31 @pytest.fixture(scope="module") 32 def icon_grid_simple_path(pytestconfig): 33 url = "https://github.com/agoscinski/icon-testfiles/raw/refs/heads/main/icon_grid_0013_R02B04_R.nc" 34 filename = "icon_grid_simple.nc" 35 cache_dir = pytestconfig.cache.mkdir("downloaded_files") 36 icon_grid_path = cache_dir / filename 37 38 # Check if the file is already cached 39 if icon_grid_path.exists(): 40 LOGGER.info("Found icon grid in cache, reusing it.") 41 else: 42 # File is not cached, download and save it 43 LOGGER.info("Downloading and caching icon grid.") 44 download_file(url, icon_grid_path) 45 46 return icon_grid_path 47 48 49 @pytest.fixture 50 def icon_filepath_executable() -> str: 51 which_icon = subprocess.run(["which", "icon"], capture_output=True, check=False) 52 if which_icon.returncode: 53 msg = "Could not find icon executable." 54 raise FileNotFoundError(msg) 55 56 return which_icon.stdout.decode().strip() 57 58 59 @pytest.fixture(scope="session") 60 def minimal_config() -> models.ConfigWorkflow: 61 return models.ConfigWorkflow( 62 name="minimal", 63 rootdir=pathlib.Path("minimal"), 64 cycles=[models.ConfigCycle(name="minimal", tasks=[models.ConfigCycleTask(name="some_task")])], 65 tasks=[models.ConfigShellTask(name="some_task", command="some_command")], 66 data=models.ConfigData( 67 available=[models.ConfigAvailableData(name="foo", type=models.DataType.FILE, src=pathlib.Path("foo.txt"))], 68 generated=[models.ConfigGeneratedData(name="bar", type=models.DataType.DIR, src=pathlib.Path("bar"))], 69 ), 70 parameters={}, 71 ) 72 73 74 @pytest.fixture(scope="session") 75 def minimal_invert_task_io_config() -> models.ConfigWorkflow: 76 return models.ConfigWorkflow( 77 name="minimal", 78 rootdir=pathlib.Path("minimal"), 79 cycles=[ 80 models.ConfigCycle( 81 name="minimal", 82 tasks=[ 83 models.ConfigCycleTask( 84 name="task_b", 85 inputs=[models.ConfigCycleTaskInput(name="output_a", port="None")], 86 outputs=[models.ConfigCycleTaskOutput(name="output_b")], 87 ), 88 models.ConfigCycleTask( 89 name="task_a", 90 inputs=[models.ConfigCycleTaskInput(name="available", port="None")], 91 outputs=[models.ConfigCycleTaskOutput(name="output_a")], 92 ), 93 ], 94 ), 95 ], 96 tasks=[ 97 models.ConfigShellTask(name="task_a", command="command_a"), 98 models.ConfigShellTask(name="task_b", command="command_b"), 99 ], 100 data=models.ConfigData( 101 available=[ 102 models.ConfigAvailableData(name="available", type=models.DataType.FILE, src=pathlib.Path("foo.txt")) 103 ], 104 generated=[ 105 models.ConfigGeneratedData(name="output_a", type=models.DataType.DIR, src=pathlib.Path("bar")), 106 models.ConfigGeneratedData(name="output_b", type=models.DataType.DIR, src=pathlib.Path("bar")), 107 ], 108 ), 109 parameters={}, 110 ) 111 112 113 # configs that are tested for parsing 114 ALL_CONFIG_CASES = ["small", "parameters", "large"] 115 116 117 @pytest.fixture(params=ALL_CONFIG_CASES) 118 def config_case(request) -> str: 119 return request.param 120 121 122 @pytest.fixture 123 def pprinter() -> pretty_print.PrettyPrinter: 124 return pretty_print.PrettyPrinter() 125 126 127 def generate_config_paths(test_case: str): 128 return { 129 "yml": pathlib.Path(f"tests/cases/{test_case}/config/config.yml"), 130 "txt": pathlib.Path(f"tests/cases/{test_case}/data/config.txt"), 131 "svg": pathlib.Path(f"tests/cases/{test_case}/svg/config.svg"), 132 } 133 134 135 @pytest.fixture 136 def config_paths(config_case) -> dict[str, pathlib.Path]: 137 return generate_config_paths(config_case) 138 139 140 def pytest_addoption(parser): 141 parser.addoption("--reserialize", action="store_true", default=False) 142 143 144 def serialize_worklfow(config_paths: dict[str, pathlib.Path], workflow: workflow.Workflow) -> None: 145 config_paths["txt"].write_text(pretty_print.PrettyPrinter().format(workflow)) 146 147 148 def serialize_nml(config_paths: dict[str, pathlib.Path], workflow: workflow.Workflow) -> None: 149 nml_refdir = config_paths["txt"].parent / "ICON_namelists" 150 for task in workflow.tasks: 151 if isinstance(task, core_tasks.icon_task.IconTask): 152 task.dump_namelists(directory=nml_refdir) 153 154 155 def pytest_configure(config): 156 if config.getoption("reserialize"): 157 LOGGER.info("Regenerating serialized references") 158 for config_case in ALL_CONFIG_CASES: 159 config_paths = generate_config_paths(config_case) 160 wf = workflow.Workflow.from_config_file(str(config_paths["yml"])) 161 serialize_worklfow(config_paths=config_paths, workflow=wf) 162 serialize_nml(config_paths=config_paths, workflow=wf)