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

'''
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, createProject
from testconfig import TESTMOLES3_DB_CONNECTION
from ea_model.ceda_metadatamodel.ceda_observationcollection.ceda_observationcollection import CEDA_ObservationCollection
from ea_model.moles3_4.observationcollection.mo_observationcollection import MO_ObservationCollection
from ea_model.moles3_4.utilities.mo_responsiblepartyinfo import MO_ResponsiblePartyInfo
from ea_model.ceda_metadatamodel.ceda_observation.ceda_observation import CEDA_Observation
from ea_model.moles3_4.observation.mo_observation import MO_Observation
from ea_model.moles3_4.utilities.mo_rolevalue import MO_RoleValue


class Moles3EPBTest(TestCase):

    def setUp(self):    
        Moles3EPBTest.molesDB = DbManager(MOLES3_DB_CONNECTION, MOLES3_DB_SCRIPT)
        #Moles3EPBTest.molesDB = DbManager(TESTMOLES3_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() 
        #self.checkExtractProjectObservationCollections()
        #self.checkExtractObservationCollectionsForObservation()
        self.checkObservationAuthors()

    def checkObservationAuthors(self):
        session = Moles3EPB.getNewMolesSession();
        coll = Moles3EPB.observationAuthor(1, session)
        colls = coll.all()
        print colls

    def checkExtractObservationCollectionsForObservation(self):
        session = Moles3EPB.getNewMolesSession();
        #coll = session.query(CEDA_ObservationCollection). \
        #    join(MO_ObservationCollection).join(MO_ObservationCollection.member)
        coll = session.query(CEDA_ObservationCollection)
        colls = coll.all()
        print colls

    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")
        session.delete(observation)
        session.commit()           

    def checkExtractProjectObservationCollections(self):
        session = Moles3EPB.getNewMolesSession();

        p_1 = createProject()
        p_2 = createProject()
        p_3 = createProject()        
        p_4 = createProject()   
        self._create([p_1, p_2, p_3, p_4], session)  
        
        o_1 = createObservation()
        o_1.inSupportOf = p_1               
        o_2 = createObservation()
        o_2.inSupportOf = p_2        
        o_3 = createObservation()                
        o_3.inSupportOf = p_3
        self._create([o_1, o_2, o_3], session)

        oc_1 = createObservationCollection()
        oc_2 = createObservationCollection()
        oc_3 = createObservationCollection()        
        self._create([oc_1, oc_2, oc_3], session)
        oc_1.member.append(o_1)
        session.commit()
        
        oc_2.member.append(o_1)
        oc_2.member.append(o_2)
        session.commit()
        
        oc_3.member.append(o_2)
        oc_3.member.append(o_3)        
        session.commit()

        objects = [p_1, p_2, p_3, p_4, o_1, o_2, o_3, oc_1, oc_2, oc_3]


                

        try:            
            res = Moles3EPB.extractProjectObservationCollections(p_1, session)
            print res.count()
            #for item in res:
            #    print item
            #self.assertTrue(res.count() == 2, "Cannot find a CEDA_Observationcollection")
            #res = Moles3EPB.extractProjectObservationCollections(p_2, session)
            #self.assertTrue(res.count() == 2, "Cannot find a CEDA_Observationcollection")
            #res = Moles3EPB.extractProjectObservationCollections(p_3, session)
            #self.assertTrue(res.count() == 1, "Cannot find a CEDA_Observationcollection")                        
            #res = Moles3EPB.extractProjectObservationCollections(p_4, session)
            #self.assertTrue(res.count() == 0, "Cannot find a CEDA_Observationcollection")            
        except Exception as e:
            print e
            self._delete(objects, session)
            self.assertTrue(True==False, "Error!")
        self._delete(objects, session)                

                    
    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

    def _create(self, objects, session):
        for item in objects:
            Moles3EPB.addCedaObject(item, session)
        session.commit()

    def _delete(self, objects, session):
        for item in objects:
            session.delete(item)
        session.commit()

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