#!/usr/bin/env python  import iixr import os  # Remove old test files.  for filename in ("test", "testF", "testFI", "testI", "testP"):     try:         os.remove(filename)     except OSError:         pass  try:     for filename in os.listdir("test_index"):         os.remove(os.path.join("test_index", filename))     os.rmdir("test_index") except OSError:     pass  # Test basic data types.  numbers = [12345678, 0, 1, 127, 128, 255, 256]  f = open("test", "wb") w = iixr.FileWriter(f) for number in numbers:     w.write_number(number) w.close()  f = open("test", "rb") r = iixr.FileReader(f) for number in numbers:     n = r.read_number()     print number == n, number, n r.close()  # Test positions.  all_doc_positions = [     [         (123, [1, 3, 5, 15, 25]),         (124, [0, 100]),         (125, [11, 99, 199]),         (130, [77, 78, 80, 82, 89])     ],     [         (78, [9]),         (196, [10, 11]),         (197, [17, 21, 30])     ]     ]  f = open("testP", "wb") w = iixr.PositionWriter(f) for doc_positions in all_doc_positions:     for docnum, positions in doc_positions:         w.write_positions(docnum, positions)     w.reset() w.close()  f = open("testP", "rb") r = iixr.PositionReader(f) for doc_positions in all_doc_positions:     for docnum, positions in doc_positions:         d, p = r.read_positions()         print docnum == d, docnum, d         print positions == p, positions, p     r.reset() r.close()  # Test position index files.  indexed_positions = [     [         (1234, 0, 100),         (2345, 700, 100),         (3456, 1900, 50)     ],     [         (4567, 2800, 20)     ]     ]  offsets = [] f = open("testPI", "wb") w = iixr.PositionIndexWriter(f) for term_positions in indexed_positions:     offset = None     doc_frequency = 0     w.reset()     for docnum, pos_offset, count in term_positions:         io = w.write_positions(docnum, pos_offset, count)         if offset is None:             offset = io         doc_frequency += count     offsets.append((offset, doc_frequency)) w.close()  f = open("testPI", "rb") r = iixr.PositionIndexReader(f) offsets.reverse() indexed_positions.reverse() for (offset, doc_frequency), term_positions in zip(offsets, indexed_positions):     found_positions = r.read_term_positions(offset, doc_frequency)     for (docnum, pos_offset, count), (dn, po, c) in zip(term_positions, found_positions):         print docnum == dn, docnum, dn         print pos_offset == po, pos_offset, po         print count == c, count, c r.close()  # Test position dictionaries.  f = open("testP", "wb") w = iixr.PositionWriter(f) f2 = open("testPI", "wb") w2 = iixr.PositionIndexWriter(f2) wd = iixr.PositionDictionaryWriter(w, w2, 2) offsets = [] for doc_positions in all_doc_positions:     offset, frequency, doc_frequency = wd.write_term_positions(doc_positions)     offsets.append((offset, doc_frequency)) wd.close()  f = open("testP", "rb") r = iixr.PositionReader(f) f2 = open("testPI", "rb") r2 = iixr.PositionIndexReader(f2) rd = iixr.PositionDictionaryReader(r, r2) offsets.reverse() all_doc_positions.reverse() for (offset, doc_frequency), doc_positions in zip(offsets, all_doc_positions):     dp = list(rd.read_term_positions(offset, doc_frequency))     print doc_positions == dp, doc_positions, dp rd.close()  # Test fields.  doc_fields = [     (123, ["testing", "fields", "stored", "compressed"]),     (456, ["fields", "for a second", "document"]),     (789, ["field value"]),     (1234, []),     (2345, ["abc", "def"]),     (3456, ["apple", "banana", "cherry"]),     (4567, ["drue", "eple"])     ]  f = open("testF", "wb") w = iixr.FieldWriter(f) for docnum, fields in doc_fields:     w.write_fields(docnum, list(enumerate(fields))) w.close()  f = open("testF", "rb") r = iixr.FieldReader(f) for docnum, fields in doc_fields:     dn, df = r.read_fields()     print docnum == dn, docnum, dn     print list(enumerate(fields)) == df, list(enumerate(fields)), df r.close()  # Test field index files.  indexed_docs = [     (123, 100000987),     (456, 100004321),     (789, 100008765)     ]  f = open("testFI", "wb") w = iixr.FieldIndexWriter(f) for docnum, offset in indexed_docs:     w.write_document(docnum, offset) w.close()  f = open("testFI", "rb") r = iixr.FieldIndexReader(f) for docnum, offset in indexed_docs:     dn, o = r.read_document()     print docnum == dn, docnum, dn     print offset == o, offset, o r.close()  # Test field dictionaries.  f = open("testF", "wb") w = iixr.FieldWriter(f) f2 = open("testFI", "wb") w2 = iixr.FieldIndexWriter(f2) wd = iixr.FieldDictionaryWriter(w, w2, 3) for docnum, fields in doc_fields:     wd.write_fields(docnum, list(enumerate(fields))) wd.close()  f = open("testF", "rb") r = iixr.FieldReader(f) f2 = open("testFI", "rb") r2 = iixr.FieldIndexReader(f2) rd = iixr.FieldDictionaryReader(r, r2) doc_fields_reversed = doc_fields[:] doc_fields_reversed.reverse() for docnum, fields in doc_fields_reversed:     df = dict(rd.get_fields(docnum))     print dict(enumerate(fields)) == df, dict(enumerate(fields)), df for docnum in (13579, 246810):     df = rd.get_fields(docnum)     print df is None, df  # (Test sequential access.)  rd.rewind() for docnum, fields in doc_fields:     dn, df = rd.read_fields()     print docnum == dn, docnum, dn     print list(enumerate(fields)) == df, list(enumerate(fields)), df rd.close()  # Test terms.  terms = [     # term       offset      frequency  doc_frequency     ("aardvark",  100000123,  1,         1),     ("anteater",  100000456,  2,         1),     ("badger",    100000789, 13,         7),     ("bull",     1000001234, 59,        17),     ("bulldog",  1000002345, 99,        80),     ("cat",      1000003456, 89,        28)     ]  f = open("test", "wb") w = iixr.TermWriter(f) for term, offset, frequency, doc_frequency in terms:     w.write_term(term, offset, frequency, doc_frequency) w.close()  f = open("test", "rb") r = iixr.TermReader(f) for term, offset, frequency, doc_frequency in terms:     t, o, fr, df = r.read_term()     print term == t, term, t     print offset == o, offset, o     print frequency == fr, frequency, fr     print doc_frequency == df, doc_frequency, df r.close()  # Test terms in index files.  indexed_terms = [     # term       offset      frequency  doc_frequency   info_offset     ("aardvark",  100000123,  1,         1,             200000321),     ("anteater",  100000456,  2,         1,             200000654),     ("badger",    100000789, 13,         7,             200000987),     ("bull",     1000001234, 59,        17,             200004321),     ("bulldog",  1000002345, 99,        80,             200005432),     ("cat",      1000003456, 89,        28,             200006543)     ]  f = open("test", "wb") w = iixr.TermIndexWriter(f) for term, offset, frequency, doc_frequency, info_offset in indexed_terms:     w.write_term(term, offset, frequency, doc_frequency, info_offset) w.close()  f = open("test", "rb") r = iixr.TermIndexReader(f) for term, offset, frequency, doc_frequency, info_offset in indexed_terms:     t, o, fr, df, i = r.read_term()     print term == t, term, t     print offset == o, offset, o     print frequency == fr, frequency, fr     print doc_frequency == df, doc_frequency, df     print info_offset == i, info_offset, i r.close()  # Test dictionaries with only term data.  f = open("test", "wb") w = iixr.TermWriter(f) f2 = open("testI", "wb") w2 = iixr.TermIndexWriter(f2) f3 = open("testP", "wb") w3 = iixr.PositionWriter(f3) f4 = open("testPI", "wb") w4 = iixr.PositionIndexWriter(f4) wp = iixr.PositionDictionaryWriter(w3, w4, 2) wd = iixr.TermDictionaryWriter(w, w2, wp, 3) for term, offset, frequency, doc_frequency in terms:     wd._write_term(term, offset, frequency, doc_frequency) wd.close()  f = open("test", "rb") r = iixr.TermReader(f) f2 = open("testI", "rb") r2 = iixr.TermIndexReader(f2) f3 = open("testP", "rb") r3 = iixr.PositionReader(f3) f4 = open("testPI", "rb") r4 = iixr.PositionIndexReader(f4) rp = iixr.PositionDictionaryReader(r3, r4) rd = iixr.TermDictionaryReader(r, r2, rp) terms_reversed = terms[:] terms_reversed.reverse() for term, offset, frequency, doc_frequency in terms_reversed:     o, fr, df = rd._find_term(term)     print offset == o, offset, o     print frequency == fr, frequency, fr     print doc_frequency == df, doc_frequency, df for term in ("dog", "dingo"):     t = rd._find_term(term)     print t is None, t  # (Test term prefix searching.)  print rd.find_terms("a") == ["aardvark", "anteater"], rd.find_terms("a"), ["aardvark", "anteater"] print rd.find_terms("bu") == ["bull", "bulldog"], rd.find_terms("bu"), ["bull", "bulldog"] print rd.find_terms("c") == ["cat"], rd.find_terms("c"), ["cat"] print rd.find_terms("d") == [], rd.find_terms("d"), [] rd.close()  # Test dictionaries with term and position data.  terms_with_positions = [     ("aardvark",  [(1, [2, 45, 96]), (20, [13])]),     ("anteater",  [(1, [43, 44])]),     ("badger",    [(7, [2, 22, 196]), (19, [55, 1333]), (21, [0])]),     ("bull",      [(6, [128]), (16, [12]), (26, [1, 3, 5, 7, 9]), (36, [2, 4, 6, 8, 10])]),     ("bulldog",   [(43, [17, 19, 256, 512])]),     ("cat",       [(123, [12, 145, 196]), (1200, [113])])     ]  position_dict_tests = [     ("badger", 19, [55, 1333]),     ("badger", 20, None),     ("bull", 6, [128]),     ("bull", 26, [1, 3, 5, 7, 9]),     ("cat", 111, None),     ("cat", 123, [12, 145, 196]),     ("cat", 1234, None)     ]  f = open("test", "wb") w = iixr.TermWriter(f) f2 = open("testI", "wb") w2 = iixr.TermIndexWriter(f2) f3 = open("testP", "wb") w3 = iixr.PositionWriter(f3) f4 = open("testPI", "wb") w4 = iixr.PositionIndexWriter(f4) wp = iixr.PositionDictionaryWriter(w3, w4, 2) wd = iixr.TermDictionaryWriter(w, w2, wp, 3) for term, doc_positions in terms_with_positions:     wd.write_term_positions(term, doc_positions) wd.close()  f = open("test", "rb") r = iixr.TermReader(f) f2 = open("testI", "rb") r2 = iixr.TermIndexReader(f2) f3 = open("testP", "rb") r3 = iixr.PositionReader(f3) f4 = open("testPI", "rb") r4 = iixr.PositionIndexReader(f4) rp = iixr.PositionDictionaryReader(r3, r4) rd = iixr.TermDictionaryReader(r, r2, rp) terms_reversed = terms_with_positions[:] terms_reversed.reverse() for term, doc_positions in terms_reversed:     dp = list(rd.find_positions(term))     print doc_positions == dp, doc_positions, dp for term in ("aaa", "dog", "dingo"):     dp = rd.find_positions(term)     print dp is None, dp  # (Test iterators.)  for term, docnum, positions in position_dict_tests:     dp = rd.find_positions(term)     pos = dp.from_document(docnum)     print positions is None and pos is None or pos is not None and positions == list(pos), positions, pos  # (Test sequential access.)  rd.rewind() for term, doc_positions in terms_with_positions:     t, fr, df, dp = rd.read_term()     dp = list(dp)     print term == t, term, t     print doc_positions == dp, doc_positions, dp rd.close()  # Test high-level index operations (including merging).  docs = [     (1, "The cat sat on the mat"),     (2, "Every good boy deserves football"),     (13, "One good turn deserves another"),     (14, "Every man for himself"),     (25, "Red sky at night shepherd's delight"),     (36, "She sells sea shells on the sea shore")     ]  doc_tests = [     ("Every", 2, [(2, [0]), (14, [0])]),     ("good", 2, [(2, [1]), (13, [1])]),     ("deserves", 2, [(2, [3]), (13, [3])]),     ("sea", 2, [(36, [2, 6])])     ]  position_tests = [     ("Every", 14, [0]),     ("sea", 36, [2, 6]),     ("shells", 1, None),     ("shells", 37, None)     ]  index = iixr.Index("test_index") wi = index.get_writer(3, 2, 6) for docnum, text in docs:     doc = iixr.Document(docnum)     for position, term in enumerate(text.split()):         doc.add_position(term, position)     doc.add_field(123, text)     wi.add_document(doc) wi.close()  rd = index.get_reader() for term, frequency, doc_positions in doc_tests:     dp = list(rd.find_positions(term))     print doc_positions == dp, doc_positions, dp     fr = rd.get_frequency(term)     print frequency == fr, frequency, fr for docnum, text in docs:     df = dict(rd.get_fields(docnum))     print df[123] == text, text, df[123] for term, docnum, positions in position_tests:     dp = rd.find_positions(term)     pos = dp.from_document(docnum)     print positions is None and pos is None or pos is not None and positions == list(pos), positions, pos index.close()  # vim: tabstop=4 expandtab shiftwidth=4