mira/tests/test_subscription.py

157 lines
5.8 KiB
Python
Raw Permalink Normal View History

from mira.subscription import SubscriptionManager
class MockStorageManager:
'''
The SubscriptionManager requieres the StorageManager, but we don't
need it for the tests. So just stub it.
'''
def set_data(self, module, section, data):
pass
def get_sum():
return SubscriptionManager({
'test': {
'a@localhost': {
'thing1': {
'data': 42
},
'thing2': {
'data': 100
}
},
'b@localhost': {
'thing2': {
'data': 89
},
'thing3': {
'data': [1, 2, 4]
}
},
'd@localhost': {
'thing1': {
'data': {}
}
}
}
}, MockStorageManager())
def test_get_subscriptions_for_jid():
sum = get_sum()
assert sum.get_subscriptions_for_jid('prod', 'a@localhost') == {}
assert sum.get_subscriptions_for_jid('test', 'z@localhost') == {}
subs = sum.get_subscriptions_for_jid('test', 'a@localhost')
assert len(subs.keys()) == 2
assert 'thing1' in subs and subs['thing1']['data'] == 42
assert 'thing2' in subs and subs['thing2']['data'] == 100
def test_get_subscriptions_for_keyword():
sum = get_sum()
assert sum.get_subscriptions_for_keyword('prod', 'thing1') == {}
assert sum.get_subscriptions_for_keyword('test', 'thing4') == {}
subs = sum.get_subscriptions_for_keyword('test', 'thing2')
assert 'a@localhost' in subs and subs['a@localhost'] == 100
assert 'b@localhost' in subs and subs['b@localhost'] == 89
def test_get_subscriptions_for_keywords():
sum = get_sum()
assert sum.get_subscriptions_for_keywords('prod', 'thing1') == {}
assert sum.get_subscriptions_for_keywords('test', 'thing4') == {}
subs = sum.get_subscriptions_for_keywords('test', ['thing2', 'thing3'])
assert 'a@localhost' in subs and 'thing2'in subs['a@localhost'] and subs['a@localhost']['thing2'] == 100
assert not 'thing3' in subs['a@localhost']
assert 'b@localhost' in subs and 'thing2' in subs['b@localhost'] and subs['b@localhost']['thing2'] == 89
assert 'b@localhost' in subs and 'thing3' in subs['b@localhost'] and subs['b@localhost']['thing3'] == [1, 2, 4]
def test_get_subscription_keywords():
sum = get_sum()
assert sum.get_subscription_keywords('prod') == []
assert not set(sum.get_subscription_keywords('test')) - set(['thing1', 'thing2', 'thing3'])
def test_is_subscribed_to_keyword():
sum = get_sum()
assert not sum.is_subscribed_to_keyword('prod', 'a@localhost', 'thing1')
assert not sum.is_subscribed_to_keyword('test', 'a@localhost', 'thing4')
assert sum.is_subscribed_to_keyword('test', 'a@localhost', 'thing1')
def test_is_subscribed_to_data():
sum = get_sum()
assert not sum.is_subscribed_to_data('prod', 'b@localhost', 'thing1', 1)
assert not sum.is_subscribed_to_data('test', 'b@localhost', 'thing4', 1)
assert not sum.is_subscribed_to_data('test', 'b@localhost', 'thing3', 10)
assert sum.is_subscribed_to_data('test', 'b@localhost', 'thing3', 1)
def test_is_subscribed_to_data_func():
sum = get_sum()
func1 = lambda x: x % 2 == 0
func2 = lambda x: x == 10
assert not sum.is_subscribed_to_data_func('prod', 'b@localhost', 'thing1', func1)
assert not sum.is_subscribed_to_data_func('test', 'b@localhost', 'thing4', func1)
assert not sum.is_subscribed_to_data_func('test', 'b@localhost', 'thing3', func2)
assert sum.is_subscribed_to_data_func('test', 'b@localhost', 'thing3', func1)
def test_add_subscription():
sum = get_sum()
sum.add_subscription('test', 'c@localhost', 'thing1')
assert sum.is_subscribed_to_keyword('test', 'c@localhost', 'thing1')
sum.add_subscription('test', 'a@localhost', 'thing4')
assert sum.is_subscribed_to_keyword('test', 'a@localhost', 'thing4')
sum.add_subscription('prod', 'a@localhost', 'thing4')
assert sum.is_subscribed_to_keyword('prod', 'a@localhost', 'thing4')
sum.add_subscription('prod', 'a@localhost', 'thing5', 60)
subs = sum.get_subscriptions_for_jid('prod', 'a@localhost')
assert sum.is_subscribed_to_keyword('prod', 'a@localhost', 'thing5')
assert subs and 'thing5' in subs and subs['thing5']['data'] == 60
def test_append_subscription_data():
sum = get_sum()
sum.add_subscription('test', 'c@localhost', 'thing1', [])
sum.append_subscription_data('test', 'c@localhost', 'thing1', 1)
subs = sum.get_subscriptions_for_jid('test', 'c@localhost')
assert sum.is_subscribed_to_keyword('test', 'c@localhost', 'thing1')
assert subs and 'thing1' in subs and subs['thing1']['data'] == [1]
sum.append_subscription_data('test', 'c@localhost', 'thing1', 5)
assert subs['thing1']['data'] == [1, 5]
def test_remove_subscription():
sum = get_sum()
sum.remove_subscription('test', 'd@localhost', 'thing1')
assert not sum.is_subscribed_to_keyword('test', 'd@localhost', 'thing1')
assert not sum.get_subscriptions_for_jid('test', 'd@localhost')
def test_filter_subscription_data_items():
sum = get_sum()
func = lambda x: not x % 2 == 0
sum.filter_subscription_data_items('test', 'b@localhost', 'thing3', func)
assert sum.is_subscribed_to_data('test', 'b@localhost', 'thing3', 1)
assert not sum.is_subscribed_to_data('test', 'b@localhost', 'thing3', 2)
assert not sum.is_subscribed_to_data('test', 'b@localhost', 'thing3', 4)
def test_remove_subscription_data_item():
sum = get_sum()
sum.remove_subscription_data_item('test', 'b@localhost', 'thing3', 4)
assert sum.is_subscribed_to_data('test', 'b@localhost', 'thing3', 1)
assert sum.is_subscribed_to_data('test', 'b@localhost', 'thing3', 2)
assert not sum.is_subscribed_to_data('test', 'b@localhost', 'thing3', 4)