1# -*- coding: utf-8 -*- 2# Copyright 2013 Google Inc. All Rights Reserved. 3# 4# Licensed under the Apache License, Version 2.0 (the "License"); 5# you may not use this file except in compliance with the License. 6# You may obtain a copy of the License at 7# 8# http://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, 12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13# See the License for the specific language governing permissions and 14# limitations under the License. 15"""Tests for cat command.""" 16 17from __future__ import absolute_import 18 19import gslib.tests.testcase as testcase 20from gslib.tests.util import ObjectToURI as suri 21from gslib.tests.util import RUN_S3_TESTS 22 23 24class TestCat(testcase.GsUtilIntegrationTestCase): 25 """Integration tests for cat command.""" 26 27 def test_cat_range(self): 28 """Tests cat command with various range arguments.""" 29 key_uri = self.CreateObject(contents='0123456789') 30 # Test various invalid ranges. 31 stderr = self.RunGsUtil(['cat', '-r -', suri(key_uri)], 32 return_stderr=True, expected_status=1) 33 self.assertIn('Invalid range', stderr) 34 stderr = self.RunGsUtil(['cat', '-r a-b', suri(key_uri)], 35 return_stderr=True, expected_status=1) 36 self.assertIn('Invalid range', stderr) 37 stderr = self.RunGsUtil(['cat', '-r 1-2-3', suri(key_uri)], 38 return_stderr=True, expected_status=1) 39 self.assertIn('Invalid range', stderr) 40 stderr = self.RunGsUtil(['cat', '-r 1.7-3', suri(key_uri)], 41 return_stderr=True, expected_status=1) 42 self.assertIn('Invalid range', stderr) 43 44 # Test various valid ranges. 45 stdout = self.RunGsUtil(['cat', '-r 1-3', suri(key_uri)], 46 return_stdout=True) 47 self.assertEqual('123', stdout) 48 stdout = self.RunGsUtil(['cat', '-r 8-', suri(key_uri)], 49 return_stdout=True) 50 self.assertEqual('89', stdout) 51 stdout = self.RunGsUtil(['cat', '-r 0-0', suri(key_uri)], 52 return_stdout=True) 53 self.assertEqual('0', stdout) 54 stdout = self.RunGsUtil(['cat', '-r -3', suri(key_uri)], 55 return_stdout=True) 56 self.assertEqual('789', stdout) 57 58 def test_cat_version(self): 59 """Tests cat command on versioned objects.""" 60 bucket_uri = self.CreateVersionedBucket() 61 # Create 2 versions of an object. 62 uri1 = self.CreateObject(bucket_uri=bucket_uri, contents='data1') 63 uri2 = self.CreateObject(bucket_uri=bucket_uri, 64 object_name=uri1.object_name, contents='data2') 65 stdout = self.RunGsUtil(['cat', suri(uri1)], return_stdout=True) 66 # Last version written should be live. 67 self.assertEqual('data2', stdout) 68 # Using either version-specific URI should work. 69 stdout = self.RunGsUtil(['cat', uri1.version_specific_uri], 70 return_stdout=True) 71 self.assertEqual('data1', stdout) 72 stdout = self.RunGsUtil(['cat', uri2.version_specific_uri], 73 return_stdout=True) 74 self.assertEqual('data2', stdout) 75 if RUN_S3_TESTS: 76 # S3 GETs of invalid versions return 400s. 77 # Also, appending between 1 and 3 characters to the version_id can 78 # result in a success (200) response from the server. 79 stderr = self.RunGsUtil(['cat', uri2.version_specific_uri + '23456'], 80 return_stderr=True, expected_status=1) 81 self.assertIn('BadRequestException: 400', stderr) 82 else: 83 # Attempting to cat invalid version should result in an error. 84 stderr = self.RunGsUtil(['cat', uri2.version_specific_uri + '23'], 85 return_stderr=True, expected_status=1) 86 self.assertIn('No URLs matched', stderr) 87 88 def test_cat_multi_arg(self): 89 """Tests cat command with multiple arguments.""" 90 bucket_uri = self.CreateBucket() 91 data1 = '0123456789' 92 data2 = 'abcdefghij' 93 obj_uri1 = self.CreateObject(bucket_uri=bucket_uri, contents=data1) 94 obj_uri2 = self.CreateObject(bucket_uri=bucket_uri, contents=data2) 95 stdout, stderr = self.RunGsUtil( 96 ['cat', suri(obj_uri1), suri(bucket_uri) + 'nonexistent'], 97 return_stdout=True, return_stderr=True, expected_status=1) 98 # First object should print, second should produce an exception. 99 self.assertIn(data1, stdout) 100 self.assertIn('NotFoundException', stderr) 101 102 stdout, stderr = self.RunGsUtil( 103 ['cat', suri(bucket_uri) + 'nonexistent', suri(obj_uri1)], 104 return_stdout=True, return_stderr=True, expected_status=1) 105 106 # If first object is invalid, exception should halt output immediately. 107 self.assertNotIn(data1, stdout) 108 self.assertIn('NotFoundException', stderr) 109 110 # Two valid objects should both print successfully. 111 stdout = self.RunGsUtil(['cat', suri(obj_uri1), suri(obj_uri2)], 112 return_stdout=True) 113 self.assertIn(data1 + data2, stdout) 114