Anonymous avatar Anonymous committed 3e6dc21

removed old tests, added first functional tests

Comments (0)

Files changed (11)

     def oai_query(self,
                   offset=0,
                   batch_size=20,
-                  sets=[],
+                  needed_sets=[],
                   disallowed_sets=[],
                   allowed_sets=[],
                   from_date=None,
         # filter sets
 
         setclauses = []
-        for set_id in sets:
+        for set_id in needed_sets:
             alias = self._setrefs.alias()
             setclauses.append(
                 sql.and_(
             ut = time.mktime(until)-self.filter_data.delay
             until = datetime.fromtimestamp(ut)
             
-        if set is None:
-            filtersets = []
-        else:
-            filtersets = [set]
-
-        sets = self.config.sets_allowed
-        filtersets += self.config.filter_sets
-        notsets = self.config.sets_disallowed    
+        needed_sets = self.config.sets_needed
+        if not set is None:
+            needed_sets.append(set)
+        allowed_sets = self.config.sets_allowed
+        disallowed_sets = self.config.sets_disallowed    
         
         return self.db.oai_query(offset=cursor,
                                  batch_size=batch_size,
-                                 sets=sets,
-                                 not_sets=notsets,
-                                  filter_sets=filtersets,
+                                 needed_sets=needed_sets,
+                                 disallowed_sets=disallowed_sets,
+                                 allowed_sets=allowed_sets,
                                  from_date=from_,
                                  until_date=until,
                                  identifier=identifier
                  metadata_prefixes = ['oai_dc'],
                  batch_size = 100,
                  content_type = None,
+                 sets_needed = [],
                  sets_allowed = [],
                  sets_disallowed = [],
                  sets_deleted = [],
         self.metadata_prefixes = metadata_prefixes
         self.batch_size = batch_size
         self.content_type = content_type
+        self.sets_needed = sets_needed
         self.sets_allowed = sets_allowed
         self.sets_disallowed = sets_disallowed
         self.sets_deleted = sets_deleted
         self.filter_sets = filter_sets
         self.delay = delay
         self.base_asset_path = base_asset_path or tempfile.gettempdir()
-
-    def get_oai_id(self, internal_id):
-        return 'oai:%s' % internal_id
-
-    def get_internal_id(self, oai_id):
-        return oai_id[4:]
-
-    def get_setspec_id(self, internal_set_id):
-        return 'set_%s' % internal_set_id
-
-    def get_internal_set_id(self, oai_setspec_id):
-        return oai_setspec_id[4:]
-
-    def get_asset_path(self, internal_id, asset):
-        return os.path.abspath(
-            os.path.join(self.base_asset_path,
-                         internal_id,
-                         asset['filename']))
 from unittest import TestCase, TestSuite, makeSuite
 import doctest
 import datetime
+import urllib2
 
 from lxml import etree
+import wsgi_intercept
+from wsgi_intercept.urllib2_intercept import install_opener
 
 from moai.utils import XPath
 from moai.database import Database
+from moai.server import Server, FeedConfig
+from moai.wsgi import MOAIWSGIApp
+
+install_opener()
 
 FLAGS = doctest.NORMALIZE_WHITESPACE + doctest.ELLIPSIS
 GLOBS = {}
         self.assertEquals([r['id'] for r in self.db.oai_query()],
                           [u'oai:ham', u'oai:spam', u'oai:spamspamspam'])
         # only set ham
-        self.assertEquals([r['id'] for r in self.db.oai_query(sets=[u'ham'])],
-                          [u'oai:ham'])
+        self.assertEquals([r['id'] for r in self.db.oai_query(
+            needed_sets=[u'ham'])], [u'oai:ham'])
         # only set spam
-        self.assertEquals([r['id'] for r in self.db.oai_query(sets=[u'spam'])],
-                          [u'oai:spam', u'oai:spamspamspam'])
+        self.assertEquals([r['id'] for r in self.db.oai_query(
+            needed_sets=[u'spam'])], [u'oai:spam', u'oai:spamspamspam'])
         # records in spam set and test set
         self.assertEquals([r['id'] for r in self.db.oai_query(
-            sets=[u'test', u'spam'])], [u'oai:spam'])
+            needed_sets=[u'test', u'spam'])], [u'oai:spam'])
         # only allow records from certain sets
         self.assertEquals([r['id'] for r in self.db.oai_query(
             allowed_sets=[u'test'])], [u'oai:ham', u'oai:spam'])
                           [u'oai:ham', u'oai:spam', u'oai:spamspamspam'])
         # only allow records from certain sets, combined with set
         self.assertEquals([r['id'] for r in self.db.oai_query(
-            allowed_sets=[u'test'], sets=['spam'])],
+            allowed_sets=[u'test'], needed_sets=['spam'])],
                           [u'oai:spam'])
         self.assertEquals([r['id'] for r in self.db.oai_query(
-            allowed_sets=[u'spam'], sets=['test'])],
+            allowed_sets=[u'spam'], needed_sets=['test'])],
                           [u'oai:spam'])
         # certain records should always be disallowed
         self.assertEquals([r['id'] for r in self.db.oai_query(
         self.assertEquals([r['id'] for r in self.db.oai_query(
             batch_size=1, offset=2)], [u'oai:spamspamspam'])
         
-
+class ServerTest(TestCase):
+    def setUp(self):
+        self.database = Database()
+        self.config = FeedConfig('Test Server',
+                                 'http://test',
+                                 admin_emails=['testuser@localhost'],
+                                 metadata_prefixes=['oai_dc', 'mods'])
+        self.server = Server('http://test', self.database, self.config)
+        wsgi_intercept.add_wsgi_intercept('test', 80,
+                                          lambda : MOAIWSGIApp(self.server))
         
+    def tearDown(self):
+        del self.database
+        del self.config
+        del self.server
+        wsgi_intercept.remove_wsgi_intercept('test', 80)
+        
+    def test_identify(self):
+        xml = urllib2.urlopen('http://test?verb=Identify').read()
+        doc = etree.fromstring(xml)
+        xpath = XPath(doc, nsmap=
+                      {"oai" :"http://www.openarchives.org/OAI/2.0/"})
+        self.assertEquals(xpath.string('//oai:repositoryName'),u'Test Server')
+        
+    
 def suite():
     test_suite = TestSuite()
     test_suite.addTest(makeSuite(XPathUtilTest))
     test_suite.addTest(makeSuite(DatabaseTest))
+    test_suite.addTest(makeSuite(ServerTest))
     return test_suite

moai/tests/__init__.py

-#

moai/tests/test_assets.py

-import os
-import tempfile
-import logging
-import shutil
-import StringIO
-from unittest import TestCase, TestSuite, makeSuite, main
-
-from moai.provider.list import ListBasedContentProvider
-from moai.content import DictBasedContentObject
-from moai.update import DatabaseUpdater
-from moai.database.btree import BTreeDatabase
-from moai.server import Server, FeedConfig
-from moai.http.cgi import CGIRequest
-from moai.tests.test_database import DATA
-from moai.core import MOAI
-
-class AssetTest(TestCase):
-
-    def setUp(self):
-        # init moai, so metadata formats get registered
-        m = MOAI(logging, debug=True)
-
-        self.db = BTreeDatabase()
-        provider = ListBasedContentProvider(DATA)
-        updater = DatabaseUpdater(provider, DictBasedContentObject, self.db, logging)
-        updater.update_provider()
-        updater.update_database()
-        
-        config = FeedConfig('test', 'A test Repository',
-                            'http://localhost/repo/test',
-                            logging)
-        
-        self.server = Server('http://localhost/repo', self.db)
-        self.server.add_config(config)
-
-        self.asset_dir = os.path.join(tempfile.gettempdir(),
-                                 'id:2')
-        if not os.path.isdir(self.asset_dir):
-            os.mkdir(self.asset_dir)
-        fp = open(os.path.join(self.asset_dir, 'test.txt'), 'w')
-        fp.write('This is an asset')
-        fp.close()
-        
-        
-    def tearDown(self):
-        del self.db
-        del self.server
-
-        if os.path.isfile(self.asset_dir):
-            shutil.rmtree(self.asset_dir)
-
-    def request(self, url, **kwargs):
-        req = CGIRequest('http://localhost/repo/test' + url, **kwargs)
-        req.stream = StringIO.StringIO()
-        self.server.handle_request(req)
-        req.stream.seek(0)
-        return req.stream.read()
-
-    def test_asset_200(self):
-        # downloading this asset should work
-        response = self.request('/asset/id:2/test.txt')
-        self.assertEquals(response.splitlines()[0],
-                          'Status: 200 OK')
-    def test_asset_404(self):
-        # a non existing asset should return a 404
-        response = self.request('/asset/id:2/foo')
-        self.assertEquals(response.splitlines()[0],
-                          'Status: 404 File not Found')
-    def test_asset_403(self):
-        # let's make a feed which does not include id:2
-        # because it disallows everythin in the datasets set
-        config = FeedConfig('test', 'A test Repository',
-                            'http://localhost/repo/test',
-                            logging,
-                            sets_disallowed=[u'datasets'])
-        
-        self.server = Server('http://localhost/repo', self.db)
-        self.server.add_config(config)
-        
-        # first we'll test that the record is not in the feed
-        response = self.request('', verb='ListIdentifiers', metadataPrefix='oai_dc')
-        self.assertEquals(response.count('<identifier>'), 2)
-        self.assertFalse('<identifier>oai:id:2</identifier>' in response)
-
-        # now, we when we get the asset, it should return a forbidden
-        response = self.request('/asset/id:2/test.txt')
-        self.assertEquals(response.splitlines()[0],
-                          'Status: 403 Forbidden')
-
-        
-   
-def test_suite():
-    return TestSuite((makeSuite(AssetTest), ))
-
-
-if __name__ == '__main__':
-    main(defaultTest='test_suite')
-    pass

moai/tests/test_database.py

-from datetime import datetime
-from unittest import TestCase, TestSuite, makeSuite, main
-import logging
-
-from moai.provider.list import ListBasedContentProvider
-from moai.content import DictBasedContentObject
-from moai.update import DatabaseUpdater
-from moai.database.btree import BTreeDatabase
-from moai.database.sqlite import SQLiteDatabase
-from moai.error import UnknownRecordID
-
-from sqlalchemy.exc import IntegrityError
-
-class BtreeDatabaseTest(TestCase):
-
-    def setUp(self):
-        self.db = BTreeDatabase()
-        provider = ListBasedContentProvider(DATA)
-        updater = DatabaseUpdater(provider, DictBasedContentObject, 
-                                  self.db, logging)
-        updater.update_provider()
-        updater.update_database()
-        
-    def tearDown(self):
-        del self.db
-
-    def testGetRecord(self):
-        self.assertEquals(self.db.get_record(u'bla'), None)
-        record = self.db.get_record(u'id:1')
-        self.assertEquals(type(record.get('id')), unicode)
-        self.assertEquals(type(record.get('when_modified')), datetime)
-        self.assertEquals(type(record.get('deleted')), bool)
-        self.assertEquals(type(record.get('is_set')), bool)
-        self.assertEquals(type(record.get('content_type')), unicode)
-
-    def testGetMetadata(self):
-        self.assertEquals(self.db.get_metadata(u'bla'), None)
-        record = self.db.get_metadata(u'id:1')
-        self.assertEquals(record.get('abstract'),
-                          [u'A test publication'])
-
-    def testGetSets(self):
-        self.assertEquals(self.db.get_sets(u'bla'), [])
-        sets = self.db.get_sets(u'id:1')
-        self.assertEquals(len(sets), 3)
-
-    def testGetAssets(self):
-        self.assertEquals(self.db.get_assets(u'bla'), [])
-        self.assertEquals(self.db.get_assets(u'id:1'), [])
-        assets = self.db.get_assets(u'id:2')
-        self.assertEquals(len(assets), 1)
-        asset = assets[0]
-        self.assertEquals(type(asset.get('filename')), unicode)
-        self.assertEquals(type(asset.get('url')), unicode)
-        self.assertEquals(type(asset.get('mimetype')), unicode)
-        self.assertEquals(type(asset.get('md5')), unicode)
-        self.assertEquals(type(asset.get('absolute_uri')), unicode)
-        self.assertEquals(type(asset.get('metadata')), dict)
-        self.assertTrue(asset['metadata'].has_key('type'))
-        self.assertEquals(asset['metadata']['type'], [u'preprint'])
-
-    def testSetAddRemove(self):
-        # we have 5 sets to begin with
-        result = list(self.db.oai_sets(offset=0, batch_size=100))
-        self.assertEquals(len(result), 5)
-        self.db.add_set(u'added set', u'An added set', description=u'A set description')
-        self.db.flush_update()
-        result = list(self.db.oai_sets(offset=0, batch_size=100))
-        self.assertEquals(len(result), 6)
-        self.assertEquals(self.db.get_set(u'added set')['name'], 'An added set')
-        self.db.remove_set(u'added set')
-        result = list(self.db.oai_sets(offset=0, batch_size=100))
-        self.assertEquals(len(result), 5)
-
-    def testRecordAddRemove(self):
-        # we have 3 records to begin with
-        result = list(self.db.oai_query(offset=0, batch_size=100))
-        self.assertEquals(len(result), 3)
-        self.db.remove_content(u'id:1')
-        result = list(self.db.oai_query(offset=0, batch_size=100))
-        self.assertEquals(len(result), 2)
-        
-    def testOAISets(self):
-        result = list(self.db.oai_sets(offset=0, batch_size=100))
-        self.assertEquals(len(result), 5)
-        result = list(self.db.oai_sets(offset=2, batch_size=100))
-        self.assertEquals(len(result), 3)
-        result = list(self.db.oai_sets(offset=4, batch_size=100))
-        self.assertEquals(len(result), 1)
-
-    def testOAIQueryBatching(self):
-        result = list(self.db.oai_query(offset=0, batch_size=100))
-        self.assertEquals(len(result), 3)
-        result = list(self.db.oai_query(offset=1, batch_size=100))
-        self.assertEquals(len(result), 2)
-        result = list(self.db.oai_query(offset=2, batch_size=100))
-        self.assertEquals(len(result), 1)
-        result = list(self.db.oai_query(offset=3, batch_size=100))
-        self.assertEquals(result, [])
-        result = list(self.db.oai_query(offset=10, batch_size=100))
-        self.assertEquals(result, [])
-        result = list(self.db.oai_query(batch_size=2))
-        self.assertEquals(len(result), 2)
-        result = list(self.db.oai_query(offset=0, batch_size=0))
-        self.assertEquals(result, [])
-        result = list(self.db.oai_query(batch_size=-1))
-        self.assertEquals(result, [])
-
-    def testOAIQueryDateStamps(self):
-        result = list(self.db.oai_query())
-        self.assertEquals(len(result), 3)
-        result = list(self.db.oai_query(from_date=datetime(2000, 01, 01)))
-        self.assertEquals(len(result), 3)
-        result = list(self.db.oai_query(from_date=datetime(2005, 01, 01)))
-        self.assertEquals(len(result), 2)
-        result = list(self.db.oai_query(from_date=datetime(2007, 01, 01)))
-        self.assertEquals(len(result), 1)
-        result = list(self.db.oai_query(from_date=datetime(2020, 01, 01)))
-        self.assertEquals(len(result), 0)
-        result = list(self.db.oai_query(until_date=datetime(2000, 01, 01)))
-        self.assertEquals(len(result), 0)
-        result = list(self.db.oai_query(until_date=datetime(2005, 01, 01)))
-        self.assertEquals(len(result), 1)
-        result = list(self.db.oai_query(until_date=datetime(2007, 01, 01)))
-        self.assertEquals(len(result), 2)
-        result = list(self.db.oai_query(until_date=datetime(2020, 01, 01)))
-        self.assertEquals(len(result), 3)
-        result = list(self.db.oai_query(from_date=datetime(2003, 01, 01),
-                                        until_date=datetime(2005, 01, 01)))
-        self.assertEquals(len(result), 1)
-
-    def testOAIQuerySets(self):
-        result = list(self.db.oai_query(sets=[u'stuff']))
-        self.assertEquals(len(result), 3)
-        result = list(self.db.oai_query(sets=[u'publications']))
-        self.assertEquals(len(result), 2)
-        result = list(self.db.oai_query(sets=[u'datasets']))
-        self.assertEquals(len(result), 1)
-        result = list(self.db.oai_query(sets=[u'datasets',
-                                              u'publications']))
-        self.assertEquals(len(result), 3)
-        result = list(self.db.oai_query(not_sets=[u'stuff']))
-        self.assertEquals(len(result), 0)
-        result = list(self.db.oai_query(not_sets=[u'publications']))
-        self.assertEquals(len(result), 1)
-        result = list(self.db.oai_query(not_sets=[u'datasets']))
-        self.assertEquals(len(result), 2)
-        result = list(self.db.oai_query(sets=[u'stuff'], 
-                                        not_sets=['publications']))
-        self.assertEquals(len(result), 1)
-        result = list(self.db.oai_query(sets=[u'publications'], 
-                                        filter_sets=['top']))
-        self.assertEquals(len(result), 1)
-
-    def testEmptyDatabase(self):
-        result = list(self.db.oai_sets()) 
-        self.assertEquals(len(result), 5)
-        result = list(self.db.oai_query())
-        self.assertEquals(len(result), 3)
-
-        self.db.empty_database()
-
-        result = list(self.db.oai_sets()) 
-        self.assertEquals(result, [])
-        result = list(self.db.oai_query())
-        self.assertEquals(result, [None])
-
-
-class SQLiteDatabaseTest(BtreeDatabaseTest):
-    def setUp(self):
-        self.db = SQLiteDatabase()
-        provider = ListBasedContentProvider(DATA)
-        updater = DatabaseUpdater(provider, DictBasedContentObject, 
-                                  self.db, logging)
-        updater.update_provider()
-        updater.update_database()
-
-    def tearDown(self):
-        del self.db
-
-    def testEmptyDatabase(self):
-        result = list(self.db.oai_sets())
-        self.assertEquals(len(result), 5)
-        result = list(self.db.oai_query())
-        self.assertEquals(len(result), 3) 
-
-        self.db.empty_database()
-              
-        result = list(self.db.oai_sets())
-        self.assertEquals(result, [])
-        result = list(self.db.oai_query())
-        self.assertEquals(result, [])
-        
-        
-def test_suite():
-    suite = TestSuite()
-    suite.addTest(makeSuite(BtreeDatabaseTest))
-    suite.addTest(makeSuite(SQLiteDatabaseTest))
-    return suite
-
-
-if __name__ == '__main__':
-    main(defaultTest='test_suite')
-
-
-DATA = [{'id': u'id:1',
-         'label': u'Publication 1',
-         'content_type': u'publication',
-         'when_modified': datetime(2008, 01, 01, 14, 30, 00),
-         'deleted': False,
-         'sets': [u'stuff', u'publications', u'top'],
-         'is_set': False,
-         'abstract': [u'A test publication']
-         },
-        {'id': u'id:2',
-         'label': u'Dataset 1',
-         'content_type': u'dataset',
-         'when_modified': datetime(2004, 01, 01, 14, 30, 00),
-         'deleted': False,
-         'sets': [u'stuff', u'datasets', u'dynamic'],
-         'is_set': False,
-         'assets': [{
-             u'filename': u'test.txt',
-             u'url': u'http://example.org/assets/test.txt',
-             u'mimetype': u'text/plain',
-             u'md5': u'730652c70a12db042b8842f5049390d4',
-             u'absolute_uri': u'file:///tmp/test.txt',
-             u'metadata': {u'access': [u'public'],
-                           u'type': [u'preprint']}}]
-
-         },
-        {'id': u'id:3',
-         'label': u'Publication 2',
-         'content_type': u'publication',
-         'when_modified': datetime(2006, 01, 01, 14, 30, 00),
-         'deleted': False,
-         'sets': [u'stuff', u'publications'],
-         'is_set': False,
-         },
-        {'id': u'stuff',
-         'label': u'Stuff',
-         'content_type': u'collection',
-         'when_modified': datetime(2006, 01, 01, 14, 30, 00),
-         'deleted': False,
-         'sets': [],
-         'is_set': True,
-         },
-        {'id': u'publications',
-         'label': u'publication set',
-         'content_type': u'collection',
-         'when_modified': datetime(2006, 01, 01, 14, 30, 00),
-         'deleted': False,
-         'sets': [],
-         'is_set': True,
-         'abstract': [u'A test publication']
-         },
-        {'id': u'datasets',
-         'label': u'datasets',
-         'content_type': u'collection',
-         'when_modified': datetime(2006, 01, 01, 14, 30, 00),
-         'deleted': False,
-         'sets': [],
-         'is_set': True,
-         },
-        {'id': u'top',
-         'label': u'top publications',
-         'content_type': u'collection',
-         'when_modified': datetime(2006, 01, 01, 14, 30, 00),
-         'deleted': False,
-         'sets': [],
-         'is_set': True,
-         },
-        ]
-
-

moai/tests/test_docs.py

-import unittest
-import doctest
-import datetime
-
-FLAGS = doctest.NORMALIZE_WHITESPACE + doctest.ELLIPSIS
-GLOBS = {'datetime':datetime}
-
-def test_suite():
-
-    suite = unittest.TestSuite()
-    suite.addTests([
-        doctest.DocFileSuite('README.txt',
-                             package='moai',
-                             globs=GLOBS,
-                             optionflags=FLAGS),
-        ])
-
-    return suite
-

moai/tests/test_oai.py

-from unittest import TestCase, TestSuite, makeSuite, main
-import logging
-import StringIO
-
-from moai.provider.list import ListBasedContentProvider
-from moai.content import DictBasedContentObject
-from moai.update import DatabaseUpdater
-from moai.database.btree import BTreeDatabase
-from moai.database.sqlite import SQLiteDatabase
-from moai.server import Server, FeedConfig
-from moai.http.cgi import CGIRequest
-from moai.tests.test_database import DATA
-from moai.core import MOAI
-
-class OAIBtreeTest(TestCase):
-
-    def get_database(self):
-        return BTreeDatabase()
-
-    def setUp(self):
-        # init moai, so metadata formats get registered
-        m = MOAI(logging, debug=True)
-        self.db = self.get_database()
-        provider = ListBasedContentProvider(DATA)
-        updater = DatabaseUpdater(provider, DictBasedContentObject, self.db, logging)
-        updater.update_provider()
-        updater.update_database()
-        config = FeedConfig('test', 'A test Repository',
-                            'http://localhost/repo/test',
-                            logging)
-        self.server = Server('http://localhost/repo', self.db)
-        self.server.add_config(config)
-
-
-    def request(self, **kwargs):
-        req = CGIRequest(u'http://localhost/repo/test', **kwargs)
-        req.stream = StringIO.StringIO()
-        self.server.handle_request(req)
-        req.stream.seek(0)
-        return req.stream.read()
-        
-    def tearDown(self):
-        del self.db
-        del self.server
-
-    def test_identify(self):
-        response = self.request(verb=u'Identify')
-        self.assertEquals('Status: 200 OK', response.splitlines()[0])
-        self.assertTrue(
-            '<repositoryName>A test Repository</repositoryName>' in response)
-
-    def test_list_identifiers(self):
-        response = self.request(verb=u'ListIdentifiers', metadataPrefix=u'oai_dc')
-        self.assertEquals(response.count('<identifier>'), 3)
-
-    def test_list_records(self):
-        response = self.request(verb=u'ListRecords', metadataPrefix=u'oai_dc')
-        self.assertEquals(response.count('<identifier>'), 3)
-        self.assertEquals(response.count('<metadata>'), 3)
-
-        response = self.request(verb=u'ListRecords', metadataPrefix=u'oai_dc', until='2004-01-01T14:30:00Z')
-        self.assertEquals(response.count('<identifier>'), 1)
-
-        response = self.request(verb=u'ListRecords', metadataPrefix=u'oai_dc', until='2004-01-01T14:29:59Z')
-        self.assertEquals(response.count('<identifier>'), 0)
-
-        response = self.request(verb=u'ListRecords', metadataPrefix=u'oai_dc', from_='2008-01-01T14:30:00Z')
-        self.assertEquals(response.count('<identifier>'), 1)
-
-        response = self.request(verb=u'ListRecords', metadataPrefix=u'oai_dc', from_='2008-01-01T14:30:01Z')
-        self.assertEquals(response.count('<identifier>'), 0)
-
-        response = self.request(verb=u'ListRecords', metadataPrefix=u'oai_dc',
-                                until='2004-01-01T14:30:00Z',
-                                from_='2004-01-01T14:30:00Z')
-        self.assertEquals(response.count('<identifier>'), 1)
-
-
-    def test_list_metadata_formats(self):
-        response = self.request(verb=u'ListMetadataFormats')
-        self.assertTrue('<metadataPrefix>oai_dc</metadataPrefix>' in response)
-        response = self.request(verb=u'ListMetadataFormats', identifier='oai:id:1')
-        self.assertTrue('<metadataPrefix>oai_dc</metadataPrefix>' in response)
-
-    def test_get_record(self):
-        response = self.request(verb=u'GetRecord', identifier=u'oai:id:1',
-                                metadataPrefix=u'oai_dc')
-        self.assertEquals(response.count('<identifier>'), 1)
-        self.assertTrue('<identifier>oai:id:1</identifier>' in response)
-        
-        response = self.request(verb=u'GetRecord', identifier=u'oai:id:3',
-                                metadataPrefix=u'oai_dc')
-        self.assertEquals(response.count('<identifier>'), 1)
-        self.assertTrue('<identifier>oai:id:3</identifier>' in response)
-        
-        response = self.request(verb=u'GetRecord', identifier=u'oai:id:2',
-                                metadataPrefix=u'oai_dc')
-        self.assertEquals(response.count('<identifier>'), 1)
-        self.assertTrue('<identifier>oai:id:2</identifier>' in response)
-
-        response = self.request(verb=u'GetRecord', identifier=u'invalid"id',
-                                metadataPrefix=u'oai_dc')
-        self.assertTrue('code="idDoesNotExist"' in response)
-
-    def test_disallowed_set_record(self):
-        config = FeedConfig('test', 'A test Repository',
-                            'http://localhost/repo/test',
-                            logging,
-                            sets_disallowed=[u'datasets'])
-        self.server = Server('http://localhost/repo', self.db)
-        self.server.add_config(config)
-        # identifier 2 should not be in the identifier list
-        response = self.request(verb=u'ListIdentifiers', metadataPrefix=u'oai_dc')
-        self.assertFalse('<identifier>oai:id:2</identifier>' in response)
-        response = self.request(verb=u'GetRecord', identifier=u'oai:id:2',
-                                metadataPrefix=u'oai_dc')
-        self.assertTrue('<error code="idDoesNotExist">oai:id:2</error>' in response)
-
-    def test_deleted_set_record(self):
-        config = FeedConfig('test', 'A test Repository',
-                            'http://localhost/repo/test',
-                            logging,
-                            sets_deleted=[u'datasets'])
-        self.server = Server('http://localhost/repo', self.db)
-        self.server.add_config(config)
-        # identifier 2 should be in the identifier list
-        response = self.request(verb=u'ListIdentifiers', metadataPrefix=u'oai_dc')
-        self.assertTrue('<identifier>oai:id:2</identifier>' in response)
-        response = self.request(verb=u'GetRecord', identifier=u'oai:id:2',
-                                metadataPrefix=u'oai_dc')
-        self.assertTrue('<header status="deleted">' in response)
-        self.assertFalse('<metadata>' in response)
-
-class OAISQLiteTest(OAIBtreeTest):
-    
-    def get_database(self):
-        return SQLiteDatabase()
-
-   
-def test_suite():
-    suite = TestSuite()
-    suite.addTest(makeSuite(OAIBtreeTest))
-    suite.addTest(makeSuite(OAISQLiteTest))
-    return suite
-
-if __name__ == '__main__':
-    main(defaultTest='test_suite')

moai/tests/test_updater.py

-import logging
-import os
-import tempfile
-import StringIO
-import copy
-from datetime import datetime
-from unittest import TestCase, TestSuite, makeSuite, main
-
-from moai.provider.list import ListBasedContentProvider
-from moai.content import DictBasedContentObject
-from moai.update import DatabaseUpdater
-from moai.database.sqlite import SQLiteDatabase
-from moai.database.btree import BTreeDatabase
-from sqlalchemy.exc import IntegrityError
-from moai.tests.test_database import DATA
-from moai.server import Server, FeedConfig
-from moai.http.cgi import CGIRequest
-from moai.core import MOAI
-
-
-class UpdaterTest(TestCase):
-
-    def setUp(self):
-        self.test_data = copy.deepcopy(DATA)
-        self.db = SQLiteDatabase()
-        self.provider = ListBasedContentProvider(self.test_data)
-        self.updater = DatabaseUpdater(self.provider, DictBasedContentObject, 
-                                       self.db, logging)
-        self.updater.flush_threshold = -1
-
-    def tearDown(self):
-        del self.db
-
-    def request(self, url, **kwargs):
-        req = CGIRequest('http://localhost/repo/test' + url, **kwargs)
-        req.stream = StringIO.StringIO()
-        self.server.handle_request(req)
-        req.stream.seek(0)
-        return req.stream.read()
-
-    def setup_server(self):
-        # init moai, so metadata formats get registered
-        m = MOAI(logging, debug=True)
-        
-        config = FeedConfig('test', 'A test Repository',
-                            'http://localhost/repo/test',
-                            logging)
-        self.server = Server('http://localhost/repo', self.db)
-        self.server.add_config(config)
-
-    def setup_test_provider(self, data, supress_errors=True):
-        if 'server' not in dir(self):
-            self.setup_server()
-
-        self.db.empty_database()
-
-        provider = ListBasedContentProvider(data)
-        self.updater = DatabaseUpdater(provider, DictBasedContentObject, self.db, logging)
-        self.updater.update_provider()
-        if supress_errors:
-            self.updater.update_database(supress_errors=True)
-        else:
-            self.assertRaises(Exception, self.updater.update_database)
-
-    def test_update_provider(self):
-        # Check updating provider
-        result = self.updater.update_provider()
-        expRes = [0, 1, 2, 3, 4, 5, 6]
-        self.assertEqual(result, expRes)
-
-        # Check updating database 
-        result = self.updater.update_database_iterate()
-        update_list1 = []
-        for line in result:
-            update_list1.append(line)
-
-        # Modify data for tests below
-        self.test_data[5]['when_modified'] = ''
-        self.test_data[6].pop('when_modified')
-
-        # Check updating provider with modified data and from_date 
-        from_date = datetime(2006, 01, 01) 
-        result = self.updater.update_provider(from_date)
-        expRes = [0, 2, 3, 4]
-        self.assertEqual(result, expRes)
-
-        # Update the database with failing data, fails last 2 records
-        result = self.updater.update_database_iterate()
-        update_list2 = []
-        try:
-            for line in result:
-                update_list2.append(line)
-        except AssertionError:
-            pass
-        self.assertEqual(update_list1[:-2], update_list2)
-
-        # The same, but continue on errors
-        result = self.updater.update_database_iterate(supress_errors=True)
-        update_list2 = []
-        for line in result:
-            update_list2.append(line)
-        self.assertEqual(update_list1[:-2], update_list2[:-2])
-        self.assertNotEqual(update_list1[-2:], update_list2[-2:])
-
-    def test_update_database(self):
-        self.updater.update_provider()
-
-        # Normal update works, but fails on non-uique ids
-        self.updater.update_database()
-        self.assertRaises(IntegrityError, self.updater.update_database)
-        # But suppress errors works, and emptying the db works
-        self.updater.update_database(supress_errors=True)
-        self.db.empty_database()
-        self.updater.update_database()
-
-        # Same tests with threshold = 1
-        self.updater.flush_threshold = 1
-        self.db.empty_database()
-
-        # Normal update works, but fails on non-uique ids
-        self.updater.update_database()
-        self.assertRaises(IntegrityError, self.updater.update_database)
-        # But suppress errors works, and emptying the db works
-        self.updater.update_database(supress_errors=True)
-        self.db.empty_database()
-        self.updater.update_database()
-
-    def test_xml_compatibility(self):
-        # These fail when the content is sanatized in the IContentObject implementation
-
-        # Failure in metadata
-        test_data = copy.deepcopy(DATA)
-        test_data[0]['abstract'] = [u'A test publi\u000bcation']
-        self.setup_test_provider(test_data)
-        
-        response = self.request('', verb='ListRecords', metadataPrefix='oai_dc')
-        self.assertFalse('<identifier>oai:id:1</identifier>' in response)
-        self.assertTrue('<identifier>oai:id:2</identifier>' in response)
-        self.assertTrue('<identifier>oai:id:3</identifier>' in response)
-
-        # Failure in record
-        test_data = copy.deepcopy(DATA)
-        test_data[1]['id'] = u"i\u000bd:2" 
-        self.setup_test_provider(test_data)
-        
-        response = self.request('', verb='ListRecords', metadataPrefix='oai_dc')
-        self.assertTrue('<identifier>oai:id:1</identifier>' in response)
-        self.assertFalse('<identifier>oai:id:2</identifier>' in response)
-        self.assertFalse(u'<identifier>oai:i\u000bd:2</identifier>' in response)
-        self.assertTrue('<identifier>oai:id:3</identifier>' in response)
-
-        # Failure in sets
-        test_data = copy.deepcopy(DATA)
-        test_data[2]['sets'][0] = u"st\u000buff"
-        self.setup_test_provider(test_data)
-
-        response = self.request('', verb='ListRecords', metadataPrefix='oai_dc')
-        self.assertTrue('<identifier>oai:id:1</identifier>' in response)
-        self.assertTrue('<identifier>oai:id:2</identifier>' in response)
-        self.assertFalse('<identifier>oai:id:3</identifier>' in response)
-
-        # Check the supress_errors on _xml_comp_error()
-        test_data = copy.deepcopy(DATA)
-        test_data[0]['abstract'] = [u'A test publi\u000bcation'] 
-        self.setup_test_provider(test_data, False)
-
-
-def test_suite():
-    suite = TestSuite()
-    suite.addTest(makeSuite(UpdaterTest))
-    return suite
-
-
-if __name__ == '__main__':
-    main(defaultTest='test_suite')
-
-
     install_requires=[
     'pyoai',
     'WSGIUtils',
+    'wsgi_intercept',
     'webob',
     'pasteScript',
     'sqlalchemy',
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.