<s> import logging <EOL> import os <EOL> import time <EOL> import uuid <EOL> import resource <EOL> import tempfile <EOL> import shutil <EOL> import threading <EOL> import ufora . native . Json as NativeJson <EOL> import traceback <EOL> import unittest <EOL> import ufora . config . Setup as Setup <EOL> import ufora . native . Json as JsonNative <EOL> import ufora . distributed . SharedState . SharedState as SharedState <EOL> import ufora . distributed . SharedState . tests . SharedStateTestHarness as SharedStateTestHarness <EOL> DEFAULT_PASS_COUNT = <NUM_LIT:5> <EOL> DEFAULT_THREAD_COUNT = <NUM_LIT:10> <EOL> class TestSharedState ( unittest . TestCase ) : <EOL> def clientThread ( self , harness , passIndex , threadIndex , totalThreadCount , <EOL> successList , failureList , connectedThreads , connectedThreadEvent , <EOL> viewList , <EOL> blockUntilConnected <EOL> ) : <EOL> try : <EOL> logging . info ( "<STR_LIT>" , threadIndex ) <EOL> try : <EOL> view = harness . newView ( ) <EOL> viewList . append ( view ) <EOL> finally : <EOL> connectedThreads . append ( threadIndex ) <EOL> if len ( connectedThreads ) == totalThreadCount : <EOL> connectedThreadEvent . set ( ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> if blockUntilConnected : <EOL> connectedThreadEvent . wait ( ) <EOL> logging . info ( "<STR_LIT>" , threadIndex ) <EOL> logging . info ( "<STR_LIT>" , threadIndex ) <EOL> harness . subscribeToKeyspace ( view , NativeJson . Json ( "<STR_LIT:data>" ) ) <EOL> harness . writeToKeyspace ( <EOL> view , <EOL> NativeJson . Json ( "<STR_LIT:data>" ) , <EOL> NativeJson . Json ( str ( threadIndex ) ) , <EOL> NativeJson . Json ( str ( passIndex ) ) <EOL> ) <EOL> t0 = time . time ( ) <EOL> while True : <EOL> itemDict = dict ( harness . getAllItemsFromView ( view , NativeJson . Json ( "<STR_LIT:data>" ) ) ) <EOL> allAreCurPass = True <EOL> if len ( itemDict ) == totalThreadCount : <EOL> for i in itemDict : <EOL> if itemDict [ i ] != NativeJson . Json ( str ( passIndex ) ) : <EOL> allAreCurPass = False <EOL> if allAreCurPass : <EOL> successList . append ( threadIndex ) <EOL> return <EOL> time . sleep ( <NUM_LIT> ) <EOL> if time . time ( ) - t0 > <NUM_LIT> : <EOL> failureList . append ( "<STR_LIT>" % itemDict ) <EOL> return <EOL> view . flush ( ) <EOL> except : <EOL> failureList . append ( "<STR_LIT>" + traceback . format_exc ( ) ) <EOL> def rapidReconnectionTest ( self , harness , totalThreadCount , passCount , blockUntilConnected ) : <EOL> for passIndex in range ( passCount ) : <EOL> successes = [ ] <EOL> failures = [ ] <EOL> logging . info ( "<STR_LIT>" , passIndex ) <EOL> threads = [ ] <EOL> connectedThreads = [ ] <EOL> viewList = [ ] <EOL> allConnectedEvent = threading . Event ( ) <EOL> for threadIndex in range ( totalThreadCount ) : <EOL> threads . append ( <EOL> threading . Thread ( <EOL> target = self . clientThread , <EOL> args = ( harness , passIndex , threadIndex , <EOL> totalThreadCount , successes , failures , connectedThreads , <EOL> allConnectedEvent , <EOL> viewList , <EOL> blockUntilConnected <EOL> ) <EOL> ) <EOL> ) <EOL> threads [ - <NUM_LIT:1> ] . start ( ) <EOL> for t in threads : <EOL> t . join ( ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> viewList = [ ] <EOL> logging . debug ( "<STR_LIT>" ) <EOL> for f in failures : <EOL> logging . warn ( "<STR_LIT>" , f ) <EOL> assert len ( successes ) == totalThreadCount , "<STR_LIT>" % ( passIndex , len ( successes ) , totalThreadCount , len ( failures ) ) <EOL> successes = [ ] <EOL> failures = [ ] <EOL> time . sleep ( <NUM_LIT> ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> def getHarness ( self , ** kwds ) : <EOL> return SharedStateTestHarness . SharedStateTestHarness ( ** kwds ) <EOL> def test_rapid_connections_InMemory_BlockUntilConnected ( self ) : <EOL> harness = self . getHarness ( inMemory = True , port = Setup . config ( ) . sharedStatePort ) <EOL> try : <EOL> self . rapidReconnectionTest ( <EOL> harness , <EOL> totalThreadCount = DEFAULT_THREAD_COUNT , <EOL> passCount = DEFAULT_PASS_COUNT , <EOL> blockUntilConnected = True <EOL> ) <EOL> finally : <EOL> time . sleep ( <NUM_LIT> ) <EOL> harness . teardown ( ) <EOL> def test_rapid_connections_UsingSockets_BlockUntilConnected ( self ) : <EOL> harness = self . getHarness ( inMemory = False , port = Setup . config ( ) . sharedStatePort ) <EOL> try : <EOL> self . rapidReconnectionTest ( <EOL> harness , <EOL> totalThreadCount = DEFAULT_THREAD_COUNT , <EOL> passCount = DEFAULT_PASS_COUNT , <EOL> blockUntilConnected = True <EOL> ) <EOL> finally : <EOL> harness . teardown ( ) <EOL> def test_rapid_connections_InMemory ( self ) : <EOL> harness = self . getHarness ( inMemory = True , port = Setup . config ( ) . sharedStatePort ) <EOL> try : <EOL> self . rapidReconnectionTest ( <EOL> harness , <EOL> totalThreadCount = DEFAULT_THREAD_COUNT , <EOL> passCount = DEFAULT_PASS_COUNT , <EOL> blockUntilConnected = False <EOL> ) <EOL> finally : <EOL> time . sleep ( <NUM_LIT> ) <EOL> harness . teardown ( ) <EOL> def test_rapid_connections_UsingSockets ( self ) : <EOL> harness = self . getHarness ( inMemory = False , port = Setup . config ( ) . sharedStatePort ) <EOL> try : <EOL> self . rapidReconnectionTest ( <EOL> harness , <EOL> totalThreadCount = DEFAULT_THREAD_COUNT , <EOL> passCount = DEFAULT_PASS_COUNT , <EOL> blockUntilConnected = False <EOL> ) <EOL> finally : <EOL> harness . teardown ( ) <EOL> def test_listener ( self ) : <EOL> harness = self . getHarness ( inMemory = True ) <EOL> try : <EOL> v1 = harness . newView ( ) <EOL> v2 = harness . newView ( ) <EOL> l = SharedState . Listener ( v2 ) <EOL> space = SharedState . Keyspace ( "<STR_LIT>" , NativeJson . Json ( "<STR_LIT>" ) , <NUM_LIT:1> ) <EOL> rng = SharedState . KeyRange ( space , <NUM_LIT:0> , None , None , True , False ) <EOL> v1 . subscribe ( rng ) <EOL> v2 . subscribe ( rng ) <EOL> keysToWrite = [ JsonNative . Json ( str ( x ) ) for x in range ( <NUM_LIT> ) ] <EOL> def writer ( ) : <EOL> with SharedState . Transaction ( v1 ) : <EOL> for keyName in keysToWrite : <EOL> key = SharedState . Key ( space , ( keyName , ) ) <EOL> v1 [ key ] = JsonNative . Json ( "<STR_LIT:value>" ) <EOL> changedKeys = [ ] <EOL> def reader ( ) : <EOL> while len ( changedKeys ) < len ( keysToWrite ) : <EOL> updates = l . get ( ) <EOL> for updateType , update in updates : <EOL> if updateType == "<STR_LIT>" : <EOL> for key in update : <EOL> changedKeys . append ( key [ <NUM_LIT:0> ] ) <EOL> writerThread = threading . Thread ( target = writer ) <EOL> readerThread = threading . Thread ( target = reader ) <EOL> writerThread . start ( ) <EOL> readerThread . start ( ) <EOL> writerThread . join ( ) <EOL> readerThread . join ( ) <EOL> self . assertEqual ( set ( keysToWrite ) , set ( changedKeys ) ) <EOL> finally : <EOL> time . sleep ( <NUM_LIT> ) <EOL> harness . teardown ( ) <EOL> def test_flush ( self ) : <EOL> harness = self . getHarness ( inMemory = True ) <EOL> try : <EOL> v1 = harness . newView ( ) <EOL> space = SharedState . Keyspace ( "<STR_LIT>" , NativeJson . Json ( "<STR_LIT>" ) , <NUM_LIT:1> ) <EOL> rng = SharedState . KeyRange ( space , <NUM_LIT:0> , None , None , True , False ) <EOL> v1 . subscribe ( rng ) <EOL> keysToWrite = [ str ( x ) for x in range ( <NUM_LIT> ) ] <EOL> value = "<STR_LIT:value>" * <NUM_LIT:100> * <NUM_LIT> <EOL> with SharedState . Transaction ( v1 ) : <EOL> self . assertRaises ( UserWarning , v1 . flush , True ) <EOL> with SharedState . Transaction ( v1 ) : <EOL> for keyName in keysToWrite : <EOL> key = SharedState . Key ( space , ( NativeJson . Json ( keyName ) , ) ) <EOL> v1 [ key ] = NativeJson . Json ( value ) <EOL> v1 . flush ( ) <EOL> finally : <EOL> time . sleep ( <NUM_LIT> ) <EOL> harness . teardown ( ) <EOL> def test_file_management ( self ) : <EOL> tempDir = tempfile . mkdtemp ( ) <EOL> curOpenFiles = len ( os . listdir ( '<STR_LIT>' % os . getpid ( ) ) ) <EOL> OPEN_FILE_LIMIT = <NUM_LIT:200> <EOL> if curOpenFiles >= OPEN_FILE_LIMIT : <EOL> os . system ( "<STR_LIT>" % os . getpid ( ) ) <EOL> self . assertTrue ( curOpenFiles < OPEN_FILE_LIMIT , "<STR_LIT>" % curOpenFiles ) <EOL> soft , hard = resource . getrlimit ( resource . RLIMIT_OFILE ) <EOL> harness = self . getHarness ( inMemory = True , <EOL> cachePathOverride = tempDir , <EOL> maxOpenFiles = <NUM_LIT:15> ) <EOL> try : <EOL> v1 = harness . newView ( ) <EOL> resource . setrlimit ( resource . RLIMIT_OFILE , ( curOpenFiles + <NUM_LIT:30> , hard ) ) <EOL> for ix in range ( <NUM_LIT> ) : <EOL> space = SharedState . Keyspace ( "<STR_LIT>" , <EOL> NativeJson . Json ( "<STR_LIT>" % ix ) , <EOL> <NUM_LIT:1> ) <EOL> rng = SharedState . KeyRange ( space , <NUM_LIT:0> , None , None , True , False ) <EOL> v1 . subscribe ( rng ) <EOL> key = SharedState . Key ( space , ( NativeJson . Json ( '<STR_LIT>' % ix ) , ) ) <EOL> with SharedState . Transaction ( v1 ) : <EOL> v1 [ key ] = NativeJson . Json ( '<STR_LIT>' % ix ) <EOL> finally : <EOL> time . sleep ( <NUM_LIT> ) <EOL> harness . teardown ( ) <EOL> resource . setrlimit ( resource . RLIMIT_OFILE , ( soft , hard ) ) <EOL> try : <EOL> shutil . rmtree ( tempDir ) <EOL> except : <EOL> pass <EOL> def test_require_subscription_manager ( self ) : <EOL> tempDir = tempfile . mkdtemp ( ) <EOL> try : <EOL> testKeyspace = SharedState . Keyspace ( "<STR_LIT>" , <EOL> NativeJson . Json ( "<STR_LIT>" ) , <EOL> <NUM_LIT:1> ) <EOL> harness = self . getHarness ( inMemory = True , cachePathOverride = tempDir ) <EOL> v1 = harness . newView ( ) <EOL> v1 . subscribe ( SharedState . KeyRange ( testKeyspace , <NUM_LIT:0> , None , None , True , False ) ) <EOL> for ix in range ( <NUM_LIT:1000> ) : <EOL> with SharedState . Transaction ( v1 ) : <EOL> key = SharedState . Key ( testKeyspace , ( NativeJson . Json ( "<STR_LIT:key>" ) , ) ) <EOL> v1 [ key ] = NativeJson . Json ( str ( ix ) ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> v2 = harness . newView ( ) <EOL> v2 . setMustSubscribe ( False ) <EOL> with SharedState . Transaction ( v2 ) : <EOL> key = SharedState . Key ( testKeyspace , ( NativeJson . Json ( "<STR_LIT:key>" ) , ) ) <EOL> value = '<STR_LIT>' + uuid . uuid4 ( ) . hex <EOL> v2 [ key ] = NativeJson . Json ( value ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> def toTry ( ) : <EOL> with SharedState . Transaction ( v2 ) : <EOL> key = SharedState . Key ( testKeyspace , ( NativeJson . Json ( "<STR_LIT:key>" ) , ) ) <EOL> value = '<STR_LIT>' + uuid . uuid4 ( ) . hex <EOL> v2 [ key ] = NativeJson . Json ( value ) <EOL> self . assertRaises ( UserWarning , toTry ) <EOL> finally : <EOL> time . sleep ( <NUM_LIT> ) <EOL> harness . teardown ( ) <EOL> shutil . rmtree ( tempDir ) <EOL> def test_require_subscription_view ( self ) : <EOL> try : <EOL> testKeyspace = SharedState . Keyspace ( "<STR_LIT>" , <EOL> NativeJson . Json ( "<STR_LIT>" ) , <EOL> <NUM_LIT:1> ) <EOL> harness = self . getHarness ( inMemory = True ) <EOL> v1 = harness . newView ( ) <EOL> def toTry ( ) : <EOL> with SharedState . Transaction ( v1 ) : <EOL> key = SharedState . Key ( testKeyspace , ( "<STR_LIT:key>" , ) ) <EOL> v1 [ key ] = NativeJson . Json ( '<STR_LIT>' ) <EOL> self . assertRaises ( UserWarning , toTry ) <EOL> finally : <EOL> time . sleep ( <NUM_LIT> ) <EOL> harness . teardown ( ) <EOL> def test_persistence ( self ) : <EOL> tempDir = tempfile . mkdtemp ( ) <EOL> try : <EOL> testValues = { } <EOL> def testWrite ( passIx ) : <EOL> try : <EOL> testKeyspaces = [ <EOL> SharedState . Keyspace ( "<STR_LIT>" , <EOL> NativeJson . Json ( "<STR_LIT>" % ix ) , <NUM_LIT:3> ) for ix in range ( <NUM_LIT:3> ) <EOL> ] <EOL> harness = self . getHarness ( inMemory = True , cachePathOverride = tempDir ) <EOL> v1 = harness . newView ( ) <EOL> for space in testKeyspaces : <EOL> rng = SharedState . KeyRange ( space , <NUM_LIT:0> , None , None , True , False ) <EOL> v1 . subscribe ( rng ) <EOL> for ix in range ( <NUM_LIT:9> ) : <EOL> keyspaceIx = ( ix + passIx ) % len ( testKeyspaces ) <EOL> space = testKeyspaces [ keyspaceIx ] <EOL> with SharedState . Transaction ( v1 ) : <EOL> key = SharedState . Key ( space , <EOL> ( NativeJson . Json ( "<STR_LIT>" % ix ) , NativeJson . Json ( '<STR_LIT>' ) , NativeJson . Json ( '<STR_LIT>' ) ) ) <EOL> value = JsonNative . Json ( <EOL> '<STR_LIT>' % <EOL> ( ix , keyspaceIx , passIx , uuid . uuid4 ( ) . hex ) <EOL> ) <EOL> v1 [ key ] = value <EOL> testValues [ key ] = value <EOL> v1 . flush ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> harness . teardown ( ) <EOL> harness = self . getHarness ( inMemory = True , cachePathOverride = tempDir ) <EOL> v1 = harness . newView ( ) <EOL> for space in testKeyspaces : <EOL> rng = SharedState . KeyRange ( space , <NUM_LIT:0> , None , None , True , False ) <EOL> v1 . subscribe ( rng ) <EOL> for key , value in testValues . iteritems ( ) : <EOL> self . assertSharedStateKeyResolvesToValue ( key , v1 , testValues [ key ] ) <EOL> except : <EOL> logging . warn ( "<STR_LIT>" , traceback . format_exc ( ) ) <EOL> raise <EOL> finally : <EOL> time . sleep ( <NUM_LIT> ) <EOL> harness . teardown ( ) <EOL> for ix in range ( <NUM_LIT:2> ) : <EOL> testWrite ( ix ) <EOL> finally : <EOL> shutil . rmtree ( tempDir ) <EOL> def assertSharedStateKeyResolvesToValue ( self , key , view , value ) : <EOL> passes = <NUM_LIT:0> <EOL> lastVal = None <EOL> while passes < <NUM_LIT:100> : <EOL> with SharedState . Transaction ( view ) : <EOL> if view [ key ] is not None : <EOL> lastVal = view [ key ] . value ( ) <EOL> self . assertEqual ( value , lastVal ) <EOL> return <EOL> passes += <NUM_LIT:1> <EOL> time . sleep ( <NUM_LIT> ) <EOL> assert False , "<STR_LIT>" % ( key , value , lastVal ) </s>
