<s> from stream_framework . tests . feeds . base import TestBaseFeed , implementation <EOL> import pytest <EOL> from stream_framework . feeds . cassandra import CassandraFeed <EOL> from stream_framework . utils import datetime_to_epoch <EOL> from stream_framework . activity import Activity <EOL> class CustomActivity ( Activity ) : <EOL> @ property <EOL> def serialization_id ( self ) : <EOL> '''<STR_LIT>''' <EOL> if self . object_id >= <NUM_LIT:10> ** <NUM_LIT:10> or self . verb . id >= <NUM_LIT:10> ** <NUM_LIT:3> : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> if not self . time : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> milliseconds = str ( int ( datetime_to_epoch ( self . time ) * <NUM_LIT:1000> ) ) <EOL> serialization_id_str = '<STR_LIT>' % ( <EOL> milliseconds , self . object_id % <NUM_LIT:100> , self . verb . id ) <EOL> serialization_id = int ( serialization_id_str ) <EOL> return serialization_id <EOL> class CassandraCustomFeed ( CassandraFeed ) : <EOL> activity_class = CustomActivity <EOL> @ pytest . mark . usefixtures ( "<STR_LIT>" ) <EOL> class TestCassandraBaseFeed ( TestBaseFeed ) : <EOL> feed_cls = CassandraFeed <EOL> def test_add_insert_activity ( self ) : <EOL> pass <EOL> def test_add_remove_activity ( self ) : <EOL> pass <EOL> @ pytest . mark . usefixtures ( "<STR_LIT>" ) <EOL> class TestCassandraCustomFeed ( TestBaseFeed ) : <EOL> feed_cls = CassandraCustomFeed <EOL> activity_class = CustomActivity <EOL> def test_add_insert_activity ( self ) : <EOL> pass <EOL> def test_add_remove_activity ( self ) : <EOL> pass <EOL> @ implementation <EOL> def test_custom_activity ( self ) : <EOL> assert self . test_feed . count ( ) == <NUM_LIT:0> <EOL> self . feed_cls . insert_activity ( <EOL> self . activity <EOL> ) <EOL> self . test_feed . add ( self . activity ) <EOL> assert self . test_feed . count ( ) == <NUM_LIT:1> <EOL> assert self . activity == self . test_feed [ : <NUM_LIT:10> ] [ <NUM_LIT:0> ] <EOL> assert type ( self . activity ) == type ( self . test_feed [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) </s>
