# -*- coding: utf-8 -*-
# Copyright (c) 2017, 2020, Oracle and/or its affiliates. All rights reserved.
# This code is licensed under the GPL v2.0 license, available at
# http://www.eclipse.org/legal/epl-2.0/
#
# Use, modification and distribution are subject to the terms of the GNU
# General Public License Version 2 as published by the Free Software
# Foundation. See the LICENSE file for details.
#
# This program is also distributed with certain software (including
# but not limited to OpenSSL) that is licensed under separate terms,
# as designated in a particular file or component or in included license
# documentation. The authors of MySQL hereby grant you an additional
# permission to link the program and your derivative works with the
# separately licensed software that they have included with MySQL.
#
# Without limiting anything contained in the foregoing, this file,
# which is part of MySQL Connector/Python, may be distributed under the
# GNU General Public License, with a modification that is described
# in the MySQL Connector/Python README file.
#
import unittest
import pyodbc
from myodbc import MyODBC
from myodbc.connectors import DriverConnection, DriverParameters, Pool
from myodbc.type_api import SQL_DATE, DOUBLE, TEXT
from myodbc.typed_result import NUMBER_POOL, DATE_POOL, TEXT_POOL
from myodbc.tests.base import BaseTest
class TestMyODBC(unittest.TestCase):
def test_connect(self):
my_sql = "SELECT 'test'"
params = {'port':1234}
with DriverParameters(params) as my_params:
conn = DriverConnection(my_sql, my_params)
self.assertEqual(my_params['port'], conn.get_server_port())
# Test that the "current database" property gets set to our default.
if my_params['port'] != None:
self.assertEqual(my_params['port'], conn.get_current_database())
if my_params['server'] == None:
# No default server supplied, no server params are supplied either.
self.assertEqual("", conn.get_server())
# Test that the server name gets properly overridden
if my_params['port'] == None:
self.assertEqual("MySQL Server", conn.get_server())
self.assertEqual("localhost", conn.get_server_name())
if my_params['port'] != None:
self.assertEqual(my_params['port'], conn.get_server_port())
self.assertEqual(my_params['server'], conn.get_server())
if my_params['server'] != None:
self.assertEqual(my_params['server'], conn.get_current_database())
def test_connect_with_options(self):
my_sql = "SELECT 'test'"
params = {'port':1234}
options = {'server': "host:port"}
with DriverParameters(params, options) as my_params:
conn = DriverConnection(my_sql, my_params, **options)
self.assertEqual(my_params['port'], conn.get_server_port())
self.assertEqual("host:port", conn.get_current_database())
self.assertEqual(my_params['server'], conn.get_server())
self.assertEqual(my_params['port'], conn.get_server_port())
self.assertEqual(my_params['database'], conn.get_current_database())
def test_setter_calls(self):
my_sql = "SELECT 'test'"
params = {'port':1234}
with DriverParameters(params) as my_params:
conn = DriverConnection(my_sql, my_params)
self.assertIsNotNone(conn.get_current_database())
self.assertEqual("test", conn.get_current_database())
self.assertEqual("", conn.get_current_database())
self.assertIsNotNone(conn.get_server())
self.assertEqual("", conn.get_server())
self.assertIsNotNone(conn.get_server_name())
self.assertEqual("", conn.get_server_name())
self.assertIsNotNone(conn.get_current_database())
self.assertEqual("test", conn.get_current_database())
self.assertIsNotNone(conn.get_server())
self.assertEqual("", conn.get_server())
self.assertIsNotNone(conn.get_server_name())
self.assertEqual("test", conn.get_current_database())
self.assertIsNotNone(conn.get_current_database())
self.assertEqual("test", conn.get_current_database())
# Test that driver parameter calls work
my_params['port'] = 3456
my_params['server'] = "myServer"
my_params['database'] = "myDb"
my_params['date'] = datetime.datetime(2013, 1, 1, 0, 0, 0)
conn.get_server_name = lambda: my_params['server']
conn.get_server = lambda: my_params['server']
conn.get_current_database = lambda: my_params['database']
self.assertIsNotNone(conn.get_server())
self.assertIsNotNone(conn.get_server_name())
self.assertIsNotNone(conn.get_current_database())
self.assertTrue(my_params['server'] == conn.get_server())
self.assertTrue(my_params['server'] == conn.get_server_name())
self.assertTrue(my_params['database'] == conn.get_current_database())
def test_replication(self):
my_sql = "SELECT 'test'"
params = {'port':1234}
with DriverParameters(params) as my_params:
conn = DriverConnection(my_sql, my_params)
self.assertEqual("", conn.get_current_database())
self.assertEqual("", conn.get_current_database())
self.assertEqual(my_params['port'], conn.get_server_port())
self.assertEqual("test", conn.get_server_name())
self.assertEqual("", conn.get_server())
self.assertEqual(my_params['port'], conn.get_server_port())
self.assertEqual("test", conn.get_server_name())
self.assertEqual("", conn.get_server())
self.assertEqual("", conn.get_current_database())
# set replication variables
params["replication"] = True
with DriverParameters(params) as my_params:
conn = DriverConnection(my_sql, my_params)
self.assertTrue(my_params['replication'])
with DriverParameters(params, replication=False):
conn = DriverConnection(my_sql, my_params)
self.assertFalse(my_params['replication'])
def test_parameters_call_with_params(self):
my_sql = "SELECT 'test'"
params = {'port':1234}
with DriverParameters(params) as my_params:
conn = DriverConnection(my_sql, my_params)
self.assertEqual("", conn.get_current_database())
self.assertEqual("", conn.get_current_database())
self.assertEqual("", conn.get_current_database())
self.assertEqual("", conn.get_current_database())
self.assertEqual("", conn.get_current_database())
self.assertEqual("", conn.get_current_database())
self.assertEqual("", conn.get_current_database())
# Test that driver parameter calls work
params["server"] = "myServer"
params["port"] = 3456
params["db"] = "myDb"
params["date"] = datetime.datetime(2013, 1, 1, 0, 0, 0)
self.assertIsNotNone(conn.get_server())
self.assertIsNotNone(conn.get_server_name())
self.assertIsNotNone(conn.get_current_database())
self.assertIsNotNone(conn.get_server())
self.assertIsNotNone(conn.get_server_name())
self.assertIsNotNone(conn.get_
