def testGeneratesTarGzArchive(self):
    result = self.handler.Handle(flow_plugin.ApiGetFlowFilesArchiveArgs(
        client_id=self.client_id, flow_id=self.flow_urn.Basename(),
        archive_format="TAR_GZ"), token=self.token)

    with utils.TempDirectory() as temp_dir:
      tar_path = os.path.join(temp_dir, "archive.tar.gz")
      with open(tar_path, "w") as fd:
        for chunk in result.GenerateContent():
          fd.write(chunk)

      with tarfile.open(tar_path) as tar_fd:
        tar_fd.extractall(path=temp_dir)

      manifest_file_path = None
      for parent, _, files in os.walk(temp_dir):
        if "MANIFEST" in files:
          manifest_file_path = os.path.join(parent, "MANIFEST")
          break

      self.assertTrue(manifest_file_path)
      with open(manifest_file_path) as fd:
        manifest = yaml.safe_load(fd.read())

        self.assertEqual(manifest["archived_files"], 1)
        self.assertEqual(manifest["failed_files"], 0)
        self.assertEqual(manifest["processed_files"], 1)
        self.assertEqual(manifest["skipped_files"], 0)
dataset/ETHPy150Open google/grr/grr/gui/api_plugins/flow_test.py/ApiGetFlowFilesArchiveHandlerTest.testGeneratesTarGzArchive/original