Skip to content
Snippets Groups Projects
Verified Commit 9bb1ac20 authored by Timm Fitschen's avatar Timm Fitschen
Browse files

TST: sss and insertFilesInDir

parent a4083e17
No related branches found
No related tags found
No related merge requests found
Show whitespace changes
Inline Side-by-side
tests/test_file.py
+ 46
− 35
View file @ 9bb1ac20
......@@ -119,7 +119,6 @@ def test_pickup_file():
@with_setup(setup, teardown)
@nottest
def test_pickup_folder():
try:
# pickup_folder
......@@ -335,6 +334,8 @@ def test_file7():
@with_setup(setup, teardown)
def test_consistency_file_was_modified():
try:
# insert new test file
upload_file = open("test.dat", "w")
upload_file.write("hello world\n")
upload_file.close()
......@@ -347,30 +348,37 @@ def test_consistency_file_was_modified():
assert_is_not_none(file_.id)
assert_true(file_.is_valid())
# run consistency check (no consistency warning)
c = runCheck(None, None)
assert_is_not_none(c.messages["Info", 0])
assert_equal(c.messages["Info", 0][0], "File system is consistent.")
# run consistency check with modified warning)
c = runCheck(None, "-c FILE_WAS_MODIFIED")
assert_is_not_none(c.messages["Error", 0])
assert_equal(
c.messages["Error", 0][0],
'debug/test_file_storage_consistency: File was modified.')
# download file again and check if it is still the same (just to be
# sure that the server only simulated the consistency breach
d = open(file_.download(target="test.dat.tmp"), "r")
r = d.read()
assert_equal(r, "hello world\n")
# run a passing check again
c = runCheck(None, None)
assert_is_not_none(c.messages["Info", 0])
assert_equal(c.messages["Info", 0][0], "File system is consistent.")
finally:
# clean up file record
try:
file_.delete()
except BaseException:
pass
# clean up local files
try:
d.close()
except BaseException:
......@@ -490,36 +498,37 @@ def test_insert_files_in_dir_error1():
assert_true(c.messages["Error", 0][0].startswith("Dir is not allowed"))
@nottest
@with_setup(setup, teardown)
def test_insert_files_in_dir_with_symlink():
path = get_config().get("EndToEndTests", "test_files.test_insert_files_in_dir.path")
path = get_config().get("IntegrationTests",
"test_files.test_insert_files_in_dir.local") + "testfolder/"
path_on_server = get_config().get("IntegrationTests",
"test_files.test_insert_files_in_dir.server") + "testfolder/"
try:
# =======================================================================
# os.makedirs(path)
# create file in a server-readable directory
os.makedirs(path)
path = os.path.realpath(path) + "/"
# os.makedirs(path + "subfolder/")
# upload_file = open(path + "subfolder/test.dat", "w")
# upload_file.write("hello world\n")
# upload_file.close()
#
# upload_file = open(path + "subfolder/test2.dat", "w")
# upload_file.write("hello world2\n")
# upload_file.close()
# =======================================================================
os.makedirs(path + "subfolder/")
os.symlink(path + "subfolder/", path + "linked_subfolder")
test_file = open(path + "subfolder/test2.dat", "w")
test_file.write("hello world2\n")
test_file.close()
# create a symlink to the directory where the test_file resides
os.symlink(path_on_server + "subfolder/", path + "linked_subfolder")
# call insertFilesInDir job
c = models.Container()
c.retrieve(
unique=False,
raise_exception_on_error=False,
flags={
"InsertFilesInDir": path +
"InsertFilesInDir": path_on_server +
"linked_subfolder"})
assert_is_not_none(c.messages["Warning", 6])
assert_equal(c.messages["Warning", 6][0],
"Directory or file is symbolic link: " + path +
"Directory or file is symbolic link: " + path_on_server +
"linked_subfolder")
c = models.Container()
......@@ -528,7 +537,7 @@ def test_insert_files_in_dir_with_symlink():
raise_exception_on_error=False,
flags={
"InsertFilesInDir": "--force-allow-symlinks " +
path +
path_on_server +
"linked_subfolder"})
assert_is_not_none(c.messages["Info", 0])
assert_equal(c.messages["Info", 0][0],
......@@ -542,7 +551,7 @@ def test_insert_files_in_dir_with_symlink():
raise_exception_on_error=False,
flags={
"InsertFilesInDir": "--force-allow-symlinks " +
path +
path_on_server +
"linked_subfolder"})
assert_is_not_none(c.messages["Info", 0])
assert_equal(d.messages["Info", 0][0],
......@@ -566,27 +575,29 @@ def test_insert_files_in_dir_with_symlink():
pass
@nottest
@with_setup(None, teardown)
def test_insert_files_in_dir():
path = get_config().get("EndToEndTests", "test_files.test_insert_files_in_dir.path")
path = get_config().get("IntegrationTests",
"test_files.test_insert_files_in_dir.local") + "testfolder/"
path_on_server = get_config().get("IntegrationTests",
"test_files.test_insert_files_in_dir.server") + "testfolder/"
try:
os.makedirs(path)
os.makedirs(path + "subfolder/")
upload_file = open(path + "subfolder/test.dat", "w")
upload_file.write("hello world\n")
upload_file.close()
test_file1 = open(path + "subfolder/test.dat", "w")
test_file1.write("hello world\n")
test_file1.close()
upload_file = open(path + "subfolder/test2.dat", "w")
upload_file.write("hello world2\n")
upload_file.close()
test_file2 = open(path + "subfolder/test2.dat", "w")
test_file2.write("hello world2\n")
test_file2.close()
c = models.Container()
c.retrieve(
unique=False,
raise_exception_on_error=False,
flags={
"InsertFilesInDir": path})
"InsertFilesInDir": path_on_server})
assert_is_not_none(c.messages["Info", 0])
assert_equal(c.messages["Info", 0][0],
"Files count in testfolder/: 2")
......@@ -600,7 +611,7 @@ def test_insert_files_in_dir():
unique=False,
raise_exception_on_error=False,
flags={
"InsertFilesInDir": path})
"InsertFilesInDir": path_on_server})
assert_is_not_none(c.messages["Info", 0])
assert_equal(d.messages["Info", 0][0],
"Files count in testfolder/: 2")
......@@ -611,16 +622,16 @@ def test_insert_files_in_dir():
#assert_equal(d[1].path, "/testfolder/subfolder/test2.dat")
# create a new file and call insertFilesInDir again
upload_file = open(path + "/test3.dat", "w")
upload_file.write("hello world3\n")
upload_file.close()
test_file3 = open(path + "/test3.dat", "w")
test_file3.write("hello world3\n")
test_file3.close()
e = models.Container()
e.retrieve(
unique=False,
raise_exception_on_error=False,
flags={
"InsertFilesInDir": path})
"InsertFilesInDir": path_on_server})
assert_is_not_none(e.messages["Info", 0])
assert_equal(e.messages["Info", 0][0],
"Files count in testfolder/: 3")
......@@ -636,7 +647,7 @@ def test_insert_files_in_dir():
unique=False,
raise_exception_on_error=False,
flags={
"InsertFilesInDir": path})
"InsertFilesInDir": path_on_server})
assert_is_not_none(f.messages["Info", 0])
assert_equal(f.messages["Info", 0][0],
"Files count in testfolder/: 3")
......
tests/test_permissions.py
+ 6
− 1
View file @ 9bb1ac20
......@@ -239,6 +239,8 @@ def test_query():
deny_permission(dan, "RETRIEVE:*")
switch_to_test_user()
# this fails if server is configured with
# QUERY_FILTER_ENTITIES_WITHOUT_RETRIEVE_PERMISSIONS = FALSE
with assert_raises(db.TransactionError) as cm:
db.execute_query(
"FIND TestExperiment WHICH HAS A TestConductor WHICH has a TestFirstName=Daniel",
......@@ -1250,7 +1252,10 @@ def test_check_entity_acl_roles():
reset = db.administration.get_server_property(
"CHECK_ENTITY_ACL_ROLES_MODE")
assert reset == "MUST"
if reset == "SHOULD":
db.administration.set_server_property(
"CHECK_ENTITY_ACL_ROLES_MODE", "MUST")
p = db.Property(name="TestP", datatype=db.TEXT,
description="test_check_entity_acl_roles").insert()
......
tests/test_server_side_scripting.py
+ 1
− 0
View file @ 9bb1ac20
......@@ -107,6 +107,7 @@ def test_call_err():
assert xml.xpath("/Response/script/stderr")[0].text == "err"
@mark.skip(reason="need to setup .pycaosdb.ini in home dirs of sss within docker")
def test_simple_sss():
form = dict()
form["call"] = "simple_script.py"
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment