6
from pysqlite2 import dbapi2 as sqlite3
8
from landscape.tests.helpers import LandscapeTest
10
from landscape.broker.exchangestore import ExchangeStore
13
class ExchangeStoreTest(LandscapeTest):
14
"""Unit tests for the C{ExchangeStore}."""
17
super(ExchangeStoreTest, self).setUp()
19
self.filename = self.makeFile()
20
self.store1 = ExchangeStore(self.filename)
21
self.store2 = ExchangeStore(self.filename)
23
def test_add_message_context(self):
24
"""Adding a message context works correctly."""
26
self.store1.add_message_context(123, 'abc', 'change-packages')
28
db = sqlite3.connect(self.store2._filename)
31
"SELECT operation_id, secure_id, message_type, timestamp "
32
"FROM message_context WHERE operation_id=?", (123,))
33
results = cursor.fetchall()
34
self.assertEquals(1, len(results))
36
self.assertEquals(123, row[0])
37
self.assertEquals('abc', row[1])
38
self.assertEquals('change-packages', row[2])
39
self.assertTrue(row[3] > now)
41
def test_add_message_context_with_duplicate_operation_id(self):
42
"""Only one message context with a given operation-id is permitted."""
43
self.store1.add_message_context(123, "abc", "change-packages")
45
(sqlite3.IntegrityError, sqlite3.OperationalError),
46
self.store1.add_message_context, 123, "def", "change-packages")
48
def test_get_message_context(self):
50
Accessing a C{MessageContext} with an existing C{operation-id} works.
53
self.store1.add_message_context(234, 'bcd', 'change-packages')
54
context = self.store2.get_message_context(234)
55
self.assertEquals(234, context.operation_id)
56
self.assertEquals('bcd', context.secure_id)
57
self.assertEquals('change-packages', context.message_type)
58
self.assertTrue(context.timestamp > now)
60
def test_get_message_context_with_nonexistent_operation_id(self):
61
"""Attempts to access a C{MessageContext} with a non-existent
62
C{operation-id} result in C{None}."""
63
self.assertIs(None, self.store1.get_message_context(999))
65
def test_message_context_remove(self):
66
"""C{MessageContext}s are deleted correctly."""
67
context = self.store1.add_message_context(
68
345, 'opq', 'change-packages')
70
self.assertIs(None, self.store1.get_message_context(345))
72
def test_all_operation_ids_for_empty_database(self):
74
Calling C{all_operation_ids} on an empty database returns an empty
77
self.assertEquals([], self.store1.all_operation_ids())
79
def test_all_operation_ids(self):
80
"""C{all_operation_ids} works correctly."""
81
self.store1.add_message_context(456, 'cde', 'change-packages')
82
self.assertEquals([456], self.store2.all_operation_ids())
83
self.store2.add_message_context(567, 'def', 'change-packages')
84
self.assertEquals([456, 567], self.store1.all_operation_ids())