68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
"""Tests for the Inventory implementation"""
|
|
import os
|
|
import random
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
from minode import sql, shared, structure
|
|
|
|
|
|
# + __bool__
|
|
# + __contains__
|
|
# + __getitem__
|
|
# + __setitem__
|
|
# cleanup
|
|
# + get
|
|
# filter
|
|
# select
|
|
# + vectors_to_send
|
|
|
|
|
|
class TestObjectsSQL(unittest.TestCase):
|
|
"""A test case for the sqlite inventory"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
shared.data_directory = tempfile.gettempdir()
|
|
cls.objects = sql.Inventory()
|
|
|
|
def test_set_get(self):
|
|
"""Put some objects and check presence and getting"""
|
|
obj = structure.Object(
|
|
int(time.time()), 42, 1, 1, object_payload=b'HELLO')
|
|
self.assertFalse(obj.vector in self.objects)
|
|
with self.assertRaises(KeyError):
|
|
self.objects[obj.vector]
|
|
self.assertIsNone(self.objects.get(obj.vector))
|
|
self.objects[obj.vector] = obj
|
|
self.assertTrue(self.objects)
|
|
self.assertTrue(obj.vector in self.objects)
|
|
obj1 = self.objects[obj.vector]
|
|
self.assertEqual(obj.vector, obj1.vector)
|
|
self.assertEqual(obj.data, obj1.data)
|
|
|
|
def test_vectors_to_send(self):
|
|
"""Check vectors_to_send method"""
|
|
needed = set()
|
|
for _ in range(10):
|
|
# wrong stream
|
|
obj = structure.Object(
|
|
int(time.time() + 10), 42, 1, random.randint(1, 3),
|
|
object_payload=os.urandom(32))
|
|
self.objects[obj.vector] = obj
|
|
# expired
|
|
obj = structure.Object(
|
|
int(time.time() - 10), 42, 1, 4,
|
|
object_payload=os.urandom(32))
|
|
self.objects[obj.vector] = obj
|
|
# interesting
|
|
obj = structure.Object(
|
|
int(time.time() + 10), 42, 1, 4,
|
|
object_payload=os.urandom(32))
|
|
self.objects[obj.vector] = obj
|
|
needed.add(obj.vector)
|
|
|
|
self.assertEqual(set(self.objects.vectors_to_send(4)), needed)
|
|
self.assertTrue(set(self.objects.vectors_to_send()).difference(needed))
|