This tests uploading a small stream from stdin.
(self, s3_utils, shared_bucket)
| 1869 | |
| 1870 | class TestStreams(BaseParameterizedS3ClientTest): |
| 1871 | def test_upload(self, s3_utils, shared_bucket): |
| 1872 | """ |
| 1873 | This tests uploading a small stream from stdin. |
| 1874 | """ |
| 1875 | p = aws( |
| 1876 | 's3 cp - s3://%s/stream' % shared_bucket, |
| 1877 | input_data=b'This is a test', |
| 1878 | ) |
| 1879 | self.assert_no_errors(p) |
| 1880 | assert s3_utils.key_exists(shared_bucket, 'stream') |
| 1881 | contents = s3_utils.get_key_contents(shared_bucket, 'stream') |
| 1882 | assert contents == 'This is a test' |
| 1883 | |
| 1884 | def test_unicode_upload(self, s3_utils, shared_bucket): |
| 1885 | """ |
nothing calls this directly
no test coverage detected