bitter/tests/test_crawlers.py

from unittest import TestCase

import os
import types
import datetime
import time

from bitter import utils, easy
from bitter.crawlers import QueueException
from bitter import config as c

class TestCrawlers(TestCase):

    def setUp(self):
        self.wq = easy(os.path.join(os.path.dirname(__file__), 'credentials.json'))

    def test_create_worker(self):
        assert len(self.wq.queue)==1

    def test_get_limits(self):
        w1 = list(self.wq.queue)[0]
        print(w1.limits)
        limitslook = w1.get_limit(['statuses', 'lookup'])
        assert limitslook['remaining'] == limitslook['limit']

    def test_set_limits(self):
        w1 = list(self.wq.queue)[0]
        w1.set_limit(['test', 'test2'], {'remaining': 0})
        assert w1.get_limit(['test', 'test2']) == {'remaining': 0}

    def test_await(self):
        w1 = list(self.wq.queue)[0]
        w1.set_limit(['test', 'wait'], {'remaining': 0, 'reset': time.time()+2})
        assert w1.get_wait(['test', 'wait']) > 1
        time.sleep(2)
        assert w1.get_wait(['test', 'wait']) == 0
        assert w1.get_wait(['statuses', 'lookup']) == 0

    def test_is_limited(self):
        w1 = list(self.wq.queue)[0]
        assert not w1.is_limited(['statuses', 'lookup'])
        w1.set_limit(['test', 'limited'], {'remaining': 0, 'reset': time.time()+100})
        assert  w1.is_limited(['test', 'limited'])

    def test_call(self):
        w1 = list(self.wq.queue)[0]
        l1 = w1.get_limit(['users', 'lookup'])
        resp = self.wq.users.lookup(screen_name='balkian')
        l2 = w1.get_limit(['users', 'lookup'])
        assert l1['remaining']-l2['remaining'] == 1

    def test_consume(self):
        w1 = list(self.wq.queue)[0]
        l1 = w1.get_limit(['friends', 'list'])
        self.wq.wait = False
        for i in range(l1['remaining']):
            print(i)
            resp = self.wq.friends.list(screen_name='balkian')
        # l2 = w1.get_limit(['users', 'lookup'])
        # assert l2['remaining'] == 0
        # self.wq.users.lookup(screen_name='balkian')
        
        failed = False
        try:
            # resp = self.wq.friends.list(screen_name='balkian')
            self.wq.next(['friends', 'list'])
        except QueueException:
            failed = True
        assert failed
        l2 = w1.get_limit(['friends', 'list'])
        assert self.wq.get_wait(['friends', 'list']) > (l2['reset']-time.time())
        assert self.wq.get_wait(['friends', 'list']) < (l2['reset']-time.time()+2)