Skip to content
Snippets Groups Projects
Select Git revision
  • 57d3b5dd2919371bf44bfe9f498fbea7f5a21bf0
  • main default protected
  • f-better-sss-bin-dir
  • dev protected
  • f-remove-dropoffbox
  • f-sss4grpc
  • f-refactor-compose
  • f-real-id
  • f-doip
  • f-filesystem-import
  • henrik-tmp
  • f-filesystem-link
  • f-filesystem-directory
  • f-filesystem-core
  • f-filesystem-cleanup
  • f-string-ids
  • f-filesystem-main
  • f-linkahead-rename-before
  • f-linkahead-rename
  • f-rename-li
  • f-experiment-trino
  • v0.13.0 protected
  • v0.12.3 protected
  • v0.12.2 protected
  • v0.12.1 protected
  • v0.12.0 protected
  • v0.11.0 protected
  • v0.10.0 protected
  • v0.9.0 protected
  • v0.8.1 protected
  • v0.8.0 protected
  • v0.7.3 protected
  • v0.7.2 protected
  • v0.7.1 protected
  • v0.6.0 protected
  • v0.5.0 protected
  • v0.4.0 protected
  • v0.3.0 protected
  • working_sss protected
  • v0.1 protected
40 results

test_move_files.py

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    test_macros.py 13.90 KiB
    #!/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 caoscrawler.scanner import load_definition
    
    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()
            definition = 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:
      crawler-version: 0.3.1
      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 = """
    ---
    metadata:
      crawler-version: 0.3.1
    ---
    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:
      crawler-version: 0.3.1
      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"
    
    
    def test_macros_in_macros(register_macros, macro_store_reset):
        """
        Test that macros can be used in macro definitions.
        """
        cfood = _temp_file_load("""
    ---
    metadata:
      crawler-version: 0.3.1
      macros:
        - !defmacro
          name: one_macro
          params:
            a: 25
          definition:
            macro_sub_$a:
              b: $a
              another_param: 3
        - !defmacro
          name: test_macrodef
          params: {}
          definition:
            macro_top: !macro
              one_macro:
              - a: 17
              - {}
              - a: 98
              not_macro:
                a: 26
    ---
    extroot: !macro
        test_macrodef:
        """)
    
        assert "test_macro" not in cfood["extroot"]
        assert cfood["extroot"]["macro_top"]["not_macro"]["a"] == 26
        d = cfood["extroot"]["macro_top"]
        assert d["macro_sub_17"]["b"] == "17"
        assert d["macro_sub_17"]["another_param"] == 3
        assert d["macro_sub_25"]["b"] == "25"
        assert d["macro_sub_25"]["another_param"] == 3
        assert d["macro_sub_98"]["b"] == "98"
        assert d["macro_sub_98"]["another_param"] == 3
    
    
    @pytest.mark.xfail(
        reason="This is discussed in the following issue"
        "https://gitlab.com/caosdb/caosdb-crawler/-/issues/74."
    )
    def test_silent_overwrite(register_macros, macro_store_reset):
        cfood = _temp_file_load("""
    ---
    metadata:
      crawler-version: 0.3.1
      macros:
        - !defmacro
          name: one_macro
          params:
            a: 25
          definition:
            macro_sub:
              b: $a
              another_param: 3
        - !defmacro
          name: test_macrodef
          params: {}
          definition:
            macro_top: !macro
              one_macro:
              - a: 17
              - a: 98
    ---
    extroot: !macro
        test_macrodef:
        """)
    
        assert len(cfood["extroot"]["macro_top"]) == 2
    
    
    def test_circular_macro_definition(register_macros, macro_store_reset):
        """Test the (ab-)use of macros to create an infinite loop."""
        cfood = _temp_file_load("""
    ---
    metadata:
      crawler-version: 0.3.1
      macros:
        - !defmacro
          name: test_one
          params: {}
          definition: !macro
            test_two:
        - !defmacro
          name: test_two
          params: {}
          definition: !macro
            test_one:
        - !defmacro
          name: test_three
          params: {}
          definition: !macro
            test_two:
        - !defmacro
          name: test_four
          params: {}
          definition: !macro
            test_four:
    ---
    extroot: !macro
      test_one:
    extroot2: !macro
      test_three:
    extroot3: !macro
      test_four:
        """)
        # macros in macros can be used, but there are no circles; they stop at the first one.
        assert "test_one" not in cfood["extroot"]
        assert cfood["extroot"]["test_two"] is None
        assert "test_three" not in cfood["extroot2"]
        assert "test_one" not in cfood["extroot2"]
        assert cfood["extroot2"]["test_two"] is None
        # No recursion
        assert cfood["extroot3"]["test_four"] is None
    
    
    def test_use_macro_twice():
        """Test that the same macro can be used twice with different parameters in
        the same CFood element if the name depends on the parameters.
    
        """
    
        cfood = _temp_file_load("""
    ---
    metadata:
      crawler-version: 0.3.1
      macros:
        - !defmacro
          name: test_twice
          params:
            macro_name: default_name
            a: 4
          definition:
            $macro_name:
              something:
                a: $a
    ---
    extroot: !macro
      test_twice:
      - macro_name: once
      - macro_name: twice
        a: 5
      - {}
        """)
        for name in ["once", "twice", "default_name"]:
            assert name in cfood["extroot"]
        assert cfood["extroot"]["once"]["something"]["a"] == "4"
        assert cfood["extroot"]["twice"]["something"]["a"] == "5"
        assert cfood["extroot"]["default_name"]["something"]["a"] == "4"
        # Code sample to generate the expanded macro:
        # with open("expanded_test_macro.yaml", "w") as f:
        #     f.write(yaml.dump(cfood))
    
    
    def test_documentation_example_2():
    
        cfood = _temp_file_load("""
    ---
    metadata:
      crawler-version: 0.3.1
      macros:
      - !defmacro
        name: MarkdownFile
        params:
          name: null
          filename: null
        definition:
          ${name}_filename:
            type: SimpleFile
            match: $filename
            records:
              $name:
                parents:
                - MarkdownFile
                role: File
                path: ${name}_filename
                file: ${name}_filename
    ---
    ExperimentalData:
      type: Directory
      match: ExperimentalData
      subtree: !macro
        MarkdownFile:
        - name: README
          filename: ^README.md$
        """)
    
        # Code sample to generate the expanded macro:
        # with open("expanded_test_macro.yaml", "w") as f:
        #     f.write(yaml.dump(cfood))
    
    
    def test_documentation_example_1():
    
        cfood = _temp_file_load("""
    ---
    metadata:
      crawler-version: 0.3.1
      macros:
      - !defmacro
        name: SimulationDatasetFile
        params:
          match: null
          recordtype: null
          nodename: null
        definition:
          $nodename:
            match: $match
            type: SimpleFile
            records:
              File:
                parents:
                - $recordtype
                role: File
                path: $$$nodename
                file: $$$nodename
              Simulation:
                $recordtype: +$File
    ---
    SimulationData:
      type: Directory
      match: SimulationData
      subtree: !macro
        SimulationDatasetFile:
        - match: .*
          recordtype: DatasetFile
          nodename: Dataset
        """)
    
        # Code sample to generate the expanded macro:
        # with open("expanded_test_macro.yaml", "w") as f:
        #     f.write(yaml.dump(cfood))
    
    
    @pytest.mark.xfail(
        reason="Wait until this feature is implemented"
        "https://gitlab.com/caosdb/caosdb-crawler/-/issues/21."
    )
    def test_def_replacements():
        """Test that parameters in macro definitions can be used
        for defining subsequent parameters.
        """
    
        cfood = _temp_file_load("""
    ---
    metadata:
      crawler-version: 0.3.1
      macros:
        - !defmacro
          name: test_def_replacements
          params:
            macro_name: default_name
            z: $macro_name
            a: $macro_name
            v: $z
          definition:
            $macro_name:
              macro_name: $macro_name
              z: $z
              a: $a
              v: $v
    ---
    extroot: !macro
      test_def_replacements:
      - macro_name: once
      - macro_name: twice
        z: 5
      - {}
        """)
        assert cfood["extroot"]["once"]["z"] == "once"
        assert cfood["extroot"]["once"]["a"] == "once"
        assert cfood["extroot"]["once"]["v"] == "once"
        assert cfood["extroot"]["once"]["macro_name"] == "once"
        assert cfood["extroot"]["twice"]["z"] == "5"
        assert cfood["extroot"]["twice"]["a"] == "5"
        assert cfood["extroot"]["twice"]["v"] == "5"
        assert cfood["extroot"]["twice"]["macro_name"] == "twice"
        assert cfood["extroot"]["default_name"]["z"] == "default_name"
        assert cfood["extroot"]["default_name"]["a"] == "default_name"
        assert cfood["extroot"]["default_name"]["v"] == "default_name"
        assert cfood["extroot"]["default_name"]["macro_name"] == "default_name"
    
    
    def test_list_macro_application(register_macros, macro_store_reset):
        dat = yaml.load("""
    defs:
    - !defmacro
      name: test
      params:
        a: 2
      definition:
        expanded_$a:
          param: $a
    - !defmacro
      name: test2
      params:
        a: 2
      definition:
        expanded_${a}_test2:
          param: $a
    
    testnode:
      obl: !macro
        test:
        - a: 4
        - a: 2
        test2:
          a: 4
    """, Loader=yaml.SafeLoader)
        assert dat["testnode"]["obl"]["expanded_4"]["param"] == "4"
        assert dat["testnode"]["obl"]["expanded_2"]["param"] == "2"
        assert dat["testnode"]["obl"]["expanded_4_test2"]["param"] == "4"
    
    
    def test_variable_in_macro_definition(register_macros, macro_store_reset):
        dat = yaml.load("""
    defs:
    - !defmacro
      name: test
      params:
        a: 2
        b: $a
      definition:
        expanded_$a:
          param: $a
          param_b: $b
    
    testnode:
      obl: !macro
        test:
        - a: 4
        - a: 2
          b: 4
    """, Loader=yaml.SafeLoader)
        assert dat["testnode"]["obl"]["expanded_4"]["param"] == "4"
        assert dat["testnode"]["obl"]["expanded_4"]["param_b"] == "4"
        assert dat["testnode"]["obl"]["expanded_2"]["param"] == "2"
        assert dat["testnode"]["obl"]["expanded_2"]["param_b"] == "4"