source: TI01-discovery/branches/ingestAutomation-upgrade/OAIBatch/testfeeddocumentingester.py @ 4948

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI01-discovery/branches/ingestAutomation-upgrade/OAIBatch/testfeeddocumentingester.py
Revision 4948, 2.9 KB checked in by cbyrom, 11 years ago (diff)

Adjust ingest scripts to use the new ndg common client suite + add new
testdata to improve universality of tests.

Line 
1'''
2A test class for the feeddocumentingester module.
3
4@author C Byrom, Tessella Jan 2009
5'''
6import unittest, time, datetime
7from feeddocumentingester import FeedDocumentIngester
8from ndg.common.src.models.vocabtermdata import VocabTermData as VTD
9import ndg.common.unittests.testconstants as tc
10from ndg.common.unittests.testutils import testUtils as tu
11
12class testFeedDocumentIngester(unittest.TestCase):
13   
14    VALID_INTERVAL = 200.0
15    INVALID_INTERVAL = 'sausages'
16    VALID_DATE = '2000-12-02'
17    INVALID_DATE = '200x0-12-02'
18    VALID_FEED_URL = 'http://chinook.badc.rl.ac.uk:8080/exist/atom/content/db/atoms'
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.ingester = FeedDocumentIngester()
26        self.createdAtoms = []    # array to store paths to atoms created - in order to then delete them
27
28    def testSetPollInterval(self):
29        try:
30            self.ingester.setPollInterval(self.VALID_INTERVAL)
31            self.assertEquals(self.VALID_INTERVAL, self.ingester.pollInterval)
32        except:
33            self.fail("Should not have thrown an exception")
34
35    def testInvalidSetPollInterval(self):
36        try:
37            self.ingester.setPollInterval(self.INVALID_INTERVAL)
38            self.fail("Should have thrown an exception")
39        except Exception, e:
40            self.assertTrue(e.message.find('invalid literal for float') > -1)
41
42    def testSetIngestFromDate(self):
43        try:
44            self.ingester.setIngestFromDate(self.VALID_DATE)
45            self.assertTrue(self.ingester.ingestFromDate < datetime.datetime.now())
46        except:
47            self.fail("Should not have thrown an exception")
48
49    def testInvalidSetIngestFromDate(self):
50        try:
51            self.ingester.setIngestFromDate(self.INVALID_DATE)
52            self.fail("Should have thrown an exception")
53        except Exception, e:
54            self.assertTrue(e.message.find('not in ISO8601 format') > -1)
55
56
57    def testProcessFeed(self):
58        try:
59            # firstly create doc in eXist to ensure there is an entry to ingest
60            # NB, for this test to work, we need use a data entity - since this
61            # is the only type of document that we'll ingest via the feed
62            atom = tu(tc.EXIST_DBCONFIG_FILE).createAtomInEXist(tc.xmlString)
63            self.createdAtoms.append(atom.getFullPath())
64
65            # ensure a feed client is available
66            self.ingester.setupPoll()
67            self.ingester.setIngestFromDate(self.VALID_DATE)
68
69            self.ingester.processFeed(self.VALID_FEED_URL)
70        except Exception, e:
71            print e.message
72            self.fail("Should not have thrown an exception")
73           
74           
75    def tearDown(self):
76        for path in self.createdAtoms:
77            # delete the test file, in case it has been used in a test
78            self.utils.deleteDoc(path)
79
80if __name__ == '__main__':
81
82    unittest.main()
Note: See TracBrowser for help on using the repository browser.