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.4 2003-11-14 00:11:19 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         self.nuke_database()
  72
  73     def nuke_database(self):
  74         # clear out the database - easiest way is to nuke and re-created it
  75         db_nuke()
  76         db_create()
  77
  78 class postgresqlDBTest(postgresqlOpener, DBTest):
  79     def setUp(self):
  80         postgresqlOpener.setUp(self)
  81         DBTest.setUp(self)
  82
  83     def tearDown(self):
  84         DBTest.tearDown(self)
  85         postgresqlOpener.tearDown(self)
  86
  87     def testFilteringIntervalSort(self):
  88         # PostgreSQL sorts NULLs differently to other databases (others
  89         # treat it as lower than real values, PG treats it as higher)
  90         ae, filt = self.filteringSetup()
  91         # ascending should sort None, 1:10, 1d
  92         ae(filt(None, {}, ('+','foo'), (None,None)), ['4', '1', '2', '3'])
  93         # descending should sort 1d, 1:10, None
  94         ae(filt(None, {}, ('-','foo'), (None,None)), ['3', '2', '1', '4'])
  95
  96 class postgresqlROTest(postgresqlOpener, ROTest):
  97     def setUp(self):
  98         postgresqlOpener.setUp(self)
  99         ROTest.setUp(self)
 100
 101     def tearDown(self):
 102         ROTest.tearDown(self)
 103         postgresqlOpener.tearDown(self)
 104
 105 class postgresqlSchemaTest(postgresqlOpener, SchemaTest):
 106     def setUp(self):
 107         postgresqlOpener.setUp(self)
 108         SchemaTest.setUp(self)
 109
 110     def tearDown(self):
 111         SchemaTest.tearDown(self)
 112         postgresqlOpener.tearDown(self)
 113
 114 class postgresqlClassicInitTest(postgresqlOpener, ClassicInitTest):
 115     backend = 'postgresql'
 116     extra_config = "POSTGRESQL_DATABASE = {'database': 'rounduptest'}"
 117     def setUp(self):
 118         postgresqlOpener.setUp(self)
 119         ClassicInitTest.setUp(self)
 120
 121     def tearDown(self):
 122         ClassicInitTest.tearDown(self)
 123         postgresqlOpener.tearDown(self)
 124
 125 def test_suite():
 126     suite = unittest.TestSuite()
 127     if not hasattr(backends, 'postgresql'):
 128         return suite
 129
 130     # Check if we can run postgresql tests
 131     print 'Including postgresql tests'
 132     suite.addTest(unittest.makeSuite(postgresqlDBTest))
 133     suite.addTest(unittest.makeSuite(postgresqlROTest))
 134     suite.addTest(unittest.makeSuite(postgresqlSchemaTest))
 135     suite.addTest(unittest.makeSuite(postgresqlClassicInitTest))
 136     return suite
 137