source: mauRepo/MolesManager/trunk/src/libs/migration/tests/moles3epbmothods.py @ 8205

'''
Created on 8 Mar 2012

@author: mnagni
'''
import unittest
from testconfig import MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT
from MolesManager.moles3epb import Moles3EPB
from libs.commons_db import DbManager
from unittest import TestCase
import logging
from logging import StreamHandler
from test_utils import createObservationCollection, createObservation
from sqlalchemy.dialects import postgres


class Moles3EPBTest(TestCase):

    def setUp(self):    
        Moles3EPBTest.molesDB = DbManager(MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT)
        Moles3EPB.overrrideDBManager(Moles3EPBTest.molesDB)
        self._initSearchIndexes(Moles3EPBTest.molesDB)        
        self.logging = logging.getLogger('Moles3EPBTest')
        self.logging.addHandler(StreamHandler())
        self.logging.setLevel(logging.DEBUG)

    def tearDown(self):
        pass
        #self._dropAllTables()

    def runTest(self):
        self.checkExtractCollectionIdentifierByTitle()   
        self.checkExtractObservationByTitleKeywords() 

    def checkExtractCollectionIdentifierByTitle(self):
        session = Moles3EPB.getNewMolesSession();
        observationCollection = createObservationCollection()
        
        self.logging.info('Stores an new CEDA_ObservationCollection')
        Moles3EPB.addCedaObject(observationCollection, session)
        session.commit()

        my_identifier = Moles3EPB.extractCollectionIdentifierByTitle('test_title', session)            
        self.assertTrue(my_identifier.count() > 0, "Cannot find MD_Identifier")
        my_identifier = Moles3EPB.extractCollectionIdentifierByTitle('fake_title', session)        
        self.assertTrue(my_identifier.count() == 0, "Cannot find MD_Identifier")            
        
        self.logging.info('Deletes it')
        session.delete(observationCollection)
        session.commit()

    def checkExtractObservationByTitleKeywords(self):
        session = Moles3EPB.getNewMolesSession();
        observation = createObservation()
        
        self.logging.info('Stores an new CEDA_Observation')
        Moles3EPB.addCedaObject(observation, session)
        session.commit()
        
        obs = Moles3EPB.extractObservationByTitleKeywords('test_code', session)
        self.assertTrue(obs.count() > 1, "Cannot find a CEDA_Observation")
        obs = Moles3EPB.extractObservationByTitleKeywords('dummy_code', session)
        self.assertTrue(obs.count() == 0, "Should NOT find a CEDA_Observation")        

    def _initSearchIndexes(self, db_manager):
        #To Be Done - CHECK IF THE COLUMN ALREADY EXISTS!
        # We don't want sqlalchemy to know about this column so we add it externally.
        try:
            db_manager._engine.execute("alter table md_identifier add column code_search_vector tsvector")                 

            # This indexes the tsvector column
            db_manager._engine.execute("create index md_identifier_code_search_index on md_identifier using gin(code_search_vector)")
            
            # This sets up the trigger that keeps the tsvector column up to date.
            db_manager._engine.execute("create trigger md_identifier_code_search_update before update or insert on md_identifier \
                for each row execute procedure tsvector_update_trigger('code_search_vector', 'pg_catalog.english', 'code')")            
        except Exception as e:
            pass



if __name__ == "__main__":
    #import sys;sys.argv = ['', 'Test.testName']
    unittest.main()