157 lines
5.8 KiB
Python
157 lines
5.8 KiB
Python
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)
|