<s> """<STR_LIT>""" <EOL> import os <EOL> import StringIO <EOL> import tarfile <EOL> import zipfile <EOL> import yaml <EOL> from grr . gui import api_test_lib <EOL> from grr . gui . api_plugins import flow as flow_plugin <EOL> from grr . lib import action_mocks <EOL> from grr . lib import aff4 <EOL> from grr . lib import flags <EOL> from grr . lib import flow <EOL> from grr . lib import flow_runner <EOL> from grr . lib import output_plugin <EOL> from grr . lib import rdfvalue <EOL> from grr . lib import test_lib <EOL> from grr . lib import throttle <EOL> from grr . lib import type_info <EOL> from grr . lib import utils <EOL> from grr . lib . aff4_objects import collects as aff4_collections <EOL> from grr . lib . flows . general import administrative <EOL> from grr . lib . flows . general import discovery <EOL> from grr . lib . flows . general import file_finder <EOL> from grr . lib . flows . general import processes <EOL> from grr . lib . flows . general import transfer <EOL> from grr . lib . hunts import standard_test <EOL> from grr . lib . output_plugins import email_plugin <EOL> from grr . lib . rdfvalues import client as rdf_client <EOL> from grr . lib . rdfvalues import paths as rdf_paths <EOL> class ApiGetFlowStatusHandlerTest ( test_lib . GRRBaseTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( ApiGetFlowStatusHandlerTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> self . handler = flow_plugin . ApiGetFlowStatusHandler ( ) <EOL> def testIsDisabledByDefault ( self ) : <EOL> self . assertFalse ( self . handler . enabled_by_default ) <EOL> def testParameterValidation ( self ) : <EOL> """<STR_LIT>""" <EOL> bad_flowid = flow_plugin . ApiGetFlowStatusArgs ( <EOL> client_id = self . client_id . Basename ( ) , flow_id = "<STR_LIT>" ) <EOL> with self . assertRaises ( ValueError ) : <EOL> self . handler . Render ( bad_flowid , token = self . token ) <EOL> with self . assertRaises ( type_info . TypeValueError ) : <EOL> flow_plugin . ApiGetFlowStatusArgs ( <EOL> client_id = "<STR_LIT>" , flow_id = "<STR_LIT>" ) <EOL> class ApiGetFlowStatusHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def Run ( self ) : <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> client_urn = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> with aff4 . FACTORY . Open ( client_urn , mode = "<STR_LIT>" , <EOL> token = self . token ) as client_obj : <EOL> client_obj . DeleteAttribute ( client_obj . Schema . CERT ) <EOL> flow_id = flow . GRRFlow . StartFlow ( <EOL> flow_name = discovery . Interrogate . __name__ , client_id = client_urn , <EOL> token = self . token ) <EOL> flow_obj = aff4 . FACTORY . Open ( flow_id , aff4_type = flow . GRRFlow . __name__ , <EOL> token = self . token ) <EOL> flow_state = flow_obj . Get ( flow_obj . Schema . FLOW_STATE ) <EOL> with aff4 . FACTORY . Create ( <EOL> flow_state . context . output_urn , <EOL> aff4_type = aff4_collections . RDFValueCollection . __name__ , <EOL> token = self . token ) as collection : <EOL> collection . Add ( rdf_client . ClientSummary ( ) ) <EOL> self . Check ( "<STR_LIT:GET>" , "<STR_LIT>" % ( client_urn . Basename ( ) , <EOL> flow_id . Basename ( ) ) , <EOL> replace = { flow_id . Basename ( ) : "<STR_LIT>" } ) <EOL> class ApiGetFlowHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def Run ( self ) : <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> client_urn = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> with aff4 . FACTORY . Open ( client_urn , mode = "<STR_LIT>" , <EOL> token = self . token ) as client_obj : <EOL> client_obj . DeleteAttribute ( client_obj . Schema . CERT ) <EOL> flow_id = flow . GRRFlow . StartFlow ( <EOL> flow_name = discovery . Interrogate . __name__ , client_id = client_urn , <EOL> token = self . token ) <EOL> self . Check ( "<STR_LIT:GET>" , "<STR_LIT>" % ( client_urn . Basename ( ) , <EOL> flow_id . Basename ( ) ) , <EOL> replace = { flow_id . Basename ( ) : "<STR_LIT>" } ) <EOL> class ApiListClientFlowsHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def Run ( self ) : <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> client_urn = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_id_1 = flow . GRRFlow . StartFlow ( <EOL> flow_name = discovery . Interrogate . __name__ , client_id = client_urn , <EOL> output = "<STR_LIT>" , token = self . token ) <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_id_2 = flow . GRRFlow . StartFlow ( <EOL> flow_name = processes . ListProcesses . __name__ , client_id = client_urn , <EOL> output = "<STR_LIT>" , token = self . token ) <EOL> self . Check ( "<STR_LIT:GET>" , "<STR_LIT>" % client_urn . Basename ( ) , <EOL> replace = { flow_id_1 . Basename ( ) : "<STR_LIT>" , <EOL> flow_id_2 . Basename ( ) : "<STR_LIT>" } ) <EOL> class ApiListFlowResultsHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ApiListFlowResultsHandlerRegressionTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def Run ( self ) : <EOL> runner_args = flow_runner . FlowRunnerArgs ( <EOL> flow_name = transfer . GetFile . __name__ ) <EOL> flow_args = transfer . GetFileArgs ( <EOL> pathspec = rdf_paths . PathSpec ( <EOL> path = "<STR_LIT>" , <EOL> pathtype = rdf_paths . PathSpec . PathType . OS ) ) <EOL> client_mock = test_lib . SampleHuntMock ( ) <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_urn = flow . GRRFlow . StartFlow ( client_id = self . client_id , <EOL> args = flow_args , <EOL> runner_args = runner_args , <EOL> token = self . token ) <EOL> for _ in test_lib . TestFlowHelper ( flow_urn , client_mock = client_mock , <EOL> client_id = self . client_id , <EOL> token = self . token ) : <EOL> pass <EOL> self . Check ( "<STR_LIT:GET>" , <EOL> "<STR_LIT>" % ( self . client_id . Basename ( ) , <EOL> flow_urn . Basename ( ) ) , <EOL> replace = { flow_urn . Basename ( ) : "<STR_LIT>" } ) <EOL> class ApiListFlowLogsHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ApiListFlowLogsHandlerRegressionTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def Run ( self ) : <EOL> flow_urn = flow . GRRFlow . StartFlow ( <EOL> flow_name = processes . ListProcesses . __name__ , <EOL> client_id = self . client_id , <EOL> token = self . token ) <EOL> with aff4 . FACTORY . Open ( flow_urn , mode = "<STR_LIT>" , token = self . token ) as flow_obj : <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_obj . Log ( "<STR_LIT>" ) <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_obj . Log ( "<STR_LIT>" ) <EOL> base_url = "<STR_LIT>" % ( self . client_id . Basename ( ) , <EOL> flow_urn . Basename ( ) ) <EOL> replace = { flow_urn . Basename ( ) : "<STR_LIT>" } <EOL> self . Check ( "<STR_LIT:GET>" , base_url , replace = replace ) <EOL> self . Check ( "<STR_LIT:GET>" , base_url + "<STR_LIT>" , replace = replace ) <EOL> self . Check ( "<STR_LIT:GET>" , base_url + "<STR_LIT>" , replace = replace ) <EOL> class ApiGetFlowResultsExportCommandHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ApiGetFlowResultsExportCommandHandlerRegressionTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def Run ( self ) : <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_urn = flow . GRRFlow . StartFlow ( <EOL> flow_name = processes . ListProcesses . __name__ , <EOL> client_id = self . client_id , <EOL> token = self . token ) <EOL> self . Check ( "<STR_LIT:GET>" , <EOL> "<STR_LIT>" % ( <EOL> self . client_id . Basename ( ) , flow_urn . Basename ( ) ) , <EOL> replace = { flow_urn . Basename ( ) : "<STR_LIT>" } ) <EOL> class ApiListFlowOutputPluginsHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ApiListFlowOutputPluginsHandlerRegressionTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def Run ( self ) : <EOL> email_descriptor = output_plugin . OutputPluginDescriptor ( <EOL> plugin_name = email_plugin . EmailOutputPlugin . __name__ , <EOL> plugin_args = email_plugin . EmailOutputPluginArgs ( <EOL> email_address = "<STR_LIT>" , emails_limit = <NUM_LIT> ) ) <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_urn = flow . GRRFlow . StartFlow ( <EOL> flow_name = processes . ListProcesses . __name__ , <EOL> client_id = self . client_id , <EOL> output_plugins = [ email_descriptor ] , <EOL> token = self . token ) <EOL> self . Check ( "<STR_LIT:GET>" , "<STR_LIT>" % ( <EOL> self . client_id . Basename ( ) , flow_urn . Basename ( ) ) , <EOL> replace = { flow_urn . Basename ( ) : "<STR_LIT>" } ) <EOL> class DummyFlowWithSingleReply ( flow . GRRFlow ) : <EOL> """<STR_LIT>""" <EOL> @ flow . StateHandler ( ) <EOL> def Start ( self , unused_response = None ) : <EOL> self . CallState ( next_state = "<STR_LIT>" ) <EOL> @ flow . StateHandler ( ) <EOL> def SendSomething ( self , unused_response = None ) : <EOL> self . SendReply ( rdfvalue . RDFString ( "<STR_LIT>" ) ) <EOL> class ApiListFlowOutputPluginLogsHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ApiListFlowOutputPluginLogsHandlerRegressionTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def Run ( self ) : <EOL> email_descriptor = output_plugin . OutputPluginDescriptor ( <EOL> plugin_name = email_plugin . EmailOutputPlugin . __name__ , <EOL> plugin_args = email_plugin . EmailOutputPluginArgs ( <EOL> email_address = "<STR_LIT>" , emails_limit = <NUM_LIT> ) ) <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_urn = flow . GRRFlow . StartFlow ( <EOL> flow_name = DummyFlowWithSingleReply . __name__ , <EOL> client_id = self . client_id , <EOL> output_plugins = [ email_descriptor ] , <EOL> token = self . token ) <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> for _ in test_lib . TestFlowHelper ( flow_urn , token = self . token ) : <EOL> pass <EOL> self . Check ( "<STR_LIT:GET>" , "<STR_LIT>" <EOL> "<STR_LIT>" % ( self . client_id . Basename ( ) , <EOL> flow_urn . Basename ( ) ) , <EOL> replace = { flow_urn . Basename ( ) : "<STR_LIT>" } ) <EOL> class ApiListFlowOutputPluginErrorsHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ApiListFlowOutputPluginErrorsHandlerRegressionTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def Run ( self ) : <EOL> failing_descriptor = output_plugin . OutputPluginDescriptor ( <EOL> plugin_name = standard_test . FailingDummyHuntOutputPlugin . __name__ ) <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_urn = flow . GRRFlow . StartFlow ( <EOL> flow_name = DummyFlowWithSingleReply . __name__ , <EOL> client_id = self . client_id , <EOL> output_plugins = [ failing_descriptor ] , <EOL> token = self . token ) <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> for _ in test_lib . TestFlowHelper ( flow_urn , token = self . token ) : <EOL> pass <EOL> self . Check ( "<STR_LIT:GET>" , "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> self . client_id . Basename ( ) , flow_urn . Basename ( ) ) , <EOL> replace = { flow_urn . Basename ( ) : "<STR_LIT>" } ) <EOL> class ApiCreateFlowHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ApiCreateFlowHandlerRegressionTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def Run ( self ) : <EOL> def ReplaceFlowId ( ) : <EOL> flows_dir_fd = aff4 . FACTORY . Open ( self . client_id . Add ( "<STR_LIT>" ) , <EOL> token = self . token ) <EOL> flow_urn = list ( flows_dir_fd . ListChildren ( ) ) [ <NUM_LIT:0> ] <EOL> return { flow_urn . Basename ( ) : "<STR_LIT>" } <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> self . Check ( "<STR_LIT:POST>" , <EOL> "<STR_LIT>" % self . client_id . Basename ( ) , <EOL> { "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : processes . ListProcesses . __name__ , <EOL> "<STR_LIT:args>" : { <EOL> "<STR_LIT>" : "<STR_LIT:.>" , <EOL> "<STR_LIT>" : True <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : False , <EOL> } , <EOL> } <EOL> } , replace = ReplaceFlowId ) <EOL> class ApiCancelFlowHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ApiCancelFlowHandlerRegressionTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def Run ( self ) : <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> flow_urn = flow . GRRFlow . StartFlow ( <EOL> flow_name = processes . ListProcesses . __name__ , <EOL> client_id = self . client_id , <EOL> token = self . token ) <EOL> self . Check ( "<STR_LIT:POST>" , "<STR_LIT>" % ( <EOL> self . client_id . Basename ( ) , flow_urn . Basename ( ) ) , <EOL> replace = { flow_urn . Basename ( ) : "<STR_LIT>" } ) <EOL> class ApiListFlowDescriptorsHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> """<STR_LIT>""" <EOL> handler = "<STR_LIT>" <EOL> def Run ( self ) : <EOL> with utils . Stubber ( flow . GRRFlow , "<STR_LIT>" , { <EOL> "<STR_LIT>" : processes . ListProcesses , <EOL> "<STR_LIT>" : file_finder . FileFinder , <EOL> "<STR_LIT>" : administrative . RunReport <EOL> } ) : <EOL> self . CreateAdminUser ( "<STR_LIT:test>" ) <EOL> self . Check ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . Check ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> self . Check ( "<STR_LIT:GET>" , "<STR_LIT>" ) <EOL> class ApiStartGetFileOperationHandlerRegressionTest ( <EOL> api_test_lib . ApiCallHandlerRegressionTest ) : <EOL> handler = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> super ( ApiStartGetFileOperationHandlerRegressionTest , self ) . setUp ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def Run ( self ) : <EOL> def ReplaceFlowId ( ) : <EOL> flows_dir_fd = aff4 . FACTORY . Open ( self . client_id . Add ( "<STR_LIT>" ) , <EOL> token = self . token ) <EOL> flow_urn = list ( flows_dir_fd . ListChildren ( ) ) [ <NUM_LIT:0> ] <EOL> return { flow_urn . Basename ( ) : "<STR_LIT>" } <EOL> with test_lib . FakeTime ( <NUM_LIT> ) : <EOL> self . Check ( <EOL> "<STR_LIT:POST>" , <EOL> "<STR_LIT>" % self . client_id . Basename ( ) , <EOL> { "<STR_LIT>" : self . client_id . Basename ( ) , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] } , <EOL> replace = ReplaceFlowId ) <EOL> class ApiStartGetFileOperationHandlerTest ( test_lib . GRRBaseTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( ApiStartGetFileOperationHandlerTest , self ) . setUp ( ) <EOL> self . client_ids = self . SetupClients ( <NUM_LIT:4> ) <EOL> self . handler = flow_plugin . ApiStartGetFileOperationHandler ( ) <EOL> def testClientLookup ( self ) : <EOL> """<STR_LIT>""" <EOL> args = flow_plugin . ApiStartGetFileOperationArgs ( <EOL> hostname = "<STR_LIT>" , paths = [ "<STR_LIT>" ] ) <EOL> result = self . handler . Render ( args , token = self . token ) <EOL> self . assertIn ( "<STR_LIT>" , result [ "<STR_LIT>" ] ) <EOL> def testThrottle ( self ) : <EOL> """<STR_LIT>""" <EOL> args = flow_plugin . ApiStartGetFileOperationArgs ( <EOL> hostname = "<STR_LIT>" , paths = [ "<STR_LIT>" ] ) <EOL> result = self . handler . Render ( args , token = self . token ) <EOL> self . assertIn ( "<STR_LIT>" , result [ "<STR_LIT>" ] ) <EOL> with self . assertRaises ( throttle . ErrorFlowDuplicate ) : <EOL> self . handler . Render ( args , token = self . token ) <EOL> class ApiGetFlowFilesArchiveHandlerTest ( test_lib . GRRBaseTest ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( ApiGetFlowFilesArchiveHandlerTest , self ) . setUp ( ) <EOL> self . handler = flow_plugin . ApiGetFlowFilesArchiveHandler ( ) <EOL> self . client_id = self . SetupClients ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> self . flow_urn = flow . GRRFlow . StartFlow ( <EOL> flow_name = file_finder . FileFinder . __name__ , <EOL> client_id = self . client_id , <EOL> paths = [ os . path . join ( self . base_path , "<STR_LIT>" ) ] , <EOL> action = file_finder . FileFinderAction ( action_type = "<STR_LIT>" ) , <EOL> token = self . token ) <EOL> action_mock = action_mocks . ActionMock ( <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> for _ in test_lib . TestFlowHelper ( <EOL> self . flow_urn , action_mock , client_id = self . client_id , <EOL> token = self . token ) : <EOL> pass <EOL> def testGeneratesZipArchive ( self ) : <EOL> result = self . handler . Handle ( flow_plugin . ApiGetFlowFilesArchiveArgs ( <EOL> client_id = self . client_id , flow_id = self . flow_urn . Basename ( ) , <EOL> archive_format = "<STR_LIT>" ) , token = self . token ) <EOL> out_fd = StringIO . StringIO ( ) <EOL> for chunk in result . GenerateContent ( ) : <EOL> out_fd . write ( chunk ) <EOL> zip_fd = zipfile . ZipFile ( out_fd , "<STR_LIT:r>" ) <EOL> manifest = None <EOL> for name in zip_fd . namelist ( ) : <EOL> if name . endswith ( "<STR_LIT>" ) : <EOL> manifest = yaml . safe_load ( zip_fd . read ( name ) ) <EOL> self . assertEqual ( manifest [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( manifest [ "<STR_LIT>" ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( manifest [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( manifest [ "<STR_LIT>" ] , <NUM_LIT:0> ) <EOL> def testGeneratesTarGzArchive ( self ) : <EOL> result = self . handler . Handle ( flow_plugin . ApiGetFlowFilesArchiveArgs ( <EOL> client_id = self . client_id , flow_id = self . flow_urn . Basename ( ) , <EOL> archive_format = "<STR_LIT>" ) , token = self . token ) <EOL> with utils . TempDirectory ( ) as temp_dir : <EOL> tar_path = os . path . join ( temp_dir , "<STR_LIT>" ) <EOL> with open ( tar_path , "<STR_LIT:w>" ) as fd : <EOL> for chunk in result . GenerateContent ( ) : <EOL> fd . write ( chunk ) <EOL> with tarfile . open ( tar_path ) as tar_fd : <EOL> tar_fd . extractall ( path = temp_dir ) <EOL> manifest_file_path = None <EOL> for parent , _ , files in os . walk ( temp_dir ) : <EOL> if "<STR_LIT>" in files : <EOL> manifest_file_path = os . path . join ( parent , "<STR_LIT>" ) <EOL> break <EOL> self . assertTrue ( manifest_file_path ) <EOL> with open ( manifest_file_path ) as fd : <EOL> manifest = yaml . safe_load ( fd . read ( ) ) <EOL> self . assertEqual ( manifest [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( manifest [ "<STR_LIT>" ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( manifest [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( manifest [ "<STR_LIT>" ] , <NUM_LIT:0> ) <EOL> def main ( argv ) : <EOL> test_lib . main ( argv ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> flags . StartMain ( main ) </s>
