<s> """<STR_LIT>""" <EOL> import os <EOL> from grr . client . client_actions import searching <EOL> from grr . lib import action_mocks <EOL> from grr . lib import aff4 <EOL> from grr . lib import data_store <EOL> from grr . lib import flags <EOL> from grr . lib import rdfvalue <EOL> from grr . lib import test_lib <EOL> from grr . lib . flows . general import grep <EOL> from grr . lib . rdfvalues import client as rdf_client <EOL> from grr . lib . rdfvalues import paths as rdf_paths <EOL> from grr . lib . rdfvalues import standard as rdf_standard <EOL> class GrepTests ( test_lib . FlowTestsBaseclass ) : <EOL> pass <EOL> class TestSearchFileContentWithFixture ( GrepTests ) : <EOL> def FlushVFSCache ( self ) : <EOL> test_lib . ClientVFSHandlerFixture . cache = { } <EOL> def CreateFile ( self , filename , data ) : <EOL> self . FlushVFSCache ( ) <EOL> test_lib . client_fixture . VFS . append ( <EOL> ( filename , <EOL> ( "<STR_LIT>" , <EOL> { "<STR_LIT>" : ( <EOL> "<STR_LIT:\n>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( filename , data ) ) , <EOL> "<STR_LIT>" : len ( data ) } ) ) ) <EOL> def DeleteFile ( self , filename ) : <EOL> self . FlushVFSCache ( ) <EOL> test_lib . client_fixture . VFS = [ path for path in test_lib . client_fixture . VFS <EOL> if path [ <NUM_LIT:0> ] != filename ] <EOL> def setUp ( self ) : <EOL> super ( TestSearchFileContentWithFixture , self ) . setUp ( ) <EOL> self . client_mock = action_mocks . ActionMock ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . vfs_overrider = test_lib . VFSOverrider ( <EOL> rdf_paths . PathSpec . PathType . OS , test_lib . ClientVFSHandlerFixture ) <EOL> self . vfs_overrider . Start ( ) <EOL> def tearDown ( self ) : <EOL> super ( TestSearchFileContentWithFixture , self ) . tearDown ( ) <EOL> self . vfs_overrider . Stop ( ) <EOL> def testNormalGrep ( self ) : <EOL> output_path = "<STR_LIT>" <EOL> grepspec = rdf_client . BareGrepSpec ( mode = rdf_client . GrepSpec . Mode . FIRST_HIT , <EOL> literal = "<STR_LIT:hello>" ) <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , client_id = self . client_id , <EOL> paths = [ "<STR_LIT>" ] , pathtype = rdf_paths . PathSpec . PathType . OS , <EOL> token = self . token , output = output_path , grep = grepspec ) : <EOL> pass <EOL> fd = aff4 . FACTORY . Open ( self . client_id . Add ( output_path ) , token = self . token ) <EOL> self . assertEqual ( len ( fd ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( fd [ <NUM_LIT:0> ] . offset , <NUM_LIT:3> ) <EOL> self . assertEqual ( fd [ <NUM_LIT:0> ] , "<STR_LIT>" ) <EOL> self . assertEqual ( fd [ <NUM_LIT:0> ] . length , <NUM_LIT> ) <EOL> def testMultipleHits ( self ) : <EOL> filename = "<STR_LIT>" <EOL> data = "<STR_LIT>" * <NUM_LIT:100> <EOL> self . CreateFile ( filename , data ) <EOL> output_path = "<STR_LIT>" <EOL> grepspec = rdf_client . BareGrepSpec ( mode = rdf_client . GrepSpec . Mode . ALL_HITS , <EOL> literal = "<STR_LIT>" ) <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , client_id = self . client_id , <EOL> paths = [ "<STR_LIT>" ] , <EOL> pathtype = rdf_paths . PathSpec . PathType . OS , <EOL> grep = grepspec , token = self . token , output = output_path ) : <EOL> pass <EOL> fd = aff4 . FACTORY . Open ( self . client_id . Add ( output_path ) , token = self . token ) <EOL> self . assertEqual ( len ( fd ) , <NUM_LIT:100> ) <EOL> self . assertEqual ( fd [ <NUM_LIT:15> ] . offset , <NUM_LIT> ) <EOL> self . assertEqual ( fd [ <NUM_LIT> ] , "<STR_LIT>" ) <EOL> self . assertEqual ( fd [ <NUM_LIT> ] , "<STR_LIT>" ) <EOL> self . DeleteFile ( filename ) <EOL> def testPatternAtBufsize ( self ) : <EOL> old_size = searching . Grep . BUFF_SIZE <EOL> try : <EOL> searching . Grep . BUFF_SIZE = <NUM_LIT> <EOL> filename = "<STR_LIT>" <EOL> data = "<STR_LIT:X>" * ( searching . Grep . BUFF_SIZE - len ( "<STR_LIT>" ) ) + "<STR_LIT>" + "<STR_LIT:X>" * <NUM_LIT:1000> <EOL> self . CreateFile ( filename , data ) <EOL> output_path = "<STR_LIT>" <EOL> output_urn = self . client_id . Add ( output_path ) <EOL> data_store . DB . DeleteSubject ( output_urn , token = self . token ) <EOL> grepspec = rdf_client . BareGrepSpec ( <EOL> mode = rdf_client . GrepSpec . Mode . FIRST_HIT , <EOL> literal = "<STR_LIT>" ) <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , client_id = self . client_id , <EOL> paths = [ "<STR_LIT>" ] , <EOL> pathtype = rdf_paths . PathSpec . PathType . OS , <EOL> token = self . token , output = output_path , grep = grepspec ) : <EOL> pass <EOL> fd = aff4 . FACTORY . Open ( output_urn , token = self . token ) <EOL> self . assertEqual ( len ( fd ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( fd [ <NUM_LIT:0> ] . offset , searching . Grep . BUFF_SIZE - len ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( fd [ <NUM_LIT:0> ] . length , <NUM_LIT> ) <EOL> self . DeleteFile ( filename ) <EOL> finally : <EOL> searching . Grep . BUFF_SIZE = old_size <EOL> class TestSearchFileContent ( GrepTests ) : <EOL> def testSearchFileContents ( self ) : <EOL> pattern = "<STR_LIT>" <EOL> client_mock = action_mocks . ActionMock ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> path = os . path . join ( self . base_path , pattern ) <EOL> args = grep . SearchFileContentArgs ( <EOL> paths = [ path ] , pathtype = rdf_paths . PathSpec . PathType . OS ) <EOL> args . grep . literal = rdf_standard . LiteralExpression ( <EOL> "<STR_LIT>" ) <EOL> args . grep . mode = rdf_client . GrepSpec . Mode . ALL_HITS <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , client_mock , client_id = self . client_id , <EOL> output = "<STR_LIT>" , args = args , token = self . token ) : <EOL> pass <EOL> fd = aff4 . FACTORY . Open ( <EOL> rdfvalue . RDFURN ( self . client_id ) . Add ( "<STR_LIT>" ) , <EOL> token = self . token ) <EOL> self . assertEqual ( len ( fd ) , <NUM_LIT:1> ) <EOL> first = fd [ <NUM_LIT:0> ] <EOL> self . assertEqual ( first . offset , <NUM_LIT> ) <EOL> self . assertEqual ( first . data , <EOL> "<STR_LIT>" ) <EOL> def testSearchFileContentsNoGrep ( self ) : <EOL> """<STR_LIT>""" <EOL> pattern = "<STR_LIT>" <EOL> client_mock = action_mocks . ActionMock ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> path = os . path . join ( self . base_path , pattern ) <EOL> args = grep . SearchFileContentArgs ( paths = [ path ] ) <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , client_mock , client_id = self . client_id , <EOL> output = "<STR_LIT>" , args = args , token = self . token ) : <EOL> pass <EOL> fd = aff4 . FACTORY . Open ( <EOL> rdfvalue . RDFURN ( self . client_id ) . Add ( "<STR_LIT>" ) , <EOL> token = self . token ) <EOL> self . assertEqual ( len ( fd ) , <NUM_LIT:3> ) <EOL> def testSearchFileContentDownload ( self ) : <EOL> pattern = "<STR_LIT>" <EOL> client_mock = action_mocks . ActionMock ( "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> path = os . path . join ( self . base_path , pattern ) <EOL> args = grep . SearchFileContentArgs ( paths = [ path ] , <EOL> also_download = True ) <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , client_mock , client_id = self . client_id , <EOL> output = "<STR_LIT>" , args = args , token = self . token ) : <EOL> pass <EOL> fd = aff4 . FACTORY . Open ( <EOL> rdfvalue . RDFURN ( self . client_id ) . Add ( "<STR_LIT>" ) , <EOL> token = self . token ) <EOL> self . assertEqual ( len ( fd ) , <NUM_LIT:3> ) <EOL> for log in aff4 . FACTORY . Open ( <EOL> rdfvalue . RDFURN ( self . client_id ) . Add ( "<STR_LIT>" ) . Add ( <EOL> self . base_path ) . Add ( "<STR_LIT>" ) , <EOL> token = self . token ) . OpenChildren ( ) : <EOL> self . assertTrue ( isinstance ( log , aff4 . VFSBlobImage ) ) <EOL> self . assertGreater ( len ( log ) , <NUM_LIT:0> ) <EOL> def main ( argv ) : <EOL> test_lib . GrrTestProgram ( argv = argv ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> flags . StartMain ( main ) </s>
