1. Matthew Frazier
  2. redismap

Source

redismap / tests / collections / native.py

# -*- coding: utf-8 -*-
"""
Tests for `redismap.collections.native`.
"""
from attest import Tests, raises, assert_hook
from datetime import date
from redismap import (Key, Set, SetView, List, ListView, SortedSet,
                      SortedSetView, Hash, HashView, Integer, Bytes, Date)
from .._environment import client

def sort(i):
    l = list(i)
    l.sort()
    return l


nativetests = Tests()

@nativetests.test
def set_view():
    k = Key("~set", client)
    k.delete()
    s = SetView(k, Integer)
    # length
    assert len(s) == 0
    k.sadd("1")
    k.sadd("2")
    k.sadd("3")
    assert len(s) == 3
    # itering and copying
    assert sort(s) == [1, 2, 3]
    assert s.copy() == set([1, 2, 3])
    # randmember
    assert s.randmember() in (1, 2, 3)
    k.delete()


@nativetests.test
def set_collection():
    s = Set(Key("~set", client), Integer)
    s.clear()
    assert not client.exists("~set")
    assert len(s) == 0
    # adding and removing items
    assert s.add(1)
    assert s.add(2)
    assert sort(client.smembers("~set")) == ["1", "2"]
    s.add(4)
    s.add(5)
    assert not s.add(1)
    assert sort(client.smembers("~set")) == ["1", "2", "4", "5"]
    assert len(s) == 4
    assert 2 in s
    assert 3 not in s
    # removing items
    assert s.discard(5)
    assert 5 not in s
    assert not s.discard(6)
    assert 6 not in s
    s.remove(4)
    assert 4 not in s
    with raises(KeyError):
        s.remove(3)
    # pop
    item = s.pop()
    assert item in [1, 2]
    assert item not in s
    s.pop()
    assert len(s) == 0
    assert s.pop() is None
    assert s.randmember() is None
    s.clear()


@nativetests.test
def list_view():
    k = Key("~list", client)
    k.delete()
    l = ListView(k, Bytes)
    # length
    assert len(l) == 0
    pipe = k.client.pipeline()
    for char in "hijklmnopqrs":
        pipe.rpush(k, char)
    pipe.execute()
    assert len(l) == 12
    assert l.copy() == list("hijklmnopqrs")
    assert tuple(l) == tuple("hijklmnopqrs")
    # slicing and ranging
    assert l[4:8] == ["l", "m", "n", "o"]
    assert l.range(4, 8) == ["l", "m", "n", "o", "p"]
    assert l[:4] == ["h", "i", "j", "k"]
    assert l.range(0, 4) == ["h", "i", "j", "k", "l"]
    


@nativetests.test
def list_collection():
    l = List(Key("~list", client), Bytes)
    l.clear()
    assert not client.exists("~list")
    assert len(l) == 0
    # adding items
    l.append("n")
    l.prepend("m")
    l.append("o")
    l.prepend("l")
    assert client.lrange("~list", 0, -1) == ["l", "m", "n", "o"]
    assert list(l) == ["l", "m", "n", "o"]
    # extending and preextending
    l.extend("pqrs")
    l.preextend("hijk")
    assert l.copy() == list("hijklmnopqrs")
    assert len(l) == 12
    # pop and shift
    assert l.pop() == "s"
    assert l.pop() == "r"
    assert len(l) == 10
    assert l.shift() == "h"
    assert l.shift() == "i"
    assert len(l) == 8
    # trim
    l.trim(2, 5)
    assert list(l) == ["l", "m", "n", "o"]
    l.clear()


@nativetests.test
def zset_view():
    k = Key("~zset", client)
    k.delete()
    z = SortedSetView(k, Bytes)
    assert len(z) == 0
    k.zadd("uno", 0.5)
    k.zadd("dos", 2.0)
    k.zadd("tres", 3.0)
    k.zadd("cuatro", 5.0)
    assert len(z) == 4
    # getting scores
    assert z["uno"] == 0.5
    assert z.score("dos") == 2.0
    assert z.score("tres") == 3.0
    assert z.score("cuatro") == 5.0
    assert z.score("seis") is None
    with raises(KeyError):
        z["siete"]
    assert "cuatro" in z
    assert "cinco" not in z
    # rank
    assert z.rank("uno") == 0
    assert z.revrank("dos") == 2
    assert z.rank("tres") == 2
    assert z.revrank("cuatro") == 0
    k.delete()


@nativetests.test
def zset_collection():
    z = SortedSet(Key("~zset", client), Bytes)
    z.clear()
    assert not client.exists("~zset")
    assert len(z) == 0
    # adding items
    z.add("uno", 0.5)
    z.add("dos", 2.0)
    z["tres"] = 2.0
    z["cuatro"] = 5.0
    # incr
    z["uno"] = 1.0
    assert z["uno"] == 1.0
    assert z.incr("tres") == 3.0
    assert z.incr("cuatro", -1.0) == 4.0
    # removing
    assert not z.remove("seis")
    assert z.remove("cuatro")
    assert "cuatro" not in z
    z.clear()


