mira/tests/test_subscription.py

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)