#encoding: utf-8

from __future__ import with_statement
import copy
from datetime import datetime, timedelta

import pytest

from whoosh import analysis, fields, index, qparser, query, searching, scoring
from whoosh.codec.whoosh3 import W3Codec
from whoosh.compat import b, u, text_type
from whoosh.compat import xrange, permutations, izip_longest
from whoosh.filedb.filestore import RamStorage
from whoosh.util.testing import TempIndex


def make_index():
    s = fields.Schema(key=fields.ID(stored=True),
                      name=fields.TEXT,
                      value=fields.TEXT)
    st = RamStorage()
    ix = st.create_index(s)

    w = ix.writer()
    w.add_document(key=u("A"), name=u("Yellow brown"),
                   value=u("Blue red green render purple?"))
    w.add_document(key=u("B"), name=u("Alpha beta"),
                   value=u("Gamma delta epsilon omega."))
    w.add_document(key=u("C"), name=u("One two"),
                   value=u("Three rendered four five."))
    w.add_document(key=u("D"), name=u("Quick went"),
                   value=u("Every red town."))
    w.add_document(key=u("E"), name=u("Yellow uptown"),
                   value=u("Interest rendering outer photo!"))
    w.commit()

    return ix


def _get_keys(stored_fields):
    return sorted([d.get("key") for d in stored_fields])


def _docs(q, s):
    return _get_keys([s.stored_fields(docnum) for docnum
                           in q.docs(s)])


def _run_query(q, target):
    ix = make_index()
    with ix.searcher() as s:
        assert target == _docs(q, s)


def test_empty_index():
    schema = fields.Schema(key=fields.ID(stored=True), value=fields.TEXT)
    st = RamStorage()
    with pytest.raises(index.EmptyIndexError):
        st.open_index(schema=schema)


def test_docs_method():
    ix = make_index()
    with ix.searcher() as s:
        assert _get_keys(s.documents(name="yellow")) == ["A", "E"]
        assert _get_keys(s.documents(value="red")) == ["A", "D"]
        assert _get_keys(s.documents()) == ["A", "B", "C", "D", "E"]


def test_term():
    _run_query(query.Term("name", u("yellow")), [u("A"), u("E")])
    _run_query(query.Term("value", u("zeta")), [])
    _run_query(query.Term("value", u("red")), [u("A"), u("D")])


def test_require():
    _run_query(query.Require(query.Term("value", u("red")),
                             query.Term("name", u("yellow"))),
               [u("A")])


def test_and():
    _run_query(query.And([query.Term("value", u("red")),
                          query.Term("name", u("yellow"))]),
               [u("A")])
    # Missing
    _run_query(query.And([query.Term("value", u("ochre")),
                          query.Term("name", u("glonk"))]),
               [])


def test_or():
    _run_query(query.Or([query.Term("value", u("red")),
                         query.Term("name", u("yellow"))]),
               [u("A"), u("D"), u("E")])
    # Missing
    _run_query(query.Or([query.Term("value", u("ochre")),
                         query.Term("name", u("glonk"))]),
               [])
    _run_query(query.Or([]), [])


def test_ors():
    domain = u("alfa bravo charlie delta").split()
    s = fields.Schema(num=fields.STORED, text=fields.TEXT)
    st = RamStorage()
    ix = st.create_index(s)
    with ix.writer() as w:
        for i, ls in enumerate(permutations(domain)):
            w.add_document(num=i, text=" ".join(ls))

    with ix.searcher() as s:
        qs = [query.Term("text", word) for word in domain]
        for i in xrange(1, len(domain)):
            q = query.Or(qs[:i])
            r1 = [(hit.docnum, hit.score) for hit in s.search(q, limit=None)]

            q.binary_matcher = True
            r2 = [(hit.docnum, hit.score) for hit in s.search(q, limit=None)]

            for item1, item2 in izip_longest(r1, r2):
                assert item1[0] == item2[0]
                assert item1[1] == item2[1]


def test_not():
    _run_query(query.And([query.Or([query.Term("value", u("red")),
                                    query.Term("name", u("yellow"))]),
                          query.Not(query.Term("name", u("quick")))]),
               [u("A"), u("E")])


def test_topnot():
    _run_query(query.Not(query.Term("value", "red")), [u("B"), "C", "E"])
    _run_query(query.Not(query.Term("name", "yellow")), [u("B"), u("C"),
                                                         u("D")])


def test_andnot():
    _run_query(query.AndNot(query.Term("name", u("yellow")),
                            query.Term("value", u("purple"))),
               [u("E")])


def test_andnot2():
    schema = fields.Schema(a=fields.ID(stored=True))
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(a=u("bravo"))
    w.add_document(a=u("echo"))
    w.add_document(a=u("juliet"))
    w.commit()
    w = ix.writer()
    w.add_document(a=u("kilo"))
    w.add_document(a=u("foxtrot"))
    w.add_document(a=u("charlie"))
    w.commit(merge=False)
    w = ix.writer()
    w.delete_by_term("a", u("echo"))
    w.add_document(a=u("alfa"))
    w.add_document(a=u("india"))
    w.add_document(a=u("delta"))
    w.commit(merge=False)

    with ix.searcher() as s:
        q = query.TermRange("a", u("bravo"), u("k"))
        qr = [hit["a"] for hit in s.search(q)]
        assert " ".join(sorted(qr)) == "bravo charlie delta foxtrot india juliet"

        oq = query.Or([query.Term("a", "bravo"), query.Term("a", "delta")])
        oqr = [hit["a"] for hit in s.search(oq)]
        assert " ".join(sorted(oqr)) == "bravo delta"

        anq = query.AndNot(q, oq)

        m = anq.matcher(s)
        r = s.search(anq)
        assert list(anq.docs(s)) == sorted(hit.docnum for hit in r)
        assert " ".join(sorted(hit["a"] for hit in r)) == "charlie foxtrot india juliet"


def test_variations():
    _run_query(query.Variations("value", u("render")),
               [u("A"), u("C"), u("E")])


def test_wildcard():
    _run_query(query.Or([query.Wildcard('value', u('*red*')),
                         query.Wildcard('name', u('*yellow*'))]),
               [u("A"), u("C"), u("D"), u("E")])
    # Missing
    _run_query(query.Wildcard('value', 'glonk*'), [])


def test_not2():
    schema = fields.Schema(name=fields.ID(stored=True), value=fields.TEXT)
    storage = RamStorage()
    ix = storage.create_index(schema)
    writer = ix.writer()
    writer.add_document(name=u("a"), value=u("alfa bravo charlie delta echo"))
    writer.add_document(name=u("b"),
                        value=u("bravo charlie delta echo foxtrot"))
    writer.add_document(name=u("c"),
                        value=u("charlie delta echo foxtrot golf"))
    writer.add_document(name=u("d"), value=u("delta echo golf hotel india"))
    writer.add_document(name=u("e"), value=u("echo golf hotel india juliet"))
    writer.commit()

    with ix.searcher() as s:
        p = qparser.QueryParser("value", None)
        results = s.search(p.parse("echo NOT golf"))
        assert sorted([d["name"] for d in results]) == ["a", "b"]

        results = s.search(p.parse("echo NOT bravo"))
        assert sorted([d["name"] for d in results]) == ["c", "d", "e"]

    ix.delete_by_term("value", u("bravo"))

    with ix.searcher() as s:
        results = s.search(p.parse("echo NOT charlie"))
        assert sorted([d["name"] for d in results]) == ["d", "e"]

#    def test_or_minmatch():
#        schema = fields.Schema(k=fields.STORED, v=fields.TEXT)
#        st = RamStorage()
#        ix = st.create_index(schema)
#
#        w = ix.writer()
#        w.add_document(k=1, v=u("alfa bravo charlie delta echo"))
#        w.add_document(k=2, v=u("bravo charlie delta echo foxtrot"))
#        w.add_document(k=3, v=u("charlie delta echo foxtrot golf"))
#        w.add_document(k=4, v=u("delta echo foxtrot golf hotel"))
#        w.add_document(k=5, v=u("echo foxtrot golf hotel india"))
#        w.add_document(k=6, v=u("foxtrot golf hotel india juliet"))
#        w.commit()
#
#        s = ix.searcher()
#        q = Or([Term("v", "echo"), Term("v", "foxtrot")], minmatch=2)
#        r = s.search(q)
#        assert sorted(d["k"] for d in r), [2, 3, 4, 5])


