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
17from __future__ import absolute_import
18
19import os
20
21from gslib.file_part import FilePart
22import gslib.tests.testcase as testcase
23
24
25# pylint: disable=protected-access
26class 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)
107