@nativetests.test
def zscore_converters():
    z = SortedSet(Key("~beatles", client), Bytes, Date)
    z.clear()
    # add our data
    john = date(1940, 10, 9)
    paul = date(1942, 6, 18)
    george = date(1943, 2, 25)
    ringo = date(1940, 7, 7)
    z.add("John Lennon", john)
    z.add("Paul McCartney", paul)
    z["George Harrison"] = george
    z["Ringo Starr"] = ringo
    # check the scores
    assert z["John Lennon"] == john
    assert z["Paul McCartney"] == paul
    assert z.score("George Harrison") == george
    assert z.score("Ringo Starr") == ringo
    assert z.score("Pete Best") is None
    with raises(KeyError):
        z["Bono"]
    # check that everyone ranks properly
    assert z.rank("Ringo Starr") == 0
    assert z.rank("John Lennon") == 1
    assert z.rank("Paul McCartney") == 2
    assert z.rank("George Harrison") == 3
    # score pairs
    assert z.scorepairs() == [
        ("Ringo Starr", ringo),
        ("John Lennon", john),
        ("Paul McCartney", paul),
        ("George Harrison", george)
    ]
    # counting
    assert z.count(date(1940, 1, 1), date(1940, 12, 31)) == 2
    # score ranges
    items = z.scorerange(date(1940, 1, 1), date(1940, 12, 31), withscores=True)
    assert items == [
        ("Ringo Starr", ringo),
        ("John Lennon", john)
    ]
    z.clear()


NUMBERS = [
    ("uno", 1.0),
    ("dos", 2.0),
    ("tres", 3.0),
    ("cuatro", 4.0),
    ("cinco", 5.0),
    ("seis", 6.0),
    ("siete", 7.0)
]

@nativetests.test
def zset_ranges():
    z = SortedSet(Key("~zset", client), Bytes)
    z.clear()
    for (name, score) in NUMBERS:
        z[name] = score
    # counting
    assert z.count(2.0, 6.0) == 5
    assert z.count(7.0, 8.0) == 1
    # items
    assert z.items() == "uno dos tres cuatro cinco seis siete".split()
    assert z.items(2, 4) == ["tres", "cuatro", "cinco"]
    assert z.items(2, 4, True) == ["cinco", "cuatro", "tres"]
    # scorepairs
    assert z.scorepairs() == NUMBERS
    assert z.scorepairs(2, 4) == [("tres", 3.0), ("cuatro", 4.0),
                                  ("cinco", 5.0)]
    assert z.scorepairs(2, 4, True) == [("cinco", 5.0), ("cuatro", 4.0),
                                        ("tres", 3.0)]
    # scorerange
    assert z.scorerange(1.5, 3.5) == ["dos", "tres"]
    assert z.scorerange(1.5, 3.5, reverse=True) == ["tres", "dos"]
    assert (z.scorerange(1.5, 7.5, start=1, num=3) ==
            ["tres", "cuatro", "cinco"])
    assert (z.scorerange(1.5, 7.5, start=1, num=3, reverse=True) ==
            ["seis", "cinco", "cuatro"])
    assert (z.scorerange(1.5, 3.5, withscores=True) ==
            [("dos", 2.0), ("tres", 3.0)])
    z.clear()


@nativetests.test
def hash_view():
    k = Key("~hash", client)
    k.delete()
    h = HashView(k, Integer, Integer)
    assert len(h) == 0
    k.hmset({1: 40, 2: 80, 3: 120, 4: 160})
    # getting keys
    assert h[1] == 40
    assert h.get(2) == 80
    assert 3 in h
    assert h.get(5) is None
    assert 6 not in h
    with raises(KeyError):
        h[7]
    assert h.copy() == {1: 40, 2: 80, 3: 120, 4: 160}
    # iterkeys/itervalues/iteritems
    assert sort(h.iterkeys()) == [1, 2, 3, 4]
    assert sort(h.itervalues()) == [40, 80, 120, 160]
    assert sort(h.iteritems()) == [(1, 40), (2, 80), (3, 120), (4, 160)]
    k.delete()


@nativetests.test
def hash_basics():
    h = Hash(Key("~hash", client), Integer, Integer)
    h.clear()
    assert not client.exists("~hash")
    assert len(h) == 0
    # adding keys
    h[1] = 40
    h[2] = 80
    assert client.hget("~hash", 1) == "40"
    assert client.hget("~hash", 2) == "80"
    h[3] = 100
    # setdefault
    assert h.setdefault(3, 100) == 100
    assert h[3] == 100
    assert h.setdefault(4, 150) == 150
    assert h[4] == 150
    # update
    h.update({4: 160, 5: 204})
    assert h[4] == 160
    assert h[5] == 204
    # incr
    assert h.incr(5) == 205
    assert h[5] == 205
    assert h.incr(5, -5) == 200
    assert h[5] == 200
    # pop
    assert h.pop(5) == 200
    assert 5 not in h
    assert h.pop(5, None) is None
    with raises(KeyError):
        h.pop(6)
    h.clear()