Skip to content
Snippets Groups Projects
Select Git revision
  • bb3f86ee11d4e7e4713c1b037d7bca30e01f1d81
  • main default protected
  • f-structured-query-tree
  • f-remove-chown-check
  • 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
  • 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

move_files.py

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