<s> import time <EOL> from tests import AsyncHTTPTestCase <EOL> from tests . utils import task_succeeded_events , task_failed_events <EOL> from tests . utils import HtmlTableParser <EOL> from celery . events import Event <EOL> from celery . utils import uuid <EOL> from flower . events import EventsState <EOL> class DashboardTests ( AsyncHTTPTestCase ) : <EOL> def setUp ( self ) : <EOL> self . app = super ( DashboardTests , self ) . get_app ( ) <EOL> super ( DashboardTests , self ) . setUp ( ) <EOL> def get_app ( self ) : <EOL> return self . app <EOL> def test_default_page ( self ) : <EOL> r1 = self . get ( '<STR_LIT:/>' ) <EOL> r2 = self . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( r1 . body , r2 . body ) <EOL> def test_no_workers ( self ) : <EOL> r = self . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:200> , r . code ) <EOL> self . assertIn ( '<STR_LIT>' , str ( r . body ) ) <EOL> self . assertNotIn ( '<STR_LIT>' , str ( r . body ) ) <EOL> def test_unknown_worker ( self ) : <EOL> r = self . get ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT> , r . code ) <EOL> self . assertIn ( '<STR_LIT>' , str ( r . body ) ) <EOL> def test_single_workers_offline ( self ) : <EOL> state = EventsState ( ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> state . event ( Event ( '<STR_LIT>' , hostname = '<STR_LIT>' , <EOL> local_received = time . time ( ) ) ) <EOL> state . event ( Event ( '<STR_LIT>' , hostname = '<STR_LIT>' , <EOL> local_received = time . time ( ) ) ) <EOL> self . app . events . state = state <EOL> r = self . get ( '<STR_LIT>' ) <EOL> table = HtmlTableParser ( ) <EOL> table . parse ( str ( r . body ) ) <EOL> self . assertEqual ( <NUM_LIT:200> , r . code ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( table . rows ( ) ) ) <EOL> self . assertTrue ( table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:False>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertFalse ( table . get_row ( '<STR_LIT>' ) ) <EOL> def test_single_workers_online ( self ) : <EOL> state = EventsState ( ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> state . event ( Event ( '<STR_LIT>' , hostname = '<STR_LIT>' , <EOL> local_received = time . time ( ) ) ) <EOL> self . app . events . state = state <EOL> r = self . get ( '<STR_LIT>' ) <EOL> table = HtmlTableParser ( ) <EOL> table . parse ( str ( r . body ) ) <EOL> self . assertEqual ( <NUM_LIT:200> , r . code ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( table . rows ( ) ) ) <EOL> self . assertTrue ( table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertFalse ( table . get_row ( '<STR_LIT>' ) ) <EOL> def test_task_received ( self ) : <EOL> state = EventsState ( ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> events = [ Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = uuid ( ) , name = '<STR_LIT>' , <EOL> args = '<STR_LIT>' , kwargs = "<STR_LIT>" , <EOL> retries = <NUM_LIT:0> , eta = None , hostname = '<STR_LIT>' ) ] <EOL> for i , e in enumerate ( events ) : <EOL> e [ '<STR_LIT>' ] = i <EOL> e [ '<STR_LIT>' ] = time . time ( ) <EOL> state . event ( e ) <EOL> self . app . events . state = state <EOL> r = self . get ( '<STR_LIT>' ) <EOL> table = HtmlTableParser ( ) <EOL> table . parse ( str ( r . body ) ) <EOL> self . assertEqual ( <NUM_LIT:200> , r . code ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( table . rows ( ) ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:1>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> def test_task_started ( self ) : <EOL> state = EventsState ( ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> events = [ Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , name = '<STR_LIT>' , <EOL> args = '<STR_LIT>' , kwargs = "<STR_LIT>" , <EOL> retries = <NUM_LIT:0> , eta = None , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , hostname = '<STR_LIT>' ) ] <EOL> for i , e in enumerate ( events ) : <EOL> e [ '<STR_LIT>' ] = i <EOL> e [ '<STR_LIT>' ] = time . time ( ) <EOL> state . event ( e ) <EOL> self . app . events . state = state <EOL> r = self . get ( '<STR_LIT>' ) <EOL> table = HtmlTableParser ( ) <EOL> table . parse ( str ( r . body ) ) <EOL> self . assertEqual ( <NUM_LIT:200> , r . code ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( table . rows ( ) ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:1>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> def test_task_succeeded ( self ) : <EOL> state = EventsState ( ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> events = [ Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , name = '<STR_LIT>' , <EOL> args = '<STR_LIT>' , kwargs = "<STR_LIT>" , <EOL> retries = <NUM_LIT:0> , eta = None , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , result = '<STR_LIT:4>' , <EOL> runtime = <NUM_LIT> , hostname = '<STR_LIT>' ) ] <EOL> for i , e in enumerate ( events ) : <EOL> e [ '<STR_LIT>' ] = i <EOL> e [ '<STR_LIT>' ] = time . time ( ) <EOL> state . event ( e ) <EOL> self . app . events . state = state <EOL> r = self . get ( '<STR_LIT>' ) <EOL> table = HtmlTableParser ( ) <EOL> table . parse ( str ( r . body ) ) <EOL> self . assertEqual ( <NUM_LIT:200> , r . code ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( table . rows ( ) ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:1>' , '<STR_LIT:0>' , '<STR_LIT:1>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> def test_task_failed ( self ) : <EOL> state = EventsState ( ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> events = [ Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , name = '<STR_LIT>' , <EOL> args = '<STR_LIT>' , kwargs = "<STR_LIT>" , <EOL> retries = <NUM_LIT:0> , eta = None , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , exception = "<STR_LIT>" , <EOL> traceback = '<STR_LIT>' , hostname = '<STR_LIT>' ) ] <EOL> for i , e in enumerate ( events ) : <EOL> e [ '<STR_LIT>' ] = i <EOL> e [ '<STR_LIT>' ] = time . time ( ) <EOL> state . event ( e ) <EOL> self . app . events . state = state <EOL> r = self . get ( '<STR_LIT>' ) <EOL> table = HtmlTableParser ( ) <EOL> table . parse ( str ( r . body ) ) <EOL> self . assertEqual ( <NUM_LIT:200> , r . code ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( table . rows ( ) ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:1>' , '<STR_LIT:1>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> def test_task_retried ( self ) : <EOL> state = EventsState ( ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> events = [ Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , name = '<STR_LIT>' , <EOL> args = '<STR_LIT>' , kwargs = "<STR_LIT>" , <EOL> retries = <NUM_LIT:0> , eta = None , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , exception = "<STR_LIT>" , <EOL> traceback = '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , uuid = '<STR_LIT>' , exception = "<STR_LIT>" , <EOL> traceback = '<STR_LIT>' , hostname = '<STR_LIT>' ) ] <EOL> for i , e in enumerate ( events ) : <EOL> e [ '<STR_LIT>' ] = i <EOL> e [ '<STR_LIT>' ] = time . time ( ) <EOL> state . event ( e ) <EOL> self . app . events . state = state <EOL> r = self . get ( '<STR_LIT>' ) <EOL> table = HtmlTableParser ( ) <EOL> table . parse ( str ( r . body ) ) <EOL> self . assertEqual ( <NUM_LIT:200> , r . code ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( table . rows ( ) ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:1>' , '<STR_LIT:1>' , '<STR_LIT:0>' , '<STR_LIT:1>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> def test_tasks ( self ) : <EOL> state = EventsState ( ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> state . get_or_create_worker ( '<STR_LIT>' ) <EOL> events = [ Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) , <EOL> Event ( '<STR_LIT>' , hostname = '<STR_LIT>' ) ] <EOL> for i in range ( <NUM_LIT:100> ) : <EOL> events += task_succeeded_events ( worker = '<STR_LIT>' ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> events += task_succeeded_events ( worker = '<STR_LIT>' ) <EOL> for i in range ( <NUM_LIT> ) : <EOL> events += task_failed_events ( worker = '<STR_LIT>' ) <EOL> for i , e in enumerate ( events ) : <EOL> e [ '<STR_LIT>' ] = i <EOL> e [ '<STR_LIT>' ] = time . time ( ) <EOL> state . event ( e ) <EOL> self . app . events . state = state <EOL> r = self . get ( '<STR_LIT>' ) <EOL> table = HtmlTableParser ( ) <EOL> table . parse ( str ( r . body ) ) <EOL> self . assertEqual ( <NUM_LIT:200> , r . code ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( table . rows ( ) ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:100>' , '<STR_LIT:0>' , '<STR_LIT:100>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT:True>' , '<STR_LIT:0>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:0>' , None ] , <EOL> table . get_row ( '<STR_LIT>' ) ) </s>
