test/test_postgresql.py

   1 #
   2 # Copyright (c) 2001 Bizar Software Pty Ltd (http://www.bizarsoftware.com.au/)
   3 # This module is free software, and you may redistribute it and/or modify
   4 # under the same terms as Python, so long as this copyright message and
   5 # disclaimer are retained in their original form.
   6 #
   7 # IN NO EVENT SHALL BIZAR SOFTWARE PTY LTD BE LIABLE TO ANY PARTY FOR
   8 # DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES ARISING
   9 # OUT OF THE USE OF THIS CODE, EVEN IF THE AUTHOR HAS BEEN ADVISED OF THE
  10 # POSSIBILITY OF SUCH DAMAGE.
  11 #
  12 # BIZAR SOFTWARE PTY LTD SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING,
  13 # BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
  14 # FOR A PARTICULAR PURPOSE.  THE CODE PROVIDED HEREUNDER IS ON AN "AS IS"
  15 # BASIS, AND THERE IS NO OBLIGATION WHATSOEVER TO PROVIDE MAINTENANCE,
  16 # SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
  17 #
  18 # $Id: test_postgresql.py,v 1.3 2003-11-11 11:19:18 richard Exp $
  19
  20 import sys, unittest, os, shutil, time, popen2
  21
  22 from roundup.hyperdb import DatabaseError
  23
  24 from db_test_base import DBTest, ROTest, config, SchemaTest, ClassicInitTest
  25
  26 # Postgresql connection data
  27 # NOTE: THIS MUST BE A LOCAL DATABASE
  28 config.POSTGRESQL_DATABASE = {'database': 'rounduptest'}
  29
  30 from roundup import backends
  31
  32 def db_create():
  33     """Clear all database contents and drop database itself"""
  34     name = config.POSTGRESQL_DATABASE['database']
  35     cout,cin = popen2.popen4('createdb %s'%name)
  36     cin.close()
  37     response = cout.read().split('\n')[0]
  38     if response.find('FATAL') != -1 or response.find('ERROR') != -1:
  39         raise RuntimeError, response
  40
  41 def db_nuke(fail_ok=0):
  42     """Clear all database contents and drop database itself"""
  43     name = config.POSTGRESQL_DATABASE['database']
  44     cout,cin = popen2.popen4('dropdb %s'%name)
  45     cin.close()
  46     response = cout.read().split('\n')[0]
  47     if response.endswith('does not exist') and fail_ok:
  48         return
  49     if response.find('FATAL') != -1 or response.find('ERROR') != -1:
  50         raise RuntimeError, response
  51     if os.path.exists(config.DATABASE):
  52         shutil.rmtree(config.DATABASE)
  53
  54 def db_exists(config):
  55     """Check if database already exists"""
  56     try:
  57         db = Database(config, 'admin')
  58         return 1
  59     except:
  60         return 0
  61
  62 class postgresqlOpener:
  63     if hasattr(backends, 'postgresql'):
  64         from roundup.backends import postgresql as module
  65
  66     def setUp(self):
  67         db_nuke(1)
  68         db_create()
  69
  70     def tearDown(self):
  71         db_nuke()
  72
  73 class postgresqlDBTest(postgresqlOpener, DBTest):
  74     def setUp(self):
  75         postgresqlOpener.setUp(self)
  76         DBTest.setUp(self)
  77
  78     def tearDown(self):
  79         DBTest.tearDown(self)
  80         postgresqlOpener.tearDown(self)
  81
  82     def testFilteringIntervalSort(self):
  83         # PostgreSQL sorts NULLs differently to other databases (others
  84         # treat it as lower than real values, PG treats it as higher)
  85         ae, filt = self.filteringSetup()
  86         # ascending should sort None, 1:10, 1d
  87         ae(filt(None, {}, ('+','foo'), (None,None)), ['4', '1', '2', '3'])
  88         # descending should sort 1d, 1:10, None
  89         ae(filt(None, {}, ('-','foo'), (None,None)), ['3', '2', '1', '4'])
  90
  91 class postgresqlROTest(postgresqlOpener, ROTest):
  92     def setUp(self):
  93         postgresqlOpener.setUp(self)
  94         ROTest.setUp(self)
  95
  96     def tearDown(self):
  97         ROTest.tearDown(self)
  98         postgresqlOpener.tearDown(self)
  99
 100 class postgresqlSchemaTest(postgresqlOpener, SchemaTest):
 101     def setUp(self):
 102         postgresqlOpener.setUp(self)
 103         SchemaTest.setUp(self)
 104
 105     def tearDown(self):
 106         SchemaTest.tearDown(self)
 107         postgresqlOpener.tearDown(self)
 108
 109 class postgresqlClassicInitTest(postgresqlOpener, ClassicInitTest):
 110     backend = 'postgresql'
 111     extra_config = "POSTGRESQL_DATABASE = {'database': 'rounduptest'}"
 112     def setUp(self):
 113         postgresqlOpener.setUp(self)
 114         ClassicInitTest.setUp(self)
 115
 116     def tearDown(self):
 117         ClassicInitTest.tearDown(self)
 118         postgresqlOpener.tearDown(self)
 119
 120 def test_suite():
 121     suite = unittest.TestSuite()
 122     if not hasattr(backends, 'postgresql'):
 123         return suite
 124
 125     # Check if we can run postgresql tests
 126     print 'Including postgresql tests'
 127     suite.addTest(unittest.makeSuite(postgresqlDBTest))
 128     suite.addTest(unittest.makeSuite(postgresqlROTest))
 129     suite.addTest(unittest.makeSuite(postgresqlSchemaTest))
 130     suite.addTest(unittest.makeSuite(postgresqlClassicInitTest))
 131     return suite
 132