def test_range():
    schema = fields.Schema(id=fields.ID(stored=True), content=fields.TEXT)
    st = RamStorage()
    ix = st.create_index(schema)

    w = ix.writer()
    w.add_document(id=u("A"), content=u("alfa bravo charlie delta echo"))
    w.add_document(id=u("B"), content=u("bravo charlie delta echo foxtrot"))
    w.add_document(id=u("C"), content=u("charlie delta echo foxtrot golf"))
    w.add_document(id=u("D"), content=u("delta echo foxtrot golf hotel"))
    w.add_document(id=u("E"), content=u("echo foxtrot golf hotel india"))
    w.commit()

    with ix.searcher() as s:
        qp = qparser.QueryParser("content", schema)

        q = qp.parse(u("charlie [delta TO foxtrot]"))
        assert q.__class__ == query.And
        assert q[0].__class__ == query.Term
        assert q[1].__class__ == query.TermRange
        assert q[1].start == "delta"
        assert q[1].end == "foxtrot"
        assert not q[1].startexcl
        assert not q[1].endexcl
        ids = sorted([d['id'] for d in s.search(q)])
        assert ids == [u('A'), u('B'), u('C')]

        q = qp.parse(u("foxtrot {echo TO hotel]"))
        assert q.__class__ == query.And
        assert q[0].__class__ == query.Term
        assert q[1].__class__ == query.TermRange
        assert q[1].start == "echo"
        assert q[1].end == "hotel"
        assert q[1].startexcl
        assert not q[1].endexcl
        ids = sorted([d['id'] for d in s.search(q)])
        assert ids == [u('B'), u('C'), u('D'), u('E')]

        q = qp.parse(u("{bravo TO delta}"))
        assert q.__class__ == query.TermRange
        assert q.start == "bravo"
        assert q.end == "delta"
        assert q.startexcl
        assert q.endexcl
        ids = sorted([d['id'] for d in s.search(q)])
        assert ids == [u('A'), u('B'), u('C')]

        # Shouldn't match anything
        q = qp.parse(u("[1 to 10]"))
        assert q.__class__ == query.TermRange
        assert len(s.search(q)) == 0


def test_range_clusiveness():
    schema = fields.Schema(id=fields.ID(stored=True))
    st = RamStorage()
    ix = st.create_index(schema)
    w = ix.writer()
    for letter in u("abcdefg"):
        w.add_document(id=letter)
    w.commit()

    with ix.searcher() as s:
        def check(startexcl, endexcl, string):
            q = query.TermRange("id", "b", "f", startexcl, endexcl)
            r = "".join(sorted(d['id'] for d in s.search(q)))
            assert r == string

        check(False, False, "bcdef")
        check(True, False, "cdef")
        check(True, True, "cde")
        check(False, True, "bcde")


def test_open_ranges():
    schema = fields.Schema(id=fields.ID(stored=True))
    st = RamStorage()
    ix = st.create_index(schema)
    w = ix.writer()
    for letter in u("abcdefg"):
        w.add_document(id=letter)
    w.commit()

    with ix.searcher() as s:
        qp = qparser.QueryParser("id", schema)

        def check(qstring, result):
            q = qp.parse(qstring)
            r = "".join(sorted([d['id'] for d in s.search(q)]))
            assert r == result

        check(u("[b TO]"), "bcdefg")
        check(u("[TO e]"), "abcde")
        check(u("[b TO d]"), "bcd")
        check(u("{b TO]"), "cdefg")
        check(u("[TO e}"), "abcd")
        check(u("{b TO d}"), "c")


def test_open_numeric_ranges():
    domain = range(0, 1000, 7)

    schema = fields.Schema(num=fields.NUMERIC(stored=True))
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    for i in domain:
        w.add_document(num=i)
    w.commit()

    qp = qparser.QueryParser("num", schema)
    with ix.searcher() as s:
        q = qp.parse("[100 to]")
        r = [hit["num"] for hit in s.search(q, limit=None)]
        assert r == [n for n in domain if n >= 100]

        q = qp.parse("[to 500]")
        r = [hit["num"] for hit in s.search(q, limit=None)]
        assert r == [n for n in domain if n <= 500]


def test_open_date_ranges():
    basedate = datetime(2011, 1, 24, 6, 25, 0, 0)
    domain = [basedate + timedelta(days=n) for n in xrange(-20, 20)]

    schema = fields.Schema(date=fields.DATETIME(stored=True))
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    for d in domain:
        w.add_document(date=d)
    w.commit()

    with ix.searcher() as s:
        # Without date parser
        qp = qparser.QueryParser("date", schema)
        q = qp.parse("[2011-01-10 to]")
        r = [hit["date"] for hit in s.search(q, limit=None)]
        assert len(r) > 0
        target = [d for d in domain if d >= datetime(2011, 1, 10, 6, 25)]
        assert r == target

        q = qp.parse("[to 2011-01-30]")
        r = [hit["date"] for hit in s.search(q, limit=None)]
        assert len(r) > 0
        target = [d for d in domain if d <= datetime(2011, 1, 30, 6, 25)]
        assert r == target

        # With date parser
        from whoosh.qparser.dateparse import DateParserPlugin
        qp.add_plugin(DateParserPlugin(basedate))

        q = qp.parse("[10 jan 2011 to]")
        r = [hit["date"] for hit in s.search(q, limit=None)]
        assert len(r) > 0
        target = [d for d in domain if d >= datetime(2011, 1, 10, 6, 25)]
        assert r == target

        q = qp.parse("[to 30 jan 2011]")
        r = [hit["date"] for hit in s.search(q, limit=None)]
        assert len(r) > 0
        target = [d for d in domain if d <= datetime(2011, 1, 30, 6, 25)]
        assert r == target


def test_negated_unlimited_ranges():
    # Whoosh should treat u("[to]") as if it was "*"
    schema = fields.Schema(id=fields.ID(stored=True), num=fields.NUMERIC,
                           date=fields.DATETIME)
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    from string import ascii_letters
    domain = text_type(ascii_letters)

    dt = datetime.now()
    for i, letter in enumerate(domain):
        w.add_document(id=letter, num=i, date=dt + timedelta(days=i))
    w.commit()

    with ix.searcher() as s:
        qp = qparser.QueryParser("id", schema)

        nq = qp.parse(u("NOT [to]"))
        assert nq.__class__ == query.Not
        q = nq.query
        assert q.__class__ == query.Every
        assert "".join(h["id"] for h in s.search(q, limit=None)) == domain
        assert not list(nq.docs(s))

        nq = qp.parse(u("NOT num:[to]"))
        assert nq.__class__ == query.Not
        q = nq.query
        assert q.__class__ == query.NumericRange
        assert q.start is None
        assert q.end is None
        assert "".join(h["id"] for h in s.search(q, limit=None)) == domain
        assert not list(nq.docs(s))

        nq = qp.parse(u("NOT date:[to]"))
        assert nq.__class__ == query.Not
        q = nq.query
        assert q.__class__ == query.Every
        assert "".join(h["id"] for h in s.search(q, limit=None)) == domain
        assert not list(nq.docs(s))


def test_keyword_or():
    schema = fields.Schema(a=fields.ID(stored=True), b=fields.KEYWORD)
    st = RamStorage()
    ix = st.create_index(schema)

    w = ix.writer()
    w.add_document(a=u("First"), b=u("ccc ddd"))
    w.add_document(a=u("Second"), b=u("aaa ddd"))
    w.add_document(a=u("Third"), b=u("ccc eee"))
    w.commit()

    qp = qparser.QueryParser("b", schema)
    with ix.searcher() as s:
        qr = qp.parse(u("b:ccc OR b:eee"))
        assert qr.__class__ == query.Or
        r = s.search(qr)
        assert len(r) == 2
        assert r[0]["a"] == "Third"
        assert r[1]["a"] == "First"


def test_merged():
    schema = fields.Schema(id=fields.ID(stored=True), content=fields.TEXT)
    with TempIndex(schema) as ix:
        with ix.writer() as w:
            w.add_document(id=u("alfa"), content=u("alfa"))
            w.add_document(id=u("bravo"), content=u("bravo"))

        with ix.searcher() as s:
            r = s.search(query.Term("content", u("bravo")))
            assert len(r) == 1
            assert r[0]["id"] == "bravo"

        with ix.writer() as w:
            w.add_document(id=u("charlie"), content=u("charlie"))
            w.optimize = True

        assert len(ix._segments()) == 1

        with ix.searcher() as s:
            r = s.search(query.Term("content", u("bravo")))
            assert len(r) == 1
            assert r[0]["id"] == "bravo"


