Source Code for Module oort.test.test_rdfview

  1  from nose.tools import assert_equals 
  2  from rdflib import (ConjunctiveGraph as Graph, URIRef, Literal, BNode, 
  3          Namespace, RDF) 
  4  from oort.rdfview import (RdfQuery, one, each, one_where_self_is, 
  5          each_where_self_is, collection, localized, i18n_dict, each_localized, 
  6          localized_xml, Sorter, Filter, run_queries, THIS_QUERY, selector) 
  7  from oort.util import queries 
  8   
  9   
 10  from oort.test.helper import siblingpath 
 11   
 12 -def _create_test_graph(filename='testdata.n3'): 
 13      testfilepath = siblingpath(__file__, filename) 
 14      graph = Graph() 
 15      graph.load(open(testfilepath), format='n3') 
 16      return graph 
 17  # TODO: in a setup (but once for entire test module?) 
 18  testgraph = _create_test_graph() 
 19   
 20   
 21   
 22 -def assert_all_equals(results, expected): 
 23      assert set(unicode(v) for v in results) == set(expected) 
 24   
 25 -def assert_contains(graph, query): 
 26      for spo in query.to_graph(): 
 27          assert spo in graph 
 28   
 29   
 30  #--------------------------------------- 
 31  # TODO: test some with rdflib.RDFS to make sure module namespaces work. 
 32   
 33   
 34  T = Namespace("http://example.org/oort/test#") 
 35   
 36  itemX = URIRef('tag:oort.to,2006:test:item:x') 
 37  en, sv = 'en', 'sv' 
 38   
 39   
 40 -class Part(RdfQuery): 
 41      name = one(T) 
 42   
 43 -class Item(Part): 
 44      title = localized(T) 
 45      relations = each(T.relation) >> THIS_QUERY 
 46      unaryRelation = one(T) >> Part 
 47      partlist = collection(T) >> Part 
 48      multiple_partlist = collection(multiple=True) >> Part 
 49      xmlData = localized_xml(T) 
 50      titleLang = i18n_dict(T['title']) 
 51      labels = each_localized(T.label) 
 52   
 53   
 54 -class TestItem: 
 55   
 56 -    def setup(self): 
 57          self.item = Item(testgraph, en, itemX) 
 58   
 59 -    def test_one(self): 
 60          assert self.item.name == u'Item X' 
 61   
 62 -    def test_one_relation(self): 
 63          assert self.item.unaryRelation.name == u'One Relation' 
 64   
 65 -    def test_each(self): 
 66          relItemNames = [item.name for item in self.item.relations] 
 67          assert set(relItemNames) == set(["Related 1", "Related 2"]) 
 68   
 69 -    def test_collection(self): 
 70          collItems = [item.name for item in self.item.partlist] 
 71          assert set(collItems) == set(["Part 1", "Part 2"]) 
 72   
 73 -    def test_collection_multiple(self): 
 74          collItems = [item.name for item in self.item.multiple_partlist] 
 75          assert set(collItems) == set(["Part 1", "Part 2"]) 
 76   
 77 -    def test_i18n_dict(self): 
 78          titleLangs = self.item.titleLang 
 79          assert set(titleLangs.keys()) == set(['en', 'sv']) 
 80          assert_all_equals(titleLangs.values(), [u'Example Item', u'Exempelsak']) 
 81   
 82   
 83 -def test_localized(): 
 84      for l, v in [(en, u'Example Item'), (sv, u'Exempelsak')]: 
 85          yield assert_equals, Item(testgraph, l, itemX).title, Literal(v, l) 
 86   
 87 -def test_each_localized(): 
 88      assert_all_equals(Item(testgraph, en, itemX).labels, [u'En', u'eN']) 
 89      assert_all_equals(Item(testgraph, sv, itemX).labels, [u'Sv', u'sV']) 
 90   
 91   
 92 -class RichItem(Item): 
 93      xmlData = localized_xml(T) 
 94   
 95 -def test_localized_xml(): 
 96      for l, v in [(en, u'XML Data'), (sv, u'XML-data')]: 
 97          treeData = list(RichItem(testgraph, l, itemX).xmlData) 
 98          ok = False 
 99          for stream in treeData: 
