<s> """<STR_LIT>""" <EOL> import gzip <EOL> import os <EOL> import subprocess <EOL> import time <EOL> from grr . client import client_utils_linux <EOL> from grr . client import client_utils_osx <EOL> from grr . client . client_actions import standard <EOL> from grr . client . components . rekall_support import rekall_types as rdf_rekall_types <EOL> from grr . lib import action_mocks <EOL> from grr . lib import aff4 <EOL> from grr . lib import artifact <EOL> from grr . lib import artifact_registry <EOL> from grr . lib import config_lib <EOL> from grr . lib import flags <EOL> from grr . lib import parsers <EOL> from grr . lib import rdfvalue <EOL> from grr . lib import test_lib <EOL> from grr . lib import utils <EOL> from grr . lib . flows . general import collectors <EOL> from grr . lib . rdfvalues import anomaly as rdf_anomaly <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 protodict as rdf_protodict <EOL> WMI_SAMPLE = [ <EOL> rdf_protodict . Dict ( { <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT:Name>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" <EOL> } ) , rdf_protodict . Dict ( { <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT:Name>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" <EOL> } ) , rdf_protodict . Dict ( { <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT:Name>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" , <EOL> u"<STR_LIT>" : u"<STR_LIT>" <EOL> } ) <EOL> ] <EOL> class TestCmdProcessor ( parsers . CommandParser ) : <EOL> output_types = [ "<STR_LIT>" ] <EOL> supported_artifacts = [ "<STR_LIT>" ] <EOL> def Parse ( self , cmd , args , stdout , stderr , return_val , time_taken , <EOL> knowledge_base ) : <EOL> _ = cmd , args , stdout , stderr , return_val , time_taken , knowledge_base <EOL> installed = rdf_client . SoftwarePackage . InstallState . INSTALLED <EOL> soft = rdf_client . SoftwarePackage ( name = "<STR_LIT>" , description = "<STR_LIT>" , <EOL> version = "<STR_LIT:1>" , architecture = "<STR_LIT>" , <EOL> install_state = installed ) <EOL> yield soft <EOL> soft = rdf_client . SoftwarePackage ( name = "<STR_LIT>" , description = "<STR_LIT>" , <EOL> version = "<STR_LIT:1>" , architecture = "<STR_LIT>" , <EOL> install_state = installed ) <EOL> yield soft <EOL> yield rdf_client . StatEntry ( ) <EOL> yield rdf_anomaly . Anomaly ( type = "<STR_LIT>" , <EOL> symptom = "<STR_LIT>" ) <EOL> class MultiProvideParser ( parsers . RegistryValueParser ) : <EOL> output_types = [ "<STR_LIT>" ] <EOL> supported_artifacts = [ "<STR_LIT>" ] <EOL> def Parse ( self , stat , knowledge_base ) : <EOL> _ = stat , knowledge_base <EOL> test_dict = { "<STR_LIT>" : rdfvalue . RDFString ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" : rdfvalue . RDFString ( "<STR_LIT>" ) } <EOL> yield rdf_protodict . Dict ( test_dict ) <EOL> class RekallMock ( action_mocks . MemoryClientMock ) : <EOL> def __init__ ( self , client_id , result_filename , * args , ** kwargs ) : <EOL> super ( RekallMock , self ) . __init__ ( * args , ** kwargs ) <EOL> self . result_filename = result_filename <EOL> self . client_id = client_id <EOL> def RekallAction ( self , _ ) : <EOL> ps_list_file = os . path . join ( config_lib . CONFIG [ "<STR_LIT>" ] , <EOL> self . result_filename ) <EOL> result = rdf_rekall_types . RekallResponse ( <EOL> json_messages = gzip . open ( ps_list_file ) . read ( <NUM_LIT> ) , <EOL> plugin = "<STR_LIT>" , <EOL> client_urn = self . client_id ) <EOL> return [ result , rdf_client . Iterator ( state = "<STR_LIT>" ) ] <EOL> class ArtifactTest ( test_lib . FlowTestsBaseclass ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( ArtifactTest , self ) . setUp ( ) <EOL> self . client_mock = action_mocks . ActionMock ( <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ) <EOL> def LoadTestArtifacts ( self ) : <EOL> """<STR_LIT>""" <EOL> artifact_registry . REGISTRY . AddFileSource ( os . path . join ( <EOL> config_lib . CONFIG [ "<STR_LIT>" ] , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> class MockClient ( action_mocks . MemoryClientMock ) : <EOL> def WmiQuery ( self , _ ) : <EOL> return WMI_SAMPLE <EOL> def MockClientMountPointsWithImage ( self , image_path , fs_type = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> def MockGetMountpoints ( ) : <EOL> return { "<STR_LIT:/>" : ( image_path , fs_type ) } <EOL> return utils . MultiStubber ( <EOL> ( client_utils_linux , "<STR_LIT>" , MockGetMountpoints ) , <EOL> ( client_utils_osx , "<STR_LIT>" , MockGetMountpoints ) , <EOL> ( standard , "<STR_LIT>" , utils . FastStore ( <NUM_LIT:100> ) ) ) <EOL> def RunCollectorAndGetCollection ( self , artifact_list , client_mock = None , <EOL> ** kw ) : <EOL> """<STR_LIT>""" <EOL> if client_mock is None : <EOL> client_mock = self . MockClient ( client_id = self . client_id ) <EOL> output_name = "<STR_LIT>" % int ( time . time ( ) ) <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , client_mock = client_mock , output = output_name , <EOL> client_id = self . client_id , artifact_list = artifact_list , <EOL> token = self . token , ** kw ) : <EOL> pass <EOL> output_urn = self . client_id . Add ( output_name ) <EOL> return aff4 . FACTORY . Open ( output_urn , aff4_type = "<STR_LIT>" , <EOL> token = self . token ) <EOL> class GRRArtifactTest ( ArtifactTest ) : <EOL> def testRDFMaps ( self ) : <EOL> """<STR_LIT>""" <EOL> for rdf_name , dat in artifact . GRRArtifactMappings . rdf_map . items ( ) : <EOL> _ , aff4_type , aff4_attribute , operator = dat <EOL> if operator not in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> raise artifact_registry . ArtifactDefinitionError ( <EOL> "<STR_LIT>" % <EOL> ( operator , rdf_name ) ) <EOL> if aff4_type not in aff4 . AFF4Object . classes : <EOL> raise artifact_registry . ArtifactDefinitionError ( <EOL> "<STR_LIT>" % <EOL> ( aff4_type , rdf_name ) ) <EOL> attr = getattr ( aff4 . AFF4Object . classes [ aff4_type ] . SchemaCls , <EOL> aff4_attribute ) ( ) <EOL> if not isinstance ( attr , rdfvalue . RDFValue ) : <EOL> raise artifact_registry . ArtifactDefinitionError ( <EOL> "<STR_LIT>" % <EOL> ( aff4_attribute , rdf_name ) ) <EOL> def testUploadArtifactYamlFileAndDumpToYaml ( self ) : <EOL> artifact_registry . REGISTRY . ClearRegistry ( ) <EOL> artifact_registry . REGISTRY . ClearSources ( ) <EOL> artifact_registry . REGISTRY . _CheckDirty ( ) <EOL> try : <EOL> test_artifacts_file = os . path . join ( <EOL> config_lib . CONFIG [ "<STR_LIT>" ] , "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> filecontent = open ( test_artifacts_file ) . read ( ) <EOL> artifact . UploadArtifactYamlFile ( filecontent , token = self . token ) <EOL> loaded_artifacts = artifact_registry . REGISTRY . GetArtifacts ( ) <EOL> self . assertEqual ( len ( loaded_artifacts ) , <NUM_LIT> ) <EOL> self . assertIn ( "<STR_LIT>" , [ a . name for a in loaded_artifacts ] ) <EOL> yaml_data = artifact_registry . REGISTRY . DumpArtifactsToYaml ( ) <EOL> for snippet in [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] : <EOL> self . assertIn ( snippet , yaml_data ) <EOL> finally : <EOL> artifact . ArtifactLoader ( ) . RunOnce ( ) <EOL> def testUploadArtifactYamlFileMissingDoc ( self ) : <EOL> content = """<STR_LIT>""" <EOL> with self . assertRaises ( artifact_registry . ArtifactDefinitionError ) : <EOL> artifact . UploadArtifactYamlFile ( content , token = self . token ) <EOL> def testUploadArtifactYamlFileBadList ( self ) : <EOL> content = """<STR_LIT>""" <EOL> with self . assertRaises ( artifact_registry . ArtifactDefinitionError ) : <EOL> artifact . UploadArtifactYamlFile ( content , token = self . token ) <EOL> def testUploadArtifactYamlFileMissingNamesAttribute ( self ) : <EOL> content = """<STR_LIT>""" <EOL> with self . assertRaises ( artifact_registry . ArtifactDefinitionError ) : <EOL> artifact . UploadArtifactYamlFile ( content , token = self . token ) <EOL> def testCommandArgumentOrderIsPreserved ( self ) : <EOL> content = """<STR_LIT>""" <EOL> artifact . UploadArtifactYamlFile ( content , token = self . token ) <EOL> artifact_obj = artifact_registry . REGISTRY . GetArtifacts ( <EOL> name_list = [ "<STR_LIT>" ] ) . pop ( ) <EOL> arglist = artifact_obj . sources [ <NUM_LIT:0> ] . attributes . get ( "<STR_LIT:args>" ) <EOL> self . assertEqual ( arglist , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> serialized = artifact_obj . SerializeToString ( ) <EOL> artifact_obj = artifact_registry . Artifact ( serialized ) <EOL> arglist = artifact_obj . sources [ <NUM_LIT:0> ] . attributes . get ( "<STR_LIT:args>" ) <EOL> self . assertEqual ( arglist , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> class ArtifactFlowLinuxTest ( ArtifactTest ) : <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( ArtifactFlowLinuxTest , self ) . setUp ( ) <EOL> with aff4 . FACTORY . Open ( <EOL> self . SetupClients ( <NUM_LIT:1> , system = "<STR_LIT>" , os_version = "<STR_LIT>" ) [ <NUM_LIT:0> ] , <EOL> mode = "<STR_LIT>" , <EOL> token = self . token ) as fd : <EOL> kb = fd . Get ( fd . Schema . KNOWLEDGE_BASE ) <EOL> kb . MergeOrAddUser ( rdf_client . User ( username = "<STR_LIT>" ) ) <EOL> kb . MergeOrAddUser ( rdf_client . User ( username = "<STR_LIT>" ) ) <EOL> kb . MergeOrAddUser ( rdf_client . User ( username = "<STR_LIT>" ) ) <EOL> fd . Set ( kb ) <EOL> self . LoadTestArtifacts ( ) <EOL> def testCmdArtifact ( self ) : <EOL> """<STR_LIT>""" <EOL> client_mock = self . MockClient ( "<STR_LIT>" , client_id = self . client_id ) <EOL> with utils . Stubber ( subprocess , "<STR_LIT>" , test_lib . Popen ) : <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , client_mock , client_id = self . client_id , <EOL> store_results_in_aff4 = True , use_tsk = False , <EOL> artifact_list = [ "<STR_LIT>" ] , token = self . token ) : <EOL> pass <EOL> urn = self . client_id . Add ( "<STR_LIT>" ) <EOL> fd = aff4 . FACTORY . Open ( urn , token = self . token ) <EOL> packages = fd . Get ( fd . Schema . INSTALLED_PACKAGES ) <EOL> self . assertEqual ( len ( packages ) , <NUM_LIT:2> ) <EOL> self . assertEqual ( packages [ <NUM_LIT:0> ] . __class__ . __name__ , "<STR_LIT>" ) <EOL> with aff4 . FACTORY . Open ( self . client_id . Add ( "<STR_LIT>" ) , <EOL> token = self . token ) as anomaly_coll : <EOL> self . assertEqual ( len ( anomaly_coll ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( "<STR_LIT>" in anomaly_coll [ <NUM_LIT:0> ] . symptom ) <EOL> def testFilesArtifact ( self ) : <EOL> """<STR_LIT>""" <EOL> with test_lib . VFSOverrider ( <EOL> rdf_paths . PathSpec . PathType . OS , test_lib . FakeTestDataVFSHandler ) : <EOL> self . RunCollectorAndGetCollection ( [ "<STR_LIT>" ] , <EOL> client_mock = self . client_mock ) <EOL> urn = self . client_id . Add ( "<STR_LIT>" ) . Add ( "<STR_LIT>" ) <EOL> aff4 . FACTORY . Open ( urn , aff4_type = "<STR_LIT>" , token = self . token ) <EOL> def testLinuxPasswdHomedirsArtifact ( self ) : <EOL> """<STR_LIT>""" <EOL> with test_lib . VFSOverrider ( <EOL> rdf_paths . PathSpec . PathType . OS , test_lib . FakeTestDataVFSHandler ) : <EOL> fd = self . RunCollectorAndGetCollection ( [ "<STR_LIT>" ] , <EOL> client_mock = self . client_mock ) <EOL> self . assertEqual ( len ( fd ) , <NUM_LIT:3> ) <EOL> self . assertItemsEqual ( [ x . username for x in fd ] , [ u"<STR_LIT>" , u"<STR_LIT>" , <EOL> u"<STR_LIT>" ] ) <EOL> for user in fd : <EOL> if user . username == u"<STR_LIT>" : <EOL> self . assertEqual ( user . full_name , u"<STR_LIT>" ) <EOL> self . assertEqual ( user . gid , <NUM_LIT> ) <EOL> self . assertEqual ( user . homedir , u"<STR_LIT>" ) <EOL> self . assertEqual ( user . shell , u"<STR_LIT>" ) <EOL> self . assertEqual ( user . uid , <NUM_LIT> ) <EOL> def testArtifactOutput ( self ) : <EOL> """<STR_LIT>""" <EOL> with test_lib . VFSOverrider ( <EOL> rdf_paths . PathSpec . PathType . OS , test_lib . FakeTestDataVFSHandler ) : <EOL> self . RunCollectorAndGetCollection ( [ "<STR_LIT>" ] , <EOL> client_mock = self . client_mock ) <EOL> self . RunCollectorAndGetCollection ( [ "<STR_LIT>" ] , <EOL> client_mock = self . client_mock , <EOL> split_output_by_artifact = True ) <EOL> with self . assertRaises ( RuntimeError ) as context : <EOL> self . RunCollectorAndGetCollection ( <EOL> [ "<STR_LIT>" ] , client_mock = self . client_mock , <EOL> split_output_by_artifact = True , on_no_results_error = True ) <EOL> if "<STR_LIT>" not in str ( context . exception ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> class ArtifactFlowWindowsTest ( ArtifactTest ) : <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> super ( ArtifactFlowWindowsTest , self ) . setUp ( ) <EOL> self . SetupClients ( <NUM_LIT:1> , system = "<STR_LIT>" , os_version = "<STR_LIT>" , arch = "<STR_LIT>" ) <EOL> self . LoadTestArtifacts ( ) <EOL> def testWMIQueryArtifact ( self ) : <EOL> """<STR_LIT>""" <EOL> self . RunCollectorAndGetCollection ( [ "<STR_LIT>" ] , <EOL> store_results_in_aff4 = True ) <EOL> urn = self . client_id . Add ( "<STR_LIT>" ) <EOL> fd = aff4 . FACTORY . Open ( urn , token = self . token ) <EOL> packages = fd . Get ( fd . Schema . INSTALLED_PACKAGES ) <EOL> self . assertEqual ( len ( packages ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( packages [ <NUM_LIT:0> ] . description , "<STR_LIT>" ) <EOL> def testRekallPsListArtifact ( self ) : <EOL> """<STR_LIT>""" <EOL> test_lib . WriteComponent ( token = self . token ) <EOL> fd = self . RunCollectorAndGetCollection ( <EOL> [ "<STR_LIT>" ] , RekallMock ( <EOL> self . client_id , "<STR_LIT>" ) ) <EOL> self . assertEqual ( len ( fd ) , <NUM_LIT> ) <EOL> self . assertEqual ( fd [ <NUM_LIT:0> ] . exe , "<STR_LIT>" ) <EOL> self . assertEqual ( fd [ <NUM_LIT:0> ] . pid , <NUM_LIT:4> ) <EOL> self . assertIn ( "<STR_LIT>" , [ x . exe for x in fd ] ) <EOL> def testRekallVadArtifact ( self ) : <EOL> """<STR_LIT>""" <EOL> test_lib . WriteComponent ( token = self . token ) <EOL> with aff4 . FACTORY . Open ( self . client_id , mode = "<STR_LIT>" , token = self . token ) as fd : <EOL> fd . Set ( fd . Schema . KNOWLEDGE_BASE ( <EOL> os = "<STR_LIT>" , <EOL> environ_systemdrive = r"<STR_LIT>" ) ) <EOL> fd = self . RunCollectorAndGetCollection ( <EOL> [ "<STR_LIT>" ] , RekallMock ( <EOL> self . client_id , "<STR_LIT>" ) ) <EOL> self . assertEqual ( len ( fd ) , <NUM_LIT> ) <EOL> self . assertEqual ( fd [ <NUM_LIT:0> ] . path , u"<STR_LIT>" ) <EOL> for x in fd : <EOL> self . assertEqual ( x . pathtype , "<STR_LIT>" ) <EOL> extension = x . path . lower ( ) . split ( "<STR_LIT:.>" ) [ - <NUM_LIT:1> ] <EOL> self . assertIn ( extension , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> class GrrKbTest ( ArtifactTest ) : <EOL> def setUp ( self ) : <EOL> super ( GrrKbTest , self ) . setUp ( ) <EOL> test_lib . ClientFixture ( self . client_id , token = self . token ) <EOL> def ClearKB ( self ) : <EOL> client = aff4 . FACTORY . Open ( self . client_id , mode = "<STR_LIT>" , token = self . token ) <EOL> client . Set ( client . Schema . KNOWLEDGE_BASE , rdf_client . KnowledgeBase ( ) ) <EOL> client . Flush ( ) <EOL> class GrrKbWindowsTest ( GrrKbTest ) : <EOL> def setUp ( self ) : <EOL> super ( GrrKbWindowsTest , self ) . setUp ( ) <EOL> self . SetupClients ( <NUM_LIT:1> , system = "<STR_LIT>" , os_version = "<STR_LIT>" , arch = "<STR_LIT>" ) <EOL> self . os_overrider = test_lib . VFSOverrider ( <EOL> rdf_paths . PathSpec . PathType . OS , test_lib . FakeFullVFSHandler ) <EOL> self . reg_overrider = test_lib . VFSOverrider ( <EOL> rdf_paths . PathSpec . PathType . REGISTRY , test_lib . FakeRegistryVFSHandler ) <EOL> self . os_overrider . Start ( ) <EOL> self . reg_overrider . Start ( ) <EOL> def tearDown ( self ) : <EOL> super ( GrrKbWindowsTest , self ) . tearDown ( ) <EOL> try : <EOL> self . os_overrider . Stop ( ) <EOL> self . reg_overrider . Stop ( ) <EOL> except AttributeError : <EOL> pass <EOL> def testKnowledgeBaseRetrievalWindows ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ClearKB ( ) <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , <EOL> client_id = self . client_id , token = self . token ) : <EOL> pass <EOL> client = aff4 . FACTORY . Open ( self . client_id , token = self . token ) <EOL> kb = artifact . GetArtifactKnowledgeBase ( client ) <EOL> self . assertEqual ( kb . environ_systemroot , "<STR_LIT>" ) <EOL> self . assertEqual ( kb . time_zone , "<STR_LIT>" ) <EOL> self . assertEqual ( kb . code_page , "<STR_LIT>" ) <EOL> self . assertEqual ( kb . environ_windir , "<STR_LIT>" ) <EOL> self . assertEqual ( kb . environ_allusersprofile , "<STR_LIT>" ) <EOL> self . assertEqual ( kb . environ_allusersappdata , "<STR_LIT>" ) <EOL> self . assertEqual ( kb . environ_temp , "<STR_LIT>" ) <EOL> self . assertEqual ( kb . environ_systemdrive , "<STR_LIT>" ) <EOL> self . assertItemsEqual ( [ x . username for x in kb . users ] , <EOL> [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> user = kb . GetUser ( username = "<STR_LIT>" ) <EOL> self . assertEqual ( user . username , "<STR_LIT>" ) <EOL> self . assertEqual ( user . sid , "<STR_LIT>" ) <EOL> def testKnowledgeBaseMultiProvides ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ClearKB ( ) <EOL> self . LoadTestArtifacts ( ) <EOL> with test_lib . ConfigOverrider ( { "<STR_LIT>" : <EOL> [ "<STR_LIT>" ] } ) : <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , <EOL> client_id = self . client_id , token = self . token ) : <EOL> pass <EOL> client = aff4 . FACTORY . Open ( self . client_id , token = self . token ) <EOL> kb = artifact . GetArtifactKnowledgeBase ( client ) <EOL> self . assertEqual ( kb . environ_temp , "<STR_LIT>" ) <EOL> self . assertEqual ( kb . environ_path , "<STR_LIT>" ) <EOL> def testGlobRegistry ( self ) : <EOL> """<STR_LIT>""" <EOL> paths = [ "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ] <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , paths = paths , <EOL> pathtype = rdf_paths . PathSpec . PathType . REGISTRY , <EOL> client_id = self . client_id , token = self . token ) : <EOL> pass <EOL> path = paths [ <NUM_LIT:0> ] . replace ( "<STR_LIT:\\>" , "<STR_LIT:/>" ) <EOL> fd = aff4 . FACTORY . Open ( self . client_id . Add ( "<STR_LIT>" ) . Add ( path ) , <EOL> token = self . token ) <EOL> self . assertEqual ( fd . __class__ . __name__ , "<STR_LIT>" ) <EOL> self . assertEqual ( fd . Get ( fd . Schema . STAT ) . registry_data . GetValue ( ) , <EOL> "<STR_LIT>" ) <EOL> def testGetDependencies ( self ) : <EOL> """<STR_LIT>""" <EOL> artifact_registry . REGISTRY . ClearSources ( ) <EOL> try : <EOL> test_artifacts_file = os . path . join ( <EOL> config_lib . CONFIG [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> artifact_registry . REGISTRY . AddFileSource ( test_artifacts_file ) <EOL> args = artifact . CollectArtifactDependenciesArgs ( <EOL> artifact_list = [ "<STR_LIT>" ] ) <EOL> collect_obj = artifact . CollectArtifactDependencies ( None , token = self . token ) <EOL> collect_obj . args = args <EOL> collect_obj . knowledge_base = None <EOL> collect_obj . state . Register ( "<STR_LIT>" , set ( ) ) <EOL> collect_obj . state . Register ( "<STR_LIT>" , [ ] ) <EOL> collect_obj . state . Register ( "<STR_LIT>" , <EOL> rdf_client . KnowledgeBase ( os = "<STR_LIT>" ) ) <EOL> no_deps = collect_obj . GetFirstFlowsForCollection ( ) <EOL> self . assertItemsEqual ( no_deps , [ ] ) <EOL> self . assertItemsEqual ( collect_obj . state . all_deps , [ ] ) <EOL> self . assertItemsEqual ( collect_obj . state . awaiting_deps_artifacts , [ ] ) <EOL> args = artifact . CollectArtifactDependenciesArgs ( <EOL> artifact_list = [ "<STR_LIT>" ] ) <EOL> collect_obj . args = args <EOL> no_deps = collect_obj . GetFirstFlowsForCollection ( ) <EOL> self . assertItemsEqual ( no_deps , [ "<STR_LIT>" ] ) <EOL> self . assertItemsEqual ( collect_obj . state . all_deps , [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> self . assertItemsEqual ( collect_obj . state . awaiting_deps_artifacts , <EOL> [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> finally : <EOL> artifact . ArtifactLoader ( ) . RunOnce ( ) <EOL> def testGetKBDependencies ( self ) : <EOL> """<STR_LIT>""" <EOL> artifact_registry . REGISTRY . ClearSources ( ) <EOL> try : <EOL> test_artifacts_file = os . path . join ( <EOL> config_lib . CONFIG [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> artifact_registry . REGISTRY . AddFileSource ( test_artifacts_file ) <EOL> with test_lib . ConfigOverrider ( { <EOL> "<STR_LIT>" : [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] } ) : <EOL> args = artifact . KnowledgeBaseInitializationArgs ( lightweight = True ) <EOL> kb_init = artifact . KnowledgeBaseInitializationFlow ( <EOL> None , token = self . token ) <EOL> kb_init . args = args <EOL> kb_init . state . Register ( "<STR_LIT>" , set ( ) ) <EOL> kb_init . state . Register ( "<STR_LIT>" , [ ] ) <EOL> kb_init . state . Register ( "<STR_LIT>" , <EOL> rdf_client . KnowledgeBase ( os = "<STR_LIT>" ) ) <EOL> no_deps = kb_init . GetFirstFlowsForCollection ( ) <EOL> self . assertItemsEqual ( no_deps , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertItemsEqual ( kb_init . state . all_deps , [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> self . assertItemsEqual ( kb_init . state . awaiting_deps_artifacts , <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> finally : <EOL> artifact . ArtifactLoader ( ) . RunOnce ( ) <EOL> class GrrKbLinuxTest ( GrrKbTest ) : <EOL> def setUp ( self ) : <EOL> super ( GrrKbLinuxTest , self ) . setUp ( ) <EOL> self . SetupClients ( <NUM_LIT:1> , system = "<STR_LIT>" , os_version = "<STR_LIT>" ) <EOL> def testKnowledgeBaseRetrievalLinux ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ClearKB ( ) <EOL> with test_lib . ConfigOverrider ( { <EOL> "<STR_LIT>" : [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] } ) : <EOL> with test_lib . VFSOverrider ( rdf_paths . PathSpec . PathType . OS , <EOL> test_lib . FakeTestDataVFSHandler ) : <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , <EOL> client_id = self . client_id , token = self . token ) : <EOL> pass <EOL> client = aff4 . FACTORY . Open ( self . client_id , token = self . token , mode = "<STR_LIT>" ) <EOL> kb = artifact . GetArtifactKnowledgeBase ( client ) <EOL> self . assertEqual ( kb . os_major_version , <NUM_LIT> ) <EOL> self . assertEqual ( kb . os_minor_version , <NUM_LIT:4> ) <EOL> self . assertItemsEqual ( [ x . username for x in kb . users ] , <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> user = kb . GetUser ( username = "<STR_LIT>" ) <EOL> self . assertEqual ( user . last_logon . AsSecondsFromEpoch ( ) , <NUM_LIT> ) <EOL> self . assertEqual ( user . homedir , "<STR_LIT>" ) <EOL> def testKnowledgeBaseRetrievalLinuxPasswd ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ClearKB ( ) <EOL> with test_lib . VFSOverrider ( rdf_paths . PathSpec . PathType . OS , <EOL> test_lib . FakeTestDataVFSHandler ) : <EOL> with test_lib . ConfigOverrider ( { <EOL> "<STR_LIT>" : [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : [ ] } ) : <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , <EOL> client_id = self . client_id , token = self . token ) : <EOL> pass <EOL> client = aff4 . FACTORY . Open ( self . client_id , token = self . token , mode = "<STR_LIT>" ) <EOL> kb = artifact . GetArtifactKnowledgeBase ( client ) <EOL> self . assertEqual ( kb . os_major_version , <NUM_LIT> ) <EOL> self . assertEqual ( kb . os_minor_version , <NUM_LIT:4> ) <EOL> self . assertItemsEqual ( [ x . username for x in kb . users ] , [ "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) <EOL> user = kb . GetUser ( username = "<STR_LIT>" ) <EOL> self . assertEqual ( user . last_logon . AsSecondsFromEpoch ( ) , <NUM_LIT> ) <EOL> self . assertEqual ( user . homedir , "<STR_LIT>" ) <EOL> user = kb . GetUser ( username = "<STR_LIT>" ) <EOL> self . assertEqual ( user . last_logon . AsSecondsFromEpoch ( ) , <NUM_LIT> ) <EOL> self . assertEqual ( user . homedir , "<STR_LIT>" ) <EOL> self . assertFalse ( kb . GetUser ( username = "<STR_LIT>" ) ) <EOL> def testKnowledgeBaseRetrievalLinuxNoUsers ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ClearKB ( ) <EOL> with test_lib . ConfigOverrider ( { <EOL> "<STR_LIT>" : [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] , <EOL> "<STR_LIT>" : [ "<STR_LIT>" ] } ) : <EOL> with test_lib . VFSOverrider ( rdf_paths . PathSpec . PathType . OS , <EOL> test_lib . FakeTestDataVFSHandler ) : <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , <EOL> require_complete = False , <EOL> client_id = self . client_id , token = self . token ) : <EOL> pass <EOL> client = aff4 . FACTORY . Open ( self . client_id , token = self . token , mode = "<STR_LIT>" ) <EOL> kb = artifact . GetArtifactKnowledgeBase ( client ) <EOL> self . assertEqual ( kb . os_major_version , <NUM_LIT> ) <EOL> self . assertEqual ( kb . os_minor_version , <NUM_LIT:4> ) <EOL> self . assertItemsEqual ( [ x . username for x in kb . users ] , [ ] ) <EOL> class GrrKbDarwinTest ( GrrKbTest ) : <EOL> def setUp ( self ) : <EOL> super ( GrrKbDarwinTest , self ) . setUp ( ) <EOL> self . SetupClients ( <NUM_LIT:1> , system = "<STR_LIT>" , os_version = "<STR_LIT>" ) <EOL> def testKnowledgeBaseRetrievalDarwin ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ClearKB ( ) <EOL> with test_lib . ConfigOverrider ( { "<STR_LIT>" : [ "<STR_LIT>" ] } ) : <EOL> with test_lib . VFSOverrider ( rdf_paths . PathSpec . PathType . OS , <EOL> test_lib . ClientVFSHandlerFixture ) : <EOL> for _ in test_lib . TestFlowHelper ( <EOL> "<STR_LIT>" , self . client_mock , <EOL> client_id = self . client_id , token = self . token ) : <EOL> pass <EOL> client = aff4 . FACTORY . Open ( self . client_id , token = self . token , mode = "<STR_LIT>" ) <EOL> kb = artifact . GetArtifactKnowledgeBase ( client ) <EOL> self . assertEqual ( kb . os_major_version , <NUM_LIT:10> ) <EOL> self . assertEqual ( kb . os_minor_version , <NUM_LIT:9> ) <EOL> self . assertItemsEqual ( [ x . username for x in kb . users ] , [ "<STR_LIT>" ] ) <EOL> user = kb . GetUser ( username = "<STR_LIT>" ) <EOL> self . assertEqual ( user . homedir , "<STR_LIT>" ) <EOL> def main ( argv ) : <EOL> test_lib . GrrTestProgram ( argv = argv ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> flags . StartMain ( main ) </s>
