test_scanner.py

#!/usr/bin/env python3
# encoding: utf-8
#
# This file is a part of the LinkAhead Project.
#
# Copyright (C) 2023,2024 Indiscale GmbH <info@indiscale.com>
# Copyright (C) 2023,2024 Henrik tom Wörden <h.tomwoerden@indiscale.com>
#               2021-2023 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
# Alexander Schlemmer <alexander.schlemmer@ds.mpg.de>
#
# 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/>.
#
"""
Unit test functions for the scanner.
"""

from functools import partial
from pathlib import Path
from tempfile import NamedTemporaryFile
from unittest.mock import MagicMock, Mock, patch

import linkahead as db
import pytest
import yaml
from caoscrawler.crawl import Crawler
from caoscrawler.debug_tree import DebugTree
from caoscrawler.scanner import (create_converter_registry, load_definition,
                                 scan_directory, scan_structure_elements,
                                 _load_definition_from_yaml_dict)
from caoscrawler.structure_elements import (DictElement, DictListElement,
                                            DictTextElement, File)
from pytest import raises

from utils import dircheckstr as dircheck_base

UNITTESTDIR = Path(__file__).parent

dircheckstr = partial(dircheck_base, UNITTESTDIR / "test_directories" / "examples_article")


def test_scan_structure_elements():
    tmpfi = NamedTemporaryFile(delete=False)
    with open(UNITTESTDIR / "example_datastructure.yml", "r") as f:
        data = yaml.load(f, Loader=yaml.SafeLoader)

    crawler_definition = load_definition(UNITTESTDIR / "example_cfood.yml")
    converter_registry = create_converter_registry(crawler_definition)
    recs = scan_structure_elements(DictElement(name="", value=data), crawler_definition,
                                   converter_registry)
    assert len(recs) == 4


def test_provenance_debug_data():
    # TODO rewrite the test to use a smaller example setup
    tmpfi = NamedTemporaryFile(delete=False)
    debug_tree = DebugTree()
    with open(UNITTESTDIR / "example_datastructure.yml", "r") as f:
        data = yaml.load(f, Loader=yaml.SafeLoader)

    crawler_definition = load_definition(UNITTESTDIR / "example_cfood.yml")
    converter_registry = create_converter_registry(crawler_definition)
    stuff = scan_structure_elements(DictElement(name="", value=data), crawler_definition,
                                    converter_registry, debug_tree=debug_tree)
    crawler = Crawler()
    crawler.save_debug_data(tmpfi.name, debug_tree)
    with open(tmpfi.name, "r") as f:
        provenance = yaml.load(f, Loader=yaml.SafeLoader)

    pr = provenance["provenance"]

    def check_key_count(prefix):
        return sum([1 for key in pr.keys() if key.startswith(prefix)])
    assert check_key_count("Ent") == 4


