hogumathi-app/hogumathi_app/test/unit/hogumathi_app_test/test_content_system.py

from typing import List, Dict

import pytest

from hogumathi_app.content_system import ContentSystem

def tweet_cs (tweet_id:str) -> object:
    print(f'tweet_cs: {tweet_id}')
    if tweet_id == '1':
        return {'text': 'one', 'id': '1'}

def tweets_cs (content_ids:List[str]) -> Dict[str,object]:
    print(f'tweets_cs: {content_ids}')
    if ','.join(content_ids) == 'twitter:tweet:1,twitter:tweet:2':
        return {
            'twitter:tweet:1': {'text': 'one', 'id': '1'},
            'twitter:tweet:2': {'text': 'two', 'id': '2'}
            }
    elif ','.join(content_ids) == 'twitter:tweet:2':
        return {
            'twitter:tweet:2': {'text': 'two', 'id': '2'}
        }


def video_cs (vid_id:str) -> object:
    print(f'video_cs: {vid_id}')
    if vid_id == '3':
        return {'url': 'three.mp4', 'id': '3'}

def videos_cs (content_ids:List[str]) -> Dict[str, object]:
    print(f'videos_cs: {content_ids}')
    if ','.join(content_ids) == 'youtube:video:3,youtube:video:4':
        return {
            'youtube:video:3': {'url': 'three.mp4', 'id': '3'},
            'youtube:video:4': {'url': 'four.mp4', 'id': '4'}
            }

def photo_cs (post_id:str) -> object:
    print(f'photo_cs: {post_id}')
    if post_id == '3':
        return {'url': 'three.png', 'id': '3'}


def test_content_system ():
    
    h_cs = ContentSystem()
    
    h_cs.register_content_source('twitter:tweet:', tweet_cs)
    h_cs.register_content_source('twitter:tweets', tweets_cs)
    
    t1 = h_cs.get_content('twitter:tweet:1')
    t2 = h_cs.get_content('twitter:tweet:2')
    t3 = h_cs.get_content('fake:1')
    
    assert(t1 == {'text': 'one', 'id': '1'})
    assert(t2 == None)
    assert(t3 == None)
    
    t_multi = h_cs.get_content('twitter:tweets', content_ids=['twitter:tweet:1', 'twitter:tweet:2'])
    
    assert(t_multi == {
            'twitter:tweet:1': {'text': 'one', 'id': '1'},
            'twitter:tweet:2': {'text': 'two', 'id': '2'}
            })
    
def test_bulk_content ():
    
    h_cs = ContentSystem()
    
    h_cs.register_content_source('twitter:tweet:', tweet_cs)
    h_cs.register_content_source('twitter:tweets', tweets_cs, id_pattern='')
    
    tweets = h_cs.get_all_content(['twitter:tweet:1', 'twitter:tweet:2', 'fake:1'], enable_bulk_fetch=True)
    
    
    assert(tweets == {
        'twitter:tweet:1': {'text': 'one', 'id': '1'},
        'twitter:tweet:2': {'text': 'two', 'id': '2'},
        'fake:1': None
        })

@pytest.mark.xfail(reason="need to rethink get_all_content code for this.")
def test_bulk_content_partial_miss ():
    """
    xfail: note in ContentSystem.get_content
    """
    
    h_cs = ContentSystem()
    
    def tweets_cache_cs (content_ids):
        if ','.join(content_ids) == 'twitter:tweet:1,twitter:tweet:2':
            return {
                'twitter:tweet:1': {'text': 'one', 'id': '1'},
                'twitter:tweet:2': None
            }
    
    h_cs.register_content_source('twitter:tweet:', tweet_cs)
    h_cs.register_content_source('twitter:tweets', tweets_cs, id_pattern='')
    h_cs.register_content_source('', tweets_cache_cs, id_pattern='(.+)', weight=99999)
    
    tweets = h_cs.get_all_content(['twitter:tweet:1', 'twitter:tweet:2', 'fake:1'], enable_bulk_fetch=True)
    
    
    assert(tweets == {
        'twitter:tweet:1': {'text': 'one', 'id': '1'},
        'twitter:tweet:2': {'text': 'two', 'id': '2'},
        'fake:1': None
        })
    
