tests/db_test.py

   1 from base_test import DakTestCase, fixture
   2
   3 from daklib.config import Config
   4 from daklib.dbconn import DBConn
   5
   6 from sqlalchemy import create_engine
   7 from sqlalchemy.exc import SADeprecationWarning
   8
   9 import pickle
  10 import warnings
  11
  12 # suppress some deprecation warnings in squeeze related to sqlalchemy
  13 warnings.filterwarnings('ignore', \
  14     "The SQLAlchemy PostgreSQL dialect has been renamed from 'postgres' to 'postgresql'.*", \
  15     SADeprecationWarning)
  16
  17 class DBDakTestCase(DakTestCase):
  18     def setUp(self):
  19         cnf = Config()
  20         if cnf["DB::Host"]:
  21             # TCP/IP
  22             connstr = "postgres://%s" % cnf["DB::Host"]
  23             if cnf["DB::Port"] and cnf["DB::Port"] != "-1":
  24                 connstr += ":%s" % cnf["DB::Port"]
  25             connstr += "/%s" % cnf["DB::Name"]
  26         else:
  27             # Unix Socket
  28             connstr = "postgres:///%s" % cnf["DB::Name"]
  29             if cnf["DB::Port"] and cnf["DB::Port"] != "-1":
  30                 connstr += "?port=%s" % cnf["DB::Port"]
  31
  32         pickle_file = open(fixture('db-metadata.pkl'), 'r')
  33         self.metadata = pickle.load(pickle_file)
  34         self.metadata.ddl_listeners = pickle.load(pickle_file)
  35         pickle_file.close()
  36         self.metadata.bind = create_engine(connstr)
  37         self.metadata.create_all()
  38         self.session = DBConn().session()
  39
  40     def tearDown(self):
  41         #pass
  42         self.session.close()
  43         #self.metadata.drop_all()
  44