<s> from core . models import Follow , Pin <EOL> from core . utils . request import RequestMock <EOL> from django . conf import settings <EOL> from django . contrib . auth import get_user_model <EOL> from django . core . urlresolvers import reverse <EOL> from django . test . client import Client <EOL> from django . test . testcases import TestCase <EOL> from core . feed_managers import manager <EOL> import json <EOL> import os <EOL> import logging <EOL> import copy <EOL> from stream_framework . utils . timing import timer <EOL> logger = logging . getLogger ( __name__ ) <EOL> def absolute_path ( path ) : <EOL> return os . path . join ( settings . BASE_DIR , path ) <EOL> class BaseTestCase ( TestCase ) : <EOL> fixtures = map ( <EOL> absolute_path , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def setUp ( self ) : <EOL> TestCase . setUp ( self ) <EOL> self . bogus_user = bogus_user = get_user_model ( ) . objects . get ( <EOL> username = '<STR_LIT>' ) <EOL> self . bogus_user2 = bogus_user2 = get_user_model ( ) . objects . get ( <EOL> username = '<STR_LIT>' ) <EOL> for u in [ self . bogus_user , self . bogus_user2 ] : <EOL> user_feed = manager . get_user_feed ( u . id ) <EOL> user_feed . delete ( ) <EOL> for name , feed in manager . get_feeds ( u . id ) . items ( ) : <EOL> feed . delete ( ) <EOL> rf = RequestMock ( ) <EOL> self . rf = rf <EOL> self . client = Client ( ) <EOL> self . auth_client = Client ( ) <EOL> self . auth_client2 = Client ( ) <EOL> auth_response = self . auth_client . login ( <EOL> username = '<STR_LIT>' , password = '<STR_LIT>' ) <EOL> self . assertTrue ( auth_response ) <EOL> auth_response = self . auth_client2 . login ( <EOL> username = '<STR_LIT>' , password = '<STR_LIT>' ) <EOL> self . assertTrue ( auth_response ) <EOL> class PinTest ( BaseTestCase ) : <EOL> def test_pin ( self ) : <EOL> data = dict ( <EOL> message = '<STR_LIT>' , <EOL> item = <NUM_LIT:1> , <EOL> board_name = '<STR_LIT>' , <EOL> influencer = <NUM_LIT:1> , <EOL> ) <EOL> pin_url = reverse ( '<STR_LIT>' ) + '<STR_LIT>' <EOL> response = self . client . post ( pin_url , data ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> pin_response = self . auth_client . post ( pin_url , data ) <EOL> self . assertEqual ( pin_response . status_code , <NUM_LIT:200> ) <EOL> response_data = json . loads ( pin_response . content ) <EOL> self . assertTrue ( response_data [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( response_data [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> def test_unpin ( self ) : <EOL> data = dict ( <EOL> message = '<STR_LIT>' , <EOL> item = <NUM_LIT:1> , <EOL> board_name = '<STR_LIT>' , <EOL> influencer = <NUM_LIT:1> , <EOL> ) <EOL> pin_url = reverse ( '<STR_LIT>' ) + '<STR_LIT>' <EOL> response = self . client . post ( pin_url , data ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> pin_response = self . auth_client . post ( pin_url , data ) <EOL> self . assertEqual ( pin_response . status_code , <NUM_LIT:200> ) <EOL> response_data = json . loads ( pin_response . content ) <EOL> self . assertTrue ( response_data [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( response_data [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> data [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> response = self . auth_client . post ( pin_url , data ) <EOL> pins = list ( Pin . objects . filter ( item = <NUM_LIT:1> , user = self . bogus_user ) [ : <NUM_LIT:4> ] ) <EOL> assert pins == [ ] <EOL> class FollowTest ( BaseTestCase ) : <EOL> def test_follow ( self ) : <EOL> data = dict ( <EOL> target = <NUM_LIT:2> , <EOL> ) <EOL> follow_url = reverse ( '<STR_LIT>' ) <EOL> response = self . client . post ( follow_url , data ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> follow_response = self . auth_client . post ( follow_url , data ) <EOL> self . assertEqual ( follow_response . status_code , <NUM_LIT:200> ) <EOL> response_data = json . loads ( follow_response . content ) <EOL> self . assertTrue ( response_data [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( response_data [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> def test_unfollow ( self ) : <EOL> data = dict ( <EOL> target = <NUM_LIT:2> , <EOL> ) <EOL> follow_url = reverse ( '<STR_LIT>' ) <EOL> response = self . client . post ( follow_url , data ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> follow_response = self . auth_client . post ( follow_url , data ) <EOL> self . assertEqual ( follow_response . status_code , <NUM_LIT:200> ) <EOL> response_data = json . loads ( follow_response . content ) <EOL> self . assertTrue ( response_data [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( response_data [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> data [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> response = self . auth_client . post ( follow_url , data ) <EOL> follows = list ( Follow . objects . filter ( <EOL> target = <NUM_LIT:2> , user = self . bogus_user ) [ : <NUM_LIT:4> ] ) <EOL> assert follows == [ ] <EOL> class SimpleViewTest ( BaseTestCase ) : <EOL> def test_trending ( self ) : <EOL> url = reverse ( '<STR_LIT>' ) <EOL> response = self . client . get ( url ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> response = self . auth_client . get ( url ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> def test_feed ( self ) : <EOL> url = reverse ( '<STR_LIT>' ) <EOL> response = self . client . get ( url ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT> ) <EOL> response = self . auth_client . get ( url ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> def test_profile ( self ) : <EOL> url = reverse ( '<STR_LIT>' , args = [ '<STR_LIT>' ] ) <EOL> response = self . client . get ( url ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> response = self . auth_client . get ( url ) <EOL> self . assertEqual ( response . status_code , <NUM_LIT:200> ) <EOL> class BenchmarkTest ( BaseTestCase ) : <EOL> '''<STR_LIT>''' <EOL> fixtures = BaseTestCase . fixtures + map ( absolute_path , [ '<STR_LIT>' ] ) <EOL> def test_batch_import ( self ) : <EOL> admin_user_id = <NUM_LIT:1> <EOL> pins = list ( Pin . objects . filter ( user = admin_user_id ) [ : <NUM_LIT:3> ] ) <EOL> activities = [ p . create_activity ( ) for p in pins ] <EOL> manager . batch_import ( admin_user_id , activities , <NUM_LIT:10> ) <EOL> def test_aggregated_add_many ( self ) : <EOL> t = timer ( ) <EOL> admin_user_id = <NUM_LIT:1> <EOL> aggregated = manager . get_feeds ( admin_user_id ) [ '<STR_LIT>' ] <EOL> pins = list ( Pin . objects . filter ( user = admin_user_id ) [ : <NUM_LIT:3> ] ) <EOL> activities = [ ] <EOL> base_activity = pins [ <NUM_LIT:0> ] . create_activity ( ) <EOL> sample_size = <NUM_LIT:1000> <EOL> for x in range ( <NUM_LIT:1> , sample_size ) : <EOL> activity = copy . deepcopy ( base_activity ) <EOL> activity . actor_id = x <EOL> activity . object_id = x <EOL> activities . append ( activity ) <EOL> aggregated . insert_activities ( activities ) <EOL> for activity in activities : <EOL> aggregated . add_many ( [ activity ] , trim = False ) <EOL> add_many_time = t . next ( ) <EOL> print '<STR_LIT>' % add_many_time <EOL> popular_user_time = <NUM_LIT> / sample_size * add_many_time <EOL> print '<STR_LIT>' % popular_user_time <EOL> class StreamViewTest ( BaseTestCase ) : <EOL> def pin_in_feed ( self , pin , auth_client ) : <EOL> feed_url = reverse ( '<STR_LIT>' ) <EOL> response = auth_client . get ( feed_url ) <EOL> feed_pins = response . context [ '<STR_LIT>' ] <EOL> present_in_feed = False <EOL> pin_ids = [ f . object_id for f in feed_pins ] <EOL> if feed_pins and pin . id in pin_ids : <EOL> present_in_feed = True <EOL> aggregated_feed_url = reverse ( '<STR_LIT>' ) <EOL> response = auth_client . get ( aggregated_feed_url ) <EOL> feed_pins = response . context [ '<STR_LIT>' ] <EOL> present_in_aggregated_feed = False <EOL> pin_ids = sum ( [ a . object_ids for a in feed_pins ] , [ ] ) <EOL> if feed_pins and pin . id in pin_ids : <EOL> present_in_aggregated_feed = True <EOL> return present_in_feed , present_in_aggregated_feed <EOL> def pin_in_profile ( self , pin ) : <EOL> profile_url = reverse ( '<STR_LIT>' , args = [ self . bogus_user . username ] ) <EOL> response = self . auth_client . get ( profile_url ) <EOL> pins = response . context [ '<STR_LIT>' ] <EOL> present = False <EOL> if pins and pins [ <NUM_LIT:0> ] . object_id == pin . id : <EOL> present = True <EOL> return present <EOL> def setup_bogus2_pins ( self ) : <EOL> for x in range ( <NUM_LIT:1> , <NUM_LIT:4> ) : <EOL> data = dict ( <EOL> message = '<STR_LIT>' , <EOL> item = x , <EOL> board_name = '<STR_LIT>' , <EOL> influencer = <NUM_LIT:1> , <EOL> ) <EOL> pin_url = reverse ( '<STR_LIT>' ) + '<STR_LIT>' <EOL> self . auth_client2 . post ( pin_url , data ) <EOL> def test_pin_add ( self ) : <EOL> '''<STR_LIT>''' <EOL> print '<STR_LIT>' <EOL> data = dict ( <EOL> message = '<STR_LIT>' , <EOL> item = <NUM_LIT:2> , <EOL> board_name = '<STR_LIT>' , <EOL> influencer = <NUM_LIT:1> , <EOL> ) <EOL> pin_url = reverse ( '<STR_LIT>' ) + '<STR_LIT>' <EOL> self . auth_client . post ( pin_url , data ) <EOL> print '<STR_LIT>' <EOL> last_pin = Pin . objects . all ( ) . order_by ( '<STR_LIT>' ) [ : <NUM_LIT:1> ] [ <NUM_LIT:0> ] <EOL> profile_pin = self . pin_in_profile ( last_pin ) <EOL> self . assertTrue ( profile_pin ) <EOL> feed , aggregated = self . pin_in_feed ( <EOL> last_pin , auth_client = self . auth_client2 ) <EOL> self . assertTrue ( feed ) <EOL> self . assertTrue ( aggregated ) <EOL> def test_pin_flow ( self ) : <EOL> '''<STR_LIT>''' <EOL> data = dict ( <EOL> message = '<STR_LIT>' , <EOL> item = <NUM_LIT:2> , <EOL> board_name = '<STR_LIT>' , <EOL> influencer = <NUM_LIT:1> , <EOL> ) <EOL> pin_url = reverse ( '<STR_LIT>' ) + '<STR_LIT>' <EOL> self . auth_client . post ( pin_url , data ) <EOL> last_pin = Pin . objects . all ( ) . order_by ( '<STR_LIT>' ) [ : <NUM_LIT:1> ] [ <NUM_LIT:0> ] <EOL> profile_pin = self . pin_in_profile ( last_pin ) <EOL> self . assertTrue ( profile_pin ) <EOL> feed , aggregated = self . pin_in_feed ( <EOL> last_pin , auth_client = self . auth_client2 ) <EOL> self . assertTrue ( feed ) <EOL> self . assertTrue ( aggregated ) <EOL> data [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> self . auth_client . post ( pin_url , data ) <EOL> profile_pin = self . pin_in_profile ( last_pin ) <EOL> self . assertFalse ( profile_pin ) <EOL> feed , aggregated = self . pin_in_feed ( <EOL> last_pin , auth_client = self . auth_client2 ) <EOL> self . assertFalse ( feed ) <EOL> self . assertFalse ( aggregated ) <EOL> def test_follow_flow ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . setup_bogus2_pins ( ) <EOL> pins = list ( Pin . objects . filter ( user_id = self . bogus_user2 . id ) [ : <NUM_LIT:3> ] ) <EOL> self . assertTrue ( pins ) <EOL> data = dict ( <EOL> target = self . bogus_user2 . id , <EOL> ) <EOL> follow_url = reverse ( '<STR_LIT>' ) <EOL> self . auth_client . post ( follow_url , data ) <EOL> for pin in pins : <EOL> feed , aggregated = self . pin_in_feed ( <EOL> pin , auth_client = self . auth_client ) <EOL> self . assertTrue ( feed ) <EOL> self . assertTrue ( aggregated ) <EOL> data [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> self . auth_client . post ( follow_url , data ) <EOL> for pin in pins : <EOL> feed , aggregated = self . pin_in_feed ( <EOL> pin , auth_client = self . auth_client ) <EOL> self . assertFalse ( feed ) <EOL> self . assertFalse ( aggregated ) </s>
