<s> """<STR_LIT>""" <EOL> import threading <EOL> from grr . lib import aff4 <EOL> from grr . lib import rdfvalue <EOL> from grr . lib import test_lib <EOL> from grr . lib import utils <EOL> from grr . lib . aff4_objects import sequential_collection <EOL> class TestSequentialCollection ( sequential_collection . SequentialCollection ) : <EOL> RDF_TYPE = rdfvalue . RDFInteger <EOL> class SequentialCollectionTest ( test_lib . AFF4ObjectTest ) : <EOL> def testAddScan ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> collection . Add ( rdfvalue . RDFInteger ( i ) ) <EOL> i = <NUM_LIT:0> <EOL> last_ts = <NUM_LIT:0> <EOL> for ( ts , v ) in collection . Scan ( ) : <EOL> last_ts = ts <EOL> self . assertEqual ( i , v ) <EOL> i += <NUM_LIT:1> <EOL> for j in range ( <NUM_LIT:100> ) : <EOL> collection . Add ( rdfvalue . RDFInteger ( j + <NUM_LIT:100> ) ) <EOL> for ( ts , v ) in collection . Scan ( after_timestamp = last_ts ) : <EOL> self . assertEqual ( i , v ) <EOL> i += <NUM_LIT:1> <EOL> self . assertEqual ( i , <NUM_LIT:200> ) <EOL> def testDuplicateTimestamps ( self ) : <EOL> with aff4 . FACTORY . Create ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> t = rdfvalue . RDFDatetime ( ) . Now ( ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> ts = collection . Add ( rdfvalue . RDFInteger ( i ) , timestamp = t ) <EOL> self . assertEqual ( ts [ <NUM_LIT:0> ] , t ) <EOL> i = <NUM_LIT:0> <EOL> for ( ts , _ ) in collection . Scan ( ) : <EOL> self . assertEqual ( ts , t ) <EOL> i += <NUM_LIT:1> <EOL> self . assertEqual ( i , <NUM_LIT:10> ) <EOL> def testMultiResolve ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> timestamps = [ ] <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> timestamps . append ( collection . Add ( rdfvalue . RDFInteger ( i ) ) ) <EOL> even_results = sorted ( [ r for r in collection . MultiResolve ( timestamps [ : : <NUM_LIT:2> ] ) <EOL> ] ) <EOL> self . assertEqual ( len ( even_results ) , <NUM_LIT:50> ) <EOL> self . assertEqual ( even_results [ <NUM_LIT:0> ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( even_results [ <NUM_LIT> ] , <NUM_LIT> ) <EOL> def testDelete ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> collection . Add ( rdfvalue . RDFInteger ( i ) ) <EOL> aff4 . FACTORY . Delete ( "<STR_LIT>" , <EOL> token = self . token ) <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> for _ in collection . Scan ( ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> class TestIndexedSequentialCollection ( <EOL> sequential_collection . IndexedSequentialCollection ) : <EOL> RDF_TYPE = rdfvalue . RDFInteger <EOL> class IndexedSequentialCollectionTest ( test_lib . AFF4ObjectTest ) : <EOL> def setUp ( self ) : <EOL> super ( IndexedSequentialCollectionTest , self ) . setUp ( ) <EOL> biu = sequential_collection . BackgroundIndexUpdater ( ) <EOL> sequential_collection . BACKGROUND_INDEX_UPDATER = biu <EOL> t = threading . Thread ( None , biu . UpdateLoop ) <EOL> t . daemon = True <EOL> t . start ( ) <EOL> def testAddGet ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> self . assertEqual ( collection . CalculateLength ( ) , <NUM_LIT:0> ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> collection . Add ( rdfvalue . RDFInteger ( i ) ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> self . assertEqual ( collection [ i ] , i ) <EOL> self . assertEqual ( collection . CalculateLength ( ) , <NUM_LIT:100> ) <EOL> self . assertEqual ( len ( collection ) , <NUM_LIT:100> ) <EOL> def testStaticAddGet ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> self . assertEqual ( collection . CalculateLength ( ) , <NUM_LIT:0> ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> TestIndexedSequentialCollection . StaticAdd ( <EOL> "<STR_LIT>" , self . token , <EOL> rdfvalue . RDFInteger ( i ) ) <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> self . assertEqual ( collection [ i ] , i ) <EOL> self . assertEqual ( collection . CalculateLength ( ) , <NUM_LIT:100> ) <EOL> self . assertEqual ( len ( collection ) , <NUM_LIT:100> ) <EOL> def testIndexCreate ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> for i in range ( <NUM_LIT:10> * <NUM_LIT> ) : <EOL> collection . Add ( rdfvalue . RDFInteger ( i ) ) <EOL> self . assertEqual ( collection . _index , None ) <EOL> self . assertEqual ( collection . CalculateLength ( ) , <NUM_LIT:10> * <NUM_LIT> ) <EOL> self . assertEqual ( sorted ( collection . _index . keys ( ) ) , [ <NUM_LIT:0> ] ) <EOL> with test_lib . FakeTime ( rdfvalue . RDFDatetime ( ) . Now ( ) + rdfvalue . Duration ( <EOL> "<STR_LIT>" ) ) : <EOL> _ = collection [ <NUM_LIT:0> ] <EOL> self . assertEqual ( sorted ( collection . _index . keys ( ) ) , [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( collection . CalculateLength ( ) , <NUM_LIT:10> * <NUM_LIT> ) <EOL> self . assertEqual ( <EOL> sorted ( collection . _index . keys ( ) ) , [ <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <EOL> <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> self . assertEqual ( collection . _index , None ) <EOL> _ = collection [ <NUM_LIT:0> ] <EOL> self . assertEqual ( <EOL> sorted ( collection . _index . keys ( ) ) , [ <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <EOL> <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] ) <EOL> def testIndexedReads ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> data_size = <NUM_LIT:4> * <NUM_LIT> <EOL> for i in range ( data_size ) : <EOL> collection . Add ( rdfvalue . RDFInteger ( i ) ) <EOL> with test_lib . FakeTime ( rdfvalue . RDFDatetime ( ) . Now ( ) + rdfvalue . Duration ( <EOL> "<STR_LIT>" ) ) : <EOL> for i in range ( data_size - <NUM_LIT:1> , data_size - <NUM_LIT:20> , - <NUM_LIT:1> ) : <EOL> self . assertEqual ( collection [ i ] , i ) <EOL> self . assertEqual ( collection [ <NUM_LIT> ] , <NUM_LIT> ) <EOL> self . assertEqual ( collection [ <NUM_LIT> ] , <NUM_LIT> ) <EOL> self . assertEqual ( collection [ <NUM_LIT> ] , <NUM_LIT> ) <EOL> for i in range ( data_size - <NUM_LIT> , data_size - <NUM_LIT> , - <NUM_LIT:1> ) : <EOL> self . assertEqual ( collection [ i ] , i ) <EOL> def testListing ( self ) : <EOL> test_urn = "<STR_LIT>" <EOL> with aff4 . FACTORY . Create ( test_urn , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> timestamps = [ ] <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> timestamps . append ( collection . Add ( rdfvalue . RDFInteger ( i ) ) ) <EOL> collection = aff4 . FACTORY . Open ( test_urn , token = self . token ) <EOL> with test_lib . Instrument ( sequential_collection . SequentialCollection , <EOL> "<STR_LIT>" ) as scan : <EOL> self . assertEqual ( len ( list ( collection ) ) , <NUM_LIT:100> ) <EOL> self . assertEqual ( scan . call_count , <NUM_LIT:2> ) <EOL> def testAutoIndexing ( self ) : <EOL> indexing_done = threading . Event ( ) <EOL> def UpdateIndex ( self ) : <EOL> self . UpdateIndex . old_target ( self ) <EOL> indexing_done . set ( ) <EOL> isq = sequential_collection . IndexedSequentialCollection <EOL> biu = sequential_collection . BACKGROUND_INDEX_UPDATER <EOL> with utils . MultiStubber ( <EOL> ( biu , "<STR_LIT>" , <NUM_LIT:0> ) , <EOL> ( isq , "<STR_LIT>" , rdfvalue . Duration ( "<STR_LIT>" ) ) , <EOL> ( isq , "<STR_LIT>" , <NUM_LIT:8> ) , <EOL> ( isq , "<STR_LIT>" , UpdateIndex ) ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> collection . Flush ( ) <EOL> for i in range ( <NUM_LIT> ) : <EOL> collection . Add ( rdfvalue . RDFInteger ( i ) ) <EOL> if not indexing_done . wait ( timeout = <NUM_LIT:10> ) : <EOL> self . fail ( "<STR_LIT>" ) <EOL> with aff4 . FACTORY . Open ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> _ = collection [ <NUM_LIT:0> ] <EOL> self . assertGreater ( len ( collection . _index ) , <NUM_LIT:16> ) <EOL> class GeneralIndexedCollectionTest ( test_lib . AFF4ObjectTest ) : <EOL> def testAddGet ( self ) : <EOL> with aff4 . FACTORY . Create ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> token = self . token ) as collection : <EOL> collection . Add ( rdfvalue . RDFInteger ( <NUM_LIT> ) ) <EOL> collection . Add ( rdfvalue . RDFString ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( collection [ <NUM_LIT:0> ] . __class__ , rdfvalue . RDFInteger ) <EOL> self . assertEqual ( collection [ <NUM_LIT:0> ] , <NUM_LIT> ) <EOL> self . assertEqual ( collection [ <NUM_LIT:1> ] . __class__ , rdfvalue . RDFString ) <EOL> self . assertEqual ( collection [ <NUM_LIT:1> ] , "<STR_LIT>" ) </s>
