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

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

Fix issues with the granlite tests.

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