source: ndgCommon/trunk/ndg/common/unittests/clients/xmldb/eXist/testatomclient.py @ 5012

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/ndgCommon/trunk/ndg/common/unittests/clients/xmldb/eXist/testatomclient.py@5012
Revision 5012, 9.4 KB checked in by cbyrom, 10 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 AtomClient class.
3
4@author C Byrom, Tessella Feb 2009
5'''
6
7import unittest,os, time
8from ndg.common.src.clients.xmldb.eXist.atomclient import *
9import ndg.common.src.clients.xmldb.eXist.dbconstants as dc
10import ndg.common.unittests.testconstants as tc
11from ndg.common.unittests.testutils import testUtils as tu
12from ndg.common.src.models import AtomState
13from ndg.common.src.models.vocabtermdata import VocabTermData as VTD
14import ndg.common.src.lib.utilities as utils
15
16class TestCase(unittest.TestCase): 
17
18    def setUp(self):
19        '''
20        set up data used in the tests.
21        '''
22        self.client = AtomClient(dbHostName = tc.EXIST_DB,
23                                 configFileName = tc.EXIST_DBCONFIG_FILE, 
24                                 setUpDB = True)
25        self.createdAtoms = []    # array to store paths to atoms created - in order to then delete them
26        self.utils = tu(tc.EXIST_DBCONFIG_FILE)
27       
28    def testInit(self):
29        self.assertTrue(self.client)
30        self.assertTrue(self.client.client)
31
32       
33    def testValidCheckAtomSchemaCompliance(self):
34        atom = self.utils.createAtomInEXist(tc.xmlString)
35        self.createdAtoms.append(atom.getFullPath())
36        atomPath = atom.getFullPath()
37        self.assertEqual([], self.client.checkAtomSchemaCompliance(atomPath))
38       
39    def testInvalidCheckAtomSchemaCompliance(self):
40        atom = self.utils.createAtomInEXist(tc.invalidXmlString)
41        self.createdAtoms.append(atom.getFullPath())
42        atomPath = atom.getFullPath()
43        errors = self.client.checkAtomSchemaCompliance(atomPath)
44        self.assertEqual(1, len(errors))
45        self.assertTrue(errors[0].startswith(" Invalid content was found starting with element 'contributor'"))
46       
47    def testInvalidEmailCheckAtomSchemaCompliance(self):
48        atom = self.utils.createAtomInEXist(tc.invalidEmailXmlString)
49        self.createdAtoms.append(atom.getFullPath())
50        atomPath = atom.getFullPath()
51        errors = self.client.checkAtomSchemaCompliance(atomPath)
52        self.assertEqual(2, len(errors))
53        self.assertTrue(errors[1].find("of element 'email' is not valid") > -1)
54
55    def testValidCheckAtomSchemaComplianceWithKeyword(self):
56        atom = self.utils.createAtom(tc.xmlString)
57        self.assertEqual([], self.client.checkAtomSchemaCompliance('', atom = atom))
58
59    def testInValidCheckAtomSchemaCompliance(self):
60        atom = self.utils.createAtom(tc.xmlString)
61        atom.author.name = ''
62        self.assertNotEqual([], self.client.checkAtomSchemaCompliance('', atom = atom))
63
64    def testGetAtomFileCollectionPath(self):
65        atom = self.utils.createAtomInEXist(tc.xmlString)
66        self.createdAtoms.append(atom.getFullPath())
67        path = self.client._AtomClient__getAtomFileCollectionPath(atom.datasetID, atom.ME.providerID)
68        self.assertTrue(atom.getFullPath().startswith(path))
69           
70    def testGetAtomFileCollectionPathInvalidDatasetID(self):
71        atom = self.utils.createAtomInEXist(tc.xmlString)
72        self.createdAtoms.append(atom.getFullPath())
73        self.assertEquals(None, self.client._AtomClient__getAtomFileCollectionPath(tc.INVALID_FILE, atom.ME.providerID))
74           
75    def testGetAtomFileCollectionPathInvalidProviderID(self):
76        atom = self.utils.createAtomInEXist(tc.xmlString)
77        self.createdAtoms.append(atom.getFullPath())
78        self.assertEquals(None, self.client._AtomClient__getAtomFileCollectionPath(atom.datasetID, tc.INVALID_FILE))
79
80    def testGetAtomPublicationState(self):
81        atom = self.utils.createAtomInEXist(tc.xmlString)
82        self.createdAtoms.append(atom.getFullPath())
83        state = self.client.getAtomPublicationState(atom.datasetID, atom.ME.providerID)
84        self.assertEqual(AtomState.WORKING_STATE.title, state.title)
85        self.assertEqual(AtomState.WORKING_STATE, state)
86
87
88    def testChangeAtomPublicationState(self):
89        atom = self.utils.createAtomInEXist(tc.xmlString)
90        self.createdAtoms.append(atom.getFullPath())
91        state = self.client.getAtomPublicationState(atom.datasetID, atom.ME.providerID)
92        self.assertEqual(AtomState.WORKING_STATE.title, state.title)
93        self.client.changeAtomPublicationState(atom, AtomState.OLD_STATE)
94        state = self.client.getAtomPublicationState(atom.datasetID, atom.ME.providerID)
95        self.assertEqual(AtomState.OLD_STATE, state)
96        self.client.changeAtomPublicationState(atom, AtomState.WORKING_STATE)
97       
98       
99    def testCreateNoneAtom(self):
100        self.assertRaises(ValueError,self.client.createAtom, None)
101
102    def testCreateNotAtom(self):
103        self.assertRaises(ValueError,self.client.createAtom, [1,2,4])
104
105    def testCreateAtom(self):
106        atom = self.utils.createAtom(tc.xmlString)
107        createdAtom = self.client.createAtom(atom)
108        self.createdAtoms.append(createdAtom.getFullPath())
109        self.assertNotEquals(None, createdAtom)
110
111
112    def testSetUpEXistDB(self):
113        fileName = self.client.resources.xsd.keys()[0] + '.xsd'
114        pathName = dc.SCHEMAS_COLLECTION_PATH + '/' + fileName
115        self.client.deleteDoc(pathName)
116        self.assertEquals(True, self.client.isNewDoc(pathName))
117
118        newClient = AtomClient(dbHostName = tc.EXIST_DB,
119                               configFileName = tc.EXIST_DBCONFIG_FILE, 
120                               setUpDB = True)
121
122        self.assertEquals(False, self.client.isNewDoc(pathName))
123
124
125    def testNotSetUpEXistDB(self):
126        fileName = self.client.resources.xsd.keys()[0] + '.xsd'
127        pathName = dc.SCHEMAS_COLLECTION_PATH + '/' + fileName
128        self.client.deleteDoc(pathName)
129        self.assertEquals(True, self.client.isNewDoc(pathName))
130
131        newClient = AtomClient(dbHostName = tc.EXIST_DB,
132                               configFileName = tc.EXIST_DBCONFIG_FILE, 
133                               setUpDB = False)
134
135        self.assertEquals(True, self.client.isNewDoc(pathName))
136
137           
138    def testRunAsynchAtomPublish(self):
139        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
140        pathName = createdAtom.getFullPath()
141        self.assertEquals(False, self.client.isNewDoc(pathName))
142        createdAtom.state = AtomState.PUBLISHED_STATE
143        providerCollection = dc.PROVIDER_FEED_PATH + createdAtom.ME.providerID
144        self.createdAtoms.append(pathName)
145        feed = self.client.feedClient.getAtomFeed(providerCollection)
146       
147        if self.client.isNewCollection(providerCollection):
148            self.client.createCollections([providerCollection])
149            self.client.feedClient.createAtomFeed(providerCollection,
150                           self.client.feedClient.PROVIDERLEVEL_ATOM_FEED_TITLE %createdAtom.ME.providerID)
151        self.client._AtomClient__runAsynchAtomPublish(createdAtom)
152        time.sleep(2)
153        feed2 = self.client.feedClient.getAtomFeed(providerCollection)
154        self.assertNotEqual(feed, feed2)
155
156           
157    def testPublishAtom(self):
158        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
159        self.createdAtoms.append(createdAtom.getFullPath())
160        createdAtom.state = AtomState.PUBLISHED_STATE
161        providerCollection = dc.PROVIDER_FEED_PATH + createdAtom.ME.providerID
162        feed = self.client.feedClient.getAtomFeed(providerCollection)
163       
164        if self.client.isNewCollection(providerCollection):
165            self.client.createCollections([providerCollection])
166            self.client.feedClient.createAtomFeed(providerCollection,
167                           self.client.feedClient.PROVIDERLEVEL_ATOM_FEED_TITLE %createdAtom.ME.providerID)
168        self.client._AtomClient__publishAtom(createdAtom)
169        feed2 = self.client.feedClient.getAtomFeed(providerCollection)
170        self.assertNotEqual(feed, feed2)
171
172
173    def testCreateDIFDocumentFromAtom(self):
174        atom = self.utils.createAtom(tc.xmlString)
175        atom.atomTypeID = VTD.DE_TERM
176        atom = self.client.createAtom(atom)
177        self.createdAtoms.append(atom.getFullPath())
178        filePath = self.client._createDIFDocumentFromAtom(atom, dc.DIF_COLLECTION_PATH)
179        self.createdAtoms.append(filePath)
180        self.assertEquals(False, self.client.isNewDoc(filePath))
181
182
183    def testTransformAtomIntoDIF(self):
184        atom = self.utils.createAtom(tc.xmlString)
185        atom.atomTypeID = VTD.DE_TERM
186        atom = self.client.createAtom(atom)
187        self.createdAtoms.append(atom.getFullPath())
188        doc = self.client._transformAtomIntoDIF(atom)
189        self.assertNotEquals(None, doc)
190        self.assertTrue(doc.find('DIF') > -1)
191        self.assertTrue(doc.find(atom.datasetID) > -1)
192
193
194    def testNotLoadCollectionData(self):
195        self.assertFalse(self.utils.ac.atomCollections)
196
197
198    def testLoadCollectionData(self):
199        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
200        filePath = createdAtom.getFullPath()
201        self.createdAtoms.append(filePath)
202        newClient = AtomClient(dbHostName = tc.EXIST_DB, 
203                               configFileName = tc.EXIST_DBCONFIG_FILE,
204                               loadCollectionData = True)
205        self.assertTrue(newClient.atomCollections)
206        self.assertEquals(filePath, newClient.atomCollections[createdAtom.datasetID] + \
207                          '/' + createdAtom.atomName)
208
209       
210    def tearDown(self):
211        for path in self.createdAtoms:
212            # delete the test file, in case it has been used in a test
213            self.utils.deleteDoc(path)
214
215           
216if __name__=="__main__":
217    unittest.main()
Note: See TracBrowser for help on using the repository browser.