def test_record_structure_generation():
    # TODO create a test from this that tests scan_structure
    # the cfood should be minimal but cover typical scenarios (e.g. children)
    # add also a minimal test for scan_directory; it can be very basic since the only difference
    # to scan_structure is the kind of starting structure_element (check this statement)
    # The test should not check debug tree output but actual created records

    # TODO test creation of debug information in a separate test

    dbt = DebugTree()
    scan_directory(UNITTESTDIR / "test_directories" / "examples_article",
                   UNITTESTDIR / "scifolder_cfood.yml",
                   debug_tree=dbt)
    subd = dbt.debug_tree[dircheckstr("DataAnalysis")]
    subc = dbt.debug_metadata["copied"][dircheckstr("DataAnalysis")]
    assert len(subd) == 2
    # variables store on Data Analysis node of debug tree
    assert len(subd[0]) == 4
    # record store on Data Analysis node of debug tree
    assert len(subd[1]) == 0
    assert len(subc) == 2
    assert len(subc[0]) == 4
    assert len(subc[1]) == 0

    # The data analysis node creates one variable for the node itself:
    assert subd[0]["DataAnalysis"] == "examples_article/DataAnalysis"
    assert subc[0]["DataAnalysis"] is False

    subd = dbt.debug_tree[dircheckstr("DataAnalysis", "2020_climate-model-predict")]
    subc = dbt.debug_metadata["copied"][dircheckstr("DataAnalysis", "2020_climate-model-predict")]

    assert len(subd[1]) == 1
    assert len(subd[1]["Project"].get_parents()) == 1
    assert subd[1]["Project"].get_parents()[0].name == "Project"
    assert subd[1]["Project"].get_property("date").value == "2020"
    assert subd[1]["Project"].get_property(
        "identifier").value == "climate-model-predict"

    assert len(subd[0]) == 9
    assert subd[0]["date"] == "2020"
    assert subd[0]["identifier"] == "climate-model-predict"
    assert subd[0]["Project"].__class__ == db.Record

    assert subd[0]["DataAnalysis"] == "examples_article/DataAnalysis"
    assert subc[0]["DataAnalysis"] is True
    assert subd[0]["project_dir"] == "examples_article/DataAnalysis/2020_climate-model-predict"
    assert subc[0]["project_dir"] is False

    # Check the copy flags for the first level in the hierarchy:
    assert len(subc[0]) == 9
    assert len(subc[1]) == 1
    assert subc[1]["Project"] is False
    assert subc[0]["Project"] is False
    assert subc[0]["date"] is False
    assert subc[0]["identifier"] is False

    subd = dbt.debug_tree[dircheckstr("DataAnalysis",
                                      "2020_climate-model-predict",
                                      "2020-02-08_prediction-errors")]
    subc = dbt.debug_metadata["copied"][dircheckstr("DataAnalysis",
                                                    "2020_climate-model-predict",
                                                    "2020-02-08_prediction-errors")]
    assert len(subd[0]) == 12
    assert subd[0]["date"] == "2020-02-08"
    assert subd[0]["identifier"] == "prediction-errors"
    assert subd[0]["Project"].__class__ == db.Record
    assert subd[0]["Measurement"].__class__ == db.Record

    assert len(subd[1]) == 2

    assert len(subd[1]["Project"].get_parents()) == 1
    assert subd[1]["Project"].get_parents()[0].name == "Project"
    assert subd[1]["Project"].get_property("date").value == "2020"
    assert subd[1]["Project"].get_property(
        "identifier").value == "climate-model-predict"

    assert len(subd[1]["Measurement"].get_parents()) == 1
    assert subd[1]["Measurement"].get_parents()[0].name == "Measurement"
    assert subd[1]["Measurement"].get_property("date").value == "2020-02-08"
    assert subd[1]["Measurement"].get_property(
        "identifier").value == "prediction-errors"
    assert subd[1]["Measurement"].get_property("project").value != "$Project"
    assert subd[1]["Measurement"].get_property(
        "project").value.__class__ == db.Record
    assert subd[1]["Measurement"].get_property(
        "project").value == subd[0]["Project"]

    # Check the copy flags for the second level in the hierarchy:
    assert subc[1]["Project"] is True
    assert subc[0]["Project"] is True
    assert subc[1]["Measurement"] is False
    assert subc[0]["Measurement"] is False
    assert subc[0]["date"] is False
    assert subc[0]["identifier"] is False


def test_record_generation():
    """
    Test the correct list of returned records by the scanner using the
    scifolder example from the article.
    """

    records = scan_directory(UNITTESTDIR / "test_directories" / "examples_article",
                             UNITTESTDIR / "scifolder_cfood.yml")

    def parent_filter(parent_name):
        return [p for p in records if len(p.parents) == 1 and p.parents[0].name == parent_name]

    def check_properties(records, check_props, check_additional=True):
        records_found = [0 for r in check_props]
        for rec in records:
            rec_found = 0
            # Try each record to check
            for i, check_prop in enumerate(check_props):
                matches = True
                # Verify that all props are in the record and have the right value
                for pr in check_prop:
                    if rec.get_property(pr) is None:
                        matches = False
                        break
                    if check_prop[pr] is None:
                        if rec.get_property(pr).value is not None:
                            matches = False
                            break
                    else:
                        if rec.get_property(pr).value != check_prop[pr]:
                            matches = False
                            break
                if check_additional:
                    # Verify that there are no additional props in the record
                    for rpr in rec.properties:
                        if rpr.name not in check_prop:
                            matches = False
                            break
                if matches:
                    records_found[i] += 1
        return records_found

    # Check projects:
    # Ther are two projects in mixed categories: climate_model_predict and SpeedOfLight
    projects_found = check_properties(parent_filter("Project"), [
        {"identifier": "climate-model-predict", "date": "2020"},
        {"identifier": "SpeedOfLight", "date": "2020"}
    ])
    assert projects_found == [3, 2]

    measurements = parent_filter("Measurement")
    assert len(measurements) == 11
    measurements_found = check_properties(measurements, [
        {"identifier": "prediction-errors", "date": "2020-02-08"},
        {"identifier": "average-all-exp", "date": "2020-01-04"},
        {"identifier": "average-all-exp-corr", "date": "2020-01-05"},
        {"date": "1980-01-01", "identifier": None},
        {"date": "1990-01-01", "identifier": None},
        {"date": "2000-01-01", "identifier": None},
        {"date": "2010-01-01", "identifier": None},
        {"date": "2020-01-01", "identifier": "TimeOfFlight"},
        {"date": "2020-01-02", "identifier": "Cavity"},
        {"date": "2020-01-03", "identifier": None},
        {"date": "2020-02-01", "identifier": None},
    ], False)
    for f in measurements_found:
        assert f == 1

    persons = parent_filter("Person")
    check_props = [
        {"first_name": None, "last_name": "Author" + letter} for letter in
        ("A", "B", "C", "D", "E")]
    persons_found = check_properties(persons, check_props)
    for f in persons_found:
        assert f > 0


