from robot.errors import DataError
from normalizing import NormalizedDict
class ConnectionCache:
"""Connection cache for different Robot test libraries that use connections.
This cache stores connections and allows switching between them using
generated indexes or user given aliases. Can be used for example by web
testing libraries where there's need for multiple concurrent connections.
Note that in most cases there should be only one instance of this class but
this is not enforced.
"""
def __init__(self, no_current_msg='No open connection'):
self.current = self._no_current = _NoConnection(no_current_msg)
self.current_index = None
self._connections = []
self._aliases = NormalizedDict()
self._no_current_msg = no_current_msg
def register(self, connection, alias=None):
"""Registers given connection with optional alias and returns its index.
Given connection is set to be the current connection. Alias must be
a string. The index of the first connection after initialization or
close_all or empty_cache is 1, second is 2, etc.
"""
self.current = connection
self._connections.append(connection)
self.current_index = len(self._connections)
if isinstance(alias, basestring):
self._aliases[alias] = self.current_index
return self.current_index
def switch(self, index_or_alias):
"""Switches to the connection specified by given index or alias.
If alias is given it must be a string. Indexes can be either integers
or strings that can be converted into integer. Raises a DataError
if no connection with given index or alias found.
"""
try:
index = self._get_index(index_or_alias)
except ValueError:
raise DataError("Non-existing index or alias '%s'" % index_or_alias)
self.current = self._connections[index-1]
self.current_index = index
return self.current
def close_all(self, closer_method='close'):
"""Closes connections using given closer method and empties cache.
If simply calling the closer method is not adequate for closing
connections, clients should close connections themselves and use
empty_cache afterwards.
"""
for conn in self._connections:
getattr(conn, closer_method)()
self.empty_cache()
return self.current
def empty_cache(self):
"""Empties the connections cache.
Indexes of new connections starts from 1 after this."""
self.current = self._no_current
self.current_index = None
self._connections = []
self._aliases = NormalizedDict()
def _get_index(self, index_or_alias):
try:
return self._resolve_alias(index_or_alias)
except ValueError:
return self._resolve_index(index_or_alias)
def _resolve_alias(self, alias):
if isinstance(alias, basestring):
try:
return self._aliases[alias]
except KeyError:
pass
raise ValueError
def _resolve_index(self, index):
index = int(index)
if not 0 < index <= len(self._connections):
raise ValueError
return index
class _NoConnection:
def __init__(self, msg):
self._msg = msg
def __getattr__(self, name):
if name.startswith('__') and name.endswith('__'):
raise AttributeError
raise DataError(self._msg)
def __nonzero__(self):
return False
文件路径:robotframework-2.1.2\src\robot\utils\connectioncache.py
功能:不同的使用连接的Robot测试库的连接cache,通常情况下,一般的类只有一个实例,尽管不是强制的。
这部分需要以后的深入阅读。
阅读(38968) | 评论(0) | 转发(0) |