Select Git revision
-
Timm Fitschen authored
AGPLv3 Veröffentlichung gemäß Dienstanweisung vom 15. August 2018.
Timm Fitschen authoredAGPLv3 Veröffentlichung gemäß Dienstanweisung vom 15. August 2018.
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_file.py 11.16 KiB
# encoding: utf-8
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
# Copyright (C) 2019-2021 IndiScale GmbH <info@indiscale.com>
# Copyright (C) 2019 Daniel Hornung <d.hornung@indiscale.com>
# Copyright (C) 2020-2021 Timm Fitschen <t.fitschen@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
#
"""Created on 18.09.2013.
@author: Timm Fitschen
"""
import os
import shutil
from lxml import etree
from pytest import raises, mark
import caosdb as db
from caosdb import execute_query, get_config, get_connection
from caosdb.common import models
from caosdb.utils.checkFileSystemConsistency import runCheck
def setup_module():
teardown_function(None)
def setup_function(function):
teardown_function(function)
with open("test.dat", "w") as upload_file:
upload_file.write("hello world\n")
os.makedirs("testfolder/subfolder")
with open("testfolder/test1.dat", "w") as upload_file:
upload_file.write("hello world\n")
with open("testfolder/subfolder/test2.dat", "w") as upload_file:
upload_file.write("hello world\n")
def teardown_function(function):
d = execute_query("FIND ENTITY WHICH HAS AN ID >= 100")
if len(d) > 0:
d.delete()
try:
path = get_config().get("IntegrationTests",
"test_files.test_insert_files_in_dir.local") + "testfolder/"
shutil.rmtree(path)
except Exception as e:
print(e)
try:
shutil.rmtree("testfolder")
except Exception as e:
print(e)
try:
os.remove("test.dat")
except Exception as e:
print(e)
try:
os.remove("test2.dat")
except Exception as e:
print(e)
try:
os.remove("test.dat.tmp")
except Exception as e:
print(e)
def test_file_with_space():
file_ = models.File(name="TestFile",
description="Testfile Desc",
path="testfiles/test file with spaces.dat",
file="test.dat")
file_.insert()
qfile = execute_query("FIND FILE TestFile", unique=True)
assert qfile.id == file_.id
assert qfile.path == "/testfiles/test file with spaces.dat"
qfile.download("test2.dat")
def test_consistency_file_does_not_exist():
with open("test.dat", "w") as upload_file:
upload_file.write("hello world\n")
file_ = models.File(name="TestConsistency1",
description="Testfile Desc",
path="debug/test_file_storage_consistency",
file="test.dat")
file_.insert()
assert file_.id is not None
assert file_.is_valid()
c = runCheck(None, "/debug/")
assert c.messages["Info", 0] is not None
assert c.messages["Info",
0][0] == "File system below debug/ is consistent."
# when in debug mode, the server offers a special option
#
# '-c FILE_DOES_NOT_EXIST'
#
# which simulates a accidentially removed file.
c = runCheck(None, "-c FILE_DOES_NOT_EXIST")
assert c.messages["Error", 0] is not None
assert c.messages["Error",
0][0] == 'MISSING - DEFAULT:debug/test_file_storage_consistency'
with open(file_.download(target="test.dat.tmp"), "r") as d:
r = d.read()
assert r == "hello world\n"
c = runCheck(None, None)
assert c.messages["Info", 0] is not None
assert c.messages["Info", 0][0] == "File system is consistent."
def test_consistency_file_was_modified():
# insert new test file
with open("test.dat", "w") as upload_file:
upload_file.write("hello world\n")
file_ = models.File(name="TestConsistency1",
description="Testfile Desc",
path="debug/test_file_storage_consistency",
file="test.dat")
file_.insert()
assert file_.id is not None
assert file_.is_valid()
# run consistency check (no consistency warning)
c = runCheck(None, None)
assert c.messages["Info", 0] is not None
assert c.messages["Info", 0][0] == "File system is consistent."
# when in debug mode, the server offers a special option
#
# '-c FILE_WAS_MODIFIED'
#
# which simulates a modified file.
c = runCheck(None, "-c FILE_WAS_MODIFIED")
assert c.messages["Error", 0] is not None
assert c.messages["Error",
0][0] == 'CHANGED:HASH - DEFAULT:debug/test_file_storage_consistency'
# download file again and check if it is still the same (just to be
# sure that the server only simulated the consistency breach
with open(file_.download(target="test.dat.tmp"), "r") as d:
r = d.read()
assert r == "hello world\n"
# run a passing check again
c = runCheck(None, None)
assert c.messages["Info", 0] is not None
assert c.messages["Info", 0][0] == "File system is consistent."
def test_consistency_unknown_file():
c = runCheck(None, None)
assert c.messages["Info", 0] is not None
assert c.messages["Info", 0][0] == "File system is consistent."
# when in debug mode, the server offers a special option
#
# '-c UNKNOWN_FILE'
#
# which simulates an unknown file.
c = runCheck(None, "-c UNKNOWN_FILE")
assert c.messages["Warning", 0] is not None
assert c.messages["Warning",
0][0] == 'UNKNOWN - DEFAULT:test_unknown'
c = runCheck(None, None)
assert c.messages["Info", 0] is not None
assert c.messages["Info", 0][0] == "File system is consistent."
def test_auto_create_parent_dirs():
file_ = models.File(name="TestFile",
path="A/B/C/test.dat",
file="test.dat")
file_.insert()
body = get_connection().retrieve(
entity_uri_segments=[
"FileSystem",
""],
reconnect=True).read()
print(body)
xml = etree.fromstring(body)
assert len(xml.xpath('/Response/dir')) == 1
root = xml.xpath('/Response/dir')[0]
assert root.get("name") == "/"
assert root.get("path") == "/"
assert root.get("url")[-12:] == "/FileSystem/"
assert len(root.xpath('dir')) == 1
assert len(root.xpath('file')) == 0
dir_a = root.xpath('dir')[0]
assert dir_a.get("name") == "A"
assert dir_a.get("url")[-14:] == "/FileSystem/A/"
body = get_connection().retrieve(
entity_uri_segments=[
"FileSystem",
"A", ""],
reconnect=True).read()
xml = etree.fromstring(body)
assert len(xml.xpath('/Response/dir')) == 1
dir_a = xml.xpath('/Response/dir')[0]
assert dir_a.get("name") == "A"
assert dir_a.get("url")[-14:] == "/FileSystem/A/"
assert len(dir_a.xpath('dir')) == 1
assert len(dir_a.xpath('file')) == 0
dir_b = dir_a.xpath('dir')[0]
assert dir_b.get("name") == "B"
assert dir_b.get("url")[-16:] == "/FileSystem/A/B/"
body = get_connection().retrieve(
entity_uri_segments=[
"FileSystem",
"A", "B", ""],
reconnect=True).read()
xml = etree.fromstring(body)
assert len(xml.xpath('/Response/dir')) == 1
dir_b = xml.xpath('/Response/dir')[0]
assert len(dir_b.xpath('dir')) == 1
assert len(dir_b.xpath('file')) == 0
dir_c = dir_b.xpath('dir')[0]
assert dir_c.get("name") == "C"
assert dir_c.get("url")[-18:] == "/FileSystem/A/B/C/"
body = get_connection().retrieve(
entity_uri_segments=[
"FileSystem",
"A", "B", "C", ""],
reconnect=True).read()
xml = etree.fromstring(body)
assert len(xml.xpath('/Response/dir')) == 1
dir_c = xml.xpath('/Response/dir')[0]
assert len(dir_c.xpath('dir')) == 0
assert len(dir_c.xpath('file')) == 1
file_1 = dir_c.xpath("file")[0]
assert file_1.get("name") == "test.dat"
assert file_1.get("url")[-26:] == "/FileSystem/A/B/C/test.dat"
def test_consistency_file_was_modified_2():
# this test is different from test_consistency_file_was_modified in that
# the server does actually corrupt the file.
# insert new test file
with open("test.dat", "w") as upload_file:
upload_file.write("hello world\n")
file_ = models.File(name="TestConsistency3",
description="Testfile Desc",
path="debug/test_file_storage_consistency",
file="test.dat")
file_.insert()
checksum1 = file_.checksum.lower()
qfile = execute_query("FIND FILE TestConsistency3", unique=True)
assert qfile.id == file_.id
assert qfile.checksum.lower() == checksum1
assert file_.id is not None
assert file_.is_valid()
# run consistency check (no consistency warning)
c = runCheck(None, None)
assert c.messages["Info", 0] is not None
assert c.messages["Info", 0][0] == "File system is consistent."
# when in debug mode, the server offers a special option
#
# '-c FILE_WAS_MODIFIED_2'
#
# which simulates a modified file.
c = runCheck(None, "-c FILE_WAS_MODIFIED_2")
assert c.messages["Error", 0] is not None
assert c.messages["Error",
0][0] == 'CHANGED:HASH - DEFAULT:debug/test_file_storage_consistency'
# old checksum is still stored
qfile = execute_query("FIND FILE TestConsistency3", unique=True)
assert qfile.id == file_.id
assert qfile.checksum.lower() == checksum1
# download fails because the checksum changed.
with raises(db.exceptions.ConsistencyError) as exc:
file_.download(target="test.dat.tmp")
file_.download(target="test.dat.tmp", check_hash=False)
assert file_.checksum.lower() == checksum1
checksum2 = db.File._get_checksum("test.dat.tmp").lower()
file_.checksum = checksum2
assert checksum2 != checksum1
# old checksum is still stored
qfile = execute_query("FIND FILE TestConsistency3", unique=True)
assert qfile.id == file_.id
assert qfile.checksum.lower() == checksum1
# here the client may validate the new file
file_.file = None
file_.size = None
file_.update()
assert checksum2 == file_.checksum.lower()
# new checksum is stored
qfile = execute_query("FIND FILE TestConsistency3", unique=True)
assert qfile.id == file_.id
assert qfile.checksum.lower() == checksum2
# no ConsistencyError anymore
file_.download(target="test.dat.tmp")
assert checksum2 == file_.checksum.lower()
# new checksum is still stored
qfile = execute_query("FIND FILE TestConsistency3", unique=True)
assert qfile.id == file_.id
assert qfile.checksum.lower() == checksum2