diff --git a/adodbapi/process_connect_string.py b/adodbapi/process_connect_string.py index 806c0954e8..21347d20cd 100644 --- a/adodbapi/process_connect_string.py +++ b/adodbapi/process_connect_string.py @@ -134,12 +134,4 @@ def process( macro_name, macro_code, kwargs ) # run the code in the local context kwargs[new_key] = rslt # put the result back in the keywords dict - # special processing for PyRO IPv6 host address - try: - s = kwargs["proxy_host"] - if ":" in s: # it is an IPv6 address - if s[0] != "[": # is not surrounded by brackets - kwargs["proxy_host"] = s.join(("[", "]")) # put it in brackets - except KeyError: - pass return kwargs diff --git a/adodbapi/quick_reference.md b/adodbapi/quick_reference.md index 4e3f2e4049..436ff0318a 100644 --- a/adodbapi/quick_reference.md +++ b/adodbapi/quick_reference.md @@ -339,8 +339,7 @@ Connection Methods: (non-standard) Connection Attributes -- .errorhandler # (standard extension. See PEP.) (does not work on -remote) +- .errorhandler # (standard extension. See PEP.) - .messages[] # (standard extension. See PEP) @@ -360,8 +359,7 @@ the class for future connections. - .dbms_version # string identifying the version of the db engine. -- .variantConversions # a map of ado types to the functions used to -import them.(not available on remote) +- .variantConversions # a map of ado types to the functions used to import them. - .supportsTransactions # (bool) this driver is capable of commit()/rollback(). @@ -426,10 +424,9 @@ Cursor attributes (non-standard) - .paramstyle # can be altered by the user to change paramstyle processing. (default taken from connection.) (see below) -- .rs # the internal ADO recordset (local) or raw unpickled data (remote) +- .rs # the internal ADO recordset - .converters[] # a list of input-conversion functions, one per column. - (not available on remote) - .columnNames{} # a dictionary of: ((lower-cased) column name : (column number). @@ -879,6 +876,4 @@ Convenient way to run the main and api tests using different Python versions. - setuptestframework.py: -If run as a main program, initialize a not-really-temporary directory -for the server to use for remote testing. (Otherwise, it is a subroutine -for test setup.) +A subroutine for test setup. diff --git a/adodbapi/remote.py b/adodbapi/remote.py deleted file mode 100644 index 6a192e43ff..0000000000 --- a/adodbapi/remote.py +++ /dev/null @@ -1,624 +0,0 @@ -"""adodbapi.remote - A python DB API 2.0 (PEP 249) interface to Microsoft ADO - -Copyright (C) 2002 Henrik Ekelund, version 2.1 by Vernon Cole -* http://sourceforge.net/projects/pywin32 -* http://sourceforge.net/projects/adodbapi - - This library is free software; you can redistribute it and/or - modify it under the terms of the GNU Lesser General Public - License as published by the Free Software Foundation; either - version 2.1 of the License, or (at your option) any later version. - - This library 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 - Lesser General Public License for more details. - - You should have received a copy of the GNU Lesser General Public - License along with this library; if not, write to the Free Software - Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA - - django adaptations and refactoring thanks to Adam Vandenberg - -DB-API 2.0 specification: http://www.python.org/dev/peps/pep-0249/ -""" - -__version__ = "2.6.0.4" -version = "adodbapi.remote v" + __version__ - -import array -import datetime -import os -import sys -import time - -# Pyro4 is required for server and remote operation --> https://pypi.python.org/pypi/Pyro4/ -try: - import Pyro4 -except ImportError: - print('* * * Sorry, server operation requires Pyro4. Please "pip import" it.') - exit(11) - -import adodbapi -import adodbapi.apibase as api -import adodbapi.process_connect_string -from adodbapi.apibase import ProgrammingError - -sys.excepthook = Pyro4.util.excepthook -Pyro4.config.PREFER_IP_VERSION = 0 # allow system to prefer IPv6 -Pyro4.config.COMMTIMEOUT = 40.0 # a bit longer than the default SQL server Gtimeout -Pyro4.config.SERIALIZER = "pickle" - -try: - verbose = int(os.environ["ADODBAPI_VERBOSE"]) -except: - verbose = False -if verbose: - print(version) - - -# ----------------------------------------------------------- -# conversion functions mandated by PEP 249 -def Binary(aString): - """This function constructs an object capable of holding a binary (long) string value.""" - return bytes(aString) - - -def Date(year, month, day): - return datetime.date(year, month, day) # dateconverter.Date(year,month,day) - - -def Time(hour, minute, second): - return datetime.time(hour, minute, second) # dateconverter.Time(hour,minute,second) - - -def Timestamp(year, month, day, hour, minute, second): - return datetime.datetime(year, month, day, hour, minute, second) - - -def DateFromTicks(ticks): - return Date(*time.gmtime(ticks)[:3]) - - -def TimeFromTicks(ticks): - return Time(*time.gmtime(ticks)[3:6]) - - -def TimestampFromTicks(ticks): - return Timestamp(*time.gmtime(ticks)[:6]) - - -def connect(*args, **kwargs): # --> a remote db-api connection object - """Create and open a remote db-api database connection object""" - # process the argument list the programmer gave us - kwargs = adodbapi.process_connect_string.process(args, kwargs) - # the "proxy_xxx" keys tell us where to find the PyRO proxy server - kwargs.setdefault( - "pyro_connection", "PYRO:ado.connection@%(proxy_host)s:%(proxy_port)s" - ) - if not "proxy_port" in kwargs: - try: - pport = os.environ["PROXY_PORT"] - except KeyError: - pport = 9099 - kwargs["proxy_port"] = pport - if not "proxy_host" in kwargs or not kwargs["proxy_host"]: - try: - phost = os.environ["PROXY_HOST"] - except KeyError: - phost = "[::1]" # '127.0.0.1' - kwargs["proxy_host"] = phost - ado_uri = kwargs["pyro_connection"] % kwargs - # ask PyRO make us a remote connection object - auto_retry = 3 - while auto_retry: - try: - dispatcher = Pyro4.Proxy(ado_uri) - if "comm_timeout" in kwargs: - dispatcher._pyroTimeout = float(kwargs["comm_timeout"]) - uri = dispatcher.make_connection() - break - except Pyro4.core.errors.PyroError: - auto_retry -= 1 - if auto_retry: - time.sleep(1) - else: - raise api.DatabaseError("Cannot create connection to=%s" % ado_uri) - - conn_uri = fix_uri(uri, kwargs) # get a host connection from the proxy server - while auto_retry: - try: - host_conn = Pyro4.Proxy( - conn_uri - ) # bring up an exclusive Pyro connection for my ADO connection - break - except Pyro4.core.errors.PyroError: - auto_retry -= 1 - if auto_retry: - time.sleep(1) - else: - raise api.DatabaseError( - "Cannot create ADO connection object using=%s" % conn_uri - ) - if "comm_timeout" in kwargs: - host_conn._pyroTimeout = float(kwargs["comm_timeout"]) - # make a local clone - myConn = Connection() - while auto_retry: - try: - myConn.connect( - kwargs, host_conn - ) # call my connect method -- hand him the host connection - break - except Pyro4.core.errors.PyroError: - auto_retry -= 1 - if auto_retry: - time.sleep(1) - else: - raise api.DatabaseError( - "Pyro error creating connection to/thru=%s" % repr(kwargs) - ) - except Exception as e: - raise api.DatabaseError( - "Error creating remote connection to=%s, e=%s, %s" - % (repr(kwargs), repr(e), sys.exc_info()[2]) - ) - return myConn - - -def fix_uri(uri, kwargs): - """convert a generic pyro uri with '0.0.0.0' into the address we actually called""" - u = uri.asString() - s = u.split("[::0]") # IPv6 generic address - if len(s) == 1: # did not find one - s = u.split("0.0.0.0") # IPv4 generic address - if len(s) > 1: # found a generic - return kwargs["proxy_host"].join(s) # fill in our address for the host - return uri - - -# # # # # ----- the Class that defines a connection ----- # # # # # -class Connection: - # include connection attributes required by api definition. - Warning = api.Warning - Error = api.Error - InterfaceError = api.InterfaceError - DataError = api.DataError - DatabaseError = api.DatabaseError - OperationalError = api.OperationalError - IntegrityError = api.IntegrityError - InternalError = api.InternalError - NotSupportedError = api.NotSupportedError - ProgrammingError = api.ProgrammingError - # set up some class attributes - paramstyle = api.paramstyle - - @property - def dbapi(self): # a proposed db-api version 3 extension. - "Return a reference to the DBAPI module for this Connection." - return api - - def __init__(self): - self.proxy = None - self.kwargs = {} - self.errorhandler = None - self.supportsTransactions = False - self.paramstyle = api.paramstyle - self.timeout = 30 - self.cursors = {} - - def connect(self, kwargs, connection_maker): - self.kwargs = kwargs - if verbose: - print('%s attempting: "%s"' % (version, repr(kwargs))) - self.proxy = connection_maker - ##try: - ret = self.proxy.connect(kwargs) # ask the server to hook us up - ##except ImportError, e: # Pyro is trying to import pywinTypes.comerrer - ## self._raiseConnectionError(api.DatabaseError, 'Proxy cannot connect using=%s' % repr(kwargs)) - if ret is not True: - self._raiseConnectionError( - api.OperationalError, "Proxy returns error message=%s" % repr(ret) - ) - - self.supportsTransactions = self.getIndexedValue("supportsTransactions") - self.paramstyle = self.getIndexedValue("paramstyle") - self.timeout = self.getIndexedValue("timeout") - if verbose: - print("adodbapi.remote New connection at %X" % id(self)) - - def _raiseConnectionError(self, errorclass, errorvalue): - eh = self.errorhandler - if eh is None: - eh = api.standardErrorHandler - eh(self, None, errorclass, errorvalue) - - def close(self): - """Close the connection now (rather than whenever __del__ is called). - - The connection will be unusable from this point forward; - an Error (or subclass) exception will be raised if any operation is attempted with the connection. - The same applies to all cursor objects trying to use the connection. - """ - for crsr in list(self.cursors.values())[ - : - ]: # copy the list, then close each one - crsr.close() - try: - """close the underlying remote Connection object""" - self.proxy.close() - if verbose: - print("adodbapi.remote Closed connection at %X" % id(self)) - object.__delattr__( - self, "proxy" - ) # future attempts to use closed cursor will be caught by __getattr__ - except Exception: - pass - - def __del__(self): - try: - self.proxy.close() - except: - pass - - def commit(self): - """Commit any pending transaction to the database. - - Note that if the database supports an auto-commit feature, - this must be initially off. An interface method may be provided to turn it back on. - Database modules that do not support transactions should implement this method with void functionality. - """ - if not self.supportsTransactions: - return - result = self.proxy.commit() - if result: - self._raiseConnectionError( - api.OperationalError, "Error during commit: %s" % result - ) - - def _rollback(self): - """In case a database does provide transactions this method causes the the database to roll back to - the start of any pending transaction. Closing a connection without committing the changes first will - cause an implicit rollback to be performed. - """ - result = self.proxy.rollback() - if result: - self._raiseConnectionError( - api.OperationalError, "Error during rollback: %s" % result - ) - - def __setattr__(self, name, value): - if name in ("paramstyle", "timeout", "autocommit"): - if self.proxy: - self.proxy.send_attribute_to_host(name, value) - object.__setattr__(self, name, value) # store attribute locally (too) - - def __getattr__(self, item): - if ( - item == "rollback" - ): # the rollback method only appears if the database supports transactions - if self.supportsTransactions: - return ( - self._rollback - ) # return the rollback method so the caller can execute it. - else: - raise self.ProgrammingError( - "this data provider does not support Rollback" - ) - elif item in ( - "dbms_name", - "dbms_version", - "connection_string", - "autocommit", - ): # 'messages' ): - return self.getIndexedValue(item) - elif item == "proxy": - raise self.ProgrammingError("Attempting to use closed connection") - else: - raise self.ProgrammingError('No remote access for attribute="%s"' % item) - - def getIndexedValue(self, index): - r = self.proxy.get_attribute_for_remote(index) - return r - - def cursor(self): - "Return a new Cursor Object using the connection." - myCursor = Cursor(self) - return myCursor - - def _i_am_here(self, crsr): - "message from a new cursor proclaiming its existence" - self.cursors[crsr.id] = crsr - - def _i_am_closing(self, crsr): - "message from a cursor giving connection a chance to clean up" - try: - del self.cursors[crsr.id] - except: - pass - - def __enter__(self): # Connections are context managers - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - if exc_type: - self._rollback() # automatic rollback on errors - else: - self.commit() - - def get_table_names(self): - return self.proxy.get_table_names() - - -def fixpickle(x): - """pickle barfs on buffer(x) so we pass as array.array(x) then restore to original form for .execute()""" - if x is None: - return None - if isinstance(x, dict): - # for 'named' paramstyle user will pass a mapping - newargs = {} - for arg, val in list(x.items()): - if isinstance(val, memoryview): - newval = array.array("B") - newval.fromstring(val) - newargs[arg] = newval - else: - newargs[arg] = val - return newargs - # if not a mapping, then a sequence - newargs = [] - for arg in x: - if isinstance(arg, memoryview): - newarg = array.array("B") - newarg.fromstring(arg) - newargs.append(newarg) - else: - newargs.append(arg) - return newargs - - -class Cursor: - def __init__(self, connection): - self.command = None - self.errorhandler = None ## was: connection.errorhandler - self.connection = connection - self.proxy = self.connection.proxy - self.rs = None # the fetchable data for this cursor - self.converters = NotImplemented - self.id = connection.proxy.build_cursor() - connection._i_am_here(self) - self.recordset_format = api.RS_REMOTE - if verbose: - print( - "%s New cursor at %X on conn %X" - % (version, id(self), id(self.connection)) - ) - - def prepare(self, operation): - self.command = operation - try: - del self.description - except AttributeError: - pass - self.proxy.crsr_prepare(self.id, operation) - - def __iter__(self): # [2.1 Zamarev] - return iter(self.fetchone, None) # [2.1 Zamarev] - - def __next__(self): - r = self.fetchone() - if r: - return r - raise StopIteration - - def __enter__(self): - "Allow database cursors to be used with context managers." - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - "Allow database cursors to be used with context managers." - self.close() - - def __getattr__(self, key): - if key == "numberOfColumns": - try: - return len(self.rs[0]) - except: - return 0 - if key == "description": - try: - self.description = self.proxy.crsr_get_description(self.id)[:] - return self.description - except TypeError: - return None - if key == "columnNames": - try: - r = dict( - self.proxy.crsr_get_columnNames(self.id) - ) # copy the remote columns - - except TypeError: - r = {} - self.columnNames = r - return r - - if key == "remote_cursor": - raise api.OperationalError - try: - return self.proxy.crsr_get_attribute_for_remote(self.id, key) - except AttributeError: - raise api.InternalError( - 'Failure getting attribute "%s" from proxy cursor.' % key - ) - - def __setattr__(self, key, value): - if key == "arraysize": - self.proxy.crsr_set_arraysize(self.id, value) - if key == "paramstyle": - if value in api.accepted_paramstyles: - self.proxy.crsr_set_paramstyle(self.id, value) - else: - self._raiseCursorError( - api.ProgrammingError, 'invalid paramstyle ="%s"' % value - ) - object.__setattr__(self, key, value) - - def _raiseCursorError(self, errorclass, errorvalue): - eh = self.errorhandler - if eh is None: - eh = api.standardErrorHandler - eh(self.connection, self, errorclass, errorvalue) - - def execute(self, operation, parameters=None): - if self.connection is None: - self._raiseCursorError( - ProgrammingError, "Attempted operation on closed cursor" - ) - self.command = operation - try: - del self.description - except AttributeError: - pass - try: - del self.columnNames - except AttributeError: - pass - fp = fixpickle(parameters) - if verbose > 2: - print( - ( - '%s executing "%s" with params=%s' - % (version, operation, repr(parameters)) - ) - ) - result = self.proxy.crsr_execute(self.id, operation, fp) - if result: # an exception was triggered - self._raiseCursorError(result[0], result[1]) - - def executemany(self, operation, seq_of_parameters): - if self.connection is None: - self._raiseCursorError( - ProgrammingError, "Attempted operation on closed cursor" - ) - self.command = operation - try: - del self.description - except AttributeError: - pass - try: - del self.columnNames - except AttributeError: - pass - sq = [fixpickle(x) for x in seq_of_parameters] - if verbose > 2: - print( - ( - '%s executemany "%s" with params=%s' - % (version, operation, repr(seq_of_parameters)) - ) - ) - self.proxy.crsr_executemany(self.id, operation, sq) - - def nextset(self): - try: - del self.description - except AttributeError: - pass - try: - del self.columnNames - except AttributeError: - pass - if verbose > 2: - print(("%s nextset" % version)) - return self.proxy.crsr_nextset(self.id) - - def callproc(self, procname, parameters=None): - if self.connection is None: - self._raiseCursorError( - ProgrammingError, "Attempted operation on closed cursor" - ) - self.command = procname - try: - del self.description - except AttributeError: - pass - try: - del self.columnNames - except AttributeError: - pass - fp = fixpickle(parameters) - if verbose > 2: - print( - ( - '%s callproc "%s" with params=%s' - % (version, procname, repr(parameters)) - ) - ) - return self.proxy.crsr_callproc(self.id, procname, fp) - - def fetchone(self): - try: - f1 = self.proxy.crsr_fetchone(self.id) - except Exception as e: - self._raiseCursorError(api.DatabaseError, e) - else: - if f1 is None: - return None - self.rs = [f1] - return api.SQLrows(self.rs, 1, self)[ - 0 - ] # new object to hold the results of the fetch - - def fetchmany(self, size=None): - try: - self.rs = self.proxy.crsr_fetchmany(self.id, size) - if not self.rs: - return [] - r = api.SQLrows(self.rs, len(self.rs), self) - return r - except Exception as e: - self._raiseCursorError(api.DatabaseError, e) - - def fetchall(self): - try: - self.rs = self.proxy.crsr_fetchall(self.id) - if not self.rs: - return [] - return api.SQLrows(self.rs, len(self.rs), self) - except Exception as e: - self._raiseCursorError(api.DatabaseError, e) - - def close(self): - if self.connection is None: - return - self.connection._i_am_closing(self) # take me off the connection's cursors list - try: - self.proxy.crsr_close(self.id) - except: - pass - try: - del self.description - except: - pass - try: - del self.rs # let go of the recordset - except: - pass - self.connection = ( - None # this will make all future method calls on me throw an exception - ) - self.proxy = None - if verbose: - print("adodbapi.remote Closed cursor at %X" % id(self)) - - def __del__(self): - try: - self.close() - except: - pass - - def setinputsizes(self, sizes): - pass - - def setoutputsize(self, size, column=None): - pass diff --git a/adodbapi/remote/remote.md b/adodbapi/remote/remote.md deleted file mode 100644 index 22a15b03d0..0000000000 --- a/adodbapi/remote/remote.md +++ /dev/null @@ -1,84 +0,0 @@ -**As of the present time, the remote function is unsupported. -I have received no indication that anyone is actually using this functionality, -and have not created a test system for it. This may change in the future.** - -### Remote Connection - -Also available for the adodbapi package is a second db-api interface module, -which is used to obtain proxy access to a host ADO system. The intended -use of this is to run on a Linux computer allowing it to reach ADO data -sources, such as MS SQL Server databases or \"Jet\" (a.k.a. ACCESS) .mdb -data files. - -It should take the same argument keywords as the host adodbapi server, -and will pass them to it. In addition, other connection keywords are -used to control the connection to the proxy server. - -The extended code for this can be located in the `./remote` directory of the -source code. It is not included in the pywin32 binaries. - -\[Implementation note: adodbapi version 2.5 and 2.6 use PYRO4 for proxy -communication. The will probably change in the future to use ØMQ.\] - -\-\-\-\-\-\-\-\-- - -keywords for remote connections: - -- pyro_connection: - 'PYRO:ado.connection@%(proxy_host)s:%(proxy_port)s' # - used to build the Pyro4 connection to the proxy. - You may never need to change the default. - -- proxy_host: '::1' \# the address of the ADO proxy server. - (default = IPv6 localhost) - -- proxy_port: '9099' # the IP port for the proxy connection. - -To connect to the same database as above, assuming that the Windows box -running the proxy server (an SQLEXPRESS server) was at IPv4 address -10.11.12.13, you would use something like: - -```python -import adodbapi.remote as db - -conn_args = {'host': r"\\SQLEXPRESS\ ", - 'database': "Northwind", - 'user': "guest", - 'password': "12345678"} - -conn_args['connection_string'] = """Provider=SQLOLEDB.1; - User ID=%(user)s; Password=%(password)s; - Initial Catalog=%(database)s; Data Source= %(host)s""" - -conn_args['proxy_host'] = '10.11.12.13' - -myConn = db.connect(conn_args) -``` - -In other words, you only need to add the address of the proxy server to -whatever connection string you would have used at the server itself, -then connect using adodbapi.remote.connect() rather than -adodbapi.connect(). - -\-\-- - -**Some limitations:** Remote connections do not allow varientConversion -customization, nor customized error handlers. - -\-\-\-\-\-\-\-\-\-\-\-\-\-\-\-\-- - -### Proxy Server - -The proxy server is a module for the adodbapi package. It can be run -from the command line using the \"-m\" switch. (\*) The host address and -port number can be passed on the command line or by environment -variables. (You should also set the environment variable -\"PYRO\_HMAC\_KEY\" to some unique string for your installation.) The -environment variables are \"PYRO\_HOST\" and \"PYRO\_PORT\". The command -line arguments are \"HOST=aa.bb.cc.dd\" \"PORT=nnn\". IPv6 addresses -will also work. The default is address ::0 and port 9099 (all IPv6 -interfaces). - -`C:\>python -m adodbapi.remote.server HOST=0.0.0.0` - -ado.connection server running on uri=PYRO:ado.connection\@0.0.0.0:9099 diff --git a/adodbapi/remote/server.py b/adodbapi/remote/server.py deleted file mode 100644 index 5d1114acc4..0000000000 --- a/adodbapi/remote/server.py +++ /dev/null @@ -1,404 +0,0 @@ -"""adodbapi.server - A pyro server for remote adodbapi (from Linux) - -Copyright (C) 2013 by Vernon Cole -* http://sourceforge.net/projects/adodbapi - - This library is free software; you can redistribute it and/or - modify it under the terms of the GNU Lesser General Public - License as published by the Free Software Foundation; either - version 2.1 of the License, or (at your option) any later version. - - This library 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 - Lesser General Public License for more details. - - You should have received a copy of the GNU Lesser General Public - License along with this library; if not, write to the Free Software - Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA - -DB-API 2.0 specification: http://www.python.org/dev/peps/pep-0249/ - -This module source should run correctly in CPython versions 2.6 and later, -or, after running through 2to3.py, CPython 3.2 or later. -""" - -__version__ = "2.6.2.0" -version = "adodbapi.server v" + __version__ - -PYRO_HOST = "::0" # '::0' or '0.0.0.0' means any network -PYRO_PORT = 9099 # may be altered below for Python 3 based servers -PYRO_COMMTIMEOUT = 40 # to be larger than the default database timeout -SERVICE_NAME = "ado.connection" - -import array -import datetime -import os -import sys -import time - -# Pyro4 is required for server and remote operation --> https://pypi.python.org/pypi/Pyro4/ -try: - import Pyro4 -except ImportError: - print('* * * Sorry, server operation requires Pyro4. Please "pip import" it.') - exit(11) -import adodbapi -import adodbapi.apibase as api -import adodbapi.process_connect_string - -try: - pyro_host = os.environ["PYRO_HOST"] -except: - pyro_host = PYRO_HOST -try: - pyro_port = os.environ["PYRO_PORT"] -except: - pyro_port = PYRO_PORT - -for arg in sys.argv[1:]: - if arg.lower().startswith("host"): - try: - pyro_host = arg.split("=")[1] - except Exception: - raise TypeError('Must supply value for argument="%s"' % arg) - - if arg.lower().startswith("port"): - try: - pyro_port = int(arg.split("=")[1]) - except Exception: - raise TypeError('Must supply numeric value for argument="%s"' % arg) - - if arg.lower().startswith("timeout"): - try: - PYRO_COMMTIMEOUT = int(arg.split("=")[1]) - except Exception: - raise TypeError('Must supply numeric value for argument="%s"' % arg) - - if arg.lower().startswith("--verbose"): - try: - verbose = int(arg.split("=")[1]) - except Exception: - raise TypeError('Must supply numeric value for argument="%s"' % arg) - adodbapi.adodbapi.verbose = verbose - else: - verbose = False - -print(adodbapi.adodbapi.version) -print(version) -Pyro4.config.DETAILED_TRACEBACK = True -Pyro4.config.COMMTIMEOUT = PYRO_COMMTIMEOUT -Pyro4.config.AUTOPROXY = False -Pyro4.config.SERVERTYPE = "multiplex" -Pyro4.config.PREFER_IP_VERSION = 0 # allow system to prefer IPv6 -Pyro4.config.SERIALIZERS_ACCEPTED = set( - ["serpent", "pickle"] -) # change when Py2.5 retired - -connection_list = [] -CONNECTION_TIMEOUT = datetime.timedelta(minutes=30) -CONNECTION_REMEMBER = datetime.timedelta(hours=3) -if "--debug" in sys.argv: - CONNECTION_TIMEOUT = datetime.timedelta(minutes=10) - CONNECTION_REMEMBER = datetime.timedelta(minutes=20) -HEARTBEAT_INTERVAL = CONNECTION_TIMEOUT / 10 - -KEEP_RUNNING = True # global value which will kill server when set to False - - -def unfixpickle(x): - """pickle barfs on buffer(x) so we pass as array.array(x) then restore to original form for .execute()""" - if x is None: - return None - if isinstance(x, dict): - # for 'named' paramstyle user will pass a mapping - newargs = {} - for arg, val in list(x.items()): - if isinstance(arg, array.array): - newargs[arg] = bytes(val) - else: - newargs[arg] = val - return newargs - # if not a mapping, then a sequence - newargs = [] - for arg in x: - if isinstance(arg, array.array): - newargs.append(bytes(arg)) - else: - newargs.append(arg) - return newargs - - -class ServerConnection: - def __init__(self): - self.server_connection = None - self.cursors = {} - self.last_used = datetime.datetime.now() - self.timed_out = False - - def _check_timeout(self): - if self.timed_out: - raise api.OperationalError("Remote Connection Timed Out") - - def build_cursor(self): - "Return a new Cursor Object using the connection." - self._check_timeout() - lc = self.server_connection.cursor() # get a new real cursor - self.cursors[lc.id] = lc - return lc.id - - def close(self, remember=False): - global connection_list - for c in list(self.cursors.values())[:]: - c.close() - self.server_connection.close() - self._pyroDaemon.unregister(self) - if not remember: - connection_list.remove(self) - - def connect(self, kwargs): - global connection_list - kw = adodbapi.process_connect_string.process([], kwargs, True) - - if verbose: - print("%s trying to connect %s", (version, repr(kw))) - # kwargs has been loaded with all the values we need - try: - conn = adodbapi.adodbapi.connect(kw) - if verbose: - print("result = %s", repr(conn)) - self.server_connection = conn - connection_list.append(self) - return True - except api.Error as e: - return e - - def commit(self): - try: - self.server_connection.commit() - except api.Error as e: - return str(e) - - def rollback(self): - try: - self.server_connection.rollback() - except api.Error as e: - return str(e) - - def get_table_names(self): - return self.server_connection.get_table_names() - - def get_attribute_for_remote(self, item): - self._check_timeout() - if item == "autocommit": - item = "_autocommit" - if item in ( - "paramstyle", - "messages", - "supportsTransactions", - "dbms_name", - "dbms_version", - "connection_string", - "timeout", - "_autocommit", - ): - return getattr(self.server_connection, item) - raise AttributeError('No provision for remote access to attribute="%s"' % item) - - def send_attribute_to_host( - self, name, value - ): # to change autocommit or paramstyle on host - self._check_timeout() - self.server_connection.__setattr__(name, value) - - # # # # # # following are cursor methods called by the remote (using the connection) with a cursor id "cid" # # # - - def crsr_execute(self, cid, operation, parameters=None): - self._check_timeout() - fp = unfixpickle(parameters) - try: - self.cursors[cid].execute(operation, fp) - except api.Error as e: - try: - errorclass = self.server_connection.messages[0][0] - except: - errorclass = api.Error - return errorclass, str( - e - ) # the error class should have been stored by the standard error handler - - def crsr_prepare(self, cid, operation): - self._check_timeout() - self.cursors[cid].prepare(operation) - - def crsr_executemany(self, cid, operation, seq_of_parameters): - self._check_timeout() - sq = [unfixpickle(x) for x in seq_of_parameters] - self.cursors[cid].executemany(operation, sq) - - def crsr_callproc(self, cid, procname, parameters=None): - self._check_timeout() - fp = unfixpickle(parameters) - return self.cursors[cid].callproc(procname, fp) - - def crsr_fetchone(self, cid): - self._check_timeout() - r = self.cursors[cid].fetchone() - if r is None: - return None - return r[:] - - def crsr_fetchmany(self, cid, size=None): - self._check_timeout() - rows = [] - for row in self.cursors[cid].fetchmany(size): - r = row[:] - rows.append(r) - return rows - - def crsr_fetchall(self, cid): - self._check_timeout() - rows = [] - for row in self.cursors[cid].fetchall(): - rows.append(row[:]) # [item for item in row]) - return rows - - def crsr_get_rowcount(self, cid): - return self.cursors[cid].rowcount - - def crsr_get_description(self, cid): - return self.cursors[cid].description - - def crsr_get_columnNames(self, cid): - return self.cursors[cid].columnNames - - def crsr_nextset(self, cid): - r = self.cursors[cid].nextset() - return r - - def crsr_close(self, cid): - try: - self.cursors[cid].close() - except: - pass - del self.cursors[cid] - - def crsr_set_arraysize(self, cid, value): - self.cursors[cid].arraysize = value - - def crsr_set_conversion(self, cid, index, value): - self.cursors[cid].conversion[index] = value - - def crsr_set_paramstyle(self, cid, value): - self.cursors[cid].paramstyle = value - - def crsr_get_attribute_for_remote(self, cid, item): - if verbose > 3: - print("remote %s asking for=%s" % (cid, item)) - self._check_timeout() - r = getattr(self.cursors[cid], item) # use the built-in function - if verbose > 3: - print("server replying with=%s" % repr(r)) - return r - - def suicide(self): - """shut down the server service""" - global KEEP_RUNNING - KEEP_RUNNING = False - print("Shutdown request received") - - -class ConnectionDispatcher: - def make_connection(self): - new_connection = ServerConnection() - pyro_uri = self._pyroDaemon.register(new_connection) - return pyro_uri - - -class Heartbeat_Timer: - def __init__(self, interval, work_function, tick_result_function): - self.interval = interval - self.last_tick = datetime.datetime.now() - self.work_function = work_function - self.tick_result_function = tick_result_function - - def tick(self): - now = datetime.datetime.now() - if now - self.last_tick > self.interval: - self.last_tick = now - self.work_function() - return self.tick_result_function() - - -def heartbeat_timer_work(): - global connection_list - now = datetime.datetime.now() - for conn in connection_list[:]: # step through a copy of the list - if now - conn.last_used > CONNECTION_TIMEOUT: - try: - if not conn.timed_out: - conn.timed_out = True - conn.close(remember=True) - else: - if now - conn.last_used > CONNECTION_REMEMBER: - connection_list.remove(conn) - except: - pass - - -def still_running(): - return KEEP_RUNNING - - -heartbeat_timer = Heartbeat_Timer( - HEARTBEAT_INTERVAL, heartbeat_timer_work, still_running -) - - -def serve(): - service_name = SERVICE_NAME - if "use_nameserver" in sys.argv: - # advertise self using nameserver - if pyro_host in ("::0", "0.0.0.0"): - raise Warning("Use a specified IP address when using the nameserver") - i = 10 # wait for nameserver to come up - while i: - i -= 1 - time.sleep(2) - try: - ns = Pyro4.naming.locateNS() - break - except Pyro4.errors.PyroError: - if i == 0: - print("..unable to find nameserver..") - sys.exit(1) - ns_p = Pyro4.core.Proxy(ns._pyroUri) - if ":" in pyro_host and pyro_host[0] != "[": - ph = pyro_host.join(("[", "]")) # but [] around bare IPv6 addresses - else: - ph = pyro_host - uri = "PYRO:{}@{}:{}".format(service_name, ph, int(pyro_port)) - ns_p.register(SERVICE_NAME, uri) - print("registered {} to nameserver as={}".format(SERVICE_NAME, uri)) - print("") - - daemon = Pyro4.Daemon(host=pyro_host, port=int(pyro_port)) - uri = daemon.register(ConnectionDispatcher(), service_name) - print("%s server running on uri=%s" % (service_name, uri)) - print("(call using HOST=nnn and PORT=nnn to change interface addresses)") - print("(use ^C or to interrupt...)") - - while KEEP_RUNNING: - try: - daemon.requestLoop(heartbeat_timer.tick) - except KeyboardInterrupt: - break - - -if __name__ == "__main__": - serve() - for conn in connection_list: # clean up when done - try: - conn.server_connection.close() - except: - pass diff --git a/adodbapi/test/adodbapitest.py b/adodbapi/test/adodbapitest.py index 882073c2f5..424d32ca48 100644 --- a/adodbapi/test/adodbapitest.py +++ b/adodbapi/test/adodbapitest.py @@ -71,30 +71,26 @@ def testConnection(self): assert crsr.__class__.__name__ == "Cursor" def testErrorHandlerInherits(self): - if not self.remote: - conn = self.getConnection() - mycallable = lambda connection, cursor, errorclass, errorvalue: 1 - conn.errorhandler = mycallable - crsr = conn.cursor() - assert ( - crsr.errorhandler == mycallable - ), "Error handler on crsr should be same as on connection" + conn = self.getConnection() + mycallable = lambda connection, cursor, errorclass, errorvalue: 1 + conn.errorhandler = mycallable + crsr = conn.cursor() + assert ( + crsr.errorhandler == mycallable + ), "Error handler on crsr should be same as on connection" def testDefaultErrorHandlerConnection(self): - if not self.remote: - conn = self.getConnection() - del conn.messages[:] - try: - conn.close() - conn.commit() # Should not be able to use connection after it is closed - except: - assert len(conn.messages) == 1 - assert len(conn.messages[0]) == 2 - assert conn.messages[0][0] == api.ProgrammingError + conn = self.getConnection() + del conn.messages[:] + try: + conn.close() + conn.commit() # Should not be able to use connection after it is closed + except: + assert len(conn.messages) == 1 + assert len(conn.messages[0]) == 2 + assert conn.messages[0][0] == api.ProgrammingError def testOwnErrorHandlerConnection(self): - if self.remote: # ToDo: use "skip" - return mycallable = ( lambda connection, cursor, errorclass, errorvalue: 1 ) # does not raise anything @@ -117,18 +113,15 @@ def testOwnErrorHandlerConnection(self): def testDefaultErrorHandlerCursor(self): crsr = self.getConnection().cursor() - if not self.remote: - del crsr.messages[:] - try: - crsr.execute("SELECT abbtytddrf FROM dasdasd") - except: - assert len(crsr.messages) == 1 - assert len(crsr.messages[0]) == 2 - assert crsr.messages[0][0] == api.DatabaseError + del crsr.messages[:] + try: + crsr.execute("SELECT abbtytddrf FROM dasdasd") + except: + assert len(crsr.messages) == 1 + assert len(crsr.messages[0]) == 2 + assert crsr.messages[0][0] == api.DatabaseError def testOwnErrorHandlerCursor(self): - if self.remote: # ToDo: should be a "skip" - return mycallable = ( lambda connection, cursor, errorclass, errorvalue: 1 ) # does not raise anything @@ -148,8 +141,6 @@ def testOwnErrorHandlerCursor(self): ), "Setting errorhandler to none should bring back the standard error handler" def testUserDefinedConversions(self): - if self.remote: ## Todo: should be a "skip" - return try: duplicatingConverter = lambda aStringField: aStringField * 2 assert duplicatingConverter("gabba") == "gabbagabba" @@ -158,42 +149,38 @@ def testUserDefinedConversions(self): conn = self.getConnection() # the variantConversions attribute should not exist on a normal connection object self.assertRaises(AttributeError, lambda x: conn.variantConversions[x], [2]) - if not self.remote: - # create a variantConversions attribute on the connection - conn.variantConversions = copy.copy(api.variantConversions) - crsr = conn.cursor() - tabdef = ( - "CREATE TABLE xx_%s (fldData VARCHAR(100) NOT NULL, fld2 VARCHAR(20))" - % config.tmp - ) - crsr.execute(tabdef) - crsr.execute( - "INSERT INTO xx_%s(fldData,fld2) VALUES('gabba','booga')" - % config.tmp - ) - crsr.execute( - "INSERT INTO xx_%s(fldData,fld2) VALUES('hey','yo')" % config.tmp - ) - # change converter for ALL adoStringTypes columns - conn.variantConversions[api.adoStringTypes] = duplicatingConverter - crsr.execute( - "SELECT fldData,fld2 FROM xx_%s ORDER BY fldData" % config.tmp - ) + # create a variantConversions attribute on the connection + conn.variantConversions = copy.copy(api.variantConversions) + crsr = conn.cursor() + tabdef = ( + "CREATE TABLE xx_%s (fldData VARCHAR(100) NOT NULL, fld2 VARCHAR(20))" + % config.tmp + ) + crsr.execute(tabdef) + crsr.execute( + "INSERT INTO xx_%s(fldData,fld2) VALUES('gabba','booga')" % config.tmp + ) + crsr.execute( + "INSERT INTO xx_%s(fldData,fld2) VALUES('hey','yo')" % config.tmp + ) + # change converter for ALL adoStringTypes columns + conn.variantConversions[api.adoStringTypes] = duplicatingConverter + crsr.execute("SELECT fldData,fld2 FROM xx_%s ORDER BY fldData" % config.tmp) - rows = crsr.fetchall() - row = rows[0] - self.assertEqual(row[0], "gabbagabba") - row = rows[1] - self.assertEqual(row[0], "heyhey") - self.assertEqual(row[1], "yoyo") + rows = crsr.fetchall() + row = rows[0] + self.assertEqual(row[0], "gabbagabba") + row = rows[1] + self.assertEqual(row[0], "heyhey") + self.assertEqual(row[1], "yoyo") - upcaseConverter = lambda aStringField: aStringField.upper() - assert upcaseConverter("upThis") == "UPTHIS" + upcaseConverter = lambda aStringField: aStringField.upper() + assert upcaseConverter("upThis") == "UPTHIS" - # now use a single column converter - rows.converters[1] = upcaseConverter # convert second column - self.assertEqual(row[0], "heyhey") # first will be unchanged - self.assertEqual(row[1], "YO") # second will convert to upper case + # now use a single column converter + rows.converters[1] = upcaseConverter # convert second column + self.assertEqual(row[0], "heyhey") # first will be unchanged + self.assertEqual(row[1], "YO") # second will convert to upper case finally: try: @@ -210,7 +197,7 @@ def testUserDefinedConversionForExactNumericTypes(self): # !!! no new code should use this example, to is only a test to see that the # !!! deprecated way of doing this still works. (use connection.variantConversions) # - if not self.remote and sys.version_info < (3, 0): ### Py3 need different test + if sys.version_info < (3, 0): ### Py3 need different test oldconverter = adodbapi.variantConversions[ ado_consts.adNumeric ] # keep old function to restore later @@ -331,11 +318,7 @@ def helpTestDataType( (fldId, inParam), ) except: - if self.remote: - for message in crsr.messages: - print(message) - else: - conn.printADOerrors() + conn.printADOerrors() raise crsr.execute( "SELECT fldData FROM xx_%s WHERE ?=fldID" % config.tmp, [fldId] @@ -671,11 +654,8 @@ def testFetchManyWithArraySize(self): def testErrorConnect(self): conn = self.getConnection() - kw = {} - if "proxy_host" in conn.kwargs: - kw["proxy_host"] = conn.kwargs["proxy_host"] conn.close() - self.assertRaises(api.DatabaseError, self.db, "not a valid connect string", kw) + self.assertRaises(api.DatabaseError, self.db, "not a valid connect string", {}) def testRowIterator(self): self.helpForceDropOnTblTemp() @@ -704,11 +684,7 @@ def testRowIterator(self): (fldId, inParam[0], inParam[1], inParam[2]), ) except: - if self.remote: - for message in crsr.messages: - print(message) - else: - conn.printADOerrors() + conn.printADOerrors() raise crsr.execute( "SELECT fldTwo,fldThree,fldFour FROM xx_%s WHERE ?=fldID" % config.tmp, @@ -778,11 +754,7 @@ def testFormatParamstyle(self): try: crsr.execute(sql, (fldId, inParam)) except: - if self.remote: - for message in crsr.messages: - print(message) - else: - conn.printADOerrors() + conn.printADOerrors() raise crsr.execute( "SELECT fldData, fldConst FROM xx_" + config.tmp + " WHERE %s=fldID", @@ -809,8 +781,7 @@ def testFormatParamstyle(self): assert crsr.command == sel, 'expected:"%s" but found "%s"' % (sel, crsr.command) # test the .parameters attribute - if not self.remote: # parameter list will be altered in transit - self.assertEqual(crsr.parameters, params) + self.assertEqual(crsr.parameters, params) # now make sure the data made it crsr.execute("SELECT fldData FROM xx_%s WHERE fldID=20" % config.tmp) rec = crsr.fetchone() @@ -842,11 +813,7 @@ def testNamedParamstyle(self): {"f_Val": inParam, "Id": fldId}, ) except: - if self.remote: - for message in crsr.messages: - print(message) - else: - conn.printADOerrors() + conn.printADOerrors() raise crsr.execute( "SELECT fldData FROM xx_%s WHERE fldID=:Id" % config.tmp, {"Id": fldId} @@ -892,11 +859,7 @@ def testPyformatParamstyle(self): {"f_Val": inParam, "Id": fldId}, ) except: - if self.remote: - for message in crsr.messages: - print(message) - else: - conn.printADOerrors() + conn.printADOerrors() raise crsr.execute( "SELECT fldData FROM xx_%s WHERE fldID=%%(Id)s" % config.tmp, @@ -945,11 +908,7 @@ def testAutomaticParamstyle(self): (fldId, inParam), ) except: - if self.remote: - for message in crsr.messages: - print(message) - else: - conn.printADOerrors() + conn.printADOerrors() raise trouble = "thi%s :may cause? troub:1e" crsr.execute( @@ -974,11 +933,7 @@ def testAutomaticParamstyle(self): {"f_Val": inParam, "Id": fldId}, ) except: - if self.remote: - for message in crsr.messages: - print(message) - else: - conn.printADOerrors() + conn.printADOerrors() raise crsr.execute( "SELECT fldData FROM xx_%s WHERE :Id=fldID" % config.tmp, {"Id": fldId} @@ -1168,7 +1123,6 @@ def setUp(self): self.conn.timeout = 30 # turn timeout back up self.engine = "MSSQL" self.db = config.dbSqlServerconnect - self.remote = config.connStrSQLServer[2] def tearDown(self): try: @@ -1317,7 +1271,6 @@ def setUp(self): self.conn.timeout = 30 # turn timeout back up self.engine = "ACCESS" self.db = config.dbAccessconnect - self.remote = config.connStrAccess[2] def tearDown(self): try: @@ -1350,7 +1303,6 @@ def setUp(self): self.conn.timeout = 30 # turn timeout back up self.engine = "MySQL" self.db = config.dbMySqlconnect - self.remote = config.connStrMySql[2] def tearDown(self): try: @@ -1420,7 +1372,6 @@ def setUp(self): self.conn.timeout = 30 # turn timeout back up self.engine = "PostgreSQL" self.db = config.dbPostgresConnect - self.remote = config.connStrPostgres[2] def tearDown(self): try: diff --git a/adodbapi/test/adodbapitestconfig.py b/adodbapi/test/adodbapitestconfig.py index 5f412bf1ff..9ff2521c01 100644 --- a/adodbapi/test/adodbapitestconfig.py +++ b/adodbapi/test/adodbapitestconfig.py @@ -39,7 +39,6 @@ --mssql - test against Microsoft SQL server --pg - test against PostgreSQL --mysql - test against MariaDB - --remote= - test using remote server at= (experimental) """ ) exit() @@ -66,15 +65,6 @@ # look here _first_ to find modules sys.path.insert(1, pth) -proxy_host = None -for arg in sys.argv: - if arg.startswith("--remote="): - proxy_host = arg.split("=")[1] - import adodbapi.remote as remote - - break - - # function to clean up the temporary folder -- calling program must run this function before exit. cleanup = setuptestframework.getcleanupfunction() try: @@ -113,25 +103,20 @@ SQL_HOST_NODE = "testsql.2txt.us,1430" if doAccessTest: - if proxy_host: # determine the (probably remote) database file folder - c = {"macro_find_temp_test_path": ["mdb", mdb_name], "proxy_host": proxy_host} - else: - c = {"mdb": setuptestframework.makemdb(testfolder, mdb_name)} - - # macro definition for keyword "provider" using macro "is64bit" -- see documentation - # is64bit will return true for 64 bit versions of Python, so the macro will select the ACE provider - # (If running a remote ADO service, this will test the 64-bitedness of the ADO server.) - c["macro_is64bit"] = [ - "provider", - "Microsoft.ACE.OLEDB.12.0", # 64 bit provider - "Microsoft.Jet.OLEDB.4.0", - ] # 32 bit provider - connStrAccess = "Provider=%(provider)s;Data Source=%(mdb)s" # ;Mode=ReadWrite;Persist Security Info=False;Jet OLEDB:Bypass UserInfo Validation=True" - print( - " ...Testing ACCESS connection to {} file...".format( - c.get("mdb", "remote .mdb") - ) - ) + c = { + "mdb": setuptestframework.makemdb(testfolder, mdb_name), + # macro definition for keyword "provider" using macro "is64bit" -- see documentation + # is64bit will return true for 64 bit versions of Python, so the macro will select the ACE provider + "macro_is64bit": [ + "provider", + "Microsoft.ACE.OLEDB.12.0", # 64 bit provider + "Microsoft.Jet.OLEDB.4.0", # 32 bit provider + ], + } + + # ;Mode=ReadWrite;Persist Security Info=False;Jet OLEDB:Bypass UserInfo Validation=True" + connStrAccess = "Provider=%(provider)s;Data Source=%(mdb)s" + print(" ...Testing ACCESS connection to {} file...".format(c["mdb"])) doAccessTest, connStrAccess, dbAccessconnect = tryconnection.try_connection( verbose, connStrAccess, 10, **c ) @@ -146,8 +131,6 @@ "macro_auto_security": "security", "provider": "MSOLEDBSQL; MARS Connection=True", } - if proxy_host: - c["proxy_host"] = proxy_host connStr = "Provider=%(provider)s; Initial Catalog=%(database)s; Data Source=%(host)s; %(security)s;" print(" ...Testing MS-SQL login to {}...".format(c["host"])) ( @@ -165,8 +148,6 @@ "port": "3330", # note the nonstandard port for obfuscation "driver": "MySQL ODBC 5.1 Driver", } # or _driver="MySQL ODBC 3.51 Driver - if proxy_host: - c["proxy_host"] = proxy_host c["macro_is64bit"] = [ "provider", "Provider=MSDASQL;", @@ -194,8 +175,6 @@ ] # get driver from http://www.postgresql.org/ftp/odbc/versions/ # test using positional and keyword arguments (bad example for real code) - if proxy_host: - kws["proxy_host"] = proxy_host print(" ...Testing PostgreSQL login to {}...".format(_computername)) doPostgresTest, connStrPostgres, dbPostgresConnect = tryconnection.try_connection( verbose, diff --git a/adodbapi/test/setuptestframework.py b/adodbapi/test/setuptestframework.py index cb27a63042..53dda825ce 100644 --- a/adodbapi/test/setuptestframework.py +++ b/adodbapi/test/setuptestframework.py @@ -103,9 +103,3 @@ def makemdb(testfolder, mdb_name): shutil.copy(mdbName, _accessdatasource) return _accessdatasource - - -if __name__ == "__main__": - print("Setting up a Jet database for server to use for remote testing...") - temp = maketemp() - makemdb(temp, "server_test.mdb") diff --git a/adodbapi/test/test_adodbapi_dbapi20.py b/adodbapi/test/test_adodbapi_dbapi20.py index ccf7e7cbf5..e5d36282c9 100644 --- a/adodbapi/test/test_adodbapi_dbapi20.py +++ b/adodbapi/test/test_adodbapi_dbapi20.py @@ -80,11 +80,6 @@ driver = "Microsoft.Jet.OLEDB.4.0" testmdb = setuptestframework.makemdb(testfolder) connStr = r"Provider=%s;Data Source=%s" % (driver, testmdb) -else: # try a remote connection to an SQL server - conn_kws["proxy_host"] = "25.44.77.176" - import adodbapi.remote - - db = adodbapi.remote print("Using Connection String like=%s" % connStr) print("Keywords=%s" % repr(conn_kws)) diff --git a/adodbapi/test/tryconnection.py b/adodbapi/test/tryconnection.py index 9d3901a8c0..e0f7e06025 100644 --- a/adodbapi/test/tryconnection.py +++ b/adodbapi/test/tryconnection.py @@ -1,6 +1,3 @@ -remote = False # automatic testing of remote access has been removed here - - def try_connection(verbose, *args, **kwargs): import adodbapi @@ -18,7 +15,7 @@ def try_connection(verbose, *args, **kwargs): print(" (successful)") - return True, (args, kwargs, remote), dbconnect + return True, (args, kwargs), dbconnect def try_operation_with_expected_exception(