# Path: gslib/commands/perfdiag.py
# def _GenerateFileData(fp,
#                       file_size=0,
#                       random_ratio=100,
#                       max_unique_random_bytes=5242883):
#   """Writes data into a file like object.
# 
#   Args:
#     fp: A file like object to write the data to.
#     file_size: The amount of data to write to the file.
#     random_ratio: The percentage of randomly generated data to write. This can
#         be any number between 0 and 100 (inclusive), with 0 producing uniform
#         data, and 100 producing random data.
#     max_unique_random_bytes: The maximum number of bytes to generate
#                              pseudo-randomly before beginning to repeat
#                              bytes. The default was chosen as the next prime
#                              larger than 5 MiB.
#   """
#   # Normalize the ratio.
#   random_ratio /= 100.0
#   random_bytes = os.urandom(min(file_size, max_unique_random_bytes))
#   total_bytes_written = 0
#   while total_bytes_written < file_size:
#     num_bytes = min(max_unique_random_bytes, file_size - total_bytes_written)
#     # Calculate the amount of random and sequential data to write and
#     # resolve rounding errors by adjusting the random bytes to write.
#     num_bytes_seq = int(num_bytes * (1 - random_ratio))
#     num_bytes_random = num_bytes - num_bytes_seq
# 
#     fp.write(random_bytes[:num_bytes_random])
#     fp.write(b'x' * num_bytes_seq)
#     total_bytes_written += num_bytes
# 
# Path: gslib/tests/testcase/integration_testcase.py
# def SkipForXML(reason):
#   """Skips the test if running S3 tests, or if prefer_api isn't set to json."""
#   if not USING_JSON_API or RUN_S3_TESTS:
#     return unittest.skip(reason)
#   else:
#     return lambda func: func


import os
import socket
import sys
import six
import boto
import gslib.tests.testcase as testcase
from gslib.commands.perfdiag import _GenerateFileData
from gslib.tests.testcase.integration_testcase import SkipForXML
from gslib.tests.util import ObjectToURI as suri
from gslib.tests.util import RUN_S3_TESTS
from gslib.tests.util import unittest
from gslib.utils.system_util import IS_WINDOWS
from six import add_move, MovedModule
from six.moves import mock
    stderr = self.RunGsUtil(
        ['perfdiag', '-n', '1', '-s', '3pb', '-t', 'wthru', 'gs://foobar'],
        expected_status=1,
        return_stderr=True)
    self.assertIn('in-memory tests maximum file size', stderr)

  def test_listing(self):
    bucket_uri = self.CreateBucket()
    stdout = self.RunGsUtil(
        ['perfdiag', '-n', '1', '-t', 'list',
         suri(bucket_uri)],
        return_stdout=True)
    self.assertIn('Number of listing calls made:', stdout)
    self.AssertNObjectsInBucket(bucket_uri, 0, versioned=True)

  @SkipForXML('No compressed transport encoding support for the XML API.')
  def test_gzip_write_throughput_single_process_single_thread(self):
    (stderr_default, _) = self._run_throughput_test('wthru',
                                                    1,
                                                    1,
                                                    compression_ratio=50)
    self.assertIn('Gzip compression ratio: 50', stderr_default)
    self.assertIn('Gzip transport encoding writes: True', stderr_default)
    (stderr_default, _) = self._run_throughput_test('wthru_file',
                                                    1,
                                                    1,
                                                    compression_ratio=50)
    self.assertIn('Gzip compression ratio: 50', stderr_default)
    self.assertIn('Gzip transport encoding writes: True', stderr_default)


# Path: gslib/commands/perfdiag.py
# def _GenerateFileData(fp,
#                       file_size=0,
#                       random_ratio=100,
#                       max_unique_random_bytes=5242883):
#   """Writes data into a file like object.
# 
#   Args:
#     fp: A file like object to write the data to.
#     file_size: The amount of data to write to the file.
#     random_ratio: The percentage of randomly generated data to write. This can
#         be any number between 0 and 100 (inclusive), with 0 producing uniform
#         data, and 100 producing random data.
#     max_unique_random_bytes: The maximum number of bytes to generate
#                              pseudo-randomly before beginning to repeat
#                              bytes. The default was chosen as the next prime
#                              larger than 5 MiB.
#   """
#   # Normalize the ratio.
#   random_ratio /= 100.0
#   random_bytes = os.urandom(min(file_size, max_unique_random_bytes))
#   total_bytes_written = 0
#   while total_bytes_written < file_size:
#     num_bytes = min(max_unique_random_bytes, file_size - total_bytes_written)
#     # Calculate the amount of random and sequential data to write and
#     # resolve rounding errors by adjusting the random bytes to write.
#     num_bytes_seq = int(num_bytes * (1 - random_ratio))
#     num_bytes_random = num_bytes - num_bytes_seq
# 
#     fp.write(random_bytes[:num_bytes_random])
#     fp.write(b'x' * num_bytes_seq)
#     total_bytes_written += num_bytes
# 
# Path: gslib/tests/testcase/integration_testcase.py
# def SkipForXML(reason):
#   """Skips the test if running S3 tests, or if prefer_api isn't set to json."""
#   if not USING_JSON_API or RUN_S3_TESTS:
#     return unittest.skip(reason)
#   else:
#     return lambda func: func


# Path: gslib/tests/test_perfdiag.py
import os
import socket
import sys
import six
import boto
import gslib.tests.testcase as testcase
from gslib.commands.perfdiag import _GenerateFileData
from gslib.tests.testcase.integration_testcase import SkipForXML
from gslib.tests.util import ObjectToURI as suri
from gslib.tests.util import RUN_S3_TESTS
from gslib.tests.util import unittest
from gslib.utils.system_util import IS_WINDOWS
from six import add_move, MovedModule
from six.moves import mock

    stderr = self.RunGsUtil(
        ['perfdiag', '-n', '1', '-s', '3pb', '-t', 'wthru', 'gs://foobar'],
        expected_status=1,
        return_stderr=True)
    self.assertIn('in-memory tests maximum file size', stderr)

  def test_listing(self):
    bucket_uri = self.CreateBucket()
    stdout = self.RunGsUtil(
        ['perfdiag', '-n', '1', '-t', 'list',
         suri(bucket_uri)],
        return_stdout=True)
    self.assertIn('Number of listing calls made:', stdout)
    self.AssertNObjectsInBucket(bucket_uri, 0, versioned=True)

  @SkipForXML('No compressed transport encoding support for the XML API.')
  def test_gzip_write_throughput_single_process_single_thread(self):
    (stderr_default, _) = self._run_throughput_test('wthru',
                                                    1,
                                                    1,
                                                    compression_ratio=50)
    self.assertIn('Gzip compression ratio: 50', stderr_default)
    self.assertIn('Gzip transport encoding writes: True', stderr_default)
    (stderr_default, _) = self._run_throughput_test('wthru_file',
                                                    1,
                                                    1,
                                                    compression_ratio=50)
    self.assertIn('Gzip compression ratio: 50', stderr_default)
    self.assertIn('Gzip transport encoding writes: True', stderr_default)


  @SkipForXML('No compressed transport encoding support for the XML API.')