145 lines
4.0 KiB
Python
145 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
import os
|
|
import unittest
|
|
import sqlite3
|
|
import tempfile
|
|
|
|
import bottle
|
|
from bottle.ext import sqlite
|
|
|
|
''' python3 moves unicode to str '''
|
|
try:
|
|
unicode
|
|
except NameError:
|
|
unicode = str
|
|
|
|
|
|
class SQLiteTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.app = bottle.Bottle(catchall=False)
|
|
_, dbfile = tempfile.mkstemp(suffix='.sqlite')
|
|
self.plugin = self.app.install(sqlite.Plugin(dbfile=dbfile))
|
|
|
|
self.conn = sqlite3.connect(dbfile)
|
|
self.conn.execute("CREATE TABLE todo (id INTEGER PRIMARY KEY, task char(100) NOT NULL)")
|
|
self.conn.commit()
|
|
|
|
def tearDown(self):
|
|
os.unlink(self.plugin.dbfile)
|
|
|
|
def test_with_keyword(self):
|
|
@self.app.get('/')
|
|
def test(db):
|
|
self.assertEqual(type(db), type(sqlite3.connect(':memory:')))
|
|
self._request('/')
|
|
|
|
def test_without_keyword(self):
|
|
@self.app.get('/')
|
|
def test_1():
|
|
pass
|
|
self._request('/')
|
|
|
|
@self.app.get('/2')
|
|
def test_2(**kw):
|
|
self.assertFalse('db' in kw)
|
|
self._request('/2')
|
|
|
|
def test_install_conflicts(self):
|
|
self.app.install(sqlite.Plugin(keyword='db2'))
|
|
|
|
@self.app.get('/')
|
|
def test(db, db2):
|
|
pass
|
|
|
|
# I have two plugins working with different names
|
|
self._request('/')
|
|
|
|
def test_text_factory(self):
|
|
# set text factory to str, unicode (default) would cause
|
|
# PrammingError: You must not use 8-bit bytestrings .. exception
|
|
self.app.install(sqlite.Plugin(keyword='db2',text_factory=str))
|
|
|
|
@self.app.get('/')
|
|
def test(db, db2):
|
|
char = 'ööö'
|
|
db2.execute("CREATE TABLE todo (id INTEGER PRIMARY KEY, task char(100) NOT NULL)")
|
|
db2.execute("INSERT INTO todo (id,task) VALUES ('1',:TEST)", { "TEST": char })
|
|
count = len(db2.execute("SELECT * FROM todo").fetchall())
|
|
self.assertEqual(count, 1)
|
|
|
|
self._request('/')
|
|
|
|
def test_text_factory_fail(self):
|
|
self.app.install(sqlite.Plugin(keyword='db3',text_factory=unicode))
|
|
|
|
@self.app.get('/')
|
|
def test(db, db3):
|
|
char = 'ööö'
|
|
db3.execute("CREATE TABLE todo (id INTEGER PRIMARY KEY, task char(100) NOT NULL)")
|
|
try:
|
|
db3.execute("INSERT INTO todo (id,task) VALUES ('1',:TEST)", { "TEST": char })
|
|
except sqlite3.ProgrammingError as e:
|
|
pass
|
|
|
|
self._request('/')
|
|
|
|
def test_raise_sqlite_integrity_error(self):
|
|
@self.app.get('/')
|
|
def test(db):
|
|
# task can not be null, raise an IntegrityError
|
|
db.execute("INSERT INTO todo (id) VALUES (1)")
|
|
|
|
# TODO: assert HTTPError 500
|
|
self._request('/')
|
|
self.assert_records(0)
|
|
|
|
def test_autocommit(self):
|
|
@self.app.get('/')
|
|
def test(db):
|
|
self._insert_into(db)
|
|
|
|
self._request('/')
|
|
self.assert_records(1)
|
|
|
|
def test_not_autocommit(self):
|
|
@self.app.get('/', sqlite={'autocommit': False})
|
|
def test(db):
|
|
self._insert_into(db)
|
|
|
|
self._request('/')
|
|
self.assert_records(0)
|
|
|
|
def test_commit_on_redirect(self):
|
|
@self.app.get('/')
|
|
def test(db):
|
|
self._insert_into(db)
|
|
bottle.redirect('/')
|
|
|
|
self._request('/')
|
|
self.assert_records(1)
|
|
|
|
def test_commit_on_abort(self):
|
|
@self.app.get('/')
|
|
def test(db):
|
|
self._insert_into(db)
|
|
bottle.abort()
|
|
|
|
self._request('/')
|
|
self.assert_records(0)
|
|
|
|
def _request(self, path, method='GET'):
|
|
return self.app({'PATH_INFO': path, 'REQUEST_METHOD': method},
|
|
lambda x, y: None)
|
|
|
|
def _insert_into(self, db):
|
|
sql = "INSERT INTO todo (task) VALUES ('PASS')"
|
|
db.execute(sql)
|
|
|
|
def assert_records(self, count):
|
|
cursor = self.conn.execute("SELECT COUNT(*) FROM todo")
|
|
self.assertEqual((count,), cursor.fetchone())
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|