Skip to content

Commit

Permalink
Add unit tests for dbapi module
Browse files Browse the repository at this point in the history
Used dbapi-compliance [1] package to test module according to pep-249
specification. Not implemented features are skipped in the tests.

Added dbapi-compliance package to test.sh requirements and appveyor.yml

[1] https://github.com/baztian/dbapi-compliance/
  • Loading branch information
artembo committed Oct 18, 2020
1 parent d9235af commit 4f51516
Show file tree
Hide file tree
Showing 4 changed files with 144 additions and 3 deletions.
2 changes: 1 addition & 1 deletion appveyor.yml
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ install:
# install runtime dependencies
- "%PYTHON%\\python.exe -m pip install -r requirements.txt"
# install testing dependencies
- "%PYTHON%\\python.exe -m pip install pyyaml%PYYAML%"
- "%PYTHON%\\python.exe -m pip install pyyaml%PYYAML% dbapi-compliance==1.15.0"

build: off

Expand Down
3 changes: 2 additions & 1 deletion test.sh
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,8 @@ pip install "${PYTHON_MSGPACK:-msgpack==1.0.0}"
python -c 'import msgpack; print(msgpack.version)'

# Install testing dependencies.
pip install pyyaml
pip install -r requirements.txt
pip install pyyaml dbapi-compliance==1.15.0

# Run tests.
python setup.py test
3 changes: 2 additions & 1 deletion unit/suites/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,12 @@
from .test_reconnect import TestSuite_Reconnect
from .test_mesh import TestSuite_Mesh
from .test_execute import TestSuite_Execute
from .test_dbapi import TestSuite_DBAPI

test_cases = (TestSuite_Schema_UnicodeConnection,
TestSuite_Schema_BinaryConnection,
TestSuite_Request, TestSuite_Protocol, TestSuite_Reconnect,
TestSuite_Mesh, TestSuite_Execute)
TestSuite_Mesh, TestSuite_Execute, TestSuite_DBAPI)

def load_tests(loader, tests, pattern):
suite = unittest.TestSuite()
Expand Down
139 changes: 139 additions & 0 deletions unit/suites/test_dbapi.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
# -*- coding: utf-8 -*-

from __future__ import print_function

import sys
import unittest

import dbapi20

import tarantool
from tarantool import dbapi
from .lib.tarantool_server import TarantoolServer


class TestSuite_DBAPI(dbapi20.DatabaseAPI20Test):
table_prefix = 'dbapi20test_' # If you need to specify a prefix for tables

ddl0 = 'create table %s (id INTEGER PRIMARY KEY AUTOINCREMENT, ' \
'name varchar(20))'
ddl1 = 'create table %sbooze (name varchar(20) primary key)' % table_prefix
ddl2 = 'create table %sbarflys (name varchar(20) primary key, ' \
'drink varchar(30))' % table_prefix

ddl_params = [
{'id': None, 'name': 'Michael'},
{'id': None, 'name': 'Mary'},
{'id': None, 'name': 'John'},
{'id': None, 'name': 'Ruth'},
{'id': None, 'name': 'Rachel'}
]

@classmethod
def setUpClass(self):
print(' DBAPI '.center(70, '='), file=sys.stderr)
print('-' * 70, file=sys.stderr)
self.srv = TarantoolServer()
self.srv.script = 'unit/suites/box.lua'
self.srv.start()
self.con = tarantool.Connection(self.srv.host, self.srv.args['primary'])
self.driver = dbapi
self.connect_kw_args = dict(
host=self.srv.host,
port=self.srv.args['primary'])

def setUp(self):
# prevent a remote tarantool from clean our session
if self.srv.is_started():
self.srv.touch_lock()
self.con.flush_schema()

# grant full access to guest
self.srv.admin("box.schema.user.grant('guest', 'create,read,write,"
"execute', 'universe')")

@classmethod
def tearDownClass(self):
self.con.close()
self.srv.stop()
self.srv.clean()

def test_rowcount(self):
con = self._connect()
try:
cur = con.cursor()
self.executeDDL1(cur)
dbapi20._failUnless(self,cur.rowcount in (-1,1), # Bug #543885
'cursor.rowcount should be -1 or 0 after executing no-result '
'statements' + str(cur.rowcount)
)
cur.execute("%s into %sbooze values ('Victoria Bitter')" % (
self.insert, self.table_prefix
))
dbapi20._failUnless(self,cur.rowcount in (-1,1),
'cursor.rowcount should == number or rows inserted, or '
'set to -1 after executing an insert statement'
)
cur.execute("select name from %sbooze" % self.table_prefix)
dbapi20._failUnless(self,cur.rowcount in (-1,1),
'cursor.rowcount should == number of rows returned, or '
'set to -1 after executing a select statement'
)
self.executeDDL2(cur)
dbapi20._failUnless(self,cur.rowcount in (-1,1), # Bug #543885
'cursor.rowcount should be -1 or 0 after executing no-result '
'statements'
)
finally:
con.close()

@unittest.skip('Not implemented')
def test_Binary(self):
pass

@unittest.skip('Not implemented')
def test_STRING(self):
pass

@unittest.skip('Not implemented')
def test_BINARY(self):
pass

@unittest.skip('Not implemented')
def test_NUMBER(self):
pass

@unittest.skip('Not implemented')
def test_DATETIME(self):
pass

@unittest.skip('Not implemented')
def test_ROWID(self):
pass

@unittest.skip('Not implemented')
def test_Date(self):
pass

@unittest.skip('Not implemented')
def test_Time(self):
pass

@unittest.skip('Not implemented')
def test_Timestamp(self):
pass

@unittest.skip('Not implemented as optional.')
def test_nextset(self):
pass

@unittest.skip('To do')
def test_callproc(self):
pass

def test_setoutputsize(self): # Do nothing
pass

@unittest.skip('To do')
def test_description(self):
pass

0 comments on commit 4f51516

Please sign in to comment.