def test_hooks_bulk_content ():
    
    h_cs = ContentSystem()
    
    h_cs.register_content_source('twitter:tweet:', tweet_cs)
    h_cs.register_content_source('twitter:tweets', tweets_cs, id_pattern='')
    
    hooked_content = []
    def got_content(content_id, content):
        hooked_content.append([content_id, content])
    
    h_cs.register_hook('got_content', got_content)
    
    content = h_cs.get_all_content(['twitter:tweet:1', 'twitter:tweet:2', 'fake:1'], enable_bulk_fetch=True)
    
    assert(content == {
        'twitter:tweet:1': {'text': 'one', 'id': '1'},
        'twitter:tweet:2': {'text': 'two', 'id': '2'},
        'fake:1': None
    })
    
    print(f'hooked_content: {hooked_content}')
    assert(hooked_content == [
        ['twitter:tweets', {
            'twitter:tweet:1': {'text': 'one', 'id': '1'},
            'twitter:tweet:2': {'text': 'two', 'id': '2'}
            }],
        ['twitter:tweet:1', {'text': 'one', 'id': '1'}],
        ['twitter:tweet:2', {'text': 'two', 'id': '2'}]
        ])
        
def test_hooks_bulk_content_multi_bulk ():
    
    h_cs = ContentSystem()
    
    h_cs.register_content_source('twitter:tweet:', tweet_cs)
    h_cs.register_content_source('twitter:tweets', tweets_cs, id_pattern='')
    h_cs.register_content_source('youtube:video:', video_cs)
    h_cs.register_content_source('youtube:videos', videos_cs, id_pattern='')
    h_cs.register_content_source('instagram:post:', photo_cs)
    
    hooked_content = []
    def got_content(content_id, content):
        hooked_content.append([content_id, content])
    
    h_cs.register_hook('got_content', got_content)
    
    content = h_cs.get_all_content(['twitter:tweet:1', 'twitter:tweet:2',
                                    'youtube:video:3', 'youtube:video:4',
                                    'instagram:post:3',
                                    'fake:1'], enable_bulk_fetch=True)
    
    assert(content == {
        'twitter:tweet:1': {'text': 'one', 'id': '1'},
        'twitter:tweet:2': {'text': 'two', 'id': '2'},
        'youtube:video:3': {'url': 'three.mp4', 'id': '3'},
        'youtube:video:4': {'url': 'four.mp4', 'id': '4'},
        'instagram:post:3': {'url': 'three.png', 'id': '3'},
        'fake:1': None
    })
    
    print(f'hooked_content: {hooked_content}')
    assert(hooked_content == [
        ['instagram:post:3', {'url': 'three.png', 'id': '3'}],
        ['twitter:tweets', {
            'twitter:tweet:1': {'text': 'one', 'id': '1'},
            'twitter:tweet:2': {'text': 'two', 'id': '2'}
            }],
        ['twitter:tweet:1', {'text': 'one', 'id': '1'}],
        ['twitter:tweet:2', {'text': 'two', 'id': '2'}],
        ['youtube:videos', {
            'youtube:video:3': {'url': 'three.mp4', 'id': '3'},
            'youtube:video:4': {'url': 'four.mp4', 'id': '4'}
            }],
        ['youtube:video:3', {'url': 'three.mp4', 'id': '3'}],
        ['youtube:video:4', {'url': 'four.mp4', 'id': '4'}]
        ])

