connectioncache.py

Go to the documentation of this file.

#  Copyright 2008-2015 Nokia Networks
#  Copyright 2016-     Robot Framework Foundation
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
 
from .normalizing import NormalizedDict
from .robottypes import is_string
 
 
 
class ConnectionCache:
 
    def __init__(self, no_current_msg='No open connection.'):
        self._no_current_no_current = NoConnection(no_current_msg)
        self.currentcurrent = self._no_current_no_current  #: Current active connection.
        self._connections_connections = []
        self._aliases_aliases = NormalizedDict()
 
    @property
    current_index = property
    
    def current_index(self):
        if not self:
            return None
        for index, conn in enumerate(self):
            if conn is self.currentcurrent:
                return index + 1
 
    @current_index.setter
    
    def current_index(self, index):
        self.currentcurrent = self._connections_connections[index - 1] \
            if index is not None else self._no_current_no_current
 
    
    def register(self, connection, alias=None):
        self.currentcurrent = connection
        self._connections_connections.append(connection)
        index = len(self._connections_connections)
        if is_string(alias):
            self._aliases_aliases[alias] = index
        return index
 
    
    def switch(self, alias_or_index):
        self.currentcurrent = self.get_connectionget_connection(alias_or_index)
        return self.currentcurrent
 
    
    def get_connection(self, alias_or_index=None):
        if alias_or_index is None:
            if not self:
                self.currentcurrent.raise_error()
            return self.currentcurrent
        try:
            index = self.resolve_alias_or_indexresolve_alias_or_index(alias_or_index)
        except ValueError as err:
            raise RuntimeError(err.args[0])
        return self._connections_connections[index-1]
 
    __getitem__ = get_connection
 
    
    def close_all(self, closer_method='close'):
        for conn in self._connections_connections:
            getattr(conn, closer_method)()
        self.empty_cacheempty_cache()
        return self.currentcurrent
 
    
    def empty_cache(self):
        self.currentcurrent = self._no_current_no_current
        self._connections_connections = []
        self._aliases_aliases = NormalizedDict()
 
    def __iter__(self):
        return iter(self._connections_connections)
 
    def __len__(self):
        return len(self._connections_connections)
 
    def __bool__(self):
        return self.currentcurrent is not self._no_current_no_current
 
    def resolve_alias_or_index(self, alias_or_index):
        for resolver in self._resolve_alias_resolve_alias, self._resolve_index_resolve_index, self._is_connection_is_connection:
            try:
                return resolver(alias_or_index)
            except ValueError:
                pass
        raise ValueError(f"Non-existing index or alias '{alias_or_index}'.")
 
    def _resolve_alias(self, alias):
        if is_string(alias) and alias in self._aliases_aliases:
            return self._aliases_aliases[alias]
        raise ValueError
 
    def _resolve_index(self, index):
        try:
            index = int(index)
        except TypeError:
            raise ValueError
        if not 0 < index <= len(self._connections_connections):
            raise ValueError
        return index
 
    def _is_connection(self, conn):
        return self._connections_connections.index(conn) + 1
 
 
class NoConnection:
 
    def __init__(self, message):
        self.messagemessage = message
 
    def __getattr__(self, name):
        if name.startswith('__') and name.endswith('__'):
            raise AttributeError
        self.raise_errorraise_error()
 
    def raise_error(self):
        raise RuntimeError(self.messagemessage)
 
    def __bool__(self):
        return False