1  from pathlib import Path
 2  
 3  import pytest
 4  
 5  from sirocco.core import Workflow
 6  from sirocco.core._tasks.icon_task import IconTask
 7  from sirocco.vizgraph import VizGraph
 8  from sirocco.workgraph import AiidaWorkGraph
 9  
10  
11  def test_parse_config_file(config_paths, pprinter):
12      reference_str = config_paths["txt"].read_text()
13      test_str = pprinter.format(Workflow.from_config_file(config_paths["yml"]))
14      if test_str != reference_str:
15          new_path = Path(config_paths["txt"]).with_suffix(".new.txt")
16          new_path.write_text(test_str)
17          assert (
18              reference_str == test_str
19          ), f"Workflow graph doesn't match serialized data. New graph string dumped to {new_path}."
20  
21  
22  def test_vizgraph(config_paths):
23      VizGraph.from_config_file(config_paths["yml"]).draw(file_path=config_paths["svg"])
24  
25  
26  @pytest.mark.requires_icon
27  @pytest.mark.usefixtures("icon_filepath_executable", "icon_grid_simple_path")
28  def test_icon():
29      # Test is performed by fixtures
30      pass
31  
32  
33  # configs that are tested for running workgraph
34  @pytest.mark.slow
35  @pytest.mark.usefixtures("aiida_localhost")
36  @pytest.mark.usefixtures("config_case")
37  @pytest.mark.parametrize(
38      "config_case",
39      [
40          "small",
41          "parameters",
42      ],
43  )
44  def test_run_workgraph(config_paths):
45      """Tests end-to-end the parsing from file up to running the workgraph.
46  
47      Automatically uses the aiida_profile fixture to create a new profile. Note to debug the test with your profile
48      please run this in a separate file as the profile is deleted after test finishes.
49      """
50      core_workflow = Workflow.from_config_file(str(config_paths["yml"]))
51      aiida_workflow = AiidaWorkGraph(core_workflow)
52      output_node = aiida_workflow.run()
53      assert (
54          output_node.is_finished_ok
55      ), f"Not successful run. Got exit code {output_node.exit_code} with message {output_node.exit_message}."
56  
57  
58  # configs containing task using icon plugin
59  @pytest.mark.usefixtures("config_case")
60  @pytest.mark.parametrize(
61      "config_case",
62      ["large"],
63  )
64  def test_nml_mod(config_paths, tmp_path):
65      nml_refdir = config_paths["txt"].parent / "ICON_namelists"
66      wf = Workflow.from_config_file(config_paths["yml"])
67      # Create core mamelists
68      for task in wf.tasks:
69          if isinstance(task, IconTask):
70              task.dump_namelists(directory=tmp_path)
71      # Compare against reference
72      for nml in nml_refdir.glob("*"):
73          ref_nml = nml.read_text()
74          test_nml = (tmp_path / nml.name).read_text()
75          if test_nml != ref_nml:
76              new_path = nml.with_suffix(".new")
77              new_path.write_text(test_nml)
78              assert ref_nml == test_nml, f"Namelist {nml.name} differs between ref and test"