![[tokkee]](http://tokkee.org/images/avatar.png){kind=avatar}
index 5841f8419e2fa9add8c1d1aefc662bc8dc1f33f0..eb7104105de465b57ae04e77ac3db8a33ddbffd1 100644 (file)
--- a/test/test_postgresql.py
+++ b/test/test_postgresql.py
# FOR A PARTICULAR PURPOSE. THE CODE PROVIDED HEREUNDER IS ON AN "AS IS"
# BASIS, AND THERE IS NO OBLIGATION WHATSOEVER TO PROVIDE MAINTENANCE,
# SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
-#
-# $Id: test_postgresql.py,v 1.4 2003-11-14 00:11:19 richard Exp $
+#
+# $Id: test_postgresql.py,v 1.13 2006-08-23 12:57:10 schlatterbeck Exp $
-import sys, unittest, os, shutil, time, popen2
+import unittest
from roundup.hyperdb import DatabaseError
from db_test_base import DBTest, ROTest, config, SchemaTest, ClassicInitTest
+from db_test_base import ConcurrentDBTest, FilterCacheTest
-# Postgresql connection data
-# NOTE: THIS MUST BE A LOCAL DATABASE
-config.POSTGRESQL_DATABASE = {'database': 'rounduptest'}
-
-from roundup import backends
-
-def db_create():
- """Clear all database contents and drop database itself"""
- name = config.POSTGRESQL_DATABASE['database']
- cout,cin = popen2.popen4('createdb %s'%name)
- cin.close()
- response = cout.read().split('\n')[0]
- if response.find('FATAL') != -1 or response.find('ERROR') != -1:
- raise RuntimeError, response
-
-def db_nuke(fail_ok=0):
- """Clear all database contents and drop database itself"""
- name = config.POSTGRESQL_DATABASE['database']
- cout,cin = popen2.popen4('dropdb %s'%name)
- cin.close()
- response = cout.read().split('\n')[0]
- if response.endswith('does not exist') and fail_ok:
- return
- if response.find('FATAL') != -1 or response.find('ERROR') != -1:
- raise RuntimeError, response
- if os.path.exists(config.DATABASE):
- shutil.rmtree(config.DATABASE)
-
-def db_exists(config):
- """Check if database already exists"""
- try:
- db = Database(config, 'admin')
- return 1
- except:
- return 0
+from roundup.backends import get_backend, have_backend
class postgresqlOpener:
- if hasattr(backends, 'postgresql'):
- from roundup.backends import postgresql as module
+ if have_backend('postgresql'):
+ module = get_backend('postgresql')
def setUp(self):
- db_nuke(1)
- db_create()
+ pass
def tearDown(self):
self.nuke_database()
def nuke_database(self):
- # clear out the database - easiest way is to nuke and re-created it
- db_nuke()
- db_create()
+ # clear out the database - easiest way is to nuke and re-create it
+ self.module.db_nuke(config)
class postgresqlDBTest(postgresqlOpener, DBTest):
def setUp(self):
DBTest.tearDown(self)
postgresqlOpener.tearDown(self)
- def testFilteringIntervalSort(self):
- # PostgreSQL sorts NULLs differently to other databases (others
- # treat it as lower than real values, PG treats it as higher)
- ae, filt = self.filteringSetup()
- # ascending should sort None, 1:10, 1d
- ae(filt(None, {}, ('+','foo'), (None,None)), ['4', '1', '2', '3'])
- # descending should sort 1d, 1:10, None
- ae(filt(None, {}, ('-','foo'), (None,None)), ['3', '2', '1', '4'])
-
class postgresqlROTest(postgresqlOpener, ROTest):
def setUp(self):
postgresqlOpener.setUp(self)
ROTest.tearDown(self)
postgresqlOpener.tearDown(self)
+class postgresqlConcurrencyTest(postgresqlOpener, ConcurrentDBTest):
+ backend = 'postgresql'
+ def setUp(self):
+ postgresqlOpener.setUp(self)
+ ConcurrentDBTest.setUp(self)
+
+ def tearDown(self):
+ ConcurrentDBTest.tearDown(self)
+ postgresqlOpener.tearDown(self)
+
+class postgresqlFilterCacheTest(postgresqlOpener, FilterCacheTest):
+ backend = 'postgresql'
+ def setUp(self):
+ postgresqlOpener.setUp(self)
+ FilterCacheTest.setUp(self)
+
+ def tearDown(self):
+ FilterCacheTest.tearDown(self)
+ postgresqlOpener.tearDown(self)
+
class postgresqlSchemaTest(postgresqlOpener, SchemaTest):
def setUp(self):
postgresqlOpener.setUp(self)
class postgresqlClassicInitTest(postgresqlOpener, ClassicInitTest):
backend = 'postgresql'
- extra_config = "POSTGRESQL_DATABASE = {'database': 'rounduptest'}"
def setUp(self):
postgresqlOpener.setUp(self)
ClassicInitTest.setUp(self)
ClassicInitTest.tearDown(self)
postgresqlOpener.tearDown(self)
+from session_common import RDBMSTest
+class postgresqlSessionTest(postgresqlOpener, RDBMSTest):
+ def setUp(self):
+ postgresqlOpener.setUp(self)
+ RDBMSTest.setUp(self)
+ def tearDown(self):
+ RDBMSTest.tearDown(self)
+ postgresqlOpener.tearDown(self)
+
def test_suite():
suite = unittest.TestSuite()
- if not hasattr(backends, 'postgresql'):
+ if not have_backend('postgresql'):
+ print "Skipping postgresql tests"
return suite
- # Check if we can run postgresql tests
+ # make sure we start with a clean slate
+ if postgresqlOpener.module.db_exists(config):
+ postgresqlOpener.module.db_nuke(config, 1)
+
+ # TODO: Check if we can run postgresql tests
print 'Including postgresql tests'
suite.addTest(unittest.makeSuite(postgresqlDBTest))
suite.addTest(unittest.makeSuite(postgresqlROTest))
suite.addTest(unittest.makeSuite(postgresqlSchemaTest))
suite.addTest(unittest.makeSuite(postgresqlClassicInitTest))
+ suite.addTest(unittest.makeSuite(postgresqlSessionTest))
+ suite.addTest(unittest.makeSuite(postgresqlConcurrencyTest))
+ suite.addTest(unittest.makeSuite(postgresqlFilterCacheTest))
return suite
+# vim: set et sts=4 sw=4 :