def test_multireader():
    sc = fields.Schema(id=fields.ID(stored=True), content=fields.TEXT)
    st = RamStorage()
    ix = st.create_index(sc)
    w = ix.writer()
    w.add_document(id=u("alfa"), content=u("alfa"))
    w.add_document(id=u("bravo"), content=u("bravo"))
    w.add_document(id=u("charlie"), content=u("charlie"))
    w.add_document(id=u("delta"), content=u("delta"))
    w.add_document(id=u("echo"), content=u("echo"))
    w.add_document(id=u("foxtrot"), content=u("foxtrot"))
    w.add_document(id=u("golf"), content=u("golf"))
    w.add_document(id=u("hotel"), content=u("hotel"))
    w.add_document(id=u("india"), content=u("india"))
    w.commit()

    with ix.searcher() as s:
        r = s.search(query.Term("content", u("bravo")))
        assert len(r) == 1
        assert r[0]["id"] == "bravo"

    w = ix.writer()
    w.add_document(id=u("juliet"), content=u("juliet"))
    w.add_document(id=u("kilo"), content=u("kilo"))
    w.add_document(id=u("lima"), content=u("lima"))
    w.add_document(id=u("mike"), content=u("mike"))
    w.add_document(id=u("november"), content=u("november"))
    w.add_document(id=u("oscar"), content=u("oscar"))
    w.add_document(id=u("papa"), content=u("papa"))
    w.add_document(id=u("quebec"), content=u("quebec"))
    w.add_document(id=u("romeo"), content=u("romeo"))
    w.commit()
    assert len(ix._segments()) == 2

    #r = ix.reader()
    #assert r.__class__.__name__ == "MultiReader"
    #pr = r.postings("content", u("bravo"))

    with ix.searcher() as s:
        r = s.search(query.Term("content", u("bravo")))
        assert len(r) == 1
        assert r[0]["id"] == "bravo"


def test_posting_phrase():
    schema = fields.Schema(name=fields.ID(stored=True), value=fields.TEXT)
    storage = RamStorage()
    ix = storage.create_index(schema)
    writer = ix.writer()
    writer.add_document(name=u("A"),
                        value=u("Little Miss Muffet sat on a tuffet"))
    writer.add_document(name=u("B"), value=u("Miss Little Muffet tuffet"))
    writer.add_document(name=u("C"), value=u("Miss Little Muffet tuffet sat"))
    writer.add_document(name=u("D"),
                        value=u("Gibberish blonk falunk miss muffet sat " +
                                "tuffet garbonzo"))
    writer.add_document(name=u("E"), value=u("Blah blah blah pancakes"))
    writer.commit()

    with ix.searcher() as s:
        def names(results):
            return sorted([fields['name'] for fields in results])

        q = query.Phrase("value", [u("little"), u("miss"), u("muffet"),
                                   u("sat"), u("tuffet")])
        m = q.matcher(s)
        assert m.__class__.__name__ == "SpanNear2Matcher"

        r = s.search(q)
        assert names(r) == ["A"]
        assert len(r) == 1

        q = query.Phrase("value", [u("miss"), u("muffet"), u("sat"),
                                   u("tuffet")])
        assert names(s.search(q)) == ["A", "D"]

        q = query.Phrase("value", [u("falunk"), u("gibberish")])
        r = s.search(q)
        assert not names(r)
        assert len(r) == 0

        q = query.Phrase("value", [u("gibberish"), u("falunk")], slop=2)
        assert names(s.search(q)) == ["D"]

        q = query.Phrase("value", [u("blah")] * 4)
        assert not names(s.search(q))  # blah blah blah blah

        q = query.Phrase("value", [u("blah")] * 3)
        m = q.matcher(s)
        assert names(s.search(q)) == ["E"]


def test_phrase_score():
    schema = fields.Schema(name=fields.ID(stored=True), value=fields.TEXT)
    storage = RamStorage()
    ix = storage.create_index(schema)
    writer = ix.writer()
    writer.add_document(name=u("A"),
                        value=u("Little Miss Muffet sat on a tuffet"))
    writer.add_document(name=u("D"),
                        value=u("Gibberish blonk falunk miss muffet sat " +
                                "tuffet garbonzo"))
    writer.add_document(name=u("E"), value=u("Blah blah blah pancakes"))
    writer.add_document(name=u("F"),
                        value=u("Little miss muffet little miss muffet"))
    writer.commit()

    with ix.searcher() as s:
        q = query.Phrase("value", [u("little"), u("miss"), u("muffet")])
        m = q.matcher(s)
        assert m.id() == 0
        score1 = m.weight()
        assert score1 > 0
        m.next()
        assert m.id() == 3
        assert m.weight() > score1


def test_stop_phrase():
    schema = fields.Schema(title=fields.TEXT(stored=True))
    storage = RamStorage()
    ix = storage.create_index(schema)
    writer = ix.writer()
    writer.add_document(title=u("Richard of York"))
    writer.add_document(title=u("Lily the Pink"))
    writer.commit()

    with ix.searcher() as s:
        qp = qparser.QueryParser("title", schema)
        q = qp.parse(u("richard of york"))
        assert q.__unicode__() == "(title:richard AND title:york)"
        assert len(s.search(q)) == 1
        #q = qp.parse(u("lily the pink"))
        #assert len(s.search(q)), 1)
        assert len(s.find("title", u("lily the pink"))) == 1


def test_phrase_order():
    tfield = fields.TEXT(stored=True, analyzer=analysis.SimpleAnalyzer())
    schema = fields.Schema(text=tfield)
    storage = RamStorage()
    ix = storage.create_index(schema)

    writer = ix.writer()
    for ls in permutations(["ape", "bay", "can", "day"], 4):
        writer.add_document(text=u(" ").join(ls))
    writer.commit()

    with ix.searcher() as s:
        def result(q):
            r = s.search(q, limit=None, sortedby=None)
            return sorted([d['text'] for d in r])

        q = query.Phrase("text", ["bay", "can", "day"])
        assert result(q) == [u('ape bay can day'), u('bay can day ape')]


def test_phrase_sameword():
    schema = fields.Schema(id=fields.STORED, text=fields.TEXT)
    storage = RamStorage()
    ix = storage.create_index(schema)

    writer = ix.writer()
    writer.add_document(id=1, text=u("The film Linda Linda Linda is good"))
    writer.add_document(id=2, text=u("The model Linda Evangelista is pretty"))
    writer.commit()

    with ix.searcher() as s:
        r = s.search(query.Phrase("text", ["linda", "linda", "linda"]),
                     limit=None)
        assert len(r) == 1
        assert r[0]["id"] == 1


def test_phrase_multi():
    schema = fields.Schema(id=fields.STORED, text=fields.TEXT)
    ix = RamStorage().create_index(schema)

    domain = u("alfa bravo charlie delta echo").split()
    w = None
    for i, ls in enumerate(permutations(domain)):
        if w is None:
            w = ix.writer()
        w.add_document(id=i, text=u(" ").join(ls))
        if not i % 30:
            w.commit()
            w = None
    if w is not None:
        w.commit()

    with ix.searcher() as s:
        q = query.Phrase("text", ["alfa", "bravo"])
        _ = s.search(q)


def test_missing_field_scoring():
    schema = fields.Schema(name=fields.TEXT(stored=True),
                           hobbies=fields.TEXT(stored=True))
    with TempIndex(schema) as ix:
        with ix.writer() as w:
            w.add_document(name=u('Frank'), hobbies=u('baseball, basketball'))

        with ix.reader() as r:
            assert r.field_length("hobbies") == 2
            assert r.field_length("name") == 1

        with ix.writer() as w:
            w.add_document(name=u('Jonny'))

        with ix.searcher() as s:
            assert s.field_length("hobbies") == 2
            assert s.field_length("name") == 2

            parser = qparser.MultifieldParser(['name', 'hobbies'], schema)
            q = parser.parse(u("baseball"))
            result = s.search(q)
            assert len(result) == 1


def test_search_fieldname_underscores():
    s = fields.Schema(my_name=fields.ID(stored=True), my_value=fields.TEXT)
    st = RamStorage()
    ix = st.create_index(s)

    w = ix.writer()
    w.add_document(my_name=u("Green"), my_value=u("It's not easy being green"))
    w.add_document(my_name=u("Red"),
                   my_value=u("Hopping mad like a playground ball"))
    w.commit()

    qp = qparser.QueryParser("my_value", schema=s)
    with ix.searcher() as s:
        r = s.search(qp.parse(u("my_name:Green")))
        assert r[0]['my_name'] == "Green"