def test_cache ():
    
    h_cs = ContentSystem()
    
    cache_hits = []
    
    cache = {}
    def cache_cs (content_id = None, content_ids = None):
        if content_id in cache:
            content = cache[content_id]
            cache_hits.append([content_id, content])
            return content
        
    h_cs.register_content_source('twitter:tweet:', tweet_cs)
    h_cs.register_content_source('twitter:tweets', tweets_cs, id_pattern='')
    
    h_cs.register_content_source('', cache_cs, id_pattern='(.+)', weight=99999)
    

    def cache_hook (content_id, content):
        # FIXME we might need a skip hook mechanism
        # this will be invoked even with a cache hit.
        # perhaps pass source_id into the hook
        if content_id.endswith('s') or content_id in cache:
            # FIXME exclusion list/patterns
            # we should allow bulk fetches to get cached items
            return
        cache[content_id] = content
    
    h_cs.register_hook('got_content', cache_hook)
    
    t1 = h_cs.get_content('twitter:tweet:1')
    t2 = h_cs.get_content('twitter:tweet:2')
    t3 = h_cs.get_content('fake:1')
    
    assert(t1 == {'text': 'one', 'id': '1'})
    assert(t2 == None)
    assert(t3 == None)
    
    assert(cache_hits == [])
    
    t1_2 = h_cs.get_content('twitter:tweet:1')
    t2_2 = h_cs.get_content('twitter:tweet:2')
    t3_2 = h_cs.get_content('fake:1')
    
    assert(t1_2 == t1)
    assert(t2_2 == None)
    assert(t2_2 == None)
    
    print(f'cache_hits = {cache_hits}')
    
    assert(cache_hits == [
        ['twitter:tweet:1', {'text': 'one', 'id': '1'}]
        ])


def test_cache_bulk ():
    
    h_cs = ContentSystem()
    
    cache_hits = []
    
    cache = {}
    def cache_cs (content_id = None, content_ids = None):
        if content_id and content_id in cache:
            content = cache[content_id]
            cache_hits.append([content_id, content])
            return content
        elif content_ids:
            results = {}
            for content_id in content_ids:
                content = cache_cs(content_id)
                if content:
                    results[content_id] = content
            return results
        
    h_cs.register_content_source('twitter:tweet:', tweet_cs)
    h_cs.register_content_source('twitter:tweets', tweets_cs, id_pattern='')
    h_cs.register_content_source('youtube:video:', video_cs)
    h_cs.register_content_source('youtube:videos', videos_cs, id_pattern='')
    h_cs.register_content_source('instagram:post:', photo_cs)
    
    h_cs.register_content_source('', cache_cs, id_pattern='(.+)', weight=99999)
    
    
    def cache_hook (content_id, content):
        # FIXME we might need a skip hook mechanism
        # this will be invoked even with a cache hit.
        # perhaps pass source_id into the hook
        if content_id.endswith('s') or content_id in cache:
            # FIXME exclusion list/patterns
            # we should allow bulk fetches to get cached items
            return
        cache[content_id] = content
    
    h_cs.register_hook('got_content', cache_hook)
    
    content = h_cs.get_all_content(['twitter:tweet:1', 'twitter:tweet:2',
                                    'youtube:video:3', 'youtube:video:4',
                                    'instagram:post:3',
                                    'fake:1'], enable_bulk_fetch=True)
    
    assert(cache_hits == [])
    
    content2 = h_cs.get_all_content(['twitter:tweet:1', 'twitter:tweet:2',
                                    'youtube:video:3', 'youtube:video:4',
                                    'instagram:post:3',
                                    'fake:1'], enable_bulk_fetch=True)
    
    assert(content == content2)
    
    print(f'cache_hits = {cache_hits}')
    
    assert(cache_hits == [
        ['instagram:post:3', {'url': 'three.png', 'id': '3'}],
        ['twitter:tweet:1', {'text': 'one', 'id': '1'}],
        ['twitter:tweet:2', {'text': 'two', 'id': '2'}],
        ['youtube:video:3', {'url': 'three.mp4', 'id': '3'}],
        ['youtube:video:4', {'url': 'four.mp4', 'id': '4'}]
        ])