Skip to content
Snippets Groups Projects
Select Git revision
  • 3067c1e2f00923be45e483846e5ff3b7be61bebd
  • main default protected
  • f-remove-deprecation-warning
  • dev
  • f-docs-pylib
  • f-parse-value
  • f-compare
  • f-string-ids
  • f-217-set-special-property
  • f-filesystem-import
  • f-filesystem-link
  • f-filesystem-directory
  • f-filesystem-core
  • f-filesystem-cleanup
  • f-check-merge-entities
  • f-compare-enid
  • f-select-subproperties
  • v0.18.0
  • v0.17.0
  • v0.16.0
  • v0.15.1
  • v0.15.0
  • v0.14.0
  • v0.13.2
  • v0.13.1
  • v0.13.0
  • linkahead-rename-step-2
  • linkahead-rename-step-1
  • v0.12.0
  • v0.11.2
  • v0.11.1
  • v0.11.0
  • v0.10.0
  • v0.9.0
  • v0.8.0
  • v0.7.4
  • v0.7.3
37 results

test_connection.py

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    test_connection.py 10.03 KiB
    # -*- 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
    #
    """Test caosdb.connection."""
    # pylint: disable=missing-docstring
    from __future__ import unicode_literals, print_function
    from builtins import bytes, str  # pylint: disable=redefined-builtin
    import re
    from pytest import raises
    from nose.tools import (assert_equal as eq, assert_raises as raiz, assert_true
                            as tru, assert_is_not_none as there, assert_false as
                            falz)
    from caosdb.exceptions import ConfigurationException, LoginFailedException
    from caosdb.connection.utils import urlencode, make_uri_path, quote
    from caosdb.connection.connection import (
        configure_connection, CaosDBServerConnection,
        _DefaultCaosDBServerConnection)
    from caosdb.connection.mockup import (MockUpServerConnection, MockUpResponse,
                                          _request_log_message)
    from caosdb.configuration import get_config, _reset_config
    from caosdb.connection.authentication.interface import CredentialsAuthenticator
    from caosdb import execute_query
    
    
    def setup_module():
        _reset_config()
    
    
    def test_quote():
        unenc = 'ö'
        eq('%C3%B6', quote(unenc))
    
        unenc = '\xf6'
        eq('%C3%B6', quote(unenc))
    
        unenc = str('ö')
        eq('%C3%B6', quote(unenc))
    
    
    def test_urlencode():
        eq(urlencode({}), '')
        eq(urlencode({'key1': 'val1'}), 'key1=val1')
        eq(urlencode({'keynoval': None}), 'keynoval=')
        eq(urlencode({'kèy': 'välüe'}), 'k%C3%A8y=v%C3%A4l%C3%BCe')
        with raiz(AttributeError):
            urlencode({bytes('asdf', 'utf-8'): 'asdf'})
        with raiz(AttributeError):
            urlencode({'asdf': bytes('asdf', 'utf-8')})
        with raiz(AttributeError):
            urlencode({None: 'asdf'})
    
    
    def test_make_uri_path():
        eq(make_uri_path(None, None), '')
        eq(make_uri_path(None, {"key": "val"}), "?key=val")
        eq(make_uri_path([]), "")
        eq(make_uri_path(['a', 'b']), 'a/b')
        eq(make_uri_path(['a', str('b')]), 'a/b')
        eq(make_uri_path(['a', 'ö']), 'a/%C3%B6')
        eq(make_uri_path(['a', str('ö')]), 'a/%C3%B6')
        eq(make_uri_path(['a', 'b'], {'key': 'val'}), 'a/b?key=val')
        eq(make_uri_path([], {"key": "val"}), "?key=val")
        eq(make_uri_path([], {"key": str("val")}), "?key=val")
        eq(make_uri_path([], {"key": None}), "?key=")
        tru(
            re.match(r"^\?(key1=val1&key2=val2|key2=val2&key1=val1)$",
                     make_uri_path([], {
                         "key1": "val1",
                         "key2": "val2"
                     })))
        eq(make_uri_path(['a', 'with/slash']), 'a/with%2Fslash')
    
    
    def test_configure_connection():
        if not get_config().has_section("Connection"):
            get_config().add_section("Connection")
        get_config().set("Connection", "url", "https://host.de")
        get_config().set("Connection", "username", "test_username")
        get_config().set("Connection", "password", "test_password")
        get_config().set("Connection", "timeout", "200")
    
        there(configure_connection)
        tru(hasattr(configure_connection, "__call__"))
        c = configure_connection()
        there(c)
        tru(isinstance(c._delegate_connection, _DefaultCaosDBServerConnection))
        tru(isinstance(c._authenticator, CredentialsAuthenticator))
    
        c = configure_connection(
            implementation=MockUpServerConnection)
        tru(isinstance(c._delegate_connection, MockUpServerConnection))
    
    
    def test_connection_interface():
        with raiz(TypeError) as cm:
            CaosDBServerConnection()
        eq(cm.exception.args[0][:55],
           "Can't instantiate abstract class CaosDBServerConnection")
    
        tru(hasattr(CaosDBServerConnection, "request"))
        tru(hasattr(CaosDBServerConnection.request, "__call__"))
    
        tru(hasattr(CaosDBServerConnection, "configure"))
        tru(hasattr(CaosDBServerConnection.configure, "__call__"))
    
    
    def test_use_mockup_implementation():
        with raiz(RuntimeError) as rerr:
            execute_query("FIND Something")
        print(rerr.exception.args[0])
        eq(rerr.exception.args[0],
           "No response for this request - GET: Entity?query=FIND%20Something")
    
    
    def test_request_log_message():
        headers = {"k0": "v0"}
        body = "this is the body"
        eq(
            _request_log_message(
                method="GET", path="path", headers=None, body=None), "GET: path")
        eq(
            _request_log_message(
                method="GET", path="path", headers=headers, body=None),
            "GET: path\n" + str(headers))
        eq(
            _request_log_message(
                method="GET", path="path", headers=headers, body=body),
            "GET: path\n" + str(headers) + "\n" + str(body))
    
    
    def test_init_response():
        response = MockUpResponse(
            status=200, headers={"sessionToken": "SessionToken"}, body="Body")
        there(response)
        return response
    
    
    def test_getter_status():
        response = test_init_response()
        eq(response.status, 200)
    
    
    def test_read():
        response = test_init_response()
        tru(hasattr(response, "read"))
        eq(response.read(), "Body")
    
    
    def test_getter_session_token():
        response = test_init_response()
        tru(hasattr(response, "getheader"))
        eq(response.getheader("sessionToken"), "SessionToken")
    
    
    def test_init_connection():
        connection = MockUpServerConnection()
        there(connection)
        return connection
    
    
    def test_resources_list():
        connection = test_init_connection()
        assert hasattr(connection, "resources")
        assert len(connection.resources) == 1
        connection.resources.append(lambda **kwargs: test_init_response())
        assert len(connection.resources) == 2
        return connection
    
    
    def test_request_basics():
        connection = test_init_connection()
        tru(hasattr(connection, "request"))
        with raiz(RuntimeError) as cm:
            connection.request(method="GET", path="asdf")
        eq(cm.exception.args[0], "No response for this request - GET: asdf")
        connection = test_resources_list()
        there(connection.request(method="GET", path="asdf"))
    
    
    def setup_two_resources():
        def r1(**kwargs):
            if kwargs["method"] == "GET":
                return MockUpResponse(status=200, headers=kwargs["headers"], body="response r1")
    
        def r2(**kwargs):
            if kwargs["path"] == "matching/path/":
                return MockUpResponse(
                    status=456, headers={"key": "val"}, body="response r2")
    
        def r3(**kwargs):
            if kwargs["path"] == "401":
                return MockUpResponse(
                    status=401, headers={}, body="please login")
    
        connection = test_init_connection()
        connection.resources.extend([r1, r2, r3])
        return connection
    
    
    def test_test_request_with_two_responses():
        connection = setup_two_resources()
        eq(connection.request(method="GET", path="any", headers={}).status, 200)
        eq(connection.request(method="POST", path="matching/path/").status, 456)
    
    
    def test_missing_implementation():
        connection = configure_connection()
        with raises(ConfigurationException) as exc_info:
            connection.configure()
        assert exc_info.value.args[0].startswith(
            "Missing CaosDBServerConnection implementation.")
    
    
    def test_bad_implementation_not_callable():
        connection = configure_connection()
        with raises(ConfigurationException) as exc_info:
            connection.configure(implementation=None)
        assert exc_info.value.args[0].startswith(
            "Bad CaosDBServerConnection implementation.")
        assert exc_info.value.args[1].args[0] == "'NoneType' object is not callable"
    
    
    def test_bad_implementation_wrong_class():
        connection = configure_connection()
        with raises(ConfigurationException) as exc_info:
            connection.configure(implementation=dict)
        assert exc_info.value.args[0].startswith(
            "Bad CaosDBServerConnection implementation.")
        assert exc_info.value.args[1].args[0] == (
            "The `implementation` callable did not return an instance of "
            "CaosDBServerConnection.")
    
    
    def test_missing_auth_method():
        connection = configure_connection()
        with raises(ConfigurationException) as exc_info:
            connection.configure(implementation=MockUpServerConnection)
        assert exc_info.value.args[0].startswith("Missing password_method.")
    
    
    def test_missing_password():
        connection = configure_connection()
        connection.configure(implementation=setup_two_resources,
                             password_method="plain")
        connection._authenticator.auth_token="[test-auth-token]"
        assert connection.retrieve(["some"]).headers["Cookie"] == "SessionToken=%5Btest-auth-token%5D;"
    
        connection.configure(implementation=setup_two_resources,
                             password_method="plain")
        with raises(LoginFailedException):
            connection.delete(["401"])
    
    
    def test_auth_token_connection():
        connection = configure_connection(auth_token="blablabla",
                                          password_method="auth_token",
                                          implementation=setup_two_resources)
        connection.retrieve(["some"]).headers["Cookie"] == "SessionToken=blablabla;"
    
        connection._logout()
        with raises(LoginFailedException) as cm:
            connection.retrieve(["some"]).headers["Cookie"] == "SessionToken=blablabla;"
        assert cm.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.")