def test_short_prefix():
    s = fields.Schema(name=fields.ID, value=fields.TEXT)
    qp = qparser.QueryParser("value", schema=s)
    q = qp.parse(u("s*"))
    assert q.__class__.__name__ == "Prefix"
    assert q.text == "s"


def test_weighting():
    from whoosh.scoring import Weighting, BaseScorer

    schema = fields.Schema(id=fields.ID(stored=True),
                           n_comments=fields.STORED)
    st = RamStorage()
    ix = st.create_index(schema)

    w = ix.writer()
    w.add_document(id=u("1"), n_comments=5)
    w.add_document(id=u("2"), n_comments=12)
    w.add_document(id=u("3"), n_comments=2)
    w.add_document(id=u("4"), n_comments=7)
    w.commit()

    # Fake Weighting implementation
    class CommentWeighting(Weighting):
        def scorer(self, searcher, fieldname, text, qf=1):
            return self.CommentScorer(searcher.stored_fields)

        class CommentScorer(BaseScorer):
            def __init__(self, stored_fields):
                self.stored_fields = stored_fields

            def score(self, matcher):
                sf = self.stored_fields(matcher.id())
                ncomments = sf.get("n_comments", 0)
                return ncomments

    with ix.searcher(weighting=CommentWeighting()) as s:
        q = query.TermRange("id", u("1"), u("4"), constantscore=False)

        r = s.search(q)
        ids = [fs["id"] for fs in r]
        assert ids == ["2", "4", "1", "3"]


def test_dismax():
    schema = fields.Schema(id=fields.STORED,
                           f1=fields.TEXT, f2=fields.TEXT, f3=fields.TEXT)
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id=1, f1=u("alfa bravo charlie delta"),
                   f2=u("alfa alfa alfa"),
                   f3=u("alfa echo foxtrot hotel india"))
    w.commit()

    with ix.searcher(weighting=scoring.Frequency()) as s:
        assert list(s.documents(f1="alfa")) == [{"id": 1}]
        assert list(s.documents(f2="alfa")) == [{"id": 1}]
        assert list(s.documents(f3="alfa")) == [{"id": 1}]

        qs = [query.Term("f1", "alfa"), query.Term("f2", "alfa"),
              query.Term("f3", "alfa")]
        dm = query.DisjunctionMax(qs)
        r = s.search(dm)
        assert r.score(0) == 3.0


def test_deleted_wildcard():
    schema = fields.Schema(id=fields.ID(stored=True))
    st = RamStorage()
    ix = st.create_index(schema)

    w = ix.writer()
    w.add_document(id=u("alfa"))
    w.add_document(id=u("bravo"))
    w.add_document(id=u("charlie"))
    w.add_document(id=u("delta"))
    w.add_document(id=u("echo"))
    w.add_document(id=u("foxtrot"))
    w.commit()

    w = ix.writer()
    w.delete_by_term("id", "bravo")
    w.delete_by_term("id", "delta")
    w.delete_by_term("id", "echo")
    w.commit()

    with ix.searcher() as s:
        r = s.search(query.Every("id"))
        assert sorted([d['id'] for d in r]) == ["alfa", "charlie", "foxtrot"]


def test_missing_wildcard():
    schema = fields.Schema(id=fields.ID(stored=True), f1=fields.TEXT,
                           f2=fields.TEXT)
    st = RamStorage()
    ix = st.create_index(schema)

    w = ix.writer()
    w.add_document(id=u("1"), f1=u("alfa"), f2=u("apple"))
    w.add_document(id=u("2"), f1=u("bravo"))
    w.add_document(id=u("3"), f1=u("charlie"), f2=u("candy"))
    w.add_document(id=u("4"), f2=u("donut"))
    w.add_document(id=u("5"))
    w.commit()

    with ix.searcher() as s:
        r = s.search(query.Every("id"))
        assert sorted([d['id'] for d in r]) == ["1", "2", "3", "4", "5"]

        r = s.search(query.Every("f1"))
        assert sorted([d['id'] for d in r]) == ["1", "2", "3"]

        r = s.search(query.Every("f2"))
        assert sorted([d['id'] for d in r]) == ["1", "3", "4"]


def test_finalweighting():
    from whoosh.scoring import Frequency

    schema = fields.Schema(id=fields.ID(stored=True),
                           summary=fields.TEXT,
                           n_comments=fields.STORED)
    st = RamStorage()
    ix = st.create_index(schema)

    w = ix.writer()
    w.add_document(id=u("1"), summary=u("alfa bravo"), n_comments=5)
    w.add_document(id=u("2"), summary=u("alfa"), n_comments=12)
    w.add_document(id=u("3"), summary=u("bravo"), n_comments=2)
    w.add_document(id=u("4"), summary=u("bravo bravo"), n_comments=7)
    w.commit()

    class CommentWeighting(Frequency):
        use_final = True

        def final(self, searcher, docnum, score):
            ncomments = searcher.stored_fields(docnum).get("n_comments", 0)
            return ncomments

    with ix.searcher(weighting=CommentWeighting()) as s:
        q = qparser.QueryParser("summary", None).parse("alfa OR bravo")
        r = s.search(q)
        ids = [fs["id"] for fs in r]
        assert ["2", "4", "1", "3"] == ids


def test_outofdate():
    schema = fields.Schema(id=fields.ID(stored=True))
    st = RamStorage()
    ix = st.create_index(schema)

    w = ix.writer()
    w.add_document(id=u("1"))
    w.add_document(id=u("2"))
    w.commit()

    s = ix.searcher()
    assert s.up_to_date()

    w = ix.writer()
    w.add_document(id=u("3"))
    w.add_document(id=u("4"))

    assert s.up_to_date()
    w.commit()
    assert not s.up_to_date()

    s = s.refresh()
    assert s.up_to_date()
    s.close()


def test_find_missing():
    schema = fields.Schema(id=fields.ID, text=fields.KEYWORD(stored=True))
    ix = RamStorage().create_index(schema)

    w = ix.writer()
    w.add_document(id=u("1"), text=u("alfa"))
    w.add_document(id=u("2"), text=u("bravo"))
    w.add_document(text=u("charlie"))
    w.add_document(id=u("4"), text=u("delta"))
    w.add_document(text=u("echo"))
    w.add_document(id=u("6"), text=u("foxtrot"))
    w.add_document(text=u("golf"))
    w.commit()

    with ix.searcher() as s:
        qp = qparser.QueryParser("text", schema)
        q = qp.parse(u("NOT id:*"))
        r = s.search(q, limit=None)
        assert list(h["text"] for h in r) == ["charlie", "echo", "golf"]


def test_ngram_phrase():
    f = fields.NGRAM(minsize=2, maxsize=2, phrase=True)
    schema = fields.Schema(text=f, path=fields.ID(stored=True))
    ix = RamStorage().create_index(schema)
    writer = ix.writer()
    writer.add_document(text=u('\u9AD8\u6821\u307E\u3067\u306F\u6771\u4EAC'
                               '\u3067\u3001\u5927\u5B66\u304B\u3089\u306F'
                               '\u4EAC\u5927\u3067\u3059\u3002'),
                        path=u('sample'))
    writer.commit()

    with ix.searcher() as s:
        p = qparser.QueryParser("text", schema)

        q = p.parse(u('\u6771\u4EAC\u5927\u5B66'))
        assert len(s.search(q)) == 1

        q = p.parse(u('"\u6771\u4EAC\u5927\u5B66"'))
        assert len(s.search(q)) == 0

        q = p.parse(u('"\u306F\u6771\u4EAC\u3067"'))
        assert len(s.search(q)) == 1


def test_ordered():
    domain = u("alfa bravo charlie delta echo foxtrot").split(" ")

    schema = fields.Schema(f=fields.TEXT(stored=True))
    ix = RamStorage().create_index(schema)
    writer = ix.writer()
    for ls in permutations(domain):
        writer.add_document(f=u(" ").join(ls))
    writer.commit()

    with ix.searcher() as s:
        q = query.Ordered([query.Term("f", u("alfa")),
                           query.Term("f", u("charlie")),
                           query.Term("f", u("echo"))])
        r = s.search(q)
        for hit in r:
            ls = hit["f"].split()
            assert "alfa" in ls
            assert "charlie" in ls
            assert "echo" in ls
            a = ls.index("alfa")
            c = ls.index("charlie")
            e = ls.index("echo")
            assert a < c and c < e, repr(ls)


