<s> """<STR_LIT>""" <EOL> import os <EOL> import logging <EOL> from grr . endtoend_tests import base <EOL> from grr . lib import aff4 <EOL> from grr . lib import config_lib <EOL> from grr . lib import maintenance_utils <EOL> class TestGetClientStats ( base . AutomatedTest ) : <EOL> """<STR_LIT>""" <EOL> platforms = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> test_output_path = "<STR_LIT>" <EOL> flow = "<STR_LIT>" <EOL> def CheckFlow ( self ) : <EOL> aff4 . FACTORY . Flush ( ) <EOL> client_stats = aff4 . FACTORY . Open ( self . client_id . Add ( self . test_output_path ) , <EOL> token = self . token ) <EOL> self . assertIsInstance ( client_stats , aff4 . ClientStats ) <EOL> stats = list ( client_stats . Get ( client_stats . Schema . STATS ) ) <EOL> self . assertGreater ( len ( stats ) , <NUM_LIT:0> ) <EOL> entry = stats [ <NUM_LIT:0> ] <EOL> self . assertGreater ( entry . RSS_size , <NUM_LIT:0> ) <EOL> self . assertGreater ( entry . VMS_size , <NUM_LIT:0> ) <EOL> self . assertGreater ( entry . boot_time , <NUM_LIT:0> ) <EOL> self . assertGreater ( entry . bytes_received , <NUM_LIT:0> ) <EOL> self . assertGreater ( entry . bytes_sent , <NUM_LIT:0> ) <EOL> self . assertGreater ( entry . memory_percent , <NUM_LIT:0> ) <EOL> self . assertGreater ( len ( list ( entry . cpu_samples ) ) , <NUM_LIT:0> ) <EOL> if not list ( entry . io_samples ) : <EOL> logging . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class TestLaunchBinaries ( base . ClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> platforms = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> flow = "<STR_LIT>" <EOL> filenames = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:hello>" } <EOL> ds_names = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:hello>" } <EOL> limit = None <EOL> def __init__ ( self , ** kwargs ) : <EOL> super ( TestLaunchBinaries , self ) . __init__ ( ** kwargs ) <EOL> self . context = [ "<STR_LIT>" % self . platform . title ( ) ] <EOL> self . binary = config_lib . CONFIG . Get ( <EOL> "<STR_LIT>" , context = self . context ) . Add ( <EOL> "<STR_LIT>" % self . ds_names [ self . platform ] ) <EOL> self . args = dict ( binary = self . binary ) <EOL> try : <EOL> aff4 . FACTORY . Open ( self . binary , aff4_type = "<STR_LIT>" , <EOL> token = self . token ) <EOL> except IOError : <EOL> print "<STR_LIT>" <EOL> source = os . path . join ( config_lib . CONFIG [ "<STR_LIT>" ] , <EOL> self . filenames [ self . platform ] ) <EOL> if not os . path . exists ( source ) : <EOL> self . fail ( "<STR_LIT>" % source ) <EOL> maintenance_utils . UploadSignedConfigBlob ( <EOL> open ( source , "<STR_LIT:rb>" ) . read ( ) , aff4_path = self . binary , <EOL> client_context = self . context , token = self . token , limit = self . limit ) <EOL> def CheckFlow ( self ) : <EOL> fd = aff4 . FACTORY . Open ( self . session_id , age = aff4 . ALL_TIMES , <EOL> token = self . token ) <EOL> logs = "<STR_LIT:\n>" . join ( <EOL> [ x . log_message for x in fd . GetLog ( ) ] ) <EOL> self . assertTrue ( "<STR_LIT>" in logs ) <EOL> class TestLaunchChunkedBinaries ( TestLaunchBinaries ) : <EOL> limit = <NUM_LIT> <EOL> ds_names = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } </s>
