<s> import os <EOL> import time <EOL> import unittest2 as unittest <EOL> import logging . config <EOL> logging . config . fileConfig ( "<STR_LIT>" ) <EOL> import shutil <EOL> from pyspider . database . sqlite import resultdb <EOL> from pyspider . result . result_worker import ResultWorker <EOL> from pyspider . libs . multiprocessing_queue import Queue <EOL> from pyspider . libs . utils import run_in_thread <EOL> class TestProcessor ( unittest . TestCase ) : <EOL> resultdb_path = '<STR_LIT>' <EOL> @ classmethod <EOL> def setUpClass ( self ) : <EOL> shutil . rmtree ( '<STR_LIT>' , ignore_errors = True ) <EOL> os . makedirs ( '<STR_LIT>' ) <EOL> def get_resultdb ( ) : <EOL> return resultdb . ResultDB ( self . resultdb_path ) <EOL> self . resultdb = get_resultdb ( ) <EOL> self . inqueue = Queue ( <NUM_LIT:10> ) <EOL> def run_result_worker ( ) : <EOL> self . result_worker = ResultWorker ( get_resultdb ( ) , self . inqueue ) <EOL> self . result_worker . run ( ) <EOL> self . process = run_in_thread ( run_result_worker ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> @ classmethod <EOL> def tearDownClass ( self ) : <EOL> if self . process . is_alive ( ) : <EOL> self . result_worker . quit ( ) <EOL> self . process . join ( <NUM_LIT:2> ) <EOL> assert not self . process . is_alive ( ) <EOL> shutil . rmtree ( '<STR_LIT>' , ignore_errors = True ) <EOL> def test_10_bad_result ( self ) : <EOL> self . inqueue . put ( ( { '<STR_LIT>' : '<STR_LIT>' } , { } ) ) <EOL> self . resultdb . _list_project ( ) <EOL> self . assertEqual ( len ( self . resultdb . projects ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( self . resultdb . count ( '<STR_LIT>' ) , <NUM_LIT:0> ) <EOL> def test_20_insert_result ( self ) : <EOL> data = { <EOL> '<STR_LIT:a>' : '<STR_LIT:b>' <EOL> } <EOL> self . inqueue . put ( ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:url>' : '<STR_LIT>' <EOL> } , data ) ) <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> self . resultdb . _list_project ( ) <EOL> self . assertEqual ( len ( self . resultdb . projects ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( self . resultdb . count ( '<STR_LIT>' ) , <NUM_LIT:1> ) <EOL> result = self . resultdb . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( result [ '<STR_LIT:result>' ] , data ) <EOL> def test_30_overwrite ( self ) : <EOL> self . inqueue . put ( ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:url>' : '<STR_LIT>' <EOL> } , "<STR_LIT:abc>" ) ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> result = self . resultdb . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( result [ '<STR_LIT:result>' ] , "<STR_LIT:abc>" ) <EOL> def test_40_insert_list ( self ) : <EOL> self . inqueue . put ( ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:url>' : '<STR_LIT>' <EOL> } , [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> result = self . resultdb . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( result [ '<STR_LIT:result>' ] , [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) </s>