def test_otherwise():
    schema = fields.Schema(id=fields.STORED, f=fields.TEXT)
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id=1, f=u("alfa one two"))
    w.add_document(id=2, f=u("alfa three four"))
    w.add_document(id=3, f=u("bravo four five"))
    w.add_document(id=4, f=u("bravo six seven"))
    w.commit()

    with ix.searcher() as s:
        q = query.Otherwise(query.Term("f", u("alfa")),
                            query.Term("f", u("six")))
        assert [d["id"] for d in s.search(q)] == [1, 2]

        q = query.Otherwise(query.Term("f", u("tango")),
                            query.Term("f", u("four")))
        assert [d["id"] for d in s.search(q)] == [2, 3]

        q = query.Otherwise(query.Term("f", u("tango")),
                            query.Term("f", u("nine")))
        assert [d["id"] for d in s.search(q)] == []


def test_fuzzyterm():
    schema = fields.Schema(id=fields.STORED, f=fields.TEXT)
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id=1, f=u("alfa bravo charlie delta"))
    w.add_document(id=2, f=u("bravo charlie delta echo"))
    w.add_document(id=3, f=u("charlie delta echo foxtrot"))
    w.add_document(id=4, f=u("delta echo foxtrot golf"))
    w.commit()

    with ix.searcher() as s:
        q = query.FuzzyTerm("f", "brave")
        assert [d["id"] for d in s.search(q)] == [1, 2]


def test_fuzzyterm2():
    schema = fields.Schema(id=fields.STORED, f=fields.TEXT(spelling=True))
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id=1, f=u("alfa bravo charlie delta"))
    w.add_document(id=2, f=u("bravo charlie delta echo"))
    w.add_document(id=3, f=u("charlie delta echo foxtrot"))
    w.add_document(id=4, f=u("delta echo foxtrot golf"))
    w.commit()

    with ix.searcher() as s:
        assert list(s.reader().terms_within("f", u("brave"), 1)) == ["bravo"]
        q = query.FuzzyTerm("f", "brave")
        assert [d["id"] for d in s.search(q)] == [1, 2]


def test_multireader_not():
    schema = fields.Schema(id=fields.STORED, f=fields.TEXT)

    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id=0, f=u("alfa bravo chralie"))
    w.add_document(id=1, f=u("bravo chralie delta"))
    w.add_document(id=2, f=u("charlie delta echo"))
    w.add_document(id=3, f=u("delta echo foxtrot"))
    w.add_document(id=4, f=u("echo foxtrot golf"))
    w.commit()

    with ix.searcher() as s:
        q = query.And([query.Term("f", "delta"),
                       query.Not(query.Term("f", "delta"))])
        r = s.search(q)
        assert len(r) == 0

    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id=5, f=u("alfa bravo chralie"))
    w.add_document(id=6, f=u("bravo chralie delta"))
    w.commit(merge=False)
    w = ix.writer()
    w.add_document(id=7, f=u("charlie delta echo"))
    w.add_document(id=8, f=u("delta echo foxtrot"))
    w.commit(merge=False)
    w = ix.writer()
    w.add_document(id=9, f=u("echo foxtrot golf"))
    w.add_document(id=10, f=u("foxtrot golf delta"))
    w.commit(merge=False)
    assert len(ix._segments()) > 1

    with ix.searcher() as s:
        q = query.And([query.Term("f", "delta"),
                       query.Not(query.Term("f", "delta"))])
        r = s.search(q)
        assert len(r) == 0


def test_boost_phrase():
    schema = fields.Schema(title=fields.TEXT(field_boost=5.0, stored=True),
                           text=fields.TEXT)
    ix = RamStorage().create_index(schema)
    domain = u("alfa bravo charlie delta").split()
    w = ix.writer()
    for ls in permutations(domain):
        t = u(" ").join(ls)
        w.add_document(title=t, text=t)
    w.commit()

    q = query.Or([query.Term("title", u("alfa")),
                  query.Term("title", u("bravo")),
                  query.Phrase("text", [u("bravo"), u("charlie"), u("delta")])
                  ])

    def boost_phrases(q):
        if isinstance(q, query.Phrase):
            q.boost *= 1000.0
            return q
        else:
            return q.apply(boost_phrases)
    q = boost_phrases(q)

    with ix.searcher() as s:
        r = s.search(q, limit=None)
        for hit in r:
            if "bravo charlie delta" in hit["title"]:
                assert hit.score > 100.0


def test_filter():
    schema = fields.Schema(id=fields.STORED, path=fields.ID, text=fields.TEXT)
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id=1, path=u("/a/1"), text=u("alfa bravo charlie"))
    w.add_document(id=2, path=u("/b/1"), text=u("bravo charlie delta"))
    w.add_document(id=3, path=u("/c/1"), text=u("charlie delta echo"))
    w.commit(merge=False)
    w = ix.writer()
    w.add_document(id=4, path=u("/a/2"), text=u("delta echo alfa"))
    w.add_document(id=5, path=u("/b/2"), text=u("echo alfa bravo"))
    w.add_document(id=6, path=u("/c/2"), text=u("alfa bravo charlie"))
    w.commit(merge=False)
    w = ix.writer()
    w.add_document(id=7, path=u("/a/3"), text=u("bravo charlie delta"))
    w.add_document(id=8, path=u("/b/3"), text=u("charlie delta echo"))
    w.add_document(id=9, path=u("/c/3"), text=u("delta echo alfa"))
    w.commit(merge=False)

    with ix.searcher() as s:
        fq = query.Or([query.Prefix("path", "/a"),
                       query.Prefix("path", "/b")])
        r = s.search(query.Term("text", "alfa"), filter=fq)
        assert [d["id"] for d in r] == [1, 4, 5]

        r = s.search(query.Term("text", "bravo"), filter=fq)
        assert [d["id"] for d in r] == [1, 2, 5, 7, ]


def test_fieldboost():
    schema = fields.Schema(id=fields.STORED, a=fields.TEXT, b=fields.TEXT)
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id=0, a=u("alfa bravo charlie"), b=u("echo foxtrot india"))
    w.add_document(id=1, a=u("delta bravo charlie"), b=u("alfa alfa alfa"))
    w.add_document(id=2, a=u("alfa alfa alfa"), b=u("echo foxtrot india"))
    w.add_document(id=3, a=u("alfa sierra romeo"), b=u("alfa tango echo"))
    w.add_document(id=4, a=u("bravo charlie delta"), b=u("alfa foxtrot india"))
    w.add_document(id=5, a=u("alfa alfa echo"), b=u("tango tango tango"))
    w.add_document(id=6, a=u("alfa bravo echo"), b=u("alfa alfa tango"))
    w.commit()

    def field_booster(fieldname, factor=2.0):
        "Returns a function which will boost the given field in a query tree"
        def booster_fn(obj):
            if obj.is_leaf() and obj.field() == fieldname:
                obj = copy.deepcopy(obj)
                obj.boost *= factor
                return obj
            else:
                return obj
        return booster_fn

    with ix.searcher() as s:
        q = query.Or([query.Term("a", u("alfa")),
                      query.Term("b", u("alfa"))])
        q = q.accept(field_booster("a", 100.0))
        assert text_type(q) == text_type("(a:alfa^100.0 OR b:alfa)")
        r = s.search(q)
        assert [hit["id"] for hit in r] == [2, 5, 6, 3, 0, 1, 4]


def test_andmaybe_quality():
    schema = fields.Schema(id=fields.STORED, title=fields.TEXT(stored=True),
                           year=fields.NUMERIC)
    ix = RamStorage().create_index(schema)

    domain = [(u('Alpha Bravo Charlie Delta'), 2000),
              (u('Echo Bravo Foxtrot'), 2000), (u('Bravo Golf Hotel'), 2002),
              (u('Bravo India'), 2002), (u('Juliet Kilo Bravo'), 2004),
              (u('Lima Bravo Mike'), 2004)]
    w = ix.writer()
    for title, year in domain:
        w.add_document(title=title, year=year)
    w.commit()

    with ix.searcher() as s:
        qp = qparser.QueryParser("title", ix.schema)
        q = qp.parse(u("title:bravo ANDMAYBE year:2004"))

        titles = [hit["title"] for hit in s.search(q, limit=None)[:2]]
        assert "Juliet Kilo Bravo" in titles

        titles = [hit["title"] for hit in s.search(q, limit=2)]
        assert "Juliet Kilo Bravo" in titles


