<s> import time <EOL> from . . import base <EOL> from girder . models . model_base import ValidationException <EOL> from girder . models . notification import ProgressState <EOL> from girder . utility . progress import ProgressContext <EOL> def setUpModule ( ) : <EOL> base . startServer ( ) <EOL> def tearDownModule ( ) : <EOL> base . stopServer ( ) <EOL> class NotificationTestCase ( base . TestCase ) : <EOL> def setUp ( self ) : <EOL> base . TestCase . setUp ( self ) <EOL> self . admin = self . model ( '<STR_LIT:user>' ) . createUser ( <EOL> email = '<STR_LIT>' , login = '<STR_LIT>' , firstName = '<STR_LIT>' , <EOL> lastName = '<STR_LIT>' , password = '<STR_LIT>' ) <EOL> def _testStream ( self , user , token = None ) : <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' ) <EOL> self . assertStatus ( resp , <NUM_LIT> ) <EOL> self . assertEqual ( <EOL> resp . json [ '<STR_LIT:message>' ] , <EOL> '<STR_LIT>' ) <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' , <EOL> user = user , token = token , isJson = False , <EOL> params = { '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> self . assertStatusOk ( resp ) <EOL> self . assertEqual ( self . getBody ( resp ) , '<STR_LIT>' ) <EOL> with ProgressContext ( <EOL> True , user = user , token = token , title = '<STR_LIT>' , total = <NUM_LIT:100> , <EOL> interval = <NUM_LIT:100> ) as progress : <EOL> progress . update ( current = <NUM_LIT:1> ) <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' , <EOL> user = user , token = token , isJson = False , <EOL> params = { '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> messages = self . getSseMessages ( resp ) <EOL> self . assertEqual ( len ( messages ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( messages [ <NUM_LIT:0> ] [ '<STR_LIT:type>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( messages [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] , <NUM_LIT:100> ) <EOL> self . assertEqual ( messages [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertFalse ( ProgressState . isComplete ( <EOL> messages [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] [ '<STR_LIT:state>' ] ) ) <EOL> progress . interval = <NUM_LIT> <EOL> time . sleep ( <NUM_LIT> ) <EOL> progress . update ( current = <NUM_LIT:2> ) <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' , <EOL> user = user , token = token , isJson = False , <EOL> params = { '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> messages = self . getSseMessages ( resp ) <EOL> self . assertEqual ( len ( messages ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( messages [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> progress . update ( current = '<STR_LIT>' ) <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' , <EOL> user = user , token = token , isJson = False , <EOL> params = { '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> messages = self . getSseMessages ( resp ) <EOL> self . assertEqual ( len ( messages ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( messages [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> progress . interval = <NUM_LIT:1000> <EOL> progress . update ( current = <NUM_LIT:3> ) <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' , <EOL> user = user , token = token , isJson = False , <EOL> params = { '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> messages = self . getSseMessages ( resp ) <EOL> self . assertEqual ( len ( messages ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( messages [ <NUM_LIT:0> ] [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] , <NUM_LIT:3> ) <EOL> try : <EOL> with ProgressContext ( <EOL> True , user = user , token = token , title = '<STR_LIT>' , <EOL> total = <NUM_LIT:100> ) as progress : <EOL> raise ValidationException ( '<STR_LIT>' ) <EOL> except ValidationException : <EOL> pass <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' , <EOL> user = user , token = token , isJson = False , <EOL> params = { '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> messages = self . getSseMessages ( resp ) <EOL> self . assertEqual ( messages [ - <NUM_LIT:1> ] [ '<STR_LIT:data>' ] [ '<STR_LIT:message>' ] , <EOL> '<STR_LIT>' ) <EOL> def testUserStream ( self ) : <EOL> self . _testStream ( self . admin ) <EOL> def testTokenStream ( self ) : <EOL> resp = self . request ( path = '<STR_LIT>' , method = '<STR_LIT:GET>' ) <EOL> self . assertStatusOk ( resp ) <EOL> token = resp . json [ '<STR_LIT>' ] <EOL> tokenDoc = self . model ( '<STR_LIT>' ) . load ( token , force = True , objectId = False ) <EOL> self . _testStream ( None , tokenDoc ) </s>
