grr/server/grr_response_server/flows/general/collectors_test.py
def testFileArtifactWithParser(self):
    """Test collecting a file artifact and parsing the response."""
    parsers.SINGLE_FILE_PARSER_FACTORY.Register("TestFile", TestFileParser)
    try:
      artifact_list = ["TestFileArtifact"]

      file_path = self.InitializeTestFileArtifact()

      # Run the ArtifactCollector to get the expected result.
      session_id = flow_test_lib.TestFlowHelper(
          aff4_flows.ArtifactCollectorFlow.__name__,
          action_mocks.FileFinderClientMock(),
          artifact_list=artifact_list,
          token=self.token,
          apply_parsers=True,
          client_id=self.client_id)
      results = flow_test_lib.GetFlowResults(self.client_id, session_id)
      expected = results[0]
      self.assertIsInstance(expected, rdf_protodict.AttributedDict)
      self.assertEqual(expected.filename, file_path)
      self.assertLen(expected.users, 1000)

      # Run the ClientArtifactCollector to get the actual result.
      cac_results = self._RunFlow(
          aff4_flows.ClientArtifactCollector,
          artifact_collector.ArtifactCollector,
          artifact_list,
          apply_parsers=True)
      self.assertLen(cac_results, 1)

      self.assertEqual(results, cac_results)
    finally:
      parsers.SINGLE_FILE_PARSER_FACTORY.Unregister("TestFile")
Test collecting a file artifact and parsing the response.
Test collecting a file artifact and parsing the response.
Test collecting a file artifact and parsing the response .
def testFileArtifactWithParser(self):
    parsers.SINGLE_FILE_PARSER_FACTORY.Register("TestFile", TestFileParser)
    try:
      artifact_list = ["TestFileArtifact"]
      file_path = self.InitializeTestFileArtifact()
      session_id = flow_test_lib.TestFlowHelper(
          aff4_flows.ArtifactCollectorFlow.__name__,
          action_mocks.FileFinderClientMock(),
          artifact_list=artifact_list,
          token=self.token,
          apply_parsers=True,
          client_id=self.client_id)
      results = flow_test_lib.GetFlowResults(self.client_id, session_id)
      expected = results[0]
      self.assertIsInstance(expected, rdf_protodict.AttributedDict)
      self.assertEqual(expected.filename, file_path)
      self.assertLen(expected.users, 1000)
      cac_results = self._RunFlow(
          aff4_flows.ClientArtifactCollector,
          artifact_collector.ArtifactCollector,
          artifact_list,
          apply_parsers=True)
      self.assertLen(cac_results, 1)
      self.assertEqual(results, cac_results)
    finally:
      parsers.SINGLE_FILE_PARSER_FACTORY.Unregister("TestFile")
def testFileArtifactWithParser ( self ) : parsers . SINGLE_FILE_PARSER_FACTORY . Register ( "TestFile" , TestFileParser ) try : artifact_list = [ "TestFileArtifact" ] file_path = self . InitializeTestFileArtifact ( ) session_id = flow_test_lib . TestFlowHelper ( aff4_flows . ArtifactCollectorFlow . __name__ , action_mocks . FileFinderClientMock ( ) , artifact_list = artifact_list , token = self . token , apply_parsers = True , client_id = self . client_id ) results = flow_test_lib . GetFlowResults ( self . client_id , session_id ) expected = results [ 0 ] self . assertIsInstance ( expected , rdf_protodict . AttributedDict ) self . assertEqual ( expected . filename , file_path ) self . assertLen ( expected . users , 1000 ) cac_results = self . _RunFlow ( aff4_flows . ClientArtifactCollector , artifact_collector . ArtifactCollector , artifact_list , apply_parsers = True ) self . assertLen ( cac_results , 1 ) self . assertEqual ( results , cac_results ) finally : parsers . SINGLE_FILE_PARSER_FACTORY . Unregister ( "TestFile" )
Test collecting a file artifact and parsing the response.
Test collecting a file artifact and parsing the response .
"""Test collecting a file artifact and parsing the response.""" # Run the ArtifactCollector to get the expected result. # Run the ClientArtifactCollector to get the actual result.
self