diff --git a/awscli/customizations/s3events.py b/awscli/customizations/s3events.py index 2cd57fef7975..2a0c31d307a7 100644 --- a/awscli/customizations/s3events.py +++ b/awscli/customizations/s3events.py @@ -12,8 +12,6 @@ # language governing permissions and limitations under the License. """Add S3 specific event streaming output arg.""" -import os - from awscli.arguments import CustomArgument STREAM_HELP_TEXT = 'Filename where the records will be saved' @@ -61,7 +59,8 @@ def replace_event_stream_docs(help_command, **kwargs): # This should never happen, but in the rare case that it does # we should be raising something with a helpful error message. raise DocSectionNotFoundError( - f'Could not find the "output" section for the command: {help_command}' + 'Could not find the "output" section for the command: %s' + % help_command ) doc.write('======\nOutput\n======\n') doc.write( @@ -99,7 +98,7 @@ class S3SelectStreamOutputArgument(CustomArgument): _DOCUMENT_AS_REQUIRED = True def __init__(self, stream_key, session, **kwargs): - super().__init__(**kwargs) + super(S3SelectStreamOutputArgument, self).__init__(**kwargs) # This is the key in the response body where we can find the # streamed contents. self._stream_key = stream_key @@ -121,11 +120,7 @@ def save_file(self, parsed, **kwargs): if self._stream_key not in parsed: return event_stream = parsed[self._stream_key] - fd = os.open( - self._output_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600 - ) - os.chmod(self._output_file, 0o600) - with os.fdopen(fd, 'wb') as fp: + with open(self._output_file, 'wb') as fp: for event in event_stream: if 'Records' in event: fp.write(event['Records']['Payload']) diff --git a/awscli/customizations/streamingoutputarg.py b/awscli/customizations/streamingoutputarg.py index 596aab8a3342..c4decbb6844d 100644 --- a/awscli/customizations/streamingoutputarg.py +++ b/awscli/customizations/streamingoutputarg.py @@ -10,8 +10,6 @@ # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. -import os - from botocore.model import Shape from awscli.arguments import BaseCLIArgument @@ -94,7 +92,7 @@ def add_to_params(self, parameters, value): service_id = self._operation_model.service_model.service_id.hyphenize() operation_name = self._operation_model.name self._session.register( - f'after-call.{service_id}.{operation_name}', self.save_file + 'after-call.%s.%s' % (service_id, operation_name), self.save_file ) def save_file(self, parsed, **kwargs): @@ -106,11 +104,7 @@ def save_file(self, parsed, **kwargs): return body = parsed[self._response_key] buffer_size = self._buffer_size - fd = os.open( - self._output_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600 - ) - os.chmod(self._output_file, 0o600) - with os.fdopen(fd, 'wb') as fp: + with open(self._output_file, 'wb') as fp: data = body.read(buffer_size) while data: fp.write(data) diff --git a/tests/functional/s3api/test_select_object_content.py b/tests/functional/s3api/test_select_object_content.py index e9078aa4b677..04140563fa09 100644 --- a/tests/functional/s3api/test_select_object_content.py +++ b/tests/functional/s3api/test_select_object_content.py @@ -15,23 +15,19 @@ import shutil import tempfile -from awscli.testutils import ( - BaseAWSCommandParamsTest, - BaseAWSHelpOutputTest, - skip_if_windows, -) +from awscli.testutils import BaseAWSCommandParamsTest, BaseAWSHelpOutputTest class TestGetObject(BaseAWSCommandParamsTest): prefix = ['s3api', 'select-object-content'] def setUp(self): - super().setUp() + super(TestGetObject, self).setUp() self.parsed_response = {'Payload': self.create_fake_payload()} self._tempdir = tempfile.mkdtemp() def tearDown(self): - super().tearDown() + super(TestGetObject, self).tearDown() shutil.rmtree(self._tempdir) def create_fake_payload(self): @@ -86,28 +82,6 @@ def test_can_stream_to_file(self): contents = f.read() self.assertEqual(contents, ('a,b,c,d\n' 'e,f,g,h\n')) - @skip_if_windows('chmod is not supported on Windows') - def test_output_file_permissions(self): - filename = os.path.join(self._tempdir, 'outfile_perms') - cmdline = self.prefix + [ - '--bucket', - 'mybucket', - '--key', - 'mykey', - '--expression', - 'SELECT * FROM S3Object', - '--expression-type', - 'SQL', - '--input-serialization', - '{"CSV": {}}', - '--output-serialization', - '{"CSV": {}}', - filename, - ] - self.assert_params_for_cmd(cmdline, ignore_params=True) - # Mask file type bits to isolate permission bits (rwxrwxrwx) - self.assertEqual(os.stat(filename).st_mode & 0o777, 0o600) - def test_errors_are_propagated(self): self.http_response.status_code = 400 self.parsed_response = { diff --git a/tests/functional/test_streaming_output.py b/tests/functional/test_streaming_output.py index 3255e7b74e9a..10aa6c74407e 100644 --- a/tests/functional/test_streaming_output.py +++ b/tests/functional/test_streaming_output.py @@ -11,23 +11,17 @@ # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. -import os - from awscli.compat import BytesIO -from awscli.testutils import ( - BaseAWSCommandParamsTest, - FileCreator, - skip_if_windows, -) +from awscli.testutils import BaseAWSCommandParamsTest, FileCreator class TestStreamingOutput(BaseAWSCommandParamsTest): def setUp(self): - super().setUp() + super(TestStreamingOutput, self).setUp() self.files = FileCreator() def tearDown(self): - super().tearDown() + super(TestStreamingOutput, self).tearDown() self.files.remove_all() def test_get_media_streaming_output(self): @@ -47,18 +41,3 @@ def test_get_media_streaming_output(self): self.assert_params_for_cmd(cmdline % outpath, params) with open(outpath, 'rb') as outfile: self.assertEqual(outfile.read(), b'testbody') - - @skip_if_windows('chmod is not supported on Windows') - def test_streaming_output_file_permissions(self): - cmdline = ( - 'kinesis-video-media get-media --stream-name test-stream ' - '--start-selector StartSelectorType=EARLIEST %s' - ) - self.parsed_response = { - 'ContentType': 'video/webm', - 'Payload': BytesIO(b'testbody'), - } - outpath = self.files.full_path('outfile') - self.assert_params_for_cmd(cmdline % outpath, ignore_params=True) - # Mask file type bits to isolate permission bits (rwxrwxrwx) - self.assertEqual(os.stat(outpath).st_mode & 0o777, 0o600)