<s> import datetime <EOL> from stream_framework . activity import Activity <EOL> from stream_framework . activity import AggregatedActivity <EOL> from stream_framework . activity import DehydratedActivity <EOL> from stream_framework . tests . utils import Pin <EOL> from stream_framework . verbs . base import Love as LoveVerb <EOL> from stream_framework . aggregators . base import RecentVerbAggregator <EOL> from stream_framework . exceptions import ActivityNotFound <EOL> from stream_framework . exceptions import DuplicateActivityException <EOL> import time <EOL> import unittest <EOL> import six <EOL> class TestActivity ( unittest . TestCase ) : <EOL> def test_serialization_length ( self ) : <EOL> activity_object = Pin ( id = <NUM_LIT:1> ) <EOL> activity = Activity ( <NUM_LIT:1> , LoveVerb , activity_object ) <EOL> assert len ( str ( activity . serialization_id ) ) == <NUM_LIT> <EOL> def test_serialization_type ( self ) : <EOL> activity_object = Pin ( id = <NUM_LIT:1> ) <EOL> activity = Activity ( <NUM_LIT:1> , LoveVerb , activity_object ) <EOL> assert isinstance ( activity . serialization_id , ( six . integer_types , float ) ) <EOL> def test_serialization_overflow_check_object_id ( self ) : <EOL> activity_object = Pin ( id = <NUM_LIT:10> ** <NUM_LIT:10> ) <EOL> activity = Activity ( <NUM_LIT:1> , LoveVerb , activity_object ) <EOL> with self . assertRaises ( TypeError ) : <EOL> activity . serialization_id <EOL> def test_serialization_overflow_check_role_id ( self ) : <EOL> activity_object = Pin ( id = <NUM_LIT:1> ) <EOL> Verb = type ( '<STR_LIT>' , ( LoveVerb , ) , { '<STR_LIT:id>' : <NUM_LIT> } ) <EOL> activity = Activity ( <NUM_LIT:1> , Verb , activity_object ) <EOL> with self . assertRaises ( TypeError ) : <EOL> activity . serialization_id <EOL> def test_dehydrated_activity ( self ) : <EOL> activity_object = Pin ( id = <NUM_LIT:1> ) <EOL> activity = Activity ( <NUM_LIT:1> , LoveVerb , activity_object ) <EOL> dehydrated = activity . get_dehydrated ( ) <EOL> self . assertTrue ( isinstance ( dehydrated , DehydratedActivity ) ) <EOL> self . assertEquals ( <EOL> dehydrated . serialization_id , activity . serialization_id ) <EOL> def test_compare_idempotent_init ( self ) : <EOL> t1 = datetime . datetime . utcnow ( ) <EOL> activity_object = Pin ( id = <NUM_LIT:1> ) <EOL> activity1 = Activity ( <NUM_LIT:1> , LoveVerb , activity_object , time = t1 ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> activity2 = Activity ( <NUM_LIT:1> , LoveVerb , activity_object , time = t1 ) <EOL> self . assertEquals ( activity1 , activity2 ) <EOL> def test_compare_apple_and_oranges ( self ) : <EOL> activity_object = Pin ( id = <NUM_LIT:1> ) <EOL> activity = Activity ( <NUM_LIT:1> , LoveVerb , activity_object ) <EOL> with self . assertRaises ( ValueError ) : <EOL> activity == activity_object <EOL> class TestAggregatedActivity ( unittest . TestCase ) : <EOL> def test_contains ( self ) : <EOL> activity = Activity ( <NUM_LIT:1> , LoveVerb , Pin ( id = <NUM_LIT:1> ) ) <EOL> aggregated = AggregatedActivity ( <NUM_LIT:1> , [ activity ] ) <EOL> self . assertTrue ( aggregated . contains ( activity ) ) <EOL> def test_duplicated_activities ( self ) : <EOL> activity = Activity ( <NUM_LIT:1> , LoveVerb , Pin ( id = <NUM_LIT:1> ) ) <EOL> aggregated = AggregatedActivity ( <NUM_LIT:1> , [ activity ] ) <EOL> with self . assertRaises ( DuplicateActivityException ) : <EOL> aggregated . append ( activity ) <EOL> def test_compare_apple_and_oranges ( self ) : <EOL> activity = AggregatedActivity ( <NUM_LIT:1> , [ Activity ( <NUM_LIT:1> , LoveVerb , Pin ( id = <NUM_LIT:1> ) ) ] ) <EOL> with self . assertRaises ( ValueError ) : <EOL> activity == Pin ( id = <NUM_LIT:1> ) <EOL> def test_contains_extraneous_object ( self ) : <EOL> activity = AggregatedActivity ( <NUM_LIT:1> , [ Activity ( <NUM_LIT:1> , LoveVerb , Pin ( id = <NUM_LIT:1> ) ) ] ) <EOL> with self . assertRaises ( ValueError ) : <EOL> activity . contains ( Pin ( id = <NUM_LIT:1> ) ) <EOL> def test_aggregated_properties ( self ) : <EOL> activities = [ ] <EOL> for x in range ( <NUM_LIT:1> , <NUM_LIT> ) : <EOL> activity_object = Pin ( id = x ) <EOL> activity = Activity ( x , LoveVerb , activity_object ) <EOL> activities . append ( activity ) <EOL> aggregator = RecentVerbAggregator ( ) <EOL> aggregated_activities = aggregator . aggregate ( activities ) <EOL> aggregated = aggregated_activities [ <NUM_LIT:0> ] <EOL> self . assertEqual ( aggregated . verbs , [ LoveVerb ] ) <EOL> self . assertEqual ( aggregated . verb , LoveVerb ) <EOL> self . assertEqual ( aggregated . actor_count , <NUM_LIT:100> ) <EOL> self . assertEqual ( aggregated . minimized_activities , <NUM_LIT> ) <EOL> self . assertEqual ( aggregated . other_actor_count , <NUM_LIT> ) <EOL> self . assertEqual ( aggregated . activity_count , <NUM_LIT:100> ) <EOL> self . assertEqual ( aggregated . object_ids , list ( range ( <NUM_LIT> , <NUM_LIT> ) ) ) <EOL> self . assertEqual ( aggregated . actor_ids , list ( range ( <NUM_LIT> , <NUM_LIT> ) ) ) <EOL> self . assertEqual ( aggregated . is_seen ( ) , False ) <EOL> self . assertEqual ( aggregated . is_read ( ) , False ) <EOL> def generate_activities ( self ) : <EOL> activities = [ ] <EOL> for x in range ( <NUM_LIT:1> , <NUM_LIT:20> ) : <EOL> activity = Activity ( x , LoveVerb , Pin ( id = x ) ) <EOL> activities . append ( activity ) <EOL> return activities <EOL> def generate_aggregated_activities ( self , activities ) : <EOL> aggregator = RecentVerbAggregator ( ) <EOL> aggregated_activities = aggregator . aggregate ( activities ) <EOL> return aggregated_activities <EOL> def test_aggregated_compare ( self ) : <EOL> activities = self . generate_activities ( ) <EOL> aggregated_activities = self . generate_aggregated_activities ( activities ) <EOL> aggregated_activities_two = self . generate_aggregated_activities ( activities ) <EOL> new_activities = self . generate_activities ( ) <EOL> aggregated_activities_three = self . generate_aggregated_activities ( new_activities ) <EOL> self . assertEqual ( aggregated_activities , aggregated_activities_two ) <EOL> self . assertNotEqual ( aggregated_activities , aggregated_activities_three ) <EOL> def test_aggregated_remove ( self ) : <EOL> activities = [ ] <EOL> for x in range ( <NUM_LIT:1> , <NUM_LIT> ) : <EOL> activity_object = Pin ( id = x ) <EOL> activity = Activity ( x , LoveVerb , activity_object ) <EOL> activities . append ( activity ) <EOL> aggregator = RecentVerbAggregator ( ) <EOL> aggregated_activities = aggregator . aggregate ( activities ) <EOL> aggregated = aggregated_activities [ <NUM_LIT:0> ] <EOL> for activity in activities : <EOL> try : <EOL> aggregated . remove ( activity ) <EOL> except ( ActivityNotFound , ValueError ) : <EOL> pass <EOL> self . assertEqual ( len ( aggregated . activities ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( aggregated . activity_count , <NUM_LIT> ) </s>
