source: ndgCommon/trunk/ndg/common/unittests/models/testatom.py @ 5012

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

Various tidy ups and fixes to the test codebase - including the removal
of uneccesary files + simplify the test constants.

Line 
1'''
2A test class for the GranuleAtom module.
3
4C Byrom Jun 2008
5Tessella
6'''
7import unittest, copy
8from ndg.common.src.models.Atom import Atom, Link
9from ndg.common.src.models.vocabtermdata import VocabTermData as VTD
10from ndg.common.src.lib.utilities import getTripleData
11import ndg.common.unittests.testconstants as tc
12from ndg.common.src.models import AtomState
13from ndg.common.unittests.testutils import testUtils as tu
14from ndg.common.src.clients.xmldb.eXist.searchclient import SearchClient
15
16class testatom(unittest.TestCase):
17    VTD = VTD()
18   
19    def setUp(self):
20        '''
21        set up data used in the tests.
22        '''
23        print "Setting up..."
24        self.atom = Atom(VTD.GRANULE_TERM, vocabTermData = self.VTD)
25        self.link = Link()
26        self.link.title = 'fred'
27        self.link.href = 'www.fred.com'
28        self.link.rel = 'related'
29        self.link2 = Link()
30        self.link2.title = 'pete'
31        self.link2.href = 'www.pete.com'
32        self.link2.rel = 'self'
33        self.createdAtoms = []    # array to store paths to atoms created - in order to then delete them
34        self.utils = tu(tc.EXIST_DBCONFIG_FILE)
35
36               
37    def testAddValidParameters(self):
38        self.atom.addParameters(tc.validParams)
39        self.assertEqual(self.atom.parameters[0].term, getTripleData(tc.validParams[0])[2])
40        self.assertEqual(self.atom.parameters[1].term, getTripleData(tc.validParams[1])[2])
41        self.assertEqual(self.atom.parameters[2].term, getTripleData(tc.validParams[2])[2])
42       
43       
44    def testAddMessyParameters(self):
45        self.atom.addParameters(tc.messyParams)
46        self.assertEqual(self.atom.parameters[0].term, 'OZONE')
47        self.assertEqual(self.atom.parameters[1].term, 'ERROR')
48        self.assertEqual(self.atom.parameters[2].term, 'HYDROGEN &')
49        self.assertEqual(self.atom.parameters[2].scheme, 'http://vocab.ndg.nerc.ac.uk/blah')
50        self.assertEqual(self.atom.parameters[2].label.lower(), 'Hydrogen Peroxide < ppbv'.lower())
51       
52    def testFromString(self):
53        self.atom.fromString(tc.xmlString)
54       
55        self.assertNotEqual(self.atom.author, None)
56        self.assertEqual(tc.validAuthor, self.atom.author.name)
57        self.assertEqual(3, len(self.atom.parameters))
58       
59        for param in self.atom.parameters:
60            foundIt = False
61            for validParam in tc.validParams:
62                if (validParam.startswith(param.term)):
63                    foundIt = True
64                    break
65           
66            self.assertEqual(True, foundIt, "Cound not find parameter (%s) in valid parameter list" %param)
67       
68        # currently only store max/min boundaries
69        #for spatial in self.atom.spatialData:
70        #    if not spatial in self.validSpatialData:
71        #        self.fail("Cound not find spatial data (%s) in valid coverage list" %spatial)
72        #
73        #for temporal in self.atom.temporalData:
74        #    if not temporal in self.validTemporalData:
75        #        self.fail("Cound not find temporal data (%s) in valid coverage list" %temporal)
76       
77        #self.assertEqual(len(self.atom.relatedLinks), 2)
78        for link in self.atom.relatedLinks:
79            foundIt = False
80            for validLink in tc.validLinks:
81                if (validLink.startswith(link.href)):
82                    foundIt = True
83                    break
84               
85        self.assertEqual(True, foundIt, "Cound not find link (%s) in valid related link list" %link.href)
86
87        self.assertEqual(tc.summaryText, self.atom.summary[0])
88        self.assertEqual(tc.csmlFile, self.atom.csmlFile)
89        self.assertEqual(None, self.atom.cdmlFile)
90        self.assertEqual(tc.title, self.atom.title)
91        # Need vocab term properly set up for this test to work
92#        self.assertEqual(tc.logoURL, self.atom.getLogos()[0].href)
93        self.assertEqual(tc.datasetID, self.atom.datasetID)
94        self.assertEqual(tc.minX, self.atom.minX)
95        self.assertEqual(tc.minY, self.atom.minY)
96        self.assertEqual(tc.maxX, self.atom.maxX)
97        self.assertEqual(tc.maxY, self.atom.maxY)
98        self.assertEqual(tc.T1, self.atom.t1)
99        self.assertEqual(tc.T2, self.atom.t2)
100       
101        # date when the granule was first ingested
102        if not self.atom.publishedDate:
103            self.fail("No published date found") 
104
105        self.assertEqual(tc.name, self.atom.atomName)
106       
107
108    def testImportExport(self):
109        self.atom.fromString(tc.xmlString)
110        xml = self.atom.toPrettyXML()
111        # NB, this is a naive test as ordering of sibling elements changes as does addition of new line characters
112        #self.assertEqual(xml, tc.xmlString)
113
114    def testCreateWithInputDict(self):
115        inputs = {
116                  'atomTypeID': VTD.ACTIVITY_TERM,
117                  'title': tc.title
118                  }
119        a = Atom(**inputs)
120        self.assertEqual(tc.title, a.title)
121        self.assertEqual(VTD.ACTIVITY_TERM, a.atomTypeID)
122        self.assertEqual(VTD.TERM_DATA[VTD.ACTIVITY_TERM].title, a.atomTypeName)
123
124    def testaddUniqueRelatedLinks(self):
125        self.atom.fromString(tc.xmlString)
126        linkNo = len(self.atom.relatedLinks)
127        self.atom.addUniqueRelatedLinks(self.link)
128        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
129        self.assertTrue(self.link in self.atom.relatedLinks)
130
131    def testAddMultipeDifferentRelatedLinks(self):
132        self.atom.fromString(tc.xmlString)
133        linkNo = len(self.atom.relatedLinks)
134        self.atom.addUniqueRelatedLinks(self.link)
135        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
136        link2 = copy.deepcopy(self.link)
137        link2.title = "blabby"
138        self.atom.addUniqueRelatedLinks(link2)
139        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
140        self.atom.addUniqueRelatedLinks(link2)
141        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
142
143    def testAddMultipeDifferentRelatedLinksSimultaneously(self):
144        self.atom.fromString(tc.xmlString)
145        linkNo = len(self.atom.relatedLinks)
146        link2 = copy.deepcopy(self.link)
147        link2.title = "blabby"
148        links = [self.link, link2]
149        self.atom.addUniqueRelatedLinks(links)
150        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
151        self.assertTrue(self.link in self.atom.relatedLinks)
152        self.assertTrue(link2 in self.atom.relatedLinks)
153
154        self.atom.addUniqueRelatedLinks(link2)
155        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
156       
157
158    def testAddMultipleUniqueRelatedLinks(self):
159        self.atom.fromString(tc.xmlString)
160        linkNo = len(self.atom.relatedLinks)
161        self.atom.addUniqueRelatedLinks(self.link)
162        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
163        self.atom.addUniqueRelatedLinks(self.link)
164        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
165        self.atom.addUniqueRelatedLinks(self.link)
166        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
167
168    def testRemoveRelatedLinks(self):
169        self.atom.fromString(tc.xmlString)
170        linkNo = len(self.atom.relatedLinks)
171        self.atom.addUniqueRelatedLinks(self.link)
172        self.assertEqual(linkNo + 1, len(self.atom.relatedLinks))
173        self.assertTrue(self.link in self.atom.relatedLinks)
174        self.atom.removeRelatedLinks(self.link)
175        self.assertEqual(linkNo, len(self.atom.relatedLinks))
176        self.assertFalse(self.link in self.atom.relatedLinks)
177
178    def testRemoveMultipleRelatedLinks(self):
179        self.atom.fromString(tc.xmlString)
180        linkNo = len(self.atom.relatedLinks)
181        link2 = copy.deepcopy(self.link)
182        link2.title = "blabby"
183        links = [self.link, link2]
184
185        self.atom.addUniqueRelatedLinks(links)
186        self.assertEqual(linkNo + 2, len(self.atom.relatedLinks))
187        self.assertTrue(self.link in self.atom.relatedLinks)
188        self.assertTrue(link2 in self.atom.relatedLinks)
189
190        self.atom.removeRelatedLinks(links)
191        self.assertEqual(linkNo, len(self.atom.relatedLinks))
192        self.assertFalse(self.link in self.atom.relatedLinks)
193        self.assertFalse(link2 in self.atom.relatedLinks)
194
195    def testAddUniqueLinks(self):
196        data = []
197        links = [self.link, self.link2, self.link]
198        self.atom.addUniqueLinks(data, links)
199        self.assertEqual(2, len(data))
200
201    def testAddUniqueLinksWithInitialLinksDefined(self):
202        data = [self.link2]
203        links = [self.link, self.link2, self.link]
204        self.atom.addUniqueLinks(data, links)
205        self.assertEqual(2, len(data))
206
207    def testIsPublished(self):
208        self.assertFalse(self.atom.isPublished())
209        self.atom.state = AtomState.PUBLISHED_STATE
210        self.assertTrue(self.atom.isPublished())
211       
212    def testStateKeyword(self):
213        self.atom = Atom(VTD.GRANULE_TERM, vocabTermData = self.VTD, 
214                         state = AtomState.OLD_STATE)
215        self.assertEqual(AtomState.OLD_STATE, self.atom.state)
216        self.assertFalse(self.atom.isPublished())
217
218    def testLookupAssociatedDEDataNoData(self):
219        atom = self.utils.createAtomInEXist(tc.xmlString)
220        self.createdAtoms.append(atom.getFullPath())
221        sc = SearchClient(dbHostName = tc.EXIST_DB,
222                          configFileName = tc.EXIST_DBCONFIG_FILE)
223
224        atom.lookupAssociatedData(VTD.DE_TERM, sc)
225        self.assertFalse(atom.dataEntities)
226
227    def testLookupAssociatedDEDataWithData(self):
228        atom = self.utils.createAtomInEXist(tc.xmlString)
229        self.createdAtoms.append(atom.getFullPath())
230        sc = SearchClient(dbHostName = tc.EXIST_DB,
231                          configFileName = tc.EXIST_DBCONFIG_FILE)
232
233        link = Link()
234        link.href = atom.atomBrowseURL or ""
235        link.title = atom.title or ""
236        link.rel = VTD.DEPLOYMENT_TERM
237        atom2 = Atom(VTD.DE_TERM, vocabTermData = self.VTD)
238        atom2.ME.providerID = tc.VALID_PROVIDER_ID
239        atom2.addUniqueRelatedLinks([link])
240        self.utils.ac.createAtom(atom2)
241        self.createdAtoms.append(atom2.getFullPath())
242       
243        sc = SearchClient(dbHostName = tc.EXIST_DB,
244                          configFileName = tc.EXIST_DBCONFIG_FILE)
245
246        atom.lookupAssociatedData(VTD.DE_TERM, sc)
247       
248        self.assertFalse(atom.dataEntities)
249
250
251    def testLookupAssociatedDeploymentDataNoData(self):
252        atom = self.utils.createAtomInEXist(tc.xmlString)
253        self.createdAtoms.append(atom.getFullPath())
254        sc = SearchClient(dbHostName = tc.EXIST_DB,
255                          configFileName = tc.EXIST_DBCONFIG_FILE)
256
257        atom.lookupAssociatedData(VTD.DEPLOYMENT_TERM, sc)
258        self.assertFalse(atom.deployments)
259
260    def testLookupAssociatedDeploymentDataWithData(self):
261        atom = self.utils.createAtomInEXist(tc.xmlString)
262        self.createdAtoms.append(atom.getFullPath())
263
264        link = Link()
265        link.href = atom.atomBrowseURL or ""
266        link.title = atom.title or ""
267        link.rel = VTD.DEPLOYMENT_TERM
268        atom2 = Atom(VTD.DE_TERM, vocabTermData = self.VTD)
269        atom2.ME.providerID = tc.VALID_PROVIDER_ID
270        atom2.addUniqueRelatedLinks([link])
271        self.utils.ac.createAtom(atom2)
272        self.createdAtoms.append(atom2.getFullPath())
273       
274        sc = SearchClient(dbHostName = tc.EXIST_DB,
275                          configFileName = tc.EXIST_DBCONFIG_FILE)
276
277        atom2.lookupAssociatedData(VTD.DEPLOYMENT_TERM, sc)
278        self.assertTrue(atom2.deployments)
279
280
281    def testLookupAssociatedDEDataWithLookupIndirectReferencesWithData(self):
282        actAtom = Atom(VTD.ACTIVITY_TERM, vocabTermData = self.VTD)
283        actAtom.ME.providerID = tc.VALID_PROVIDER_ID
284        actAtom.setDatasetID('activity1')
285        self.utils.ac.createAtom(actAtom)
286        self.createdAtoms.append(actAtom.getFullPath())
287        link = Link()
288        link.href = actAtom.atomBrowseURL or ""
289        link.title = actAtom.title or ""
290        link.rel = VTD.ACTIVITY_TERM
291       
292        obsAtom = Atom(VTD.OBS_TERM, vocabTermData = self.VTD)
293        obsAtom.ME.providerID = tc.VALID_PROVIDER_ID
294        obsAtom.setDatasetID('obs1')
295        self.utils.ac.createAtom(obsAtom)
296        self.createdAtoms.append(obsAtom.getFullPath())
297        obslink = Link()
298        obslink.href = obsAtom.atomBrowseURL or ""
299        obslink.title = obsAtom.title or ""
300        obslink.rel = VTD.OBS_TERM
301       
302        depAtom = Atom(VTD.ACTIVITY_TERM, vocabTermData = self.VTD)
303        depAtom.ME.providerID = tc.VALID_PROVIDER_ID
304        depAtom.setDatasetID('dep1')
305        depAtom.addUniqueRelatedLinks([link, obslink])
306        depAtom.subtypeID = VTD.DEPLOYMENT_TERM
307        self.utils.ac.createAtom(depAtom)
308        self.createdAtoms.append(depAtom.getFullPath())
309        link = Link()
310        link.href = depAtom.atomBrowseURL or ""
311        link.title = depAtom.title or ""
312        link.rel = VTD.DEPLOYMENT_TERM
313       
314        deAtom = Atom(VTD.DE_TERM, vocabTermData = self.VTD)
315        deAtom.ME.providerID = tc.VALID_PROVIDER_ID
316        deAtom.setDatasetID('de1')
317        deAtom.addUniqueRelatedLinks([link])
318        self.utils.ac.createAtom(deAtom)
319        self.createdAtoms.append(deAtom.getFullPath())
320        sc = SearchClient(dbHostName = tc.EXIST_DB,
321                          configFileName = tc.EXIST_DBCONFIG_FILE)
322
323        actAtom.lookupAssociatedData(VTD.DE_TERM, sc,
324                                    lookupIndirectReferences=True)
325        self.assertTrue(actAtom.dataEntities)
326
327
328    def testLookupAssociatedDeploymentDataWithLookupIndirectReferencesWithData(self):
329        actAtom = Atom(VTD.ACTIVITY_TERM, vocabTermData = self.VTD)
330        actAtom.ME.providerID = tc.VALID_PROVIDER_ID
331        actAtom.setDatasetID('activity1')
332        self.utils.ac.createAtom(actAtom)
333        self.createdAtoms.append(actAtom.getFullPath())
334        link = Link()
335        link.href = actAtom.atomBrowseURL or ""
336        link.title = actAtom.title or ""
337        link.rel = VTD.ACTIVITY_TERM
338       
339        obsAtom = Atom(VTD.OBS_TERM, vocabTermData = self.VTD)
340        obsAtom.ME.providerID = tc.VALID_PROVIDER_ID
341        obsAtom.setDatasetID('obs1')
342        self.utils.ac.createAtom(obsAtom)
343        self.createdAtoms.append(obsAtom.getFullPath())
344        obslink = Link()
345        obslink.href = obsAtom.atomBrowseURL or ""
346        obslink.title = obsAtom.title or ""
347        obslink.rel = VTD.OBS_TERM
348       
349        depAtom = Atom(VTD.ACTIVITY_TERM, vocabTermData = self.VTD)
350        depAtom.ME.providerID = tc.VALID_PROVIDER_ID
351        depAtom.setDatasetID('dep1')
352        depAtom.addUniqueRelatedLinks([link, obslink])
353        depAtom.subtypeID = VTD.DEPLOYMENT_TERM
354        self.utils.ac.createAtom(depAtom)
355        self.createdAtoms.append(depAtom.getFullPath())
356        link = Link()
357        link.href = depAtom.atomBrowseURL or ""
358        link.title = depAtom.title or ""
359        link.rel = VTD.DEPLOYMENT_TERM
360       
361        deAtom = Atom(VTD.DE_TERM, vocabTermData = self.VTD)
362        deAtom.ME.providerID = tc.VALID_PROVIDER_ID
363        deAtom.setDatasetID('de1')
364        deAtom.addUniqueRelatedLinks([link])
365        self.utils.ac.createAtom(deAtom)
366        self.createdAtoms.append(deAtom.getFullPath())
367        sc = SearchClient(dbHostName = tc.EXIST_DB,
368                          configFileName = tc.EXIST_DBCONFIG_FILE)
369
370        obsAtom.lookupAssociatedData(VTD.DEPLOYMENT_TERM, sc,
371                                    lookupIndirectReferences=True)
372        self.assertTrue(obsAtom.deployments)
373
374    def testLookupAssociatedDEDataWithLookupIndirectReferencesNoData(self):
375        atom = self.utils.createAtomInEXist(tc.xmlString)
376        self.createdAtoms.append(atom.getFullPath())
377        sc = SearchClient(dbHostName = tc.EXIST_DB,
378                          configFileName = tc.EXIST_DBCONFIG_FILE)
379
380        atom.lookupAssociatedData(VTD.DE_TERM, sc,
381                                  lookupIndirectReferences=True)
382        self.assertFalse(atom.dataEntities)
383
384    def testLookupAssociatedDeploymentsDataWithLookupIndirectReferencesNoData(self):
385        atom = self.utils.createAtomInEXist(tc.xmlString)
386        self.createdAtoms.append(atom.getFullPath())
387        sc = SearchClient(dbHostName = tc.EXIST_DB,
388                          configFileName = tc.EXIST_DBCONFIG_FILE)
389
390        atom.lookupAssociatedData(VTD.DEPLOYMENT_TERM, sc,
391                                  lookupIndirectReferences=True)
392        self.assertFalse(atom.deployments)
393
394       
395    def tearDown(self):
396        for path  in self.createdAtoms:
397            # delete the test file, in case it has been used in a test
398            self.utils.deleteDoc(path)
399       
400
401if __name__ == '__main__':
402
403    unittest.main()
Note: See TracBrowser for help on using the repository browser.