<s> import unittest <EOL> import os <EOL> import flask <EOL> from flask_mab import BanditMiddleware , add_bandit , choose_arm , reward_endpt <EOL> import flask_mab . storage <EOL> from flask_mab . bandits import EpsilonGreedyBandit <EOL> from werkzeug . http import parse_cookie <EOL> import json <EOL> from test_utils import makeBandit <EOL> from threading import Thread <EOL> from multiprocessing import Pool <EOL> from copy import copy <EOL> from Queue import Queue <EOL> class MABTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> banditStorage = flask_mab . storage . JSONBanditStorage ( '<STR_LIT>' ) <EOL> app = flask . Flask ( '<STR_LIT>' ) <EOL> flask_mab . BanditMiddleware ( app ) <EOL> app . add_bandit ( '<STR_LIT>' , makeBandit ( "<STR_LIT>" , epsilon = <NUM_LIT:0.1> ) ) <EOL> app . debug = True <EOL> @ app . route ( "<STR_LIT:/>" ) <EOL> def root ( ) : <EOL> return "<STR_LIT>" <EOL> @ app . route ( "<STR_LIT>" ) <EOL> @ choose_arm ( "<STR_LIT>" ) <EOL> def assign_arm_decorated ( color_button ) : <EOL> return flask . make_response ( "<STR_LIT>" ) <EOL> @ app . route ( "<STR_LIT>" ) <EOL> @ reward_endpt ( "<STR_LIT>" , <NUM_LIT:1.0> ) <EOL> def reward_decorated ( ) : <EOL> return flask . make_response ( "<STR_LIT>" ) <EOL> self . app = app <EOL> self . app . config [ "<STR_LIT>" ] = True <EOL> self . app_client = app . test_client ( ) <EOL> def test_routing ( self ) : <EOL> rv = self . app_client . get ( "<STR_LIT:/>" ) <EOL> assert "<STR_LIT>" in rv . data <EOL> def test_suggest_decorated ( self ) : <EOL> rv = self . app_client . get ( "<STR_LIT>" ) <EOL> assert parse_cookie ( rv . headers [ "<STR_LIT>" ] ) [ "<STR_LIT>" ] <EOL> assert "<STR_LIT>" in rv . headers . keys ( ) <EOL> chosen_arm = self . get_arm ( rv . headers ) [ "<STR_LIT>" ] <EOL> assert self . app . extensions [ '<STR_LIT>' ] . bandits [ "<STR_LIT>" ] [ chosen_arm ] [ "<STR_LIT>" ] > <NUM_LIT:0> <EOL> assert json . loads ( parse_cookie ( rv . headers [ "<STR_LIT>" ] ) [ "<STR_LIT>" ] ) [ "<STR_LIT>" ] == chosen_arm <EOL> def test_from_cookie_reward_decorated ( self ) : <EOL> first_req = self . app_client . get ( "<STR_LIT>" ) <EOL> assert "<STR_LIT>" in first_req . headers . keys ( ) <EOL> chosen_arm = json . loads ( parse_cookie ( first_req . headers [ "<STR_LIT>" ] ) [ "<STR_LIT>" ] ) [ "<STR_LIT>" ] <EOL> self . app_client . get ( "<STR_LIT>" ) <EOL> assert self . app . extensions [ '<STR_LIT>' ] . bandits [ "<STR_LIT>" ] [ chosen_arm ] [ "<STR_LIT>" ] > <NUM_LIT:0> <EOL> def get_arm ( self , headers ) : <EOL> key_vals = [ h . strip ( ) for h in headers [ "<STR_LIT>" ] . split ( '<STR_LIT:;>' ) [ <NUM_LIT:1> : ] ] <EOL> return dict ( [ tuple ( tup . split ( "<STR_LIT::>" ) ) for tup in key_vals ] ) <EOL> def test_new_session ( self ) : <EOL> first_req = self . app_client . get ( "<STR_LIT>" ) <EOL> assert first_req . headers [ '<STR_LIT>' ] . split ( '<STR_LIT:;>' ) [ <NUM_LIT:0> ] . strip ( ) == '<STR_LIT>' <EOL> self . app_client . cookie_jar . clear ( ) <EOL> second_req = self . app_client . get ( "<STR_LIT>" ) <EOL> assert second_req . headers [ '<STR_LIT>' ] . split ( '<STR_LIT:;>' ) [ <NUM_LIT:0> ] . strip ( ) == '<STR_LIT>' <EOL> def test_repeating_session ( self ) : <EOL> first_req = self . app_client . get ( "<STR_LIT>" ) <EOL> for i in xrange ( <NUM_LIT:30> ) : <EOL> req = self . app_client . get ( "<STR_LIT>" ) <EOL> assert req . headers [ '<STR_LIT>' ] . split ( '<STR_LIT:;>' ) [ <NUM_LIT:0> ] . strip ( ) == '<STR_LIT>' <EOL> def test_concurrency ( self ) : <EOL> """<STR_LIT>""" <EOL> self . app . extensions [ '<STR_LIT>' ] . bandit_storage . flush ( ) <EOL> def request_worker ( test , iden , q ) : <EOL> try : <EOL> client = test . app . test_client ( ) <EOL> first_req = client . get ( "<STR_LIT>" ) <EOL> chosen_arm = json . loads ( parse_cookie ( first_req . headers [ "<STR_LIT>" ] ) [ "<STR_LIT>" ] ) [ "<STR_LIT>" ] <EOL> assert first_req . headers [ '<STR_LIT>' ] . split ( '<STR_LIT:;>' ) [ <NUM_LIT:0> ] . strip ( ) == '<STR_LIT>' <EOL> for i in xrange ( <NUM_LIT> ) : <EOL> req = client . get ( "<STR_LIT>" ) <EOL> assert req . headers [ '<STR_LIT>' ] . split ( '<STR_LIT:;>' ) [ <NUM_LIT:1> ] . split ( '<STR_LIT::>' ) [ <NUM_LIT:1> ] == chosen_arm <EOL> assert req . headers [ '<STR_LIT>' ] . split ( '<STR_LIT:;>' ) [ <NUM_LIT:0> ] . strip ( ) == '<STR_LIT>' <EOL> client . cookie_jar . clear ( ) <EOL> final_req = client . get ( "<STR_LIT>" ) <EOL> assert final_req . headers [ '<STR_LIT>' ] . split ( '<STR_LIT:;>' ) [ <NUM_LIT:0> ] . strip ( ) == '<STR_LIT>' <EOL> q . put ( True ) <EOL> except AssertionError , e : <EOL> q . put ( e ) <EOL> jobs = [ ] <EOL> q = Queue ( ) <EOL> for i in xrange ( <NUM_LIT:4> ) : <EOL> jobs . append ( Thread ( target = request_worker , args = ( self , i , q ) ) ) <EOL> map ( lambda x : x . start ( ) , jobs ) <EOL> map ( lambda x : x . join ( ) , jobs ) <EOL> while not q . empty ( ) : <EOL> val = q . get ( ) <EOL> if isinstance ( val , AssertionError ) : <EOL> raise val <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