def test_collect_limit():
    schema = fields.Schema(id=fields.STORED, text=fields.TEXT)
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id="a", text=u("alfa bravo charlie delta echo"))
    w.add_document(id="b", text=u("bravo charlie delta echo foxtrot"))
    w.add_document(id="c", text=u("charlie delta echo foxtrot golf"))
    w.add_document(id="d", text=u("delta echo foxtrot golf hotel"))
    w.add_document(id="e", text=u("echo foxtrot golf hotel india"))
    w.commit()

    with ix.searcher() as s:
        r = s.search(query.Term("text", u("golf")), limit=10)
        assert len(r) == 3
        count = 0
        for _ in r:
            count += 1
        assert count == 3

    w = ix.writer()
    w.add_document(id="f", text=u("foxtrot golf hotel india juliet"))
    w.add_document(id="g", text=u("golf hotel india juliet kilo"))
    w.add_document(id="h", text=u("hotel india juliet kilo lima"))
    w.add_document(id="i", text=u("india juliet kilo lima mike"))
    w.add_document(id="j", text=u("juliet kilo lima mike november"))
    w.commit(merge=False)

    with ix.searcher() as s:
        r = s.search(query.Term("text", u("golf")), limit=20)
        assert len(r) == 5
        count = 0
        for _ in r:
            count += 1
        assert count == 5


def test_scorer():
    schema = fields.Schema(key=fields.TEXT(stored=True))
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(key=u("alfa alfa alfa"))
    w.add_document(key=u("alfa alfa alfa alfa"))
    w.add_document(key=u("alfa alfa"))
    w.commit()
    w = ix.writer()
    w.add_document(key=u("alfa alfa alfa alfa alfa alfa"))
    w.add_document(key=u("alfa"))
    w.add_document(key=u("alfa alfa alfa alfa alfa"))
    w.commit(merge=False)

#    dw = scoring.DebugModel()
#    s = ix.searcher(weighting=dw)
#    r = s.search(query.Term("key", "alfa"))
#    log = dw.log
#    assert log, [('key', 'alfa', 0, 3.0, 3),
#                       ('key', 'alfa', 1, 4.0, 4),
#                       ('key', 'alfa', 2, 2.0, 2),
#                       ('key', 'alfa', 0, 6.0, 6),
#                       ('key', 'alfa', 1, 1.0, 1),
#                       ('key', 'alfa', 2, 5.0, 5)])


def test_pos_scorer():
    ana = analysis.SimpleAnalyzer()
    schema = fields.Schema(id=fields.STORED, key=fields.TEXT(analyzer=ana))
    ix = RamStorage().create_index(schema)
    w = ix.writer()
    w.add_document(id=0, key=u("0 0 1 0 0 0"))
    w.add_document(id=1, key=u("0 0 0 1 0 0"))
    w.add_document(id=2, key=u("0 1 0 0 0 0"))
    w.commit()
    w = ix.writer()
    w.add_document(id=3, key=u("0 0 0 0 0 1"))
    w.add_document(id=4, key=u("1 0 0 0 0 0"))
    w.add_document(id=5, key=u("0 0 0 0 1 0"))
    w.commit(merge=False)

    def pos_score_fn(searcher, fieldname, text, matcher):
        poses = matcher.value_as("positions")
        return 1.0 / (poses[0] + 1)
    pos_weighting = scoring.FunctionWeighting(pos_score_fn)

    s = ix.searcher(weighting=pos_weighting)
    r = s.search(query.Term("key", "1"))
    assert [hit["id"] for hit in r] == [4, 2, 0, 1, 5, 3]


# def test_too_many_prefix_positions():
#     schema = fields.Schema(id=fields.STORED, text=fields.TEXT)
#     ix = RamStorage().create_index(schema)
#     with ix.writer() as w:
#         for i in xrange(200):
#             text = u("a%s" % i)
#             w.add_document(id=i, text=text)
#
#     q = query.Prefix("text", u("a"))
#     q.TOO_MANY_CLAUSES = 100
#
#     with ix.searcher() as s:
#         m = q.matcher(s)
#         assert m.supports("positions")
#         items = list(m.items_as("positions"))
#         assert [(i, [0]) for i in xrange(200)] == items


def test_collapse():
    from whoosh import collectors

    # id, text, size, tag
    domain = [("a", "blah blah blah", 5, "x"),
              ("b", "blah", 3, "y"),
              ("c", "blah blah blah blah", 2, "z"),
              ("d", "blah blah", 4, "x"),
              ("e", "bloop", 1, "-"),
              ("f", "blah blah blah blah blah", 6, "x"),
              ("g", "blah", 8, "w"),
              ("h", "blah blah", 7, "=")]

    schema = fields.Schema(id=fields.STORED, text=fields.TEXT,
                           size=fields.NUMERIC,
                           tag=fields.KEYWORD(sortable=True))
    ix = RamStorage().create_index(schema)
    with ix.writer(codec=W3Codec()) as w:
        for id, text, size, tag in domain:
            w.add_document(id=u(id), text=u(text), size=size, tag=u(tag))

    with ix.searcher() as s:
        q = query.Term("text", "blah")
        r = s.search(q, limit=None)
        assert " ".join(hit["id"] for hit in r) == "f c a d h b g"

        col = s.collector(limit=3)
        col = collectors.CollapseCollector(col, "tag")
        s.search_with_collector(q, col)
        r = col.results()
        assert " ".join(hit["id"] for hit in r) == "f c h"

        col = s.collector(limit=None)
        col = collectors.CollapseCollector(col, "tag")
        s.search_with_collector(q, col)
        r = col.results()
        assert " ".join(hit["id"] for hit in r) == "f c h b g"

        r = s.search(query.Every(), sortedby="size")
        assert " ".join(hit["id"] for hit in r) == "e c b d a f h g"

        col = s.collector(sortedby="size")
        col = collectors.CollapseCollector(col, "tag")
        s.search_with_collector(query.Every(), col)
        r = col.results()
        assert " ".join(hit["id"] for hit in r) == "e c b d h g"


def test_collapse_nocolumn():
    from whoosh import collectors

    # id, text, size, tag
    domain = [("a", "blah blah blah", 5, "x"),
              ("b", "blah", 3, "y"),
              ("c", "blah blah blah blah", 2, "z"),
              ("d", "blah blah", 4, "x"),
              ("e", "bloop", 1, "-"),
              ("f", "blah blah blah blah blah", 6, "x"),
              ("g", "blah", 8, "w"),
              ("h", "blah blah", 7, "=")]

    schema = fields.Schema(id=fields.STORED, text=fields.TEXT,
                           size=fields.NUMERIC,
                           tag=fields.KEYWORD)
    ix = RamStorage().create_index(schema)
    with ix.writer() as w:
        for id, text, size, tag in domain:
            w.add_document(id=u(id), text=u(text), size=size, tag=u(tag))

    with ix.searcher() as s:
        q = query.Term("text", "blah")
        r = s.search(q, limit=None)
        assert " ".join(hit["id"] for hit in r) == "f c a d h b g"

        col = s.collector(limit=3)
        col = collectors.CollapseCollector(col, "tag")
        s.search_with_collector(q, col)
        r = col.results()
        assert " ".join(hit["id"] for hit in r) == "f c h"

        col = s.collector(limit=None)
        col = collectors.CollapseCollector(col, "tag")
        s.search_with_collector(q, col)
        r = col.results()
        assert " ".join(hit["id"] for hit in r) == "f c h b g"

        r = s.search(query.Every(), sortedby="size")
        assert " ".join(hit["id"] for hit in r) == "e c b d a f h g"

        col = s.collector(sortedby="size")
        col = collectors.CollapseCollector(col, "tag")
        s.search_with_collector(query.Every(), col)
        r = col.results()
        assert " ".join(hit["id"] for hit in r) == "e c b d h g"


