source: ndgCommon/trunk/ndg/common/unittests/lib/testgranulite.py @ 4997

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/ndgCommon/trunk/ndg/common/unittests/lib/testgranulite.py@4997
Revision 4997, 8.9 KB checked in by cbyrom, 11 years ago (diff)

Add better error handling + add general keyword catcher to avoid
issues with slight differences between the search clients + extend test data and test suites - handling proxies better.

RevLine 
[4793]1'''
2A test class for the Granulite module.
3
4C Byrom Jun 2008
5Tessella
6'''
7import unittest, logging
8from ndg.common.src.lib.granulite import granulite
[4997]9import ndg.common.src.clients.xmldb.abstractxmldbatomclient as ac
[4793]10import ndg.common.unittests.testconstants as tc
11from ndg.common.unittests.testutils import testUtils as tu
12from ndg.common.src.models.Atom import Atom
13from ndg.common.src.models.vocabtermdata import VocabTermData as VTD
14import ndg.common.src.lib.utilities as utilities
[4943]15import ndg.common.src.clients.xmldb.eXist.dbconstants as dc
16from ndg.common.src.models.ndgObject import ndgObject as no
[4793]17
18class testGranulite(unittest.TestCase):
19
20    def setUp(self):
21        '''
22        set up data used in the tests.
23        '''
[4849]24        self.utils = tu(tc.EXIST_DBCONFIG_FILE)
[4970]25        self.g = granulite(tc.VALID_TEST_GRANULITE, eXistClient = self.utils.ac)
[4793]26
27    def createGranuliteDEs(self):
28        '''
29        Create the DEs referenced in the granulite file
30        '''
31        inputs = self.g._granulite__getGranuliteDetails()
32        self.DEs = []
33        for de in inputs['data_entity_id']:
34            data = utilities.getTripleData(de)
35            id = data[0]
36            atom = self.utils.createAtom(tc.xmlString)
37            atom.atomTypeID = VTD.DE_TERM
[4943]38            atom.setDatasetID(id)
39            atom = self.utils.ac.createAtom(atom)
[4793]40            self.DEs.append(atom)
41
42    def deleteGranuliteDEs(self):
43        '''
44        Delete the DEs referenced in the granulite file
45        '''
46        for de in self.DEs:
[4943]47            self.utils.deleteDoc(de.getFullPath())
[4793]48           
49       
50       
51    def testProcessGranulite(self):
52        atom = None
53        try:
54            # firstly, create the DEs referenced in the granulite
55            self.createGranuliteDEs()
[4997]56            atom = self.g.processGranulite(replaceAtom = True)
[4793]57            self.assertNotEqual(None, atom)
[4997]58            self.assertEqual(-9.140625, atom.minX)
59            self.assertEqual(-11.25, atom.maxY)
60            self.assertEqual(-3.1640625, atom.maxX)
61            self.assertEqual(-17.578125, atom.minY)
[4793]62
63        except Exception, e:
64            self.fail("An exception should not have been thrown")
65        finally:
66            if atom is not None:
[4943]67                self.utils.deleteDoc(atom.getFullPath())
[4793]68            self.deleteGranuliteDEs()
[4997]69           
[4793]70       
[4997]71       
72    def testDuplicateProcessGranulite(self):
73        atom = None
74        try:
75            # firstly, create the DEs referenced in the granulite
76            self.createGranuliteDEs()
77            atom = self.g.processGranulite(replaceAtom = False)
78            atom = self.g.processGranulite(replaceAtom = False, initialiseAtom = True)
79            self.fail("An exception should not have been thrown")
80
81        except ac.DuplicateError, de:
82            self.assertTrue(de.message.startswith('An atom with the specified ID'))
83        except Exception, e:
84            self.fail("An exception should not have been thrown")
85        finally:
86            if atom is not None:
87                self.utils.deleteDoc(atom.getFullPath())
88            self.deleteGranuliteDEs()
89       
[4793]90    def testProcessCSMLOrCDMLFile(self):
91        try:
92            atom = self.g.processCSMLOrCDMLFile()
93            self.assertNotEqual(None, atom)
94        except:
95            self.fail("An exception should not have been thrown")
96       
97       
98    def testGetSectionName(self):
99        sectionName = "parameters"
100        foundName = self.g._granulite__getSectionName(sectionName + "::")
101        self.assertEqual(sectionName, foundName)
102       
103    def testGetInvalidSectionName(self):
104        sectionName = "parameters"
105        foundName = self.g._granulite__getSectionName(sectionName)
106        self.assertEqual(None, foundName)
107       
108    def testInitialiseGranule(self):
109        self.g._granulite__initialiseGranule()
110        self.assertNotEqual(None, self.g._atom)
111       
112    def _initGranule(self):
113        self.g._granulite__initialiseGranule()
114        return self.g._granulite__getGranuliteDetails()
115       
116
117    def testGetGranuliteDetails(self):
118        details = self._initGranule()
119        self.assertEqual(tc.GRANULITE_TITLE, details['granule_info'][0].split('|')[2].strip())
120        self.assertEqual(tc.GRANULITE_TIME_AXIS, details['cdml_file'][0].split('|')[2].strip())
121        self.assertEqual(tc.GRANULITE_CDML_DATASETID, details['cdml_file'][0].split('|')[1].strip())
122
123    def testApplyCoreGranuliteDetails(self):
124        details = self._initGranule()
125        self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
126        self.g._granulite__applyCoreGranuliteDetails(details)
127        self.assertEqual(tc.GRANULITE_TITLE, self.g._atom.title)
128        self.assertEqual(tc.GRANULITE_DATASETID, self.g._atom.datasetID)
129        self.assertEqual(tc.GRANULITE_PROVIDER_ID, self.g._atom.ME.providerID)
130
131    def testApplyGranuliteDetails(self):
132        details = self._initGranule()
133        self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
134        self.g._granulite__applyGranuliteDetails(details)
135        self.assertEqual(tc.GRANULITE_LOGO, self.g._atom.getLogos()[0].href)
136        self.assertEquals(tc.GRANULITE_AUTHOR_NAME, self.g._atom.ME.responsibleParties[0].name)
137        self.assertEqual(self.g._atom.minX, '-8.26171875')
138        self.assertEqual(self.g._atom.minY, '-11.25')
139        self.assertEqual(self.g._atom.maxX, '-3.33984375')
140        self.assertEqual(self.g._atom.maxY, '-17.578125')
141        self.assertEqual(self.g._atom.t1, tc.GRANULITE_T1)
142        self.assertEqual(self.g._atom.t2, tc.GRANULITE_T2)
143       
144
145    def testExtractSpatialData(self):
146        self._initGranule()
147        self.g._extractSpatialData("POLYGON((-8.26171875 -11.25,-9.140625 -17.40234375,-3.33984375 -17.578125,-3.1640625 -12.65625,-8.26171875 -11.25))")
148        self.assertEqual(self.g._atom.minX, '-8.26171875')
149        self.assertEqual(self.g._atom.minY, '-11.25')
150        self.assertEqual(self.g._atom.maxX, '-3.33984375')
151        self.assertEqual(self.g._atom.maxY, '-17.578125')
152
153
154    def testUpdateDataEntity(self):
155        # firstly create a DE to use
156        try:
157            atom = Atom(VTD.DE_TERM)
158            atom.fromString(tc.xmlString)
159            atom.atomTypeID = VTD.DE_TERM
[4943]160            atom = self.utils.ac.createAtom(atom)
[4793]161   
162            # now set up the granulite
163            details = self._initGranule()
164            self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
165            self.g._granulite__applyCoreGranuliteDetails(details)
166           
167            # and now update DE
[4943]168            self.g._granulite__updateDataEntity(atom.datasetID, atom.ME.providerID)
[4793]169            atomPath = atom.getDefaultCollectionPath() + atom.atomName
[4943]170            self.assertEqual([], self.utils.ac.checkAtomSchemaCompliance(atomPath))
[4793]171   
[4943]172            xml = self.utils.ac.getNDGDoc(atom.ME.providerID, no.ATOM_DOC_TYPE,
173                                          tc.datasetID, 
174                                          targetCollection = dc.ATOM_COLLECTION_PATH)
175            updatedAtom = Atom(xmlString = xml)
176           
[4793]177            self.assertNotEqual(atom, updatedAtom)
178           
179            # there should be one extra link on the new atom
180            self.assertEqual(len(atom.relatedLinks) + 1, len(updatedAtom.relatedLinks))
181        except Exception, e:
182            self.fail("An exception shouldn't have been thrown")
183        finally:
[4943]184            self.utils.deleteDoc(atom.getFullPath())
[4793]185
186
187    def testAddGranuleToDataEntityRecords(self):
188        # firstly create a DE to use
189        atom = Atom(VTD.DE_TERM)
190        atom.fromString(tc.xmlString)
191        atom.atomTypeID = VTD.DE_TERM
[4943]192        atom = self.utils.ac.createAtom(atom)
[4793]193
194        # now set up the granulite
195        details = self._initGranule()
196        self.g.ingestGranuliteFiles = False # don't worry about reading in CSML/CDML data
197        self.g._granulite__applyCoreGranuliteDetails(details)
198       
199        # and now update DE
200        self.g._dataEntityIDs = [atom.datasetID]
201        self.g._granulite__addGranuleToDataEntityRecords()
202        atomPath = atom.getDefaultCollectionPath() + atom.atomName
[4943]203        self.assertEqual([], self.utils.ac.checkAtomSchemaCompliance(atomPath))
[4793]204
[4943]205        xml = self.utils.ac.getNDGDoc(atom.ME.providerID, no.ATOM_DOC_TYPE,
206                                      tc.datasetID, 
207                                      targetCollection = dc.ATOM_COLLECTION_PATH)
208        updatedAtom = Atom(xmlString = xml)
[4793]209        self.assertNotEqual(atom, updatedAtom)
210       
211        # there should be one extra link on the new atom
212        self.assertEqual(len(atom.relatedLinks) + 1, len(updatedAtom.relatedLinks))
[4943]213        self.utils.deleteDoc(atomPath)
[4793]214       
215       
216if __name__ == '__main__':
217
218    unittest.main()
Note: See TracBrowser for help on using the repository browser.