Select Git revision
move_files.py
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