OLD | NEW |
| (Empty) |
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 """Unit tests for FilePart class.""" | |
16 | |
17 from __future__ import absolute_import | |
18 | |
19 import os | |
20 | |
21 from gslib.file_part import FilePart | |
22 import gslib.tests.testcase as testcase | |
23 | |
24 | |
25 # pylint: disable=protected-access | |
26 class TestFilePart(testcase.GsUtilUnitTestCase): | |
27 """Unit tests for FilePart class.""" | |
28 | |
29 def test_tell(self): | |
30 filename = 'test_tell' | |
31 contents = 100 * 'x' | |
32 fpath = self.CreateTempFile(file_name=filename, contents=contents) | |
33 part_length = 23 | |
34 start_pos = 50 | |
35 fp = FilePart(fpath, start_pos, part_length) | |
36 self.assertEqual(start_pos, fp._fp.tell()) | |
37 self.assertEqual(0, fp.tell()) | |
38 | |
39 def test_seek(self): | |
40 """Tests seeking in a FilePart.""" | |
41 filename = 'test_seek' | |
42 contents = 100 * 'x' | |
43 part_length = 23 | |
44 start_pos = 50 | |
45 fpath = self.CreateTempFile(file_name=filename, contents=contents) | |
46 fp = FilePart(fpath, start_pos, part_length) | |
47 offset = 10 | |
48 | |
49 # Absolute positioning. | |
50 fp.seek(offset) | |
51 self.assertEqual(start_pos + offset, fp._fp.tell()) | |
52 self.assertEqual(offset, fp.tell()) | |
53 | |
54 # Relative positioning. | |
55 fp.seek(offset, whence=os.SEEK_CUR) | |
56 self.assertEqual(start_pos + 2 * offset, fp._fp.tell()) | |
57 self.assertEqual(2 * offset, fp.tell()) | |
58 | |
59 # Absolute positioning from EOF. | |
60 fp.seek(-offset, whence=os.SEEK_END) | |
61 self.assertEqual(start_pos + part_length - offset, fp._fp.tell()) | |
62 self.assertEqual(part_length - offset, fp.tell()) | |
63 | |
64 # Seek past EOF. | |
65 fp.seek(1, whence=os.SEEK_END) | |
66 self.assertEqual(start_pos + part_length + 1, fp._fp.tell()) | |
67 self.assertEqual(part_length + 1, fp.tell()) | |
68 | |
69 def test_read(self): | |
70 """Tests various reaad operations with FilePart.""" | |
71 filename = 'test_read' | |
72 contents = '' | |
73 for i in range(1, 256): | |
74 contents += str(i) | |
75 part_length = 23 | |
76 start_pos = 50 | |
77 fpath = self.CreateTempFile(file_name=filename, contents=contents) | |
78 | |
79 # Read in the whole file. | |
80 fp = FilePart(fpath, start_pos, part_length) | |
81 whole_file = fp.read() | |
82 self.assertEqual(contents[start_pos:(start_pos + part_length)], whole_file) | |
83 | |
84 # Read in a piece of the file from the beginning. | |
85 fp.seek(0) | |
86 offset = 10 | |
87 partial_file = fp.read(offset) | |
88 self.assertEqual( | |
89 contents[start_pos:(start_pos + offset)], | |
90 partial_file) | |
91 | |
92 # Read in the rest of the file. | |
93 remaining_file = fp.read(part_length - offset) | |
94 self.assertEqual( | |
95 contents[(start_pos + offset):(start_pos + part_length)], | |
96 remaining_file) | |
97 self.assertEqual( | |
98 contents[start_pos:(start_pos + part_length)], | |
99 partial_file + remaining_file) | |
100 | |
101 # Try to read after reaching EOF. | |
102 empty_file = fp.read(100) | |
103 self.assertEqual('', empty_file) | |
104 | |
105 empty_file = fp.read() | |
106 self.assertEqual('', empty_file) | |
OLD | NEW |