test_macros.py

#!/usr/bin/env python3
# encoding: utf-8
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2022 Alexander Schlemmer
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
# ** end header
#

from caoscrawler.macros import defmacro_constructor, macro_constructor
from caoscrawler.macros.macro_yaml_object import macro_store
from caoscrawler.crawl import Crawler

from tempfile import NamedTemporaryFile

import yaml
import pytest


@pytest.fixture
def register_macros():
    yaml.SafeLoader.add_constructor("!defmacro", defmacro_constructor)
    yaml.SafeLoader.add_constructor("!macro", macro_constructor)


@pytest.fixture
def macro_store_reset():
    macro_store.clear()


def _temp_file_load(txt: str):
    """
    Create a temporary file with txt and load the crawler
    definition using load_definition from Crawler.
    """
    definition = None
    with NamedTemporaryFile() as f:
        f.write(txt.encode())
        f.flush()
        c = Crawler()
        definition = c.load_definition(f.name)
    return definition


def test_macros(register_macros, macro_store_reset):
    dat = yaml.load("""
defs:
- !defmacro
  name: test
  params:
    a: 2
    b: bla
    c: $variable
  definition:
    expanded_$b:
      blubb: ok$a
      $b: $c

testnode:
  obl: !macro
    test:
      a: 4
      b: yea
""", Loader=yaml.SafeLoader)
    assert dat["testnode"]["obl"]["expanded_yea"]["blubb"] == "ok4"
    assert dat["testnode"]["obl"]["expanded_yea"]["yea"] == "$variable"
    assert "expanded_bla" not in dat["testnode"]["obl"]
    assert "bla" not in dat["testnode"]["obl"]["expanded_yea"]


def test_macro_list_replacment(register_macros, macro_store_reset):
    dat = yaml.load("""
defs:
- !defmacro
  name: test
  params:
    a: 2
    b: bla
    c: $variable
  definition:
    expanded_$b:
      blubb:
      - ok$a
      - $b: $c

testnode:
  obl: !macro
    test:
      a: 4
      b: yea
""", Loader=yaml.SafeLoader)
    assert isinstance(dat["testnode"]["obl"]["expanded_yea"]["blubb"], list)
    assert len(dat["testnode"]["obl"]["expanded_yea"]["blubb"]) == 2
    assert dat["testnode"]["obl"]["expanded_yea"]["blubb"][0] == "ok4"
    assert dat["testnode"]["obl"]["expanded_yea"]["blubb"][1]["yea"] == "$variable"


def test_multi_macros(register_macros, macro_store_reset):
    dat = yaml.load("""
defs:
- !defmacro
  name: test_one
  params: {}
  definition:
    replaced1: ok
- !defmacro
  name: test_two
  params: {}
  definition:
    replaced2: ok
    replaced3: ok

testnode:
    obl: !macro
      test_one:
      test_two:
""", Loader=yaml.SafeLoader)
    assert dat["testnode"]["obl"]["replaced1"] == "ok"
    assert dat["testnode"]["obl"]["replaced2"] == "ok"
    assert dat["testnode"]["obl"]["replaced3"] == "ok"


def test_multi_macros_toplevel(register_macros, macro_store_reset):
    """
    See: https://gitlab.indiscale.com/caosdb/src/caosdb-crawler/-/issues/23
    """
    dat_loader = list(yaml.safe_load_all("""
---
metadata:
  macros:
    - !defmacro
      name: test_one
      params: {}
      definition:
        replaced1: ok
    - !defmacro
      name: test_two
      params: {}
      definition:
        replaced2: ok
        replaced3: ok
---
testnode: !macro
  test_one:
  test_two:
"""))
    assert len(dat_loader) == 2
    dat = dat_loader[1]
    assert dat["testnode"]["replaced1"] == "ok"
    assert dat["testnode"]["replaced2"] == "ok"
    assert dat["testnode"]["replaced3"] == "ok"


def test_load_definition(register_macros, macro_store_reset):
    txt = """
extroot:
  type: Directory
  match: extroot
  subtree:
    SimulationData:
      type: Directory
      match: SimulationData
  """
    # Check whether simple cfoods can be loaded:
    cfood = _temp_file_load(txt)
    assert cfood["extroot"]["subtree"]["SimulationData"]["match"] == "SimulationData"

    cfood = _temp_file_load("""
---
metadata:
  macros:
    - !defmacro
      name: test_one
      params: {}
      definition:
        replaced1: ok
    - !defmacro
      name: test_two
      params:
        match_name: null
      definition:
        type: Directory
        match: $match_name
---
extroot:
  type: Directory
  match: extroot
  subtree:
    SimulationData:
      type: Directory
      match: SimulationData
extroot2: !macro  # test top level macro
  test_one:
extroot3:
  subtree:
    SimulationData: !macro
      test_two:
        match_name: SimulationData
    """)
    assert cfood["extroot"]["subtree"]["SimulationData"]["match"] == "SimulationData"
    assert cfood["extroot2"]["replaced1"] == "ok"
    assert cfood["extroot3"]["subtree"]["SimulationData"]["match"] == "SimulationData"


@pytest.mark.xfail
def test_replace_arbitrary_objects(register_macros, macro_store_reset):
    """
    See: https://gitlab.indiscale.com/caosdb/src/caosdb-crawler/-/issues/24
    """
    dat = yaml.load("""
defs:
- !defmacro
  name: test
  params:
    b: 25
    testvar_list:
    - a
    - $b
    testvar_dict:
      t1: a
      t2: $b
  definition:
    replaced1:
      $b: ok
      c: $testvar_dict
      d: $testvar_list

testnode:
  obl: !macro
    test:
""", Loader=yaml.SafeLoader)
    print(yaml.dump(dat))
    assert dat["testnode"]["obl"]["replaced1"]["c"]["t1"] == "a"
    assert dat["testnode"]["obl"]["replaced1"]["c"]["t2"] == "25"
    assert dat["testnode"]["obl"]["replaced1"]["d"][0] == "a"
    assert dat["testnode"]["obl"]["replaced1"]["d"][1] == "25"