grassyknoll.tests.test

1 """ 2 Test various sample json data with various backends. 3 """ 4 5 import unittest 6 import logging 7 import os 8 import tempfile 9 import datetime 10 from grassyknoll.tests import loadFixtures 11 from grassyknoll.collection import Collection 12 from grassyknoll.backend.dictionary import DictCollection 13 from grassyknoll.backend.litesql import SqliteCollection 14 from grassyknoll.backend.lucene import LuceneCollection 15 from grassyknoll.frontend.RestCollection import CollectionApplication 16 from grassyknoll.client.ClientCollection import ClientCollection 17 from grassyknoll.lib import Norman 18 import nsf 19 20 import wsgi_intercept.httplib2_intercept 21 wsgi_intercept.httplib2_intercept.install() 22 23 directory = os.path.normpath(os.path.dirname(__file__) + '/../../samples') 24

25 -class TestBase(unittest.TestCase):

26 "Handle generic data loading and deleting." 27 source = None 28 input_norman=Norman.IdemNorman() 29

30 - def make_collection(self):

31 raise NotImplementedError

32

33 - def setUp(self):

34 self.collection=self.make_collection() 35 items = loadFixtures(os.path.join(directory, self.source)) 36 for id, doc in items: 37 try: 38 doc=self.input_norman(doc) 39 except Norman.NormanError: 40 # just drop things that don't norman 41 logging.warn("norman error %s", id, exc_info=True) 42 else: 43 doc = Collection.CollectionDocument(doc, __id__=id) 44 self.collection.create([doc])

45

46 - def tearDown(self):

47 self.collection.close() 48 self.collection.cleanUp()

49

50 -class ShakespeareDict(TestBase):

51 source = 'shakespeare' 52

53 - def make_collection(self):

54 return DictCollection.DictCollection()

55

56 - def test(self):

57 fields = '__id__', 'act', 'contents', 'play', 'scene', 'title' 58 ids = self.collection.list() 59 assert len(self.collection) == len(ids) == 754 60 assert sorted(self.collection.retrieve([ids[0]])[0]) == list(fields)

61

62 -class ShakespeareLucene(TestBase):

63 source = 'shakespeare' 64 storage_norman = LuceneCollection.LuceneNorman() 65 storage_norman.play = LuceneCollection.SmartFieldNorman(store=True, index='tokenized', alltext=True) 66 storage_norman.title = LuceneCollection.SmartFieldNorman(store=True, index='tokenized', alltext=True) 67 storage_norman.contents = LuceneCollection.SmartFieldNorman(index='tokenized', alltext=True) 68 storage_norman.act = LuceneCollection.SmartFieldNorman(index='untokenized') 69 storage_norman.scene = LuceneCollection.SmartFieldNorman(index='untokenized') 70

71 - def make_collection(self):

72 return LuceneCollection.LuceneCollection(tempfile.mktemp(prefix='gk_shakespeare_'), 73 self.storage_norman, create=True)

74

75 - def test(self):

76 fields = '__id__', 'play', 'title' 77 ids = self.collection.list() 78 assert len(self.collection) == len(ids) == 754 79 assert sorted(self.collection.retrieve([ids[0]])[0]) == list(fields) 80 docs = self.collection.searchQuery('play:"romeo and juliet"') 81 assert len(docs) == 26 82 assert all(doc['play'] == 'Romeo and Juliet' for doc in docs) 83 docs = self.collection.searchQuery('play:"romeo and juliet" AND act:1') 84 assert len(docs) == 6 85 scores = [doc['__score__'] for doc in docs] 86 assert all(scores) and scores == sorted(scores, reverse=True) 87 docs = self.collection.searchQuery('play:"romeo and juliet" AND act:1', fields=set(['play'])) 88 assert sorted(docs[0]) == ['__score__', 'play']

89

90 -class ShakespeareLuceneRAM(ShakespeareLucene):

91 - def make_collection(self):

92 return LuceneCollection.LuceneCollection(None, self.storage_norman)

93

94 -class NSFLucene(TestBase):

95 source = 'nsf_ra' 96 storage_norman = nsf.NSFLuceneNorman 97 input_norman=nsf.NSFNorman 98

99 - def make_collection(self):

100 return LuceneCollection.LuceneCollection(tempfile.mktemp(prefix='gk_nsf_lucene_'), 101 self.storage_norman, create=True)

102

103 - def test(self):

104 ids = self.collection.list() 105 106 # XXX breaks on Client tests, see #67 107 if not isinstance(self, ClientTestBase): 108 assert len(self.collection) == len(ids) == 645 109 110 docs = self.collection.searchQuery(q='"machine learning"') 111 assert [d.id for d in docs] == ['a0305543', 'a0305567', 'a0313480'] 112 assert docs[0]['Title'].startswith('ALGORITHMS') 113 # XXX some more tests would be nice 114 assert len(self.collection.searchQuery(q='model')) == 91 115 assert len(self.collection.searchQuery(q='Date:[20030101 TO 20030601]')) == 562

116

117 -class NSFSQL(TestBase):

118 source = 'nsf_ra' 119 table = nsf.NSFTable 120 input_norman=nsf.NSFNorman 121

122 - def make_collection(self):

123 return SqliteCollection.SqliteCollection(tempfile.mktemp(prefix='gk_nsf_sqlite_'), 124 self.table, create=True)

125

126 - def test(self):

127 ids = self.collection.list() 128 129 # XXX breaks on Client tests, see #67 130 if not isinstance(self, ClientTestBase): 131 assert len(self.collection) == len(ids) == 645 132 133 assert len(self.collection.andQuery(Type='Award')) == 645 134 assert len(self.collection.andQuery(Type='Award', Award_Instr='Standard Grant')) == 457 135 assert len(self.collection.andQuery(Date=datetime.date(2003, 3, 26))) == 12 136 results=self.collection.andQuery(NSF_Program=1214, Award_Instr='Standard Grant') 137 assert len(results) == 1 138 assert results[0].id == 'a0300005'

139

140 -class ClientTestBase(TestBase):

141 - def make_collection(self):

142 underlying = super(ClientTestBase, self).make_collection() 143 wsgi_intercept.add_wsgi_intercept('localhost', 8080, 144 lambda: CollectionApplication(underlying, nsf.NSFNorman)) 145 146 return ClientCollection(base_uri="http://localhost:8080/")

147

148 -class NSFClientLucene(ClientTestBase, NSFLucene):

149 pass

150

151 -class NSFClientSQL(ClientTestBase, NSFSQL):

152 pass

153

Source Code for Module grassyknoll.tests.test_fixtures