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

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

Add new test suite for RESTful client + update various other test classes to cope with changes to client + proxy uses.

Line 
1
2import unittest,os
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.assertFalse(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.