source: ndgCommon/trunk/ndg/common/unittests/clients/xmldb/eXist/testxmlrpcclient.py @ 5010

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/ndgCommon/trunk/ndg/common/unittests/clients/xmldb/eXist/testxmlrpcclient.py@5010
Revision 5010, 10.2 KB checked in by cbyrom, 12 years ago (diff)

Adjust XMLRPC client to eXist to preserve the throwing of Faults if
problems occur whilst deleting records. NB, this mainly occurs because
the record doesn't exist - but we shouldn't really be trying to delete
non-existent records, so it's probably helpful to keep this error.

Line 
1
2import unittest,os, xmlrpclib
3from ndg.common.src.clients.xmldb.eXist.xmlrpcclient import *
4import ndg.common.src.clients.xmldb.eXist.dbconstants as dc
5import ndg.common.unittests.testconstants as tc
6from ndg.common.unittests.testutils import testUtils as tu
7from ndg.common.src.lib.ndgresources import ndgResources
8import ndg.common.src.clients.xmldb.eXist.dbconstants as dc
9from ndg.common.src.lib.utilities import loadConfigDetails
10
11class TestCase(unittest.TestCase): 
12
13
14    def setUp(self):
15        '''
16        set up data used in the tests.
17        '''
18        userid, password, host = loadConfigDetails(tc.EXIST_DBCONFIG_FILE, tc.EXIST_DB)
19        self.client = XMLRPCClient(host, userid, password)
20
21        self.createdAtoms = []    # array to store paths to atoms created - in order to then delete them
22        self.utils = tu(tc.EXIST_DBCONFIG_FILE)
23
24    def testDescribeCollection(self):
25        info = self.client.describeCollection(dc.ATOM_COLLECTION_PATH)
26        self.assertEqual(dc.ATOM_COLLECTION_PATH, info['name'] + '/')
27        self.assertTrue(len(info['collections']) > 0)
28        self.assertTrue(info['owner'])
29        self.assertTrue(info['created'])
30        self.assertTrue(info['group'])
31        self.assertTrue(info['permissions'])
32       
33    def testRunQuery(self):
34        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
35        self.createdAtoms.append(createdAtom.getFullPath())
36        res = ndgResources()
37        xquery = res.createXQuery('atomList', dc.ATOM_COLLECTION_PATH, "", "")
38        result = self.client.runQuery(xquery)[0]
39        self.assertTrue(len(result) > 0)
40        self.assertTrue(result.find('fileName') > 0)
41
42    def testInvalidRunQuery(self):
43        res = ndgResources()
44        result = self.client.runQuery(tc.INVALID_XQUERY)
45        self.assertEqual([], result)
46
47    def testEmptyExecuteQuery(self):
48        self.assertRaises(xmlrpclib.Fault, self.client._XMLRPCClient__executeQuery, "")
49
50    def testValidExecuteQuery(self):
51        # NB, this test assumes there is at least one atom available in the exist db
52        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
53        self.createdAtoms.append(createdAtom.getFullPath())
54        res = ndgResources()
55        xquery = res.createXQuery('atomList', dc.ATOM_COLLECTION_PATH, "", "")
56        result = self.client._XMLRPCClient__executeQuery(xquery)
57        self.assertTrue(result[1]['hits'] > 0)
58
59
60    def testValidExecuteQueryMultipleResults(self):
61        # NB, this test assumes there is at least one atom available in the exist db
62        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
63        self.createdAtoms.append(createdAtom.getFullPath())
64
65        res = ndgResources()
66        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
67        result = self.client._XMLRPCClient__executeQuery(xquery)
68        noHits = result[1]['hits']
69        self.assertTrue(noHits > 0)
70       
71        atom2 = self.utils.createAtom(tc.xmlString)
72        atom2.ME.providerID = tc.GRANULITE_PROVIDER_ID2
73        atom2.setDatasetID(tc.datasetID)
74        path = atom2.getFullPath()
75        self.client.createCollection(atom2.getDefaultCollectionPath())
76        self.client.createDoc(atom2.toPrettyXML(), path)
77        self.createdAtoms.append(path)
78       
79        result = self.client._XMLRPCClient__executeQuery(xquery)
80        self.assertEquals(noHits + 1, result[1]['hits'])
81
82
83    def testValidRunQueryMultipleResults(self):
84        # NB, this test assumes there is at least one atom available in the exist db
85        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
86        self.createdAtoms.append(createdAtom.getFullPath())
87
88        res = ndgResources()
89        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
90        docs = self.client.runQuery(xquery)
91        self.assertEqual(1, len(docs))
92        self.assertTrue(docs[0])
93       
94        atom2 = self.utils.createAtom(tc.xmlString)
95        atom2.ME.providerID = tc.GRANULITE_PROVIDER_ID2
96        atom2.setDatasetID(tc.datasetID)
97        path = atom2.getFullPath()
98        self.client.createCollection(atom2.getDefaultCollectionPath())
99        self.client.createDoc(atom2.toPrettyXML(), path)
100        self.createdAtoms.append(path)
101       
102        docs2 = self.client.runQuery(xquery)
103        self.assertEqual(2, len(docs2))
104        self.assertTrue(docs2[0])
105        self.assertTrue(docs2[1])
106        # NB, cannot guarantee order of results
107        self.assertTrue(docs[0] == docs2[0] or docs[0] == docs2[1])
108
109
110    def testValidRunQueryMultipleResultsTwoResultsKeyword(self):
111        # NB, this test assumes there is at least one atom available in the exist db
112        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
113        self.createdAtoms.append(createdAtom.getFullPath())
114
115        res = ndgResources()
116        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
117        docs = self.client.runQuery(xquery)
118        self.assertEqual(1, len(docs))
119        self.assertTrue(docs[0])
120       
121        atom2 = self.utils.createAtom(tc.xmlString)
122        atom2.ME.providerID = tc.GRANULITE_PROVIDER_ID2
123        atom2.setDatasetID(tc.datasetID)
124        path = atom2.getFullPath()
125        self.client.createCollection(atom2.getDefaultCollectionPath())
126        self.client.createDoc(atom2.toPrettyXML(), path)
127        self.createdAtoms.append(path)
128       
129        docs2 = self.client.runQuery(xquery)
130        self.assertEqual(2, len(docs2))
131        self.assertTrue(docs2[0])
132        self.assertTrue(docs2[1])
133        # NB, cannot guarantee order of results
134        self.assertTrue(docs[0] == docs2[0] or docs[0] == docs2[1])
135
136        docs3 = self.client.runQuery(xquery, noResults = 2)
137        self.assertEqual(2, len(docs3))
138        self.assertTrue(docs3[0])
139        self.assertTrue(docs3[1])
140        # NB, cannot guarantee order of results
141        self.assertTrue(docs[0] == docs3[0] or docs[0] == docs3[1])
142
143
144    def testValidRunQueryMultipleZeroResultsKeywordChunkedOff(self):
145        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
146        self.createdAtoms.append(createdAtom.getFullPath())
147
148        res = ndgResources()
149        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
150        docs = self.client.runQuery(xquery, noResults = 0, useChunked = False)
151        self.assertEqual(0, len(docs))
152
153
154    def testValidRunQueryMultipleZeroResultsKeywordChunkedOn(self):
155        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
156        self.createdAtoms.append(createdAtom.getFullPath())
157
158        res = ndgResources()
159        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
160        docs = self.client.runQuery(xquery, noResults = 0)
161        self.assertEqual(0, len(docs))
162
163
164    def testValidRunQueryMultipleResultsOneResultKeyword(self):
165        # NB, this test assumes there is at least one atom available in the exist db
166        createdAtom = self.utils.createAtomInEXist(tc.xmlString)
167        self.createdAtoms.append(createdAtom.getFullPath())
168
169        res = ndgResources()
170        xquery = res.createXQuery('atom', dc.ATOM_COLLECTION_PATH, "", tc.datasetID)
171       
172        atom2 = self.utils.createAtom(tc.xmlString)
173        atom2.ME.providerID = tc.GRANULITE_PROVIDER_ID2
174        atom2.setDatasetID(tc.datasetID)
175        path = atom2.getFullPath()
176        self.client.createCollection(atom2.getDefaultCollectionPath())
177        self.client.createDoc(atom2.toPrettyXML(), path)
178        self.createdAtoms.append(path)
179       
180        docs2 = self.client.runQuery(xquery, noResults = 1)
181        self.assertEqual(1, len(docs2))
182        self.assertTrue(docs2[0])
183
184
185
186    def testInvalidExecuteQuery(self):
187        # NB, this test assumes there is at least one atom available in the exist db
188        res = ndgResources()
189        result = self.client._XMLRPCClient__executeQuery(tc.INVALID_XQUERY)
190        self.assertEquals(0, result[1]['hits'])
191
192
193    def testRetrieve(self):
194        res = ndgResources()
195        xquery = res.createXQuery('atomList', dc.ATOM_COLLECTION_PATH, "", "")
196        result = self.client._XMLRPCClient__executeQuery(xquery)
197        doc = self.client._XMLRPCClient__retrieve(result[0], 0)
198        self.assertTrue(doc)
199
200
201    def testCreateCollection(self):
202        self.client.removeCollection(tc.TEST_COLLECTION_PATH)
203        self.assertTrue(self.client.createCollection(tc.TEST_COLLECTION_PATH))
204        self.assertTrue(self.client.removeCollection(tc.TEST_COLLECTION_PATH))
205
206
207    def testDuplicateCreateCollection(self):
208        self.assertTrue(self.client.createCollection(tc.TEST_COLLECTION_PATH))
209        self.assertTrue(self.client.createCollection(tc.TEST_COLLECTION_PATH))
210        self.assertTrue(self.client.removeCollection(tc.TEST_COLLECTION_PATH))
211       
212
213    def testRemoveCollection(self):
214        self.assertTrue(self.client.createCollection(tc.TEST_COLLECTION_PATH))
215        self.assertTrue(self.client.removeCollection(tc.TEST_COLLECTION_PATH))
216
217
218    def testInvalidRemoveCollection(self):
219        self.assertFalse(self.client.removeCollection(tc.TEST_COLLECTION_PATH))
220   
221    def testCreateDoc(self):
222        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME))
223        self.assertTrue(self.client.removeDoc(tc.TEST_FILE_NAME))
224   
225    def testCreateDocWithoutOverwrite(self):
226        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME))
227        self.assertRaises(xmlrpclib.Fault, self.client.createDoc, tc.xmlString, tc.TEST_FILE_NAME)
228        self.assertTrue(self.client.removeDoc(tc.TEST_FILE_NAME))
229   
230    def testCreateDocWithOverwrite(self):
231        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME))
232        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME, overwrite=1))
233        self.assertTrue(self.client.removeDoc(tc.TEST_FILE_NAME))
234
235    def testRemoveDoc(self):
236        self.assertTrue(self.client.createDoc(tc.xmlString, tc.TEST_FILE_NAME))
237        self.assertTrue(self.client.removeDoc(tc.TEST_FILE_NAME))
238
239    def testInvalidRemoveDoc(self):
240        self.assertRaises(xmlrpclib.Fault, self.client.removeDoc, tc.TEST_FILE_NAME)
241
242       
243    def tearDown(self):
244        for path  in self.createdAtoms:
245            # delete the test file, in case it has been used in a test
246            self.utils.deleteDoc(path)
247
248           
249if __name__=="__main__":
250    unittest.main()
Note: See TracBrowser for help on using the repository browser.