diff --git a/.docker-base/Dockerfile b/.docker-base/Dockerfile
index 592922471aa83d5175b1fbc1d355be8b547966a1..3476e1193312f57e86778cd51422e7fc954e6e18 100644
--- a/.docker-base/Dockerfile
+++ b/.docker-base/Dockerfile
@@ -18,8 +18,3 @@ COPY wait-for-it.sh /opt/caosdb/wait-for-it.sh
WORKDIR /opt/caosdb
RUN mkdir -p /opt/caosdb/build_docker/
CMD /bin/bash
-
-# python client
-ADD https://gitlab.com/api/v4/projects/13656973/repository/branches/dev \
- pylib_version.json
-RUN pip3 install git+https://gitlab.com/caosdb/caosdb-pylib
diff --git a/.docker/docker-compose.yml b/.docker/docker-compose.yml
index 912fa9869898f91fe476c977f177e5aa2615bc70..efaa0b27ddc0827cb5db9e01552de53eaaf81246 100644
--- a/.docker/docker-compose.yml
+++ b/.docker/docker-compose.yml
@@ -23,15 +23,20 @@ services:
- type: volume
source: scripting
target: /opt/caosdb/git/caosdb-server/scripting
+ - type: volume
+ source: authtoken
+ target: /opt/caosdb/git/caosdb-server/authtoken
ports:
# - "from_outside:from_inside"
- "10443:10443"
- "10080:10080"
environment:
DEBUG: 1
+ CAOSDB_CONFIG_AUTHTOKEN_CONFIG: "conf/core/authtoken.example.yaml"
volumes:
scripting:
extroot:
+ authtoken:
networks:
caosnet:
driver: bridge
diff --git a/.docker/pycaosdb.ini b/.docker/pycaosdb.ini
deleted file mode 100644
index 29846d6f976db29f8bc05d29240a2b4a18fa6ec5..0000000000000000000000000000000000000000
--- a/.docker/pycaosdb.ini
+++ /dev/null
@@ -1,23 +0,0 @@
-[IntegrationTests]
-test_server_side_scripting.bin_dir=/scripting-bin/
-
-# location of the files from the pyinttest perspective
-test_files.test_insert_files_in_dir.local=/extroot/test_insert_files_in_dir/
-# location of the files from the caosdb_servers perspective
-test_files.test_insert_files_in_dir.server=/opt/caosdb/mnt/extroot/test_insert_files_in_dir/
-
-
-[Connection]
-url=https://caosdb-server:10443/
-username=admin
-cacert=/cert/caosdb.cert.pem
-#cacert=/etc/ssl/cert.pem
-debug=0
-
-passwordmethod=plain
-password=caosdb
-
-ssl_insecure=True
-timeout=500
-[Container]
-debug=0
diff --git a/.docker/run.sh b/.docker/run.sh
index 00c7618f87295fbe67c7c485a9204b46adb1a438..b0e1a716f28516b83043fb3fdb6594515a0bafd4 100755
--- a/.docker/run.sh
+++ b/.docker/run.sh
@@ -1,5 +1,5 @@
#!/bin/sh
-docker-compose -f tester.yml run tester
+docker-compose -f tester.yml run tester
rv=$?
echo $rv > result
diff --git a/.docker/tester.yml b/.docker/tester.yml
index 912446bf7e3c8a3b9fe5c64afd77af3b548abf00..83db879c6072bfdea7b3212c833116b96bb54d0c 100644
--- a/.docker/tester.yml
+++ b/.docker/tester.yml
@@ -14,9 +14,13 @@ services:
- type: volume
source: scripting
target: /scripting
+ - type: volume
+ source: authtoken
+ target: /authtoken
networks:
docker_caosnet:
external: true
volumes:
scripting:
extroot:
+ authtoken:
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index b3c51e8d8b235afe9ffa688cb1b4a6a8399e0e3c..f5adb2885b56d998e9b47fe48a4258290acbdb6f 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -21,6 +21,7 @@
variables:
CI_REGISTRY_IMAGE: $CI_REGISTRY/caosdb/caosdb-pyinttest/testenv:latest
CI_REGISTRY_IMAGE_BASE: $CI_REGISTRY/caosdb/caosdb-pyinttest/base:latest
+ DEPLOY_REF: dev
stages:
- setup
@@ -42,24 +43,27 @@ stages:
# | +-(caosdb-server)-------------------+ |
# | | | |
# | | /opt/caosdb | |
-# | .---->| + /git/caosdb-server/scripting/ | |
-# | | .--->| + /mnt/extroot | |
-# | | | .->| + /cert | |
-# | | | | | | |
-# | | | | +-----------------------------------+ |
-# | | | | |
-# | | | | filesystem: |
-# | | | *- /cert -----------. |
-# | | | | |
-# | | | volumes: | |
-# | . *-- extroot ------. | |
-# | *---- scripting --. | | |
-# | | | | |
-# | +-(caosdb-pyinttest)---+ | | | |
-# | | | | | | |
-# | | /scripting |<---* | | |
-# | | /extroot |<----* . |
-# | | /cert |<------* |
+# | .------->| + /git/caosdb-server/scripting/ | |
+# | | .----->| + /git/caosdb-server/authtoken/ | |
+# | | | .--->| + /mnt/extroot | |
+# | | | | .->| + /cert | |
+# | | | | | | | |
+# | | | | | +-----------------------------------+ |
+# | | | | | |
+# | | | | | filesystem: |
+# | | | | *--- /cert -----------. |
+# | | | | | |
+# | | | | volumes: | |
+# | | | *----- extroot ------. | |
+# | | *------- scripting --. | | |
+# | *--------- authtoken -. | | | |
+# | | | | | |
+# | +-(caosdb-pyinttest)---+ | | | | |
+# | | | | | | | |
+# | | /authtoken |<---* | | | |
+# | | /scripting |<----* | | |
+# | | /extroot |<------* | |
+# | | /cert |<--------* |
# | | | |
# | +----------------------+ |
# +---------------------------------------------------+
@@ -69,9 +73,8 @@ stages:
# pipeline and a certificate is created in there. The certificat is then
# available in mounted directories in the server and pyinttest containers.
#
-# Two additional volumes in the root docker are shared by the caosdb-server and
-# the caosdb-pyintest containers.
-# These volumes are inteded to be used for testing server-side scripting and
+# Additional volumes in the root docker are shared by the caosdb-server and the caosdb-pyintest
+# containers. These volumes are intended to be used for testing server-side scripting and
# file-system features.
#
services:
@@ -81,7 +84,9 @@ test:
tags: [docker]
stage: test
image: $CI_REGISTRY_IMAGE_BASE
+ needs: ["cert"]
script:
+ - ls /
- echo $F_BRANCH
- echo $CAOSDB_TAG
- echo $CI_COMMIT_REF_NAME
@@ -92,23 +97,27 @@ test:
- if ! echo "$F_BRANCH" | grep -c "^f-" ; then
F_BRANCH=dev;
fi
+
+ - echo $F_BRANCH
+ - echo $CI_COMMIT_REF_NAME
+ - echo $CI_REGISTRY_IMAGE
+
+ - docker login -u gitlab+deploy-token-ci-pull -p $TOKEN_CI_PULL $CI_REGISTRY_INDISCALE
- if [[ "$CAOSDB_TAG" == "" ]]; then
if echo "$F_BRANCH" | grep -c "^f-" ; then
- CAOSDB_TAG=dev_F_${F_BRANCH}-latest;
+ CAOSDB_TAG=${DEPLOY_REF}_F_${F_BRANCH}-latest;
+ docker pull $CI_REGISTRY_INDISCALE/caosdb/src/caosdb-deploy:$CAOSDB_TAG || CAOSDB_TAG=${DEPLOY_REF}-latest ;
else
- CAOSDB_TAG=dev-latest;
+ CAOSDB_TAG=${DEPLOY_REF}-latest;
fi;
fi
- # TODO default $CAOSDB_TAG to dev_F_$F_BRANCH-latest
- - echo $F_BRANCH
+ - docker pull $CI_REGISTRY_INDISCALE/caosdb/src/caosdb-deploy:$CAOSDB_TAG || CAOSDB_TAG=dev-latest ;
- echo $CAOSDB_TAG
- - echo $CI_COMMIT_REF_NAME
- time docker load < /image-cache/caosdb-pyint-testenv-${CI_COMMIT_REF_NAME}.tar || true
- time docker load < /image-cache/mariadb-${F_BRANCH}.tar || true
- time docker load < /image-cache/caosdb-${F_BRANCH}.tar || true
- docker login -u gitlab-ci-token -p $CI_JOB_TOKEN $CI_REGISTRY
- - docker login -u gitlab+deploy-token-ci-pull -p $TOKEN_CI_PULL $CI_REGISTRY_INDISCALE
- docker pull $CI_REGISTRY_IMAGE
- cd .docker
# here the server and the mysql backend docker are being started
@@ -134,6 +143,7 @@ build-testenv:
tags: [cached-dind]
image: docker:19.03
stage: setup
+ needs: []
script:
- df -h
- command -v wget
@@ -165,6 +175,7 @@ cert:
tags: [docker]
stage: cert
image: $CI_REGISTRY_IMAGE
+ needs: ["build-testenv"]
artifacts:
paths:
- .docker/cert/
@@ -175,6 +186,7 @@ cert:
style:
tags: [docker]
stage: style
+ needs: ["build-testenv"]
image: $CI_REGISTRY_IMAGE
script:
- autopep8 -r --diff --exit-code .
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 12e03fb33fbbffd2b9c88296384ad1ee90b6c45d..5ae2f4f080273dc0d9e9a95dd5aaf97123386cf8 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,7 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added (for new features)
+* Tests for deeply nested SELECT queries
- Tests for [#62](https://gitlab.com/caosdb/caosdb-server/-/issues/62)
+* Tests for One-time Authentication Tokens
+* Test for [caosdb-pylib#31](https://gitlab.com/caosdb/caosdb-pylib/-/issues/31)
+* Tests for [caosdb-server#62](https://gitlab.com/caosdb/caosdb-server/-/issues/62)
in caosdb-server-project, i.e., renaming of a RecordType that should
be reflected in properties with that RT as datatype.
@@ -27,7 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed (for any bug fixes)
-- Tests for NaN Double Values (see https://gitlab.com/caosdb/caosdb-server/issues/41)
+* Tests for NaN Double Values (see https://gitlab.com/caosdb/caosdb-server/issues/41)
* Tests for name queries. [caosdb-server#51](https://gitlab.com/caosdb/caosdb-server/-/issues/51)
### Security (in case of vulnerabilities)
diff --git a/resources/ok_anonymous b/resources/ok_anonymous
new file mode 100755
index 0000000000000000000000000000000000000000..fdc5d87618d9c6b229febd4e25ec7d7ec03347cb
--- /dev/null
+++ b/resources/ok_anonymous
@@ -0,0 +1,3 @@
+#!/bin/bash
+
+echo "ok_anonymous"
diff --git a/resources/simple_script.py b/resources/simple_script.py
deleted file mode 100755
index 71bd9c05b4e86133cc356e1c15359701642a9486..0000000000000000000000000000000000000000
--- a/resources/simple_script.py
+++ /dev/null
@@ -1,80 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: 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
-#
-# 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
-#
-"""server_side_script.py.
-
-An example which implements a minimal server-side script.
-
-1) This script expects to find a *.txt file in the .upload_files dir which is
-printed to stdout.
-
-2) It executes a "Count stars" query and prints the result to stdout.
-
-3) It will return with code 0 if everything is ok, or with any code that is
-specified with the commandline option --exit
-"""
-
-import sys
-from os import listdir
-from caosdb import configure_connection, execute_query
-
-
-# parse --auth-token option and configure connection
-CODE = 0
-QUERY = "COUNT stars"
-for arg in sys.argv:
- if arg.startswith("--auth-token="):
- auth_token = arg[13:]
- configure_connection(auth_token=auth_token)
- if arg.startswith("--exit="):
- CODE = int(arg[7:])
- if arg.startswith("--query="):
- QUERY = arg[8:]
-
-
-############################################################
-# 1 # find and print *.txt file ############################
-############################################################
-
-try:
- for fname in listdir(".upload_files"):
- if fname.endswith(".txt"):
- with open(".upload_files/{}".format(fname)) as f:
- print(f.read())
-except FileNotFoundError:
- pass
-
-
-############################################################
-# 2 # query "COUNT stars" ##################################
-############################################################
-
-RESULT = execute_query(QUERY)
-print(RESULT)
-
-############################################################
-# 3 ########################################################
-############################################################
-
-sys.exit(CODE)
diff --git a/tests/test_administration.py b/tests/test_administration.py
index 0d4f4901c084474c0d703f2100f8ca16a0b1ea45..c951307472bc5a5eb7f2cea3b6da12de921bc0cd 100644
--- a/tests/test_administration.py
+++ b/tests/test_administration.py
@@ -27,7 +27,7 @@
"""
from caosdb import (administration as admin, get_config)
-from nose.tools import assert_true, assert_equal, assert_is_not_none, with_setup, assert_raises
+from nose.tools import assert_true, assert_equal, assert_is_not_none, assert_raises
from caosdb.exceptions import (ClientErrorException,
HTTPAuthorizationException,
LoginFailedException,
@@ -99,12 +99,10 @@ def test_set_server_property():
assert admin.get_server_property("AUTH_OPTIONAL") == "FALSE"
-@with_setup(setup, teardown)
def test_insert_role_success():
- assert_true(admin._insert_role(name=test_role, description=test_role_desc))
+ assert admin._insert_role(name=test_role, description=test_role_desc)
-@with_setup(setup, teardown)
def test_insert_role_failure_permission():
switch_to_normal_user()
with raises(HTTPAuthorizationException) as cm:
@@ -112,7 +110,6 @@ def test_insert_role_failure_permission():
assert cm.value.msg == "You are not permitted to insert a new role."
-@with_setup(setup, teardown)
def test_insert_role_failure_name_duplicates():
test_insert_role_success()
with assert_raises(ClientErrorException) as cm:
@@ -122,7 +119,6 @@ def test_insert_role_failure_name_duplicates():
"Role name is already in use. Choose a different name.")
-@with_setup(setup, teardown)
def test_update_role_success():
test_insert_role_success()
assert_is_not_none(
@@ -132,7 +128,6 @@ def test_update_role_success():
"asdf"))
-@with_setup(setup, teardown)
def test_update_role_failure_permissions():
test_insert_role_success()
switch_to_normal_user()
@@ -141,20 +136,17 @@ def test_update_role_failure_permissions():
assert cm.value.msg == "You are not permitted to update this role."
-@with_setup(setup, teardown)
def test_update_role_failure_non_existing():
with raises(ResourceNotFoundException) as cm:
admin._update_role(name=test_role, description=test_role_desc + "asdf")
assert cm.value.msg == "Role does not exist."
-@with_setup(setup, teardown)
def test_delete_role_success():
test_insert_role_success()
assert_true(admin._delete_role(name=test_role))
-@with_setup(setup, teardown)
def test_delete_role_failure_permissions():
test_insert_role_success()
switch_to_normal_user()
@@ -163,21 +155,18 @@ def test_delete_role_failure_permissions():
assert cm.value.msg == "You are not permitted to delete this role."
-@with_setup(setup, teardown)
def test_delete_role_failure_non_existing():
with raises(ResourceNotFoundException) as cm:
admin._delete_role(name=test_role)
assert cm.value.msg == "Role does not exist."
-@with_setup(setup, teardown)
def test_retrieve_role_success():
test_insert_role_success()
r = admin._retrieve_role(test_role)
assert_is_not_none(r)
-@with_setup(setup, teardown)
def test_retrieve_role_failure_permission():
test_insert_role_success()
switch_to_normal_user()
@@ -186,14 +175,12 @@ def test_retrieve_role_failure_permission():
assert cm.value.msg == "You are not permitted to retrieve this role."
-@with_setup(setup, teardown)
def test_retrieve_role_failure_non_existing():
with raises(ResourceNotFoundException) as cm:
admin._retrieve_role(name=test_role)
assert cm.value.msg == "Role does not exist."
-@with_setup(setup, teardown)
def test_set_permissions_success():
test_insert_role_success()
assert_true(
@@ -205,7 +192,6 @@ def test_set_permissions_success():
"BLA:BLA:BLA")]))
-@with_setup(setup, teardown)
def test_set_permissions_failure_permissions():
test_insert_role_success()
switch_to_normal_user()
@@ -217,7 +203,6 @@ def test_set_permissions_failure_permissions():
assert cm.value.msg == "You are not permitted to set this role's permissions."
-@with_setup(setup, teardown)
def test_set_permissions_failure_non_existing():
with raises(ResourceNotFoundException) as cm:
admin._set_permissions(
@@ -227,7 +212,6 @@ def test_set_permissions_failure_non_existing():
assert cm.value.msg == "Role does not exist."
-@with_setup(setup, teardown)
def test_get_permissions_success():
test_set_permissions_success()
r = admin._get_permissions(role=test_role)
@@ -235,7 +219,6 @@ def test_get_permissions_success():
assert_is_not_none(r)
-@with_setup(setup, teardown)
def test_get_permissions_failure_permissions():
test_set_permissions_success()
switch_to_normal_user()
@@ -244,14 +227,12 @@ def test_get_permissions_failure_permissions():
assert cm.value.msg == "You are not permitted to retrieve this role's permissions."
-@with_setup(setup, teardown)
def test_get_permissions_failure_non_existing():
with raises(ResourceNotFoundException) as cm:
admin._get_permissions(role="non-existing-role")
assert cm.value.msg == "Role does not exist."
-@with_setup(setup, teardown)
def test_get_roles_success():
test_insert_role_success()
r = admin._get_roles(username=test_user)
@@ -259,7 +240,6 @@ def test_get_roles_success():
return r
-@with_setup(setup, teardown)
def test_get_roles_failure_permissions():
test_insert_role_success()
switch_to_normal_user()
@@ -268,14 +248,12 @@ def test_get_roles_failure_permissions():
assert cm.value.msg == "You are not permitted to retrieve this user's roles."
-@with_setup(setup, teardown)
def test_get_roles_failure_non_existing():
with raises(ResourceNotFoundException) as cm:
admin._get_roles(username="non-existing-user")
assert cm.value.msg == "User does not exist."
-@with_setup(setup, teardown)
def test_set_roles_success():
roles_old = test_get_roles_success()
roles = {test_role}
@@ -285,7 +263,6 @@ def test_set_roles_success():
assert_is_not_none(admin._set_roles(username=test_user, roles=roles_old))
-@with_setup(setup, teardown)
def test_set_roles_success_with_warning():
test_insert_role_success()
roles = {test_role}
@@ -294,7 +271,6 @@ def test_set_roles_success_with_warning():
assert_is_not_none(admin._set_roles(username=test_user, roles=[]))
-@with_setup(setup, teardown)
def test_set_roles_failure_permissions():
roles_old = test_get_roles_success()
roles = {test_role}
@@ -320,7 +296,6 @@ def test_set_roles_failure_non_existing_user():
assert cm.value.msg == "User does not exist."
-@with_setup(setup, teardown)
def test_insert_user_success():
admin._insert_user(
name=test_user + "2",
@@ -330,7 +305,6 @@ def test_insert_user_success():
entity=None)
-@with_setup(setup, teardown)
def test_insert_user_failure_permissions():
switch_to_normal_user()
with raises(HTTPAuthorizationException) as cm:
@@ -343,7 +317,6 @@ def test_insert_user_failure_permissions():
assert cm.value.msg == "You are not permitted to insert a new user."
-@with_setup(setup, teardown)
def test_insert_user_failure_name_in_use():
test_insert_user_success()
with assert_raises(ClientErrorException) as cm:
@@ -351,13 +324,11 @@ def test_insert_user_failure_name_in_use():
assert_equal(cm.exception.msg, "User name is already in use.")
-@with_setup(setup, teardown)
def test_delete_user_success():
test_insert_user_success()
assert_is_not_none(admin._delete_user(name=test_user + "2"))
-@with_setup(setup, teardown)
def test_delete_user_failure_permissions():
test_insert_user_success()
switch_to_normal_user()
@@ -366,14 +337,12 @@ def test_delete_user_failure_permissions():
assert cm.value.msg == "You are not permitted to delete this user."
-@with_setup(setup, teardown)
def test_delete_user_failure_non_existing():
with raises(ResourceNotFoundException) as cm:
admin._delete_user(name="non_existing_user")
assert cm.value.msg == "User does not exist."
-@with_setup(setup, teardown)
def test_update_user_success_status():
assert_is_not_none(
admin._insert_user(
@@ -391,7 +360,6 @@ def test_update_user_success_status():
entity=None)
-@with_setup(setup, teardown)
def test_update_user_success_email():
assert_is_not_none(
admin._insert_user(
@@ -409,7 +377,6 @@ def test_update_user_success_email():
entity=None)
-@with_setup(setup, teardown)
def test_update_user_success_entity():
assert_is_not_none(
admin._insert_user(
@@ -422,7 +389,6 @@ def test_update_user_success_entity():
status=None, email=None, entity="21")
-@with_setup(setup, teardown)
def test_update_user_success_password():
assert_is_not_none(
admin._insert_user(
@@ -440,7 +406,6 @@ def test_update_user_success_password():
entity=None)
-@with_setup(setup, teardown)
def test_update_user_failure_permissions_status():
assert admin._insert_user(name=test_user + "2",
password="secret1P!",
@@ -458,7 +423,6 @@ def test_update_user_failure_permissions_status():
assert cm.value.msg == "You are not permitted to update this user."
-@with_setup(setup, teardown)
def test_update_user_failure_permissions_email():
assert admin._insert_user(name=test_user + "2",
password="secret1P!", status="ACTIVE",
@@ -475,7 +439,6 @@ def test_update_user_failure_permissions_email():
assert cm.value.msg == "You are not permitted to update this user."
-@with_setup(setup, teardown)
def test_update_user_failure_permissions_entity():
assert admin._insert_user(name=test_user + "2",
password="secret1P!", status="ACTIVE",
@@ -492,7 +455,6 @@ def test_update_user_failure_permissions_entity():
assert cm.value.msg == "You are not permitted to update this user."
-@with_setup(setup, teardown)
def test_update_user_failure_permissions_password():
assert admin._insert_user(name=test_user + "2",
password="secret1P!", status="ACTIVE",
@@ -509,7 +471,6 @@ def test_update_user_failure_permissions_password():
assert cm.value.msg == "You are not permitted to update this user."
-@with_setup(setup, teardown)
def test_update_user_failure_non_existing_user():
with raises(ResourceNotFoundException) as cm:
admin._update_user(
@@ -522,7 +483,6 @@ def test_update_user_failure_non_existing_user():
assert cm.value.msg == "User does not exist."
-@with_setup(setup, teardown)
def test_update_user_failure_non_existing_entity():
assert admin._insert_user(name=test_user + "2",
password="secret1P!", status="ACTIVE",
@@ -538,13 +498,11 @@ def test_update_user_failure_non_existing_entity():
assert cm.value.msg == "Entity does not exist."
-@with_setup(setup, teardown)
def test_retrieve_user_success():
test_insert_user_success()
assert_is_not_none(admin._retrieve_user(realm=None, name=test_user + "2"))
-@with_setup(setup, teardown)
def test_retrieve_user_failure_permissions():
test_insert_user_success()
switch_to_normal_user()
@@ -553,14 +511,12 @@ def test_retrieve_user_failure_permissions():
assert cm.value.msg == "You are not permitted to retrieve this user."
-@with_setup(setup, teardown)
def test_retrieve_user_failure_non_existing():
with raises(ResourceNotFoundException) as cm:
admin._retrieve_user(realm=None, name="non_existing")
assert cm.value.msg == "User does not exist."
-@with_setup(setup, teardown)
def test_login_with_inactive_user_failure():
assert_is_not_none(
admin._insert_user(
diff --git a/tests/test_affiliation.py b/tests/test_affiliation.py
index f51305cfde4644387727e29c5eab1cfa5cf4407a..df5d8979d02b38b7c06791019babace954283850 100644
--- a/tests/test_affiliation.py
+++ b/tests/test_affiliation.py
@@ -49,10 +49,9 @@ file_path = "testfile.dat"
def setup():
- try:
- db.execute_query("FIND Test*").delete()
- except Exception as e:
- print(e)
+ d = db.execute_query("FIND ENTITY WITH ID > 99")
+ if len(d) > 0:
+ d.delete()
db.RecordType(name=recty_name).insert()
db.Record(name=rec_name).add_parent(name=recty_name).insert()
db.File(name=file_name, file=file_path, path="testfile.dat").insert()
diff --git a/tests/test_authentication.py b/tests/test_authentication.py
index dc862c4e1c358376a039a8a50d0863fca2d5a112..ce2a095bab5f6b2d70afbea1a2a96af624c71389 100644
--- a/tests/test_authentication.py
+++ b/tests/test_authentication.py
@@ -5,6 +5,8 @@
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
+# Copyright (C) 2020 IndiScale GmbH <info@indiscale.com>
+# Copyright (C) 2020 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
@@ -21,40 +23,45 @@
#
# ** end header
#
-"""Created on 20.01.2015.
-
-@author: tf
-"""
import os
+import time
from sys import hexversion
from urllib.parse import urlparse
from http.client import HTTPSConnection
import ssl
-from subprocess import call, check_output
+from subprocess import call
from lxml import etree
-from pytest import skip
+from pytest import raises, mark
from caosdb.exceptions import LoginFailedException
-import caosdb as h
-from nose.tools import (assert_false, assert_true, assert_is_none,
- assert_raises, assert_equal, assert_is_not_none,
- nottest, with_setup)
-from caosdb.connection.connection import _Connection
+import caosdb as db
+from .test_server_side_scripting import request
+
+
+_USED_OTA_TOKEN = set()
def setup():
- try:
- h.execute_query("FIND Test*").delete()
- except Exception as e:
- print(e)
+ db.configure_connection()
+
+ # deactivate anonymous user
+ db.administration.set_server_property("AUTH_OPTIONAL", "FALSE")
+ d = db.execute_query("FIND Test*")
+ if len(d) > 0:
+ d.delete()
+
+
+def teardown():
+ setup()
+@mark.skipif(
+ not db.get_config().has_option("Connection", "password_method")
+ or not db.get_config().get("Connection", "password_method") == "pass",
+ reason="password_method is not pass")
def test_pass():
- if not h.get_config().has_option("Connection", "password_method") or not h.get_config(
- ).get("Connection", "password_method") == "pass":
- skip()
- assert call(["pass", h.get_config().get("Connection",
- "password_identifier")]) == 0
+ assert call(["pass", db.get_config().get("Connection",
+ "password_identifier")]) == 0
def test_https_support():
@@ -65,9 +72,9 @@ def test_https_support():
context.verify_mode = ssl.CERT_REQUIRED
if hasattr(context, "check_hostname"):
context.check_hostname = True
- context.load_verify_locations(h.get_config().get("Connection", "cacert"))
+ context.load_verify_locations(db.get_config().get("Connection", "cacert"))
- url = h.get_config().get("Connection", "url")
+ url = db.get_config().get("Connection", "url")
fullurl = urlparse(url)
http_con = HTTPSConnection(
@@ -79,27 +86,35 @@ def test_https_support():
def test_login_via_post_form_data_failure():
- with assert_raises(LoginFailedException) as cm:
- h.get_connection().post_form_data(
+ with raises(LoginFailedException):
+ db.get_connection().post_form_data(
"login", {
- "username": h.get_config().get("Connection", "username"),
+ "username": db.get_config().get("Connection", "username"),
"password": "wrongpassphrase"
})
+def test_anonymous_not_returning_auth_token():
+ # activate anonymous user
+ db.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
+
+ response = request(method="GET", headers={}, path="Entity")
+ assert response.getheader("Set-Cookie") is None # no auth token returned
+
+
def test_anonymous_setter():
- """ this test verifies that the "test_login_while_anonymous_is_active" is
+ """This test verifies that the "test_login_while_anonymous_is_active" is
effective."""
# activate anonymous user
- h.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
+ db.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
# connect without auth-token
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
context.verify_mode = ssl.CERT_REQUIRED
- context.load_verify_locations(h.get_config().get("Connection", "cacert"))
+ context.load_verify_locations(db.get_config().get("Connection", "cacert"))
- url = h.get_config().get("Connection", "url")
+ url = db.get_config().get("Connection", "url")
fullurl = urlparse(url)
http_con = HTTPSConnection(
@@ -113,15 +128,14 @@ def test_anonymous_setter():
assert xml.xpath("/Response/UserInfo/Roles/Role")[0].text == "anonymous"
-@with_setup(setup, setup)
def test_login_while_anonymous_is_active():
# activate anonymous user
- h.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
+ db.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
# logout
- h.get_connection()._logout()
+ db.get_connection()._logout()
- body = h.get_connection().retrieve(
+ body = db.get_connection().retrieve(
entity_uri_segments=["Entity"],
reconnect=True).read()
xml = etree.fromstring(body)
@@ -129,3 +143,176 @@ def test_login_while_anonymous_is_active():
# pylib did the login even though the anonymous user is active
assert xml.xpath(
"/Response/UserInfo/Roles/Role")[0].text == "administration"
+
+
+def test_authtoken_config():
+ assert db.administration.get_server_property(
+ "AUTHTOKEN_CONFIG") == "conf/core/authtoken.example.yaml"
+
+
+def get_one_time_token(testcase):
+ username = "anonymous"
+ realm = "OneTimeAuthenticationToken"
+ roles = ["administration"]
+ permissions = []
+ filename = db.get_config().get("IntegrationTests",
+ "test_authentication.{}".format(testcase))
+ assert os.path.isdir(os.path.split(filename)[0])
+ assert os.path.isfile(filename)
+ with open(filename, "r") as f:
+ auth_token = f.read()
+ while auth_token in _USED_OTA_TOKEN:
+ # wait until the server has renewed the token
+ time.sleep(1)
+ with open(filename, "r") as f:
+ auth_token = f.read()
+ else:
+ _USED_OTA_TOKEN.add(auth_token)
+
+ assert auth_token.startswith('["O","{re}","{us}",["{rol}"],{perm},'.format(
+ re=realm, us=username, rol=",".join(roles), perm=permissions))
+ return auth_token
+
+
+def test_one_time_token():
+ assert db.Info().user_info.roles == ["administration"]
+ assert db.Info().user_info.name == db.get_config().get("Connection", "username")
+
+ auth_token = get_one_time_token("admin_token_crud")
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+
+ assert db.Info().user_info.roles == ["administration"]
+ assert db.Info().user_info.name == "anonymous"
+ assert db.Info().user_info.realm == "OneTimeAuthenticationToken"
+
+ db.configure_connection()
+
+ assert db.Info().user_info.roles == ["administration"]
+ assert db.Info().user_info.name == db.get_config().get("Connection",
+ "username")
+
+
+def test_one_time_token_invalid():
+ auth_token = get_one_time_token("admin_token_crud")
+ auth_token = auth_token.replace("[]", '["permission"]')
+
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+ with raises(db.LoginFailedException) as lfe:
+ db.Info()
+ assert lfe.value.args[0] == (
+ "The authentication token is expired or you have been logged out otherwise. The auth_token "
+ "authenticator cannot log in again. You must provide a new authentication token.")
+
+ # also raises exception when anonymous is enabled
+ db.configure_connection()
+ db.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+ with raises(db.LoginFailedException) as lfe:
+ db.Info()
+ assert lfe.value.args[0] == (
+ "The authentication token is expired or you have been logged out otherwise. The auth_token "
+ "authenticator cannot log in again. You must provide a new authentication token.")
+
+
+def test_one_time_token_expired():
+ auth_token = get_one_time_token("admin_token_expired")
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+ with raises(db.LoginFailedException) as lfe:
+ db.Info()
+ assert lfe.value.args[0] == (
+ "The authentication token is expired or you have been logged out otherwise. The auth_token "
+ "authenticator cannot log in again. You must provide a new authentication token.")
+
+ # also raises exception when anonymous is enabled
+ db.configure_connection()
+ db.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+ with raises(db.LoginFailedException) as lfe:
+ db.Info()
+ assert lfe.value.args[0] == (
+ "The authentication token is expired or you have been logged out otherwise. The auth_token "
+ "authenticator cannot log in again. You must provide a new authentication token.")
+
+
+def test_one_time_token_3_attempts():
+ auth_token = get_one_time_token("admin_token_3_attempts")
+
+ # 1st
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+ assert db.get_connection()._authenticator.auth_token == auth_token
+ assert db.Info().user_info.roles == ["administration"]
+ assert db.get_connection()._authenticator.auth_token != auth_token
+ assert db.Info().user_info.name == "anonymous"
+ assert db.Info().user_info.realm == "OneTimeAuthenticationToken"
+
+ # 2nd
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+ assert db.get_connection()._authenticator.auth_token == auth_token
+ assert db.Info().user_info.roles == ["administration"]
+ assert db.get_connection()._authenticator.auth_token != auth_token
+ assert db.Info().user_info.name == "anonymous"
+ assert db.Info().user_info.realm == "OneTimeAuthenticationToken"
+
+ # 3rd
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+ assert db.get_connection()._authenticator.auth_token == auth_token
+ assert db.Info().user_info.roles == ["administration"]
+ assert db.get_connection()._authenticator.auth_token != auth_token
+ assert db.Info().user_info.name == "anonymous"
+ assert db.Info().user_info.realm == "OneTimeAuthenticationToken"
+
+ # 4th
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+ assert db.get_connection()._authenticator.auth_token == auth_token
+ with raises(db.LoginFailedException) as lfe:
+ db.Info()
+ assert lfe.value.args[0] == (
+ "The authentication token is expired or you have been logged out otherwise. The auth_token "
+ "authenticator cannot log in again. You must provide a new authentication token.")
+
+ db.configure_connection()
+ db.administration.set_server_property("AUTH_OPTIONAL", "TRUE")
+
+ # 5th attempt, also raises error when anonymous user is enabled
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+ assert db.get_connection()._authenticator.auth_token == auth_token
+ with raises(db.LoginFailedException) as lfe:
+ db.Info()
+ assert lfe.value.args[0] == (
+ "The authentication token is expired or you have been logged out otherwise. The auth_token "
+ "authenticator cannot log in again. You must provide a new authentication token.")
+
+
+def test_crud_with_one_time_token():
+ auth_token = get_one_time_token("admin_token_crud")
+ db.configure_connection(password_method="auth_token",
+ auth_token=auth_token)
+
+ # CREATE
+ rt = db.RecordType(name="TestRT")
+ rt.insert()
+ assert rt.id == db.execute_query("FIND TestRT", unique=True).id
+
+ # UPDATE
+ assert db.execute_query("FIND TestRT", unique=True).description is None
+ rt.description = "new desc"
+ rt.update()
+ assert rt.description == db.execute_query(
+ "FIND TestRT", unique=True).description
+
+ # RETRIEVE
+ rt.retrieve()
+ assert rt.id == db.execute_query("FIND TestRT", unique=True).id
+
+ rt.delete()
+ assert len(db.execute_query("FIND TestRT")) == 0
diff --git a/tests/test_datatype.py b/tests/test_datatype.py
index 3a276b04137a4be3f25e1d7f423116331e091dd7..a996004167bba6bb5bb0f12387b53381036851e5 100644
--- a/tests/test_datatype.py
+++ b/tests/test_datatype.py
@@ -26,8 +26,6 @@
@author: tf
"""
import caosdb as db
-# @UnresolvedImport
-from nose.tools import nottest, with_setup, assert_true, assert_equal
from pytest import raises
@@ -45,7 +43,6 @@ def teardown():
pass
-@with_setup(setup, teardown)
def test_override_with_non_existing_ref():
rt1 = db.RecordType("TestRecordType1").insert()
rt2 = db.RecordType("TestRecordType2").insert()
@@ -62,7 +59,6 @@ def test_override_with_non_existing_ref():
"Referenced entity does not exist.")
-@with_setup(setup, teardown)
def test_override_with_existing_ref():
rt1 = db.RecordType("TestRecordType1").insert()
rt2 = db.RecordType("TestRecordType2").insert()
@@ -75,10 +71,9 @@ def test_override_with_existing_ref():
datatype=rt2,
value="TestRecord").add_parent(rt1).insert()
- assert_true(rec2.is_valid())
+ assert rec2.is_valid() is True
-@with_setup(setup, teardown)
def test_reference_datatype_sequencial():
rt = db.RecordType(name="TestRT", description="TestRTDesc").insert()
p = db.Property(
@@ -86,22 +81,21 @@ def test_reference_datatype_sequencial():
description="RefOnTestRT",
datatype="TestRT").insert()
- assert_true(p.is_valid())
+ assert p.is_valid() is True
dt = db.execute_query("FIND TestProp", unique=True).datatype
- assert_equal(dt, "TestRT")
+ assert dt == "TestRT"
rt2 = db.RecordType(
"TestRT2",
description="TestRT2Desc").add_property(
name="TestProp",
value="TestRT").insert()
- assert_true(rt2.is_valid())
- assert_equal(rt2.get_property("TestProp").value, rt.id)
- assert_equal(rt2.get_property("TestProp").datatype, "TestRT")
+ assert rt2.is_valid() is True
+ assert rt2.get_property("TestProp").value == rt.id
+ assert rt2.get_property("TestProp").datatype == "TestRT"
-@with_setup(setup, teardown)
def test_reference_datatype_at_once():
rt = db.RecordType(name="TestRT")
rt2 = db.RecordType(name="TestRT2").add_property(name="TestProp")
@@ -113,10 +107,9 @@ def test_reference_datatype_at_once():
p = db.Property(name="TestProp", datatype="TestRT")
c = db.Container().extend([rt, rt2, rec, rec2, p]).insert()
- assert_true(c.is_valid())
+ assert c.is_valid() is True
-@with_setup(setup, teardown)
def test_generic_reference_success():
rt1 = db.RecordType(name="TestRT1").insert()
rt2 = db.RecordType(name="TestRT2").insert()
@@ -128,14 +121,13 @@ def test_generic_reference_success():
name="TestP1",
value=rec1.id).insert()
- assert_true(rt1.is_valid())
- assert_true(rt2.is_valid())
- assert_true(p.is_valid())
- assert_true(rec1.is_valid())
- assert_true(rec2.is_valid())
+ assert rt1.is_valid() is True
+ assert rt2.is_valid() is True
+ assert p.is_valid() is True
+ assert rec1.is_valid() is True
+ assert rec2.is_valid() is True
-@with_setup(setup, teardown)
def test_generic_reference_failure():
db.RecordType(name="TestRT2").insert()
db.Property(name="TestP1", datatype=db.REFERENCE).insert()
@@ -144,20 +136,16 @@ def test_generic_reference_failure():
name="TestRT2").add_property(
name="TestP1",
value="asdf")
- with raises(db.TransactionError):
- rec2.insert()
+ raises(db.TransactionError, rec2.insert)
-@with_setup(setup, teardown)
def test_unknown_datatype1():
p = db.Property(name="TestP", datatype="Non-Existing")
with raises(db.TransactionError) as te:
p.insert()
- assert_true(False)
assert te.value.get_errors()[0].msg == "Unknown datatype."
-@with_setup(setup, teardown)
def test_unknown_datatype2():
p = db.Property(name="TestP", datatype="12345687654334567")
with raises(db.TransactionError) as te:
@@ -165,7 +153,6 @@ def test_unknown_datatype2():
assert te.value.get_errors()[0].msg == "Unknown datatype."
-@with_setup(setup, teardown)
def test_unknown_datatype3():
p = db.Property(name="TestP", datatype="-134")
with raises(db.TransactionError) as te:
@@ -173,7 +160,6 @@ def test_unknown_datatype3():
assert te.value.get_errors()[0].msg == "Unknown datatype."
-@with_setup(setup, teardown)
def test_wrong_refid():
rt1 = db.RecordType(name="TestRT1").insert()
rt2 = db.RecordType(name="TestRT2").insert()
@@ -192,11 +178,31 @@ def test_wrong_refid():
value=rec2.id)
with raises(db.TransactionError):
rec3.insert()
- assert (rec3.get_property("TestP1").get_errors()[0].description ==
- 'Reference not qualified. The value of this Reference Property is to be a child of its data type.')
+ except db.TransactionError:
+ desc = ('Reference not qualified. The value of this Reference '
+ 'Property is to be a child of its data type.')
+ err = rec3.get_property("TestP1").get_errors()[0]
+ assert err.description == desc
rec4 = db.Record().add_parent(
name="TestRT3").add_property(
name="TestP1",
value=rec1.id).insert()
assert rec4.is_valid()
+
+
+def test_datatype_mismatch_in_response():
+ p = db.Property("TestDoubleProperty", datatype=db.DOUBLE).insert()
+
+ with raises(ValueError) as exc:
+ rt = db.RecordType().add_property(p, "not a double")
+ assert exc.value.args[0] == ("could not convert string to "
+ "float: 'not a double'")
+
+ # add property by name,
+ rt = db.RecordType("TestEntity").add_property(p.name, "not a double")
+
+ with raises(db.TransactionError) as exc:
+ # should not raise ValueError but transaction error.
+ rt.insert()
+ assert exc.value.get_errors()[0].msg == "Cannot parse value to double."
diff --git a/tests/test_properties.py b/tests/test_properties.py
new file mode 100644
index 0000000000000000000000000000000000000000..9219501f09a5ee5eff4b89a78f17f8e02b842151
--- /dev/null
+++ b/tests/test_properties.py
@@ -0,0 +1,63 @@
+# -*- coding: utf-8 -*-
+#
+# ** header v3.0
+# This file is a part of the CaosDB Project.
+#
+# Copyright (c) 2020 IndiScale GmbH <www.indiscale.com>
+# Copyright (c) 2020 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
+import pytest
+
+import caosdb as db
+from caosdb.common.utils import xml2str
+
+
+def setup_module():
+ d = db.execute_query("FIND Test*")
+ if len(d) > 0:
+ d.delete()
+
+
+def setup():
+ setup_module()
+
+
+def teardown():
+ setup_module()
+
+
+@pytest.mark.xfail(reason="""see pylib issue #31""")
+def test_add_properties_with_wrong_role():
+ p = db.Property(name="TestProperty1", datatype=db.TEXT).insert()
+ rt = db.RecordType(name="TestRT1").add_property("TestProperty1").insert()
+
+ wrong_role = db.Record(name="TestRT1").retrieve()
+ right_role = db.RecordType(name="TestRT1").retrieve()
+ rec = db.Record(name="fail").add_property(wrong_role)
+ rec2 = db.Record(name="ok").add_property(right_role)
+
+ assert wrong_role.get_property("TestProperty1") is not None
+ assert str(wrong_role.get_property("TestProperty1")) == str(
+ right_role.get_property("TestProperty1"))
+
+ xml = rec2.to_xml()
+ assert not xml.xpath("/Record/Property/Property")
+
+ xml = rec.to_xml()
+ # TODO this fails because the Record is treated differently than the
+ # RecordType.
+ assert not xml.xpath("/Record/Property/Property")
diff --git a/tests/test_query.py b/tests/test_query.py
index 9b188d9c94f964508ef254506ea19783fd02b5b6..8432b37d92c87d772ff07c0b4772def48b6aaab1 100644
--- a/tests/test_query.py
+++ b/tests/test_query.py
@@ -1002,3 +1002,25 @@ def test_query_by_name():
assert len(h.execute_query("FIND TestRT WITH name LIKE 'TestRec*'")) == 2
assert len(h.execute_query("FIND TestRT WITH name LIKE '*Rec*'")) == 2
assert len(h.execute_query("FIND ENTITY WITH name LIKE 'TestRec*'")) == 2
+
+
+@mark.xfail(reason="Issue: https://gitlab.com/caosdb/caosdb-server/-/issues/96")
+def test_referenced_as():
+ rt_person = db.RecordType("TestPerson").insert()
+ db.RecordType("TestParty").insert()
+
+ # three persons
+ g1 = db.Record().add_parent("TestPerson").insert()
+ g2 = db.Record().add_parent("TestPerson").insert()
+ g3 = db.Record().add_parent("TestPerson").insert()
+ guest_ids = set(g1.id, g2.id, g3.id)
+
+ party = db.Record(
+ "Diamond Jubilee of Elizabeth II").add_parent("TestParty")
+ party.add_property(rt_person, datatype=db.LIST(rt_person), name="Guests",
+ value=[g1, g2, g3])
+ party.insert()
+ guests = db.execute_query("FIND RECORD TestPerson WHICH IS REFERENCED BY "
+ "TestParty AS Guests")
+ assert len(guests) == 3
+ assert set([e.id for e in guests]) == guest_ids
diff --git a/tests/test_select.py b/tests/test_select.py
index 748b85dd07d85a32e5e4643f562204433639da04..66c3992d1519f101b7b18941924b9bd939fa0066 100644
--- a/tests/test_select.py
+++ b/tests/test_select.py
@@ -25,123 +25,304 @@
@author: tf
"""
-import caosdb as h
-
-# @UnresolvedImport
-from nose.tools import assert_true, assert_equal, assert_is_not_none, assert_is_none
+import caosdb as db
def setup_module():
- print("SETUP")
teardown_module()
- h.Property(name="TestPropertyOne", datatype=h.TEXT).insert()
- h.Property(
+ db.Property(name="TestPropertyOne", datatype=db.TEXT).insert()
+ db.Property(
name="TestPropertyTwo",
description="Desc2",
- datatype=h.TEXT).insert()
- h.RecordType(
+ datatype=db.TEXT).insert()
+ db.RecordType(
name="TestRecordType",
description="DescRecTy").add_property(
- name="TestPropertyOne").add_property(
- name="TestPropertyTwo").insert()
+ name="TestPropertyOne", value="v1").add_property(
+ name="TestPropertyTwo", value="v2").insert()
+ rt_house = db.RecordType("TestHouse", description="TestHouseDesc").insert()
+ db.RecordType("TestWindow").insert()
+ rt_person = db.RecordType("TestPerson",
+ description="TestPersonDesc").insert()
+ db.RecordType("TestParty", description="TestPartyDesc").insert()
+ db.Property("TestHeight", description="TestHeightDesc", datatype=db.DOUBLE,
+ unit="ft").insert()
+ db.Property("TestDate", description="TestDateDesc",
+ datatype=db.DATETIME).insert()
+
+ window = db.Record("Window1",
+ description="Window1Desc").add_parent("TestWindow")
+ window.add_property("TestHeight", 20.5, unit="ft")
+ window.insert()
+
+ owner = db.Record("The Queen").add_parent("TestPerson").insert()
+
+ house = db.Record("Buckingham Palace")
+ house.description = "A rather large house"
+ house.add_parent("TestHouse")
+ house.add_property(rt_person, name="TestOwner", value=owner)
+ house.add_property("TestWindow", window).insert()
+
+ g1 = db.Record().add_parent("TestPerson").insert()
+ g2 = db.Record().add_parent("TestPerson").insert()
+ g3 = db.Record().add_parent("TestPerson").insert()
+
+ party = db.Record(
+ "Diamond Jubilee of Elizabeth II").add_parent("TestParty")
+ party.add_property(rt_house, name="Location", value=house)
+ party.add_property("TestDate", "2012-02-06")
+ party.add_property(rt_person, datatype=db.LIST(rt_person), name="Guests",
+ value=[g1, g2, g3])
+ party.insert()
def teardown_module():
- print("TEARDOWN")
- try:
- h.execute_query("FIND Test*").delete()
- except BaseException:
- pass
+ d = db.execute_query("FIND Test*")
+
+ if len(d) > 0:
+ d.delete()
def test_id1():
- p1 = h.execute_query("FIND TestPropertyOne", unique=True)
- assert_true(p1.is_valid())
- assert_is_not_none(p1.name)
- assert_is_not_none(p1.datatype)
- assert_is_none(p1.description)
+ p1 = db.execute_query("FIND TestPropertyOne", unique=True)
+ assert p1.is_valid() is True
+ assert p1.name is not None
+ assert p1.datatype is not None
+ assert p1.description is None
- p1_c = h.execute_query("SELECT id FROM TestPropertyOne", unique=True)
- assert_true(p1_c.is_valid())
- assert_equal(p1_c.id, p1.id)
- assert_is_none(p1_c.name)
- assert_is_none(p1_c.datatype)
- assert_is_none(p1_c.description)
+ p1_c = db.execute_query("SELECT id FROM TestPropertyOne", unique=True)
+ assert p1_c.is_valid() is True
+ assert p1_c.id == p1.id
+ assert p1_c.name is not None, "Name is always included"
+ assert p1_c.datatype is None
+ assert p1_c.description is None
def test_id2():
- p2 = h.execute_query("FIND TestPropertyTwo", unique=True)
- assert_true(p2.is_valid())
- assert_is_not_none(p2.name)
- assert_is_not_none(p2.datatype)
- assert_is_not_none(p2.description)
+ p2 = db.execute_query("FIND TestPropertyTwo", unique=True)
+ assert p2.is_valid() is True
+ assert p2.name is not None
+ assert p2.datatype is not None
+ assert p2.description is not None
- p2_c = h.execute_query("SELECT id FROM TestPropertyTwo", unique=True)
- assert_true(p2_c.is_valid())
- assert_equal(p2_c.id, p2.id)
- assert_is_none(p2_c.name)
- assert_is_none(p2_c.datatype)
- assert_is_none(p2_c.description)
+ p2_c = db.execute_query("SELECT id FROM TestPropertyTwo", unique=True)
+ assert p2_c.is_valid() is True
+ assert p2_c.id == p2.id
+ assert p2_c.name is not None, "Name is always included"
+ assert p2_c.datatype is None
+ assert p2_c.description is None
def test_id3():
- p3s = h.execute_query("SELECT description FROM TestProperty*")
- assert_equal(len(p3s), 2)
+ p3s = db.execute_query("SELECT description FROM TestProperty*")
+ assert len(p3s) == 2
+
for e in p3s:
- assert_is_not_none(e.id)
+ assert e.id is not None
def test_name1():
- p3s = h.execute_query("SELECT description FROM TestProperty*")
- assert_equal(len(p3s), 2)
+ p3s = db.execute_query("SELECT description FROM TestProperty*")
+ assert len(p3s) == 2
+
for e in p3s:
- assert_is_not_none(e.name)
+ assert e.name is not None
def test_name2():
- p3s = h.execute_query("SELECT name FROM TestProperty*")
- assert_equal(len(p3s), 2)
+ p3s = db.execute_query("SELECT name FROM TestProperty*")
+ assert len(p3s) == 2
+
for e in p3s:
- assert_is_not_none(e.name)
- assert_is_none(e.description)
+ assert e.name is not None
+ assert e.description is None
def test_multi1():
- p1 = h.execute_query(
+ p1 = db.execute_query(
"SELECT id, name, description FROM TestPropertyOne",
unique=True)
- assert_is_not_none(p1.id)
- assert_equal(p1.name, "TestPropertyOne")
- assert_is_none(p1.description)
+ assert p1.id is not None
+ assert p1.name == "TestPropertyOne"
+ assert p1.description is None
- p2 = h.execute_query(
+ p2 = db.execute_query(
"SELECT id, name, description FROM TestPropertyTwo",
unique=True)
- assert_is_not_none(p2.id)
- assert_equal(p2.name, "TestPropertyTwo")
- assert_equal(p2.description, "Desc2")
+ assert p2.id is not None
+ assert p2.name == "TestPropertyTwo"
+ assert p2.description == "Desc2"
def test_sub1():
- rt = h.execute_query("FIND TestRecordType", unique=True)
- assert_is_not_none(rt.id)
- assert_is_not_none(rt.name)
- assert_is_not_none(rt.get_property("TestPropertyOne"))
+ rt = db.execute_query("FIND TestRecordType", unique=True)
+ assert rt.id is not None
+ assert rt.name is not None
+ assert rt.get_property("TestPropertyOne") is not None
- rt = h.execute_query(
+ rt = db.execute_query(
"SELECT TestPropertyOne FROM TestRecordType",
unique=True)
- assert_is_not_none(rt.id)
- assert_is_not_none(rt.name)
- assert_is_not_none(rt.get_property("TestPropertyOne"))
+ assert rt.id is not None
+ assert rt.name is not None
+ assert rt.get_property("TestPropertyOne") is not None
+ assert rt.get_property("TestPropertyOne").value == "v1"
+ assert rt.get_property("TestPropertyTwo") is None
def test_sub2():
- rt = h.execute_query(
+ rt = db.execute_query(
"SELECT TestPropertyTwo.description FROM TestRecordType",
unique=True)
- assert_is_not_none(rt.id)
- assert_is_not_none(rt.name)
- assert_is_not_none(rt.get_property("TestPropertyTwo"))
- assert_equal(rt.get_property("TestPropertyTwo").description, "Desc2")
- assert_is_none(rt.get_property("TestPropertyTwo").datatype)
+ assert rt.id is not None
+ assert rt.name is not None
+ assert rt.get_property("TestPropertyTwo") is not None
+ assert rt.get_property("TestPropertyTwo").description == "Desc2"
+ assert rt.get_property("TestPropertyTwo").datatype is None
+ assert rt.get_property("TestPropertyTwo").value is None
+
+
+def test_subref():
+ window = db.execute_query("FIND RECORD TestWindow", unique=True)
+ s = db.execute_query("SELECT name, TestWindow.TestHeight.value, "
+ "TestWindow.TestHeight.unit FROM RECORD TestHouse")
+
+ assert len(s) == 1
+
+ row = s.get_property_values("name", "TestWindow")[0]
+ assert row[0] == "Buckingham Palace"
+ assert row[1] == window.id
+
+ row = s.get_property_values("name", ("TestWindow", "TestHeight"))[0]
+ assert row[0] == "Buckingham Palace"
+ assert row[1] == 20.5
+
+ row = s.get_property_values(
+ "name", ("TestWindow", "TestHeight", "unit"))[0]
+ assert row[0] == "Buckingham Palace"
+ assert row[1] == "ft"
+
+
+def test_subref_deep():
+ p = db.execute_query(
+ "SELECT name, Testdate, location, location.TestWindow.Testheight FROM "
+ "RECORD TestParty", unique=True)
+ row = p.get_property_values("name", "Testdate",
+ ("location", "Testwindow", "Testheight"))
+ assert row == ("Diamond Jubilee of Elizabeth II", "2012-02-06", 20.5)
+
+
+def test_select_list():
+ guests = db.execute_query(
+ "FIND RECORD TestPerson WHICH IS REFERENCED BY TestParty")
+ s = db.execute_query("SELECT guests FROM RECORD TestParty", unique=True)
+
+ column = s.get_property_values("guests")[0]
+ assert len(column) == len(guests)
+
+ for eid in [e.id for e in guests]:
+ assert eid in column
+
+
+def test_select_unit():
+ s = db.execute_query("SELECT unit FROM RECORD TestHouse", unique=True)
+ column = s.get_property_values("unit")
+ assert column == (None,)
+
+ s = db.execute_query("SELECT unit FROM PROPERTY TestHeight", unique=True)
+ column = s.get_property_values("unit")
+ assert column == ("ft",)
+
+ s = db.execute_query("SELECT TestWindow.TestHeight.unit FROM "
+ "RECORD TestHouse", unique=True)
+ column = s.get_property_values(("TestWindow", "TestHeight", "unit"))
+ assert column == ("ft",)
+
+ s = db.execute_query("SELECT TestHeight.unit.TestWindow FROM "
+ "RECORD TestWindow", unique=True)
+ column = s.get_property_values(("TestHeight", "unit", "TestWindow"))
+ assert column == (None,)
+
+
+def test_select_description():
+ s = db.execute_query("SELECT description FROM RECORD TestPerson")
+ column = s.get_property_values("description")
+ assert column == [(None,), (None,), (None,), (None,)]
+
+ s = db.execute_query("SELECT description"
+ "FROM RECORD TestHouse", unique=True)
+ column = s.get_property_values(("description"))
+ assert column == ("A rather large house",)
+
+ s = db.execute_query("SELECT location.description"
+ "FROM RECORD TestParty", unique=True)
+ column = s.get_property_values(("location", "description"))
+ assert column == ("A rather large house",)
+
+ s = db.execute_query("SELECT TestHeight.description FROM "
+ "RECORD TestWindow", unique=True)
+ column = s.get_property_values(("TestHeight", "description"))
+ assert column == ('TestHeightDesc',)
+
+ s = db.execute_query("SELECT TestWindow.TestHeight.description FROM "
+ "RECORD TestHouse", unique=True)
+ column = s.get_property_values(("TestWindow", "TestHeight", "description"))
+ assert column == ('TestHeightDesc',)
+
+ s = db.execute_query("SELECT TestHeight.description.TestWindow FROM "
+ "RECORD TestWindow", unique=True)
+ column = s.get_property_values(("TestHeight", "description", "TestWindow"))
+ assert column == (None,)
+
+
+def test_select_id():
+ house_id = db.execute_query("FIND RECORD TestHouse", unique=True).id
+ s = db.execute_query("SELECT id FROM RECORD TestHouse", unique=True)
+ column = s.get_property_values("id")
+ assert column == (house_id,)
+
+ s = db.execute_query(
+ "SELECT location.id FROM RECORD TestHouse",
+ unique=True)
+ column = s.get_property_values("id")
+ assert column == (house_id,)
+
+ height_id = db.execute_query("FIND PROPERTY TestHeight", unique=True).id
+ s = db.execute_query("SELECT id FROM PROPERTY TestHeight", unique=True)
+ column = s.get_property_values("id")
+ assert column == (height_id,)
+
+ s = db.execute_query("SELECT TestWindow.TestHeight.id FROM "
+ "RECORD TestHouse", unique=True)
+ column = s.get_property_values(("TestWindow", "TestHeight", "id"))
+ assert column == (height_id,)
+
+ s = db.execute_query("SELECT TestHeight.id.TestWindow FROM "
+ "RECORD TestWindow", unique=True)
+ column = s.get_property_values(("TestHeight", "id", "TestWindow"))
+ assert column == (None,)
+
+
+def test_select_name():
+ s = db.execute_query("SELECT name FROM RECORD TestHouse", unique=True)
+ column = s.get_property_values("name")
+ assert column == ("Buckingham Palace",)
+
+ s = db.execute_query("SELECT location.name FROM RECORD TestHouse",
+ unique=True)
+ column = s.get_property_values("name")
+ assert column == ("Buckingham Palace",)
+
+ s = db.execute_query("SELECT name FROM PROPERTY TestHeight", unique=True)
+ column = s.get_property_values("name")
+ assert column == ("TestHeight",)
+
+ s = db.execute_query("SELECT TestWindow.TestHeight.name FROM "
+ "RECORD TestHouse", unique=True)
+ column = s.get_property_values(("TestWindow", "TestHeight", "name"))
+ assert column == ("TestHeight",)
+
+ s = db.execute_query("SELECT TestHeight.name.TestWindow FROM "
+ "RECORD TestWindow", unique=True)
+ column = s.get_property_values(("TestHeight", "name", "TestWindow"))
+ assert column == (None,)
diff --git a/tests/test_server_side_scripting.py b/tests/test_server_side_scripting.py
index 74c877bd5d5e3a056e659aa49ff6f3950a834429..cff2017861e7779da9ded0d5541a77ce3a866751 100644
--- a/tests/test_server_side_scripting.py
+++ b/tests/test_server_side_scripting.py
@@ -26,34 +26,67 @@
Integration tests for the implementation of the server-side-scripting api.
"""
from __future__ import print_function, unicode_literals
-from pytest import raises, mark
+import os
+from pytest import raises
+import json
from lxml import etree
-from caosdb import get_connection, get_config
+from http.client import HTTPSConnection
+import ssl
+from caosdb import get_connection, get_config, Info, execute_query, RecordType
from caosdb.exceptions import (ClientErrorException,
ResourceNotFoundException)
from caosdb.connection.encode import MultipartParam, multipart_encode
+from caosdb.connection.utils import urlencode, urlparse
from caosdb import administration as admin
+from caosdb.utils.server_side_scripting import run_server_side_script
-_TEST_SCRIPTS = ["not_executable", "ok", "err", "simple_script.py"]
-_SERVER_SIDE_SCRIPTING_BIN_DIR = get_config().get(
+_TEST_SCRIPTS = ["not_executable", "ok", "err", "ok_anonymous"]
+_SERVER_SIDE_SCRIPTING_BIN_DIR_LOCAL = get_config().get(
"IntegrationTests",
- "test_server_side_scripting.bin_dir")
+ "test_server_side_scripting.bin_dir.local")
+_SERVER_SIDE_SCRIPTING_BIN_DIR_SERVER = get_config().get(
+ "IntegrationTests",
+ "test_server_side_scripting.bin_dir.server")
_TEST_SCRIPTS_DIR = "./resources/"
_REMOVE_FILES_AFTERWARDS = []
+_ORIGINAL_SERVER_SCRIPTING_BIN_DIR = ""
+
+
+def clean_database():
+ admin._set_permissions("anonymous", [])
+ d = execute_query("FIND ENTITY WITH ID > 99")
+ if len(d) > 0:
+ d.delete()
+
+
+def setup():
+ clean_database()
+
+
+def teardown():
+ admin.set_server_property("SERVER_SIDE_SCRIPTING_BIN_DIR",
+ _ORIGINAL_SERVER_SCRIPTING_BIN_DIR)
+ clean_database()
def setup_module():
+ global _ORIGINAL_SERVER_SCRIPTING_BIN_DIR
+ _ORIGINAL_SERVER_SCRIPTING_BIN_DIR = admin.get_server_property(
+ "SERVER_SIDE_SCRIPTING_BIN_DIR")
+ clean_database()
+
from os import makedirs
from os.path import join, isdir, exists
from shutil import copyfile, copymode
- print("bin: " + str(_SERVER_SIDE_SCRIPTING_BIN_DIR))
+ print("bin dir (local): " + str(_SERVER_SIDE_SCRIPTING_BIN_DIR_LOCAL))
+ print("bin dir (server): " + str(_SERVER_SIDE_SCRIPTING_BIN_DIR_SERVER))
print("tests scripts: " + str(_TEST_SCRIPTS))
- if not exists(_SERVER_SIDE_SCRIPTING_BIN_DIR):
- makedirs(_SERVER_SIDE_SCRIPTING_BIN_DIR)
- _REMOVE_FILES_AFTERWARDS.append(_SERVER_SIDE_SCRIPTING_BIN_DIR)
- assert isdir(_SERVER_SIDE_SCRIPTING_BIN_DIR)
+ if not exists(_SERVER_SIDE_SCRIPTING_BIN_DIR_LOCAL):
+ makedirs(_SERVER_SIDE_SCRIPTING_BIN_DIR_LOCAL)
+ _REMOVE_FILES_AFTERWARDS.append(_SERVER_SIDE_SCRIPTING_BIN_DIR_LOCAL)
+ assert isdir(_SERVER_SIDE_SCRIPTING_BIN_DIR_LOCAL)
for script_file in _TEST_SCRIPTS:
- target = join(_SERVER_SIDE_SCRIPTING_BIN_DIR, script_file)
+ target = join(_SERVER_SIDE_SCRIPTING_BIN_DIR_LOCAL, script_file)
src = join(_TEST_SCRIPTS_DIR, script_file)
copyfile(src, target)
copymode(src, target)
@@ -70,6 +103,7 @@ def teardown_module():
rmtree(obsolete)
else:
remove(obsolete)
+ clean_database()
def test_call_script_non_existing():
@@ -80,15 +114,8 @@ def test_call_script_non_existing():
def test_call_script_not_executable():
- props = admin.get_server_properties()
- print(props)
-
- import os
- dirpath = os.getcwd()
- print("PWD={}".format(dirpath))
- print("ls ./\n{}".format(os.listdir("./")))
- print("ls ../\n{}".format(os.listdir("../")))
-
+ admin.set_server_property("SERVER_SIDE_SCRIPTING_BIN_DIR",
+ _SERVER_SIDE_SCRIPTING_BIN_DIR_SERVER)
form = dict()
form["call"] = "not_executable"
with raises(ClientErrorException) as exc_info:
@@ -97,6 +124,8 @@ def test_call_script_not_executable():
def test_call_ok():
+ admin.set_server_property("SERVER_SIDE_SCRIPTING_BIN_DIR",
+ _SERVER_SIDE_SCRIPTING_BIN_DIR_SERVER)
form = dict()
form["call"] = "ok"
r = get_connection().post_form_data("scripting", form)
@@ -108,6 +137,8 @@ def test_call_ok():
def test_call_err():
+ admin.set_server_property("SERVER_SIDE_SCRIPTING_BIN_DIR",
+ _SERVER_SIDE_SCRIPTING_BIN_DIR_SERVER)
form = dict()
form["call"] = "err"
r = get_connection().post_form_data("scripting", form)
@@ -118,29 +149,234 @@ 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():
+def test_run_server_side_script_with_file_as_positional_param():
+ RecordType("TestRT").insert()
+ _REMOVE_FILES_AFTERWARDS.append("test_file.txt")
+ with open("test_file.txt", "w") as f:
+ f.write("this is a test")
+
+ response = run_server_side_script("administration/diagnostics.py",
+ "pos0",
+ "pos1",
+ exit="123",
+ query="COUNT TestRT",
+ files={"-p2": "test_file.txt"})
+ assert response.stderr is None
+ assert response.code == 123
+ assert response.call == ('administration/diagnostics.py '
+ '--exit=123 --query=COUNT TestRT '
+ 'pos0 pos1 .upload_files/test_file.txt')
+
+ json_data = json.loads(response.stdout)
+ assert "caosdb" in json_data
+ assert "query" in json_data["caosdb"]
+ assert json_data["caosdb"]["query"] == ["COUNT TestRT", "1"]
+ assert "./.upload_files/test_file.txt" in json_data["files"]
+
+
+def test_run_server_side_script_with_additional_file():
+ RecordType("TestRT").insert()
+ _REMOVE_FILES_AFTERWARDS.append("test_file.txt")
+ with open("test_file.txt", "w") as f:
+ f.write("this is a test")
+
+ response = run_server_side_script("administration/diagnostics.py",
+ "pos0",
+ "pos1",
+ exit="123",
+ query="COUNT TestRT",
+ files={"dummykey": "test_file.txt"})
+ assert response.stderr is None
+ assert response.code == 123
+ assert response.call == ('administration/diagnostics.py '
+ '--exit=123 --query=COUNT TestRT '
+ 'pos0 pos1')
+
+ json_data = json.loads(response.stdout)
+ assert json_data["caosdb"]["query"] == ["COUNT TestRT", "1"]
+ assert "./.upload_files/test_file.txt" in json_data["files"]
+
+
+def test_diagnostics_basic():
+ RecordType("TestRT").insert()
+
form = dict()
- form["call"] = "simple_script.py"
+ form["call"] = "administration/diagnostics.py"
form["-Oexit"] = "123"
- r = get_connection().post_form_data("scripting", form)
- xml = etree.parse(r)
+ form["-Oquery"] = "COUNT TestRT"
+
+ response = get_connection().post_form_data("scripting", form)
+ xml = etree.parse(response)
print(etree.tostring(xml))
- assert xml.xpath("/Response/script/@code")[0] == "123"
+ assert response.status == 200 # ok
+ assert response.getheader("Content-Type") == 'text/xml; charset=UTF-8'
+
+ diagnostics = xml.xpath("/Response/script/stdout")[0].text
+ assert diagnostics is not None
+ diagnostics = json.loads(diagnostics)
+ assert diagnostics["python_version"] is not None
+ assert diagnostics["import"]["caosdb"][0] is True, ("caosdb not installed"
+ " in server's python path")
+ assert diagnostics["auth_token"] is not None
+ EXC_ERR = ("There shouldn't be any exception during the diagnostics of "
+ "the interaction of the server-side script and the server.")
+ assert "exception" not in diagnostics["caosdb"], EXC_ERR
+
+ assert "query" in diagnostics["caosdb"]
+ assert diagnostics["caosdb"]["query"][0] == "COUNT TestRT"
+ assert diagnostics["caosdb"]["query"][1] == "1", ("The RecordType should "
+ "have been found.")
+ assert xml.xpath("/Response/script/@code")[0] == "123", ("The script "
+ "should exit "
+ "with code 123.")
+ assert xml.xpath("/Response/script/call")[0].text.startswith(
+ "administration/diagnostics.py")
+ assert xml.xpath("/Response/script/stderr")[0].text is None
+
+
+def test_diagnostics_with_file_upload():
+ RecordType("TestRT").insert()
_REMOVE_FILES_AFTERWARDS.append("test_file.txt")
with open("test_file.txt", "w") as f:
f.write("this is a test")
+
parts = []
parts.append(MultipartParam.from_file(paramname="txt_file",
filename="test_file.txt"))
- parts.append(MultipartParam("call", "simple_script.py"))
+ parts.append(MultipartParam("call", "administration/diagnostics.py"))
body, headers = multipart_encode(parts)
- r = get_connection().insert(["scripting"], body=body, headers=headers)
- xml = etree.parse(r)
- print(etree.tostring(xml).decode("utf-8"))
+ response = get_connection().insert(
+ ["scripting"], body=body, headers=headers)
+ assert response.status == 200 # ok
+ assert response.getheader("Content-Type") == 'text/xml; charset=UTF-8'
+
+ xml = etree.parse(response)
+ print(etree.tostring(xml))
assert xml.xpath("/Response/script/@code")[0] == "0"
- assert xml.xpath(
- "/Response/script/call")[0].text.startswith("simple_script.py")
- assert "this is a test" in xml.xpath("/Response/script/stdout")[0].text
+
+ diagnostics = xml.xpath("/Response/script/stdout")[0].text
+ assert diagnostics is not None
+ diagnostics = json.loads(diagnostics)
+ assert diagnostics["python_version"] is not None
+ assert diagnostics["import"]["caosdb"][0] is True, ("caosdb not installed"
+ " in server's python path")
+ assert diagnostics["auth_token"] is not None
+ assert "exception" not in diagnostics["caosdb"], ("There shouldn't be any "
+ "exception during the "
+ "diagnostics of the "
+ "interaction of the "
+ "server-side "
+ "script and the server.")
+
+ assert xml.xpath("/Response/script/@code")[0] == "0", ("The script "
+ "should exit "
+ "with code 0.")
+ assert xml.xpath("/Response/script/call")[0].text.startswith(
+ "administration/diagnostics.py")
+ assert xml.xpath("/Response/script/stderr")[0].text is None
+
+
+def test_call_as_anonymous_with_administration_role():
+ assert Info().user_info.roles == ["administration"]
+
+ # activate anonymous user
+ admin.set_server_property("AUTH_OPTIONAL", "TRUE")
+ # give permission to call diagnostics.py
+ admin._set_permissions(
+ "anonymous", [
+ admin.PermissionRule(
+ "Grant", "SCRIPTING:EXECUTE:administration:diagnostics.py")])
+
+ form = dict()
+ form["call"] = "administration/diagnostics.py"
+
+ headers = {"Content-Type": "application/x-www-form-urlencoded"}
+ response = request(method="POST", headers=headers, path="scripting",
+ body=urlencode(form))
+
+ xml = etree.parse(response)
+
+ assert response.getheader("Set-Cookie") is None # no auth token returned
+ assert response.getheader("Content-Type") == 'text/xml; charset=UTF-8'
+
+ assert response.status == 200 # ok
+ assert xml.xpath("/Response/script/@code")[0] == "0"
+ diagnostics = xml.xpath("/Response/script/stdout")[0].text
+ assert diagnostics is not None
+ diagnostics = json.loads(diagnostics)
+ assert diagnostics["python_version"] is not None
+ assert diagnostics["import"]["caosdb"][0] is True, ("caosdb not installed"
+ " in server's python path")
+ assert diagnostics["auth_token"] is not None
+ assert "exception" not in diagnostics["caosdb"]
+
+
+def request(method, headers, path, body=None):
+ """ Connect without auth-token. This is clumsy, bc the pylib is not
+ intended to be used as anonymous user.
+ """
+ context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
+ context.verify_mode = ssl.CERT_REQUIRED
+ if hasattr(context, "check_hostname"):
+ context.check_hostname = True
+ context.load_verify_locations(get_config().get("Connection", "cacert"))
+
+ url = get_config().get("Connection", "url")
+ fullurl = urlparse(url)
+
+ http_con = HTTPSConnection(
+ str(fullurl.netloc), timeout=200, context=context)
+ http_con.request(method=method, headers=headers, url=str(fullurl.path) +
+ path, body=body)
+ return http_con.getresponse()
+
+
+def test_anonymous_script_calling_not_permitted():
+ form = dict()
+ form["call"] = "ok"
+
+ # activate anonymous user
+ admin.set_server_property("AUTH_OPTIONAL", "TRUE")
+
+ # remove all anonymous permissions
+ admin._set_permissions("anonymous", [
+ admin.PermissionRule("Grant", "SCRIPTING:EXECUTE:ok"),
+ admin.PermissionRule("Deny", "*")
+ ])
+
+ headers = {"Content-Type": "application/x-www-form-urlencoded"}
+ response = request(method="POST", headers=headers, path="scripting",
+ body=urlencode(form))
+ assert response.status == 403 # forbidden
+ assert response.getheader("Set-Cookie") is None # no auth token returned
+
+
+def test_anonymous_script_calling_success():
+ admin.set_server_property("SERVER_SIDE_SCRIPTING_BIN_DIR",
+ _SERVER_SIDE_SCRIPTING_BIN_DIR_SERVER)
+ form = dict()
+ form["call"] = "ok_anonymous"
+
+ # activate anonymous user
+ admin.set_server_property("AUTH_OPTIONAL", "TRUE")
+ # give permission to call ok_anonymous
+ admin._set_permissions("anonymous",
+ [admin.PermissionRule("Grant", "SCRIPTING:EXECUTE:ok_anonymous")])
+
+ headers = {"Content-Type": "application/x-www-form-urlencoded"}
+ response = request(method="POST", headers=headers, path="scripting",
+ body=urlencode(form))
+ assert response.status == 200 # ok
+ assert response.getheader("Set-Cookie") is None # no auth token returned
+ assert response.getheader("Content-Type") == 'text/xml; charset=UTF-8'
+
+ body = response.read()
+ xml = etree.fromstring(body)
+
+ # verify unauthenticated call to script ok_anonymous
+ assert xml.xpath("/Response/UserInfo/Roles/Role")[0].text == "anonymous"
+ assert xml.xpath("/Response/script/call")[0].text == "ok_anonymous"
+ assert xml.xpath("/Response/script/stdout")[0].text == "ok_anonymous"
assert xml.xpath("/Response/script/stderr")[0].text is None
+ assert xml.xpath("/Response/script/@code")[0] == "0"