def test_collapse_length():
    domain = u("alfa apple agnostic aplomb arc "
               "bravo big braid beer "
               "charlie crouch car "
               "delta dog "
               "echo "
               "foxtrot fold flip "
               "golf gym goop"
               ).split()

    schema = fields.Schema(key=fields.ID(sortable=True),
                           word=fields.ID(stored=True))
    ix = RamStorage().create_index(schema)
    with ix.writer(codec=W3Codec()) as w:
        for word in domain:
            w.add_document(key=word[0], word=word)

    with ix.searcher() as s:
        q = query.Every()

        def check(r):
            words = " ".join(hit["word"] for hit in r)
            assert words == "alfa bravo charlie delta echo foxtrot golf"
            assert r.scored_length() == 7
            assert len(r) == 7

        r = s.search(q, collapse="key", collapse_limit=1, limit=None)
        check(r)

        r = s.search(q, collapse="key", collapse_limit=1, limit=50)
        check(r)

        r = s.search(q, collapse="key", collapse_limit=1, limit=10)
        check(r)


def test_collapse_length_nocolumn():
    domain = u("alfa apple agnostic aplomb arc "
               "bravo big braid beer "
               "charlie crouch car "
               "delta dog "
               "echo "
               "foxtrot fold flip "
               "golf gym goop"
               ).split()

    schema = fields.Schema(key=fields.ID(),
                           word=fields.ID(stored=True))
    ix = RamStorage().create_index(schema)
    with ix.writer() as w:
        for word in domain:
            w.add_document(key=word[0], word=word)

    with ix.searcher() as s:
        q = query.Every()

        def check(r):
            words = " ".join(hit["word"] for hit in r)
            assert words == "alfa bravo charlie delta echo foxtrot golf"
            assert r.scored_length() == 7
            assert len(r) == 7

        r = s.search(q, collapse="key", collapse_limit=1, limit=None)
        check(r)

        r = s.search(q, collapse="key", collapse_limit=1, limit=50)
        check(r)

        r = s.search(q, collapse="key", collapse_limit=1, limit=10)
        check(r)


def test_collapse_order():
    from whoosh import sorting

    schema = fields.Schema(id=fields.STORED,
                           price=fields.NUMERIC(sortable=True),
                           rating=fields.NUMERIC(sortable=True),
                           tag=fields.ID(sortable=True))
    ix = RamStorage().create_index(schema)
    with ix.writer(codec=W3Codec()) as w:
        w.add_document(id="a", price=10, rating=1, tag=u("x"))
        w.add_document(id="b", price=80, rating=3, tag=u("y"))
        w.add_document(id="c", price=60, rating=1, tag=u("z"))
        w.add_document(id="d", price=30, rating=2)
        w.add_document(id="e", price=50, rating=3, tag=u("x"))
        w.add_document(id="f", price=20, rating=1, tag=u("y"))
        w.add_document(id="g", price=50, rating=2, tag=u("z"))
        w.add_document(id="h", price=90, rating=5)
        w.add_document(id="i", price=50, rating=5, tag=u("x"))
        w.add_document(id="j", price=40, rating=1, tag=u("y"))
        w.add_document(id="k", price=50, rating=4, tag=u("z"))
        w.add_document(id="l", price=70, rating=2)

    with ix.searcher() as s:
        def check(kwargs, target):
            r = s.search(query.Every(), limit=None, **kwargs)
            assert " ".join(hit["id"] for hit in r) == target

        price = sorting.FieldFacet("price", reverse=True)
        rating = sorting.FieldFacet("rating", reverse=True)
        tag = sorting.FieldFacet("tag")

        check(dict(sortedby=price), "h b l c e g i k j d f a")
        check(dict(sortedby=price, collapse=tag), "h b l c e d")
        check(dict(sortedby=price, collapse=tag, collapse_order=rating),
              "h b l i k d")


def test_collapse_order_nocolumn():
    from whoosh import sorting

    schema = fields.Schema(id=fields.STORED,
                           price=fields.NUMERIC(),
                           rating=fields.NUMERIC(),
                           tag=fields.ID())
    ix = RamStorage().create_index(schema)
    with ix.writer() as w:
        w.add_document(id="a", price=10, rating=1, tag=u("x"))
        w.add_document(id="b", price=80, rating=3, tag=u("y"))
        w.add_document(id="c", price=60, rating=1, tag=u("z"))
        w.add_document(id="d", price=30, rating=2)
        w.add_document(id="e", price=50, rating=3, tag=u("x"))
        w.add_document(id="f", price=20, rating=1, tag=u("y"))
        w.add_document(id="g", price=50, rating=2, tag=u("z"))
        w.add_document(id="h", price=90, rating=5)
        w.add_document(id="i", price=50, rating=5, tag=u("x"))
        w.add_document(id="j", price=40, rating=1, tag=u("y"))
        w.add_document(id="k", price=50, rating=4, tag=u("z"))
        w.add_document(id="l", price=70, rating=2)

    with ix.searcher() as s:
        def check(kwargs, target):
            r = s.search(query.Every(), limit=None, **kwargs)
            assert " ".join(hit["id"] for hit in r) == target

        price = sorting.FieldFacet("price", reverse=True)
        rating = sorting.FieldFacet("rating", reverse=True)
        tag = sorting.FieldFacet("tag")

        check(dict(sortedby=price), "h b l c e g i k j d f a")
        check(dict(sortedby=price, collapse=tag), "h b l c e d")
        check(dict(sortedby=price, collapse=tag, collapse_order=rating),
              "h b l i k d")


def test_coord():
    from whoosh.matching import CoordMatcher

    schema = fields.Schema(id=fields.STORED, hits=fields.STORED,
                           tags=fields.KEYWORD)
    ix = RamStorage().create_index(schema)
    with ix.writer() as w:
        w.add_document(id=0, hits=0, tags=u("blah blah blah blah"))
        w.add_document(id=1, hits=0, tags=u("echo echo blah blah"))
        w.add_document(id=2, hits=1, tags=u("bravo charlie delta echo"))
        w.add_document(id=3, hits=2, tags=u("charlie delta echo foxtrot"))
        w.add_document(id=4, hits=3, tags=u("delta echo foxtrot golf"))
        w.add_document(id=5, hits=3, tags=u("echo foxtrot golf hotel"))
        w.add_document(id=6, hits=2, tags=u("foxtrot golf hotel india"))
        w.add_document(id=7, hits=1, tags=u("golf hotel india juliet"))
        w.add_document(id=8, hits=0, tags=u("foxtrot foxtrot foo foo"))
        w.add_document(id=9, hits=0, tags=u("foo foo foo foo"))

    og = qparser.OrGroup.factory(0.99)
    qp = qparser.QueryParser("tags", schema, group=og)
    q = qp.parse("golf foxtrot echo")
    assert q.__class__ == query.Or
    assert q.scale == 0.99

    with ix.searcher() as s:
        m = q.matcher(s)
        assert type(m) == CoordMatcher

        r = s.search(q, optimize=False)
        assert [hit["id"] for hit in r] == [4, 5, 3, 6, 1, 8, 2, 7]


def test_keyword_search():
    schema = fields.Schema(tags=fields.KEYWORD)
    ix = RamStorage().create_index(schema)
    with ix.writer() as w:
        w.add_document(tags=u("keyword1 keyword2 keyword3 keyword4 keyword5"))

    with ix.searcher() as s:
        r = s.search_page(query.Term("tags", "keyword3"), 1)
        assert r


def test_groupedby_with_terms():
    schema = fields.Schema(content=fields.TEXT, organism=fields.ID)
    ix = RamStorage().create_index(schema)

    with ix.writer() as w:
        w.add_document(organism=u("mus"), content=u("IPFSTD1 IPFSTD_kdwq134 Kaminski-all Study00:00:00"))
        w.add_document(organism=u("mus"), content=u("IPFSTD1 IPFSTD_kdwq134 Kaminski-all Study"))
        w.add_document(organism=u("hs"), content=u("This is the first document we've added!"))

    with ix.searcher() as s:
        q = qparser.QueryParser("content", schema=ix.schema).parse(u("IPFSTD1"))
        r = s.search(q, groupedby=["organism"], terms=True)
        assert len(r) == 2
        assert r.groups("organism") == {"mus": [1, 0]}
        assert r.has_matched_terms()
        assert r.matched_terms() == set([('content', b('ipfstd1'))])