def test_variable_deletion_problems():
    records = scan_directory(UNITTESTDIR / "test_directories" / "example_variable_deletion",
                             UNITTESTDIR / "cfood_variable_deletion.yml")

    for record in records:
        if record.name == "Record from Data_1":
            assert record.get_property("var1").value == "bla"
            assert record.get_property("var2").value == "$test_2"
        elif record.name == "Record from Data_2":
            assert record.get_property("var1").value == "$test_1"
            assert record.get_property("var2").value == "test"
        else:
            raise RuntimeError("Wrong name")

    records = scan_directory(UNITTESTDIR / "test_directories" / "example_variable_deletion",
                             UNITTESTDIR / "cfood_variable_deletion2.yml")

    # For the following test the order of records is actually important:
    assert records[0].name == "Record from Data_1"
    assert records[1].name == "Record from Data_2"
    for record in records:
        if record.name == "Record from Data_1":
            assert record.get_property("var1").value == "bla"
            assert record.get_property("var2").value == "$test_2"
        elif record.name == "Record from Data_2":
            assert record.get_property("var1").value == "example_variable_deletion"
            assert record.get_property("var2").value == "test"
        else:
            raise RuntimeError("Wrong name")


def test_record_parents():
    """ Test the correct list of returned records by the scanner     """

    data = {
        'Experiments': {}
    }

    crawler_definition = load_definition(UNITTESTDIR / "test_parent_cfood.yml")
    converter_registry = create_converter_registry(crawler_definition)

    records = scan_structure_elements(DictElement(name="", value=data), crawler_definition,
                                      converter_registry)
    assert len(records) == 4
    for rec in records:
        if rec.name == 'e':
            assert rec.parents[0].name == 'Exp'  # default parent was overwritten
            assert len(rec.parents) == 1
        elif rec.name == 'c':
            assert rec.parents[0].name == 'Cap2'  # default parent was overwritten by second
            # converter
            assert len(rec.parents) == 1
        elif rec.name == 'p':
            assert rec.parents[0].name == 'Projekt'  # top level set parent was overwritten
            assert len(rec.parents) == 1
        elif rec.name == 's':
            assert rec.parents[0].name == 'Stuff'  # default parent stays if no parent is given on
            # lower levels
            assert len(rec.parents) == 1


def test_error_messages():
    data = {
        'Experiments': {}
    }

    broken_yaml = """
EmptyConverter:
    """
    broken_definition = _load_definition_from_yaml_dict(
        [yaml.load(broken_yaml, Loader=yaml.SafeLoader)])

    converter_registry = create_converter_registry(broken_definition)

    with pytest.raises(RuntimeError, match="Definition of converter \"EmptyConverter\" is empty"):
        scan_structure_elements(DictElement(name="", value=data), broken_definition, converter_registry)

    broken_yaml = """
Converter:
  type: DictElement
  records:
    TestRecord: "42"
    """

    broken_definition = _load_definition_from_yaml_dict(
        [yaml.load(broken_yaml, Loader=yaml.SafeLoader)])

    converter_registry = create_converter_registry(broken_definition)

    with pytest.raises(RuntimeError, match="dict expected, but found str: 42"):
      scan_structure_elements(DictElement(name="", value=data), broken_definition, converter_registry)