100              for item in stream: 
101                  if v in item: ok = True 
102          assert ok 
103   
104   
105 -class Creation(RdfQuery): 
106      createdBy = one_where_self_is(T.creation) >> Part 
107   
108 -def test_one_where_self_is(): 
109      creation = Creation(testgraph, en, itemX) 
110      assert creation.createdBy.name == "Creator" 
111      assert_contains(testgraph, creation) 
112   
113   
114 -class Owner(RdfQuery): 
115      owns = each_where_self_is(T.owner) >> Part 
116   
117 -def test_each_where_self_is(): 
118      owner = Owner(testgraph, en, itemX) 
119      owned = [part.name for part in owner.owns] 
120      assert set(owned) == set(["Part A", "Part B"]) 
121      assert_contains(testgraph, owner) 
122   
123   
124 -def test_to_json(): 
125      json = Item(testgraph, en, itemX).to_json() 
126      d = eval(json) 
127      # TODO: inspect! 
128   
129   
130 -def test_to_graph(): 
131      itemGraph = Item(testgraph, en, itemX).to_graph() 
132      # TODO: inspect! 
133   
134   
135 -class CustomItem(Item): 
136      @selector 
137 -    def randomRelation(self, testgraph, lang, subject): 
138          from random import randint 
139          assert isinstance(self, CustomItem) 
140          itemUris = [uri for uri in testgraph.subjects(RDF.type, T.Item) 
141                          if uri != subject] 
142          return Part(testgraph, lang, itemUris[randint(0, len(itemUris)-1)]) 
143   
144 -def test_selector_decorator(): 
145      assert CustomItem(testgraph, en, itemX).randomRelation.name.startswith("Related ") 
146   
147   
148 -class FilteredItem(RdfQuery): 
149      sortedKeywords = each(T.keyword) | Sorter() 
150      sortedKeywordsReversed = each(T.keyword) | Sorter(reverse=True) 
151   
152      sortedRelations_attr = each(T.relation) >> Item | \ 
153              Sorter('name', reverse=True) 
154      sortedRelations_func = each(T.relation) >> Item |\ 
155              Sorter(lambda r: getattr(r, 'name'), reverse=True) 
156   
157      someKeywords = each(T.keyword) | Filter(lambda v: v in ('q', 'r', 'y')) 
158   
159      name_upper = one(T.name) | (lambda v: unicode(v).upper()) 
160   
161 -def test_filter(): 
162      item = FilteredItem(testgraph, en, itemX) 
163      yield assert_equals, item.name_upper, "ITEM X" 
164   
165      def assert_sorted(item): 
166          keywords = "e q r t w y".split() 
167          assert item.sortedKeywords == keywords 
168          keywords.reverse() 
169          assert item.sortedKeywordsReversed == keywords 
170      yield assert_sorted, item 
171   
172      def assert_sortedrels(rels): 
173          assert ['Related 2', 'Related 1'] == [it.name for it in rels] 
174      yield assert_sortedrels, item.sortedRelations_attr 
175      yield assert_sortedrels, item.sortedRelations_func 
176   
177      def assert_some(item): 
178          assert set(item.someKeywords) == set(['q', 'r', 'y']) 
179      yield assert_some, item 
180   
181   
182 -def test_selector_filtered_by(): 
183   
184      class ItemWithSelectorAndFilter(Item): 
185          @selector.filtered_by(Sorter('name'), lambda l: l[0]) 
186          def firstRelation(self, graph, lang, subject): 
187              return self.relations 
188   
189      item = ItemWithSelectorAndFilter(testgraph, en, itemX) 
190      assert item.firstRelation.name == 'Related 1' 
191   
192   
193 -class ImplicitItem(RdfQuery): 
194      _rdfbase_ = T 
195      name = one() 
196   
197 -def test_implicit_item(): 
198      item = ImplicitItem(testgraph, en, itemX) 
199      assert item.name == "Item X" 
200      assert ImplicitItem.RDF_TYPE == T.ImplicitItem 
201   
202 -class TypedImplicitItem(RdfQuery): 
203      _rdfbase_ = T 
204      RDF_TYPE = T.OtherType 
205      name = one() 
206      partlist = collection() >> Part 
207   
208 -def test_typed_implicit_item(): 
209      item = TypedImplicitItem(testgraph, en, itemX) 
210      collItems = [part.name for part in item.partlist] 
211      assert set(collItems) == set(["Part 1", "Part 2"]) 
212      assert item.name == "Item X" 
213      assert TypedImplicitItem._rdfbase_ == T 
214      assert TypedImplicitItem.RDF_TYPE == T.OtherType 
215   
216   
217 -def test_from_dict(): 
218      data = dict( 
219          name = "Item X", 
220          title = "Example Item", 
221          titleLang = { 'en': "Example Item", 'sv': "Exempelsak" }, 
222          labels = [ "En", "eN" ], 
223          relations = [ 
224                  {'name': "Related 1"}, 
225                  {'name': "Related 2"}, 
226              ], 
227          xmlData = "<div/>", 
228      ) 
229      item = Item.from_dict(data, "en", URIRef("tag:oort.to,2006/test/item/x_fromdict")) 
230      # TODO: verify graph! 
231      item.to_graph() 
232      #print item.to_rdf(); raise Exception 
233   
234   
235 -def test_bound_with(): 
236      bound = Item.bound_with(itemX) 
237      yield assert_equals, bound.query, Item 
238      yield assert_equals, bound.__name__, 'Item' 
239      yield assert_equals, bound(testgraph, en, BNode()).name,  u'Item X' 
240      bound = Item.bound_with(itemX, sv) 
241      yield assert_equals, bound(testgraph, None, BNode()).title, Literal(u'Exempelsak', sv) 
242   
243   
244 -def test_find_by(): 
245      found = list(Item.find_by(testgraph, en, name=Literal(u'Item X'))) 
246      yield assert_equals, found[0].uri, itemX 
247      for l, v in (en, u'Example Item'), (sv, u'Exempelsak'): 
248          found = Item.find_by(testgraph, l, title=Literal(v, l)) 
249          yield assert_equals, list(found)[0].uri, itemX 
250   
251   
252  #--------------------------------------- 
253   
254 -class MarkedItem(RdfQuery): 
255      title = localized(T) 
256      relations = each(T.relation) >> 'MarkedItem' 
257      unaryRelation = one(T) >> 'MarkedPart' 
258      annotated = one(T.unaryRelation) >> 'oort.util.queries:Annotated' 
259   
260 -class MarkedPart(RdfQuery): 
261      name = one(T) 
262      related = one_where_self_is(T.unaryRelation) >> 'MarkedItem' 
263   
264 -class TestSubQueryMarker: 
265   
266 -    def setup(self): 
267          self.item = MarkedItem(testgraph, en, itemX) 
268   
269 -    def test_self_as_named(self): 
270          for rel in self.item.relations: 
271              assert type(rel) is MarkedItem 
272   
273 -    def test_named_following(self): 
274          assert type(self.item.unaryRelation) is MarkedPart 
275   
276 -    def test_cyclic(self): 
277          assert type(self.item.unaryRelation.related) is MarkedItem 
278   
279 -    def test_module_ref(self): 
280          assert type(self.item.annotated) is queries.Annotated 
281   
282   
283  #--------------------------------------- 
284   
285   
286  # FIXME: test properly, with lots of reused query instances from a use with nest(l)ed references! 
287 -def test_exec_cache(): 
288      queries = [Item, CustomItem] 
289      for item, query in zip(run_queries(queries, testgraph, en, itemX), queries): 
290          assert isinstance(item, query) 
291          test = TestItem() 
292          test.item = item 
293          yield test.test_one , 
294          yield test.test_one_relation , 
295          yield test.test_each , 
296          yield test.test_collection , 
297          yield test.test_collection_multiple , 
298          yield test.test_i18n_dict , 
299