def test_score_length():
    schema = fields.Schema(a=fields.TEXT, b=fields.TEXT)
    ix = RamStorage().create_index(schema)
    with ix.writer() as w:
        w.add_document(a=u("alfa bravo charlie"))
        w.add_document(b=u("delta echo foxtrot"))
        w.add_document(a=u("golf hotel india"))

    with ix.writer() as w:
        w.merge = False
        w.add_document(b=u("juliet kilo lima"))
        # In the second segment, there is an "a" field here, but in the
        # corresponding document in the first segment, the field doesn't exist,
        # so if the scorer is getting segment offsets wrong, scoring this
        # document will error
        w.add_document(a=u("mike november oskar"))
        w.add_document(b=u("papa quebec romeo"))

    with ix.searcher() as s:
        assert not s.is_atomic()
        p = s.postings("a", "mike")
        while p.is_active():
            docnum = p.id()
            score = p.score()
            p.next()


def test_terms_with_filter():
    schema = fields.Schema(text=fields.TEXT)
    ix = RamStorage().create_index(schema)
    with ix.writer() as w:
        w.add_document(text=u("alfa bravo charlie delta"))
        w.add_document(text=u("bravo charlie delta echo"))
        w.add_document(text=u("charlie delta echo foxtrot"))
        w.add_document(text=u("delta echo foxtrot golf"))
        w.add_document(text=u("echo foxtrot golf hotel"))
        w.add_document(text=u("foxtrot golf hotel alfa"))
        w.add_document(text=u("golf hotel alfa bravo"))
        w.add_document(text=u("hotel alfa bravo charlie"))

    with ix.searcher() as s:
        workingset = set([1, 2, 3])
        q = query.Term("text", u("foxtrot"))
        r = s.search_page(q, pagenum=1, pagelen=5, terms=True,
                          filter=workingset)

        assert r.scored_length() == 2
        assert [hit.docnum for hit in r] == [2, 3]


def test_terms_to_bytes():
    schema = fields.Schema(a=fields.TEXT, b=fields.NUMERIC, id=fields.STORED)
    ix = RamStorage().create_index(schema)
    with ix.writer() as w:
        w.add_document(id=0, a=u("alfa bravo"), b=100)
        w.add_document(id=1, a=u("bravo charlie"), b=200)
        w.add_document(id=2, a=u("charlie delta"), b=100)
        w.add_document(id=3, a=u("delta echo"), b=200)

    with ix.searcher() as s:
        t1 = query.Term("b", 200)
        t2 = query.Term("a", "bravo")
        q = query.And([t1, t2])
        r = s.search(q)
        assert [hit["id"] for hit in r] == [1]


def test_issue_334():
    schema = fields.Schema(
        kind=fields.ID(stored=True),
        name=fields.ID(stored=True),
        returns=fields.ID(stored=True),
    )
    ix = RamStorage().create_index(schema)

    with ix.writer() as w:

        with w.group():
            w.add_document(kind=u('class'), name=u('Index'))
            w.add_document(kind=u('method'), name=u('add document'),
                           returns=u('void'))
            w.add_document(kind=u('method'), name=u('add reader'),
                           returns=u('void'))
            w.add_document(kind=u('method'), name=u('close'),
                           returns=u('void'))
        with w.group():
            w.add_document(kind=u('class'), name=u('Accumulator'))
            w.add_document(kind=u('method'), name=u('add'),
                           returns=u('void'))
            w.add_document(kind=u('method'), name=u('get result'),
                           returns=u('number'))
        with w.group():
            w.add_document(kind=u('class'), name=u('Calculator'))
            w.add_document(kind=u('method'), name=u('add'),
                           returns=u('number'))
            w.add_document(kind=u('method'), name=u('add all'),
                           returns=u('number'))
            w.add_document(kind=u('method'), name=u('add some'),
                           returns=u('number'))
            w.add_document(kind=u('method'), name=u('multiply'),
                           returns=u('number'))
            w.add_document(kind=u('method'), name=u('close'),
                           returns=u('void'))
        with w.group():
            w.add_document(kind=u('class'), name=u('Deleter'))
            w.add_document(kind=u('method'), name=u('add'),
                           returns=u('void'))
            w.add_document(kind=u('method'), name=u('delete'),
                           returns=u('void'))

    with ix.searcher() as s:
        pq = query.Term('kind', 'class')
        cq = query.Term('name', 'Calculator')

        q = query.NestedChildren(pq, cq) & query.Term('returns', 'void')
        r = s.search(q)
        assert len(r) == 1
        assert r[0]["name"] == u("close")


def test_find_decimals():
    from decimal import Decimal

    schema = fields.Schema(name=fields.KEYWORD(stored=True),
                           num=fields.NUMERIC(Decimal, decimal_places=5))
    ix = RamStorage().create_index(schema)

    with ix.writer() as w:
        w.add_document(name=u("alfa"), num=Decimal("1.5"))
        w.add_document(name=u("bravo"), num=Decimal("2.1"))
        w.add_document(name=u("charlie"), num=Decimal("5.3"))
        w.add_document(name=u("delta"), num=Decimal(3))
        w.add_document(name=u("echo"), num=Decimal("3.00001"))
        w.add_document(name=u("foxtrot"), num=Decimal("3"))

    qp = qparser.QueryParser("name", ix.schema)
    q = qp.parse("num:3.0")
    assert isinstance(q, query.Term)

    with ix.searcher() as s:
        r = s.search(q)
        names = " ".join(sorted(hit["name"] for hit in r))
        assert names == "delta foxtrot"


def test_limit_scores():
    domain = u"alfa bravo charlie delta echo foxtrot golf".split()

    schema = fields.Schema(desc=fields.TEXT, parent=fields.KEYWORD(stored=True))
    with TempIndex(schema) as ix:
        with ix.writer() as w:
            count = 0
            for words in permutations(domain, 4):
                count += 1
                w.add_document(desc=u" ".join(words), parent=text_type(count))

        with ix.searcher() as s:
            q = query.And([
                query.Term("desc", u"delta", boost=30.0),
                query.Term("parent", u"545")
            ])
            r = s.search(q, limit=500)
            assert r.scored_length() == 1
            limited_score = r[0].score

            r = s.search(q, limit=None)
            assert r.scored_length() == 1
            unlimited_score = r[0].score

            assert limited_score == unlimited_score


def test_function_weighting():
    def pos_score_fn(searcher, fieldname, text, matcher):
        spans = matcher.spans()
        return 1.0 / (spans[0].start + 1)

    pos_weighting = scoring.FunctionWeighting(pos_score_fn)

    schema = fields.Schema(id=fields.STORED, text=fields.TEXT)

    with TempIndex(schema) as ix:
        with ix.writer() as w:
            w.add_document(id="a", text=u"aa bb")
            w.add_document(id="b", text=u"bb aa bb")
            w.add_document(id="c", text=u"bb bb aa bb")
            w.add_document(id="d", text=u"bb bb bb aa bb")
            w.add_document(id="e", text=u"bb bb bb bb aa bb")
            w.add_document(id="f", text=u"bb bb bb bb bb aa bb")

        with ix.writer() as w:
            w.add_document(id="g", text=u"aa bb")
            w.add_document(id="h", text=u"bb aa bb")
            w.add_document(id="i", text=u"bb bb aa bb")
            w.add_document(id="j", text=u"bb bb bb aa bb")
            w.add_document(id="k", text=u"bb bb bb bb aa bb")
            w.add_document(id="l", text=u"bb bb bb bb bb aa bb")

        with ix.writer() as w:
            w.add_document(id="m", text=u"aa bb")
            w.add_document(id="n", text=u"bb aa bb")
            w.add_document(id="o", text=u"bb bb aa bb")
            w.add_document(id="p", text=u"bb bb bb aa bb")
            w.add_document(id="q", text=u"bb bb bb bb aa bb")
            w.add_document(id="r", text=u"bb bb bb bb bb aa bb")

        with ix.writer() as w:
            w.add_document(id="s", text=u"aa bb")
            w.add_document(id="t", text=u"bb aa bb")
            w.add_document(id="u", text=u"bb bb aa bb")
            w.add_document(id="v", text=u"bb bb bb aa bb")
            w.add_document(id="w", text=u"bb bb bb bb aa bb")
            w.add_document(id="x", text=u"bb bb bb bb bb aa bb")

        with ix.searcher(weighting=pos_weighting) as s:
            q = query.Term("text", "aa")
            m = q.matcher(s, s.context())
            assert not m.supports_block_quality()

            r = s.search(q, limit=5)
            ids = "".join(([hit["id"] for hit in r]))
            assert ids == "agmsb"

            q = query.Or([query.Term("text", "aa"), query.Term("text", "bb")],
                         scale=2.0)
            m = q.matcher(s, s.context())
            assert not m.supports_block_quality()