Skip to content
Snippets Groups Projects
Select Git revision
  • e62d1799809226f34652d260ab3d9dedcb80f01e
  • main default protected
  • dev protected
  • f-linkahead-rename
  • f-real-id
  • f-filesystem-import
  • f-filesystem-link
  • f-filesystem-directory
  • f-filesystem-core
  • f-filesystem-cleanup
  • f-filesystem-main
  • f-name
  • keep_changes
  • f-permission-checks-2
  • f-mysql8-tests
  • f-retrieve-history
  • t-distinct-parents
  • v8.1.0
  • v8.0.0
  • v7.0.2
  • v7.0.1
  • v7.0.0
  • v6.0.1
  • v6.0.0
  • v5.0.0
  • v4.1.0
  • v4.0.0
  • v3.0
  • v2.0.30
29 results

calcComplementUnion.sql

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    test.py 4.55 KiB
    #!/usr/bin/env python3
    # encoding: utf-8
    #
    # ** header v3.0
    # This file is a part of the CaosDB Project.
    #
    # Copyright (C) 2021 Indiscale GmbH <info@indiscale.com>
    #               2021 Henrik tom Wörden <h.tomwoerden@indiscale.com>
    #               2021 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
    #
    
    """
    module description
    """
    
    import argparse
    import sys
    from argparse import RawTextHelpFormatter
    from newcrawler import Crawler
    import caosdb as db
    from newcrawler.identifiable_adapters import CaosDBIdentifiableAdapter
    import pytest
    from caosadvancedtools.models.parser import parse_model_from_yaml
    
    import os
    
    
    def rfp(*pathcomponents):
        """
        Return full path.
        Shorthand convenience function.
        """
        return os.path.join(os.path.dirname(__file__), *pathcomponents)
    
    
    @pytest.fixture
    def clear_database():
        """First remove Records, then RecordTypes, then Properties, finally
        files. Since there may be no entities, execute all deletions
        without raising errors.
    
        """
        db.execute_query("FIND Record").delete(
            raise_exception_on_error=False)
        db.execute_query("FIND RecordType").delete(
            raise_exception_on_error=False)
        db.execute_query("FIND Property").delete(
            raise_exception_on_error=False)
        db.execute_query("FIND File").delete(
            raise_exception_on_error=False)
    
    
    @pytest.fixture
    def usemodel():
        model = parse_model_from_yaml(rfp("model.yml"))
        model.sync_data_model(noquestion=True)
    
        
    @pytest.fixture
    def ident(crawler):
        ident = CaosDBIdentifiableAdapter()
        crawler.identifiableAdapter = ident
    
        # TODO place this definition of identifiables elsewhere
        ident.register_identifiable(
            "Person", db.RecordType()
            .add_parent(name="Person")
            .add_property(name="first_name")
            .add_property(name="last_name"))
        ident.register_identifiable(
            "Measurement", db.RecordType()
            .add_parent(name="Measurement")
            .add_property(name="identifier")
            .add_property(name="date")
            .add_property(name="project"))
        ident.register_identifiable(
            "Project", db.RecordType()
            .add_parent(name="Project")
            .add_property(name="date")
            .add_property(name="identifier"))
        return ident
    
    
    @pytest.fixture
    def crawler(ident):
        crawler = Crawler(debug=True, identifiableAdapter=ident)
        crawler.crawl_directory(rfp("..", "unittests", "test_directories", "examples_article"),
                                rfp("..", "unittests", "scifolder_cfood.yml"))
        return crawler
    
    
    def test_single_insertion(clear_database, usemodel, crawler):
        ins, ups = crawler.synchronize()
        assert len(ins) == 18
        assert len(ups) == 0
    
        
    def test_multiple_insertions(clear_database, usemodel, ident, crawler):
        ins, ups = crawler.synchronize()
    
        # Do a second run on the same data, there should be no changes:
        crawler = Crawler(debug=True, identifiableAdapter=ident)
        crawler.crawl_directory(rfp("..", "unittests", "test_directories", "examples_article"),
                                rfp("..", "unittests", "scifolder_cfood.yml"))
        ins, ups = crawler.synchronize()
        assert len(ins) == 0
        assert len(ups) == 0
    
        
    def test_identifiable_update(clear_database, usemodel, ident, crawler):
        ins, ups = crawler.synchronize()
    
        # Do a second run on the same data with a change in one
        # of the identifiables:
        crawler = Crawler(debug=True, identifiableAdapter=ident)
        crawler.crawl_directory(rfp("..", "unittests", "test_directories", "examples_article"),
                                rfp("..", "unittests", "scifolder_cfood.yml"))
        l = crawler.updateList
        for record in l:
            if record.parents[0].name == "Measurement":
                # maybe a bit weird, but add an email address to a measurement
                record.add_property(name="email", value="testperson@testaccount.test")
                break
        ins, ups = crawler.synchronize()
        breakpoint()
        assert len(ins) == 0
        assert len(ups) == 1