Skip to content
Snippets Groups Projects
Commit 3ffa003d authored by Alexander Schlemmer's avatar Alexander Schlemmer
Browse files

TST: transformed integration test into a proper pytest

parent 0cfc88e3
No related branches found
No related tags found
1 merge request!53Release 0.1
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2020 Indiscale GmbH <info@indiscale.com>
# Copyright (C) 2020 Florian Spreckelsen <f.spreckelsen@indiscale.com>
#
# 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
#
"""Clear the database before and after the integration tests."""
import caosdb as db
def clear_all():
"""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)
if __name__ == "__main__":
clear_all()
#!/usr/bin/env python3
# encoding: utf-8
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2021 Henrik tom Wörden
# 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/>.
import caosdb as db
from caosadvancedtools.models.data_model import DataModel
from caosadvancedtools.models.parser import parse_model_from_yaml
def main():
model = parse_model_from_yaml("model.yml")
model.sync_data_model(noquestion=True)
if __name__ == "__main__":
main()
...@@ -32,9 +32,10 @@ import argparse ...@@ -32,9 +32,10 @@ import argparse
import sys import sys
from argparse import RawTextHelpFormatter from argparse import RawTextHelpFormatter
from newcrawler import Crawler from newcrawler import Crawler
from unittest.mock import Mock
import caosdb as db import caosdb as db
from newcrawler.identifiable_adapters import CaosDBIdentifiableAdapter from newcrawler.identifiable_adapters import CaosDBIdentifiableAdapter
import pytest
from caosadvancedtools.models.parser import parse_model_from_yaml
import os import os
...@@ -47,54 +48,95 @@ def rfp(*pathcomponents): ...@@ -47,54 +48,95 @@ def rfp(*pathcomponents):
return os.path.join(os.path.dirname(__file__), *pathcomponents) return os.path.join(os.path.dirname(__file__), *pathcomponents)
def main(args): @pytest.fixture
ident_adapt = CaosDBIdentifiableAdapter() 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 # TODO place this definition of identifiables elsewhere
ident_adapt.register_identifiable( ident.register_identifiable(
"Person", db.RecordType() "Person", db.RecordType()
.add_parent(name="Person") .add_parent(name="Person")
#.add_property(name="first_name") .add_property(name="first_name")
.add_property(name="last_name")) .add_property(name="last_name"))
ident_adapt.register_identifiable( ident.register_identifiable(
"Measurement", db.RecordType() "Measurement", db.RecordType()
.add_parent(name="Measurement") .add_parent(name="Measurement")
#.add_property(name="identifier") .add_property(name="identifier")
.add_property(name="date") .add_property(name="date")
.add_property(name="project")) .add_property(name="project"))
ident_adapt.register_identifiable( ident.register_identifiable(
"Project", db.RecordType() "Project", db.RecordType()
.add_parent(name="Project") .add_parent(name="Project")
.add_property(name="date") .add_property(name="date")
.add_property(name="identifier")) .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
crawler = Crawler(debug=True, identifiableAdapter=ident_adapt)
crawler.copy_attributes = Mock() def test_single_insertion(clear_database, usemodel, crawler):
crawler.crawl_directory(rfp("../unittests/test_directories", "examples_article"),
rfp("../unittests/scifolder_cfood.yml"))
ins, ups = crawler.synchronize() ins, ups = crawler.synchronize()
assert len(ins) == 18 assert len(ins) == 18
assert len(ups) == 0 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: # Do a second run on the same data, there should be no changes:
crawler = Crawler(debug=True, identifiableAdapter=ident_adapt) crawler = Crawler(debug=True, identifiableAdapter=ident)
crawler.copy_attributes = Mock() crawler.crawl_directory(rfp("..", "unittests", "test_directories", "examples_article"),
crawler.crawl_directory(rfp("../unittests/test_directories", "examples_article"), rfp("..", "unittests", "scifolder_cfood.yml"))
rfp("../unittests/scifolder_cfood.yml"))
ins, ups = crawler.synchronize() ins, ups = crawler.synchronize()
assert len(ins) == 0 assert len(ins) == 0
assert len(ups) == 0 assert len(ups) == 0
def parse_args(): def test_identifiable_update(clear_database, usemodel, ident, crawler):
parser = argparse.ArgumentParser(description=__doc__, ins, ups = crawler.synchronize()
formatter_class=RawTextHelpFormatter)
# parser.add_argument("path",
# help="the subtree of files below the given path will "
# "be considered. Use '/' for everything.")
return parser.parse_args()
if __name__ == "__main__": # Do a second run on the same data with a change in one
args = parse_args() # of the identifiables:
sys.exit(main(args)) 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
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment