diff --git a/src/local_ops.py b/src/local_ops.py index 6426c4e..25e805c 100644 --- a/src/local_ops.py +++ b/src/local_ops.py @@ -706,16 +706,24 @@ def readlines( ) # Adjust buffer size return - def read_binary(self, filename: str, offset: int) -> bytes: + def read_binary( + self, + filename: str, + offset: int, + size: typing.Optional[int] = None, + ) -> bytes: assert type(filename) is str assert type(offset) is int + assert size is None or type(size) is int if offset < 0: raise ValueError("Negative 'offset' is not supported.") + if size is not None and size < 0: + raise ValueError("Negative 'size' is not supported.") with open(filename, 'rb') as file: # open in a binary mode file.seek(offset, os.SEEK_SET) - r = file.read() + r = file.read(size) assert type(r) is bytes return r diff --git a/src/os_ops.py b/src/os_ops.py index 9219bb1..f7a72f3 100644 --- a/src/os_ops.py +++ b/src/os_ops.py @@ -292,10 +292,17 @@ def readlines( assert num_lines >= 0 raise NotImplementedError() - def read_binary(self, filename: str, offset: int) -> bytes: + def read_binary( + self, + filename: str, + offset: int, + size: typing.Optional[int] = None, + ) -> bytes: assert type(filename) is str assert type(offset) is int + assert size is None or type(size) is int assert offset >= 0 + assert size is None or size >= 0 raise NotImplementedError() def isfile(self, filename: str) -> bool: diff --git a/src/remote_ops.py b/src/remote_ops.py index 0551004..40c1a43 100644 --- a/src/remote_ops.py +++ b/src/remote_ops.py @@ -931,16 +931,27 @@ def readlines( assert type(lines) is list return lines - def read_binary(self, filename: str, offset: int) -> bytes: + def read_binary( + self, + filename: str, + offset: int, + size: typing.Optional[int] = None, + ) -> bytes: assert type(filename) is str assert type(offset) is int + assert size is None or type(size) is int if offset < 0: raise ValueError("Negative 'offset' is not supported.") + if size is not None and size < 0: + raise ValueError("Negative 'size' is not supported.") filename_q = __class__._quote_path(filename) cmd_p = ["tail", "-c", "+{}".format(offset + 1), filename_q] + if size is not None: + cmd_p.append("| head -c {}".format(size)) + cmd = " ".join(cmd_p) r = self.exec_command(cmd) diff --git a/tests/test_os_ops_common.py b/tests/test_os_ops_common.py index 9f1cfc5..46db01f 100644 --- a/tests/test_os_ops_common.py +++ b/tests/test_os_ops_common.py @@ -1250,6 +1250,42 @@ def test_read_binary__spec( assert type(response5) is bytes assert len(response5) == 0 + response6 = os_ops.read_binary(filename, 0, 1) + assert type(response6) is bytes + assert len(response6) == 1 + assert response6 == response0[:1] + + r = os_ops.read_binary(filename, 10, 7) + assert type(r) is bytes + assert len(r) == 7 + assert r == response0[10:17] + + r = os_ops.read_binary(filename, 10, len(response0)) + assert type(r) is bytes + assert len(r) == len(response0) - 10 + assert r == response0[10:] + + r = os_ops.read_binary(filename, 10, 2 * len(response0)) + assert type(r) is bytes + assert len(r) == len(response0) - 10 + assert r == response0[10:] + + r = os_ops.read_binary(filename, len(response0) - 1, 2 * len(response0)) + assert type(r) is bytes + assert len(r) == 1 + assert r == response0[-1:] + assert r[0] == response0[-1] + + r = os_ops.read_binary(filename, len(response0), 1) + assert type(r) is bytes + assert len(r) == 0 + assert r == b'' + + r = os_ops.read_binary(filename, len(response0) + 1, 1) + assert type(r) is bytes + assert len(r) == 0 + assert r == b'' + os_ops.remove_file(filename) return @@ -1278,6 +1314,31 @@ def test_read_binary__spec__negative_offset( os_ops.remove_file(filename) return + def test_read_binary__spec__negative_size( + self, + os_ops_descr: OsOpsDescr, + name_with_surprize: tagNameWithSurprize, + ): + """ + Test OsOperations::read_binary with negative size. + """ + assert type(os_ops_descr) is OsOpsDescr + assert isinstance(os_ops_descr.os_ops, OsOperations) + assert type(name_with_surprize) is __class__.tagNameWithSurprize + + os_ops = os_ops_descr.os_ops + assert isinstance(os_ops, OsOperations) + + filename = os_ops.mkstemp(name_with_surprize.value) + + with pytest.raises( + ValueError, + match=re.escape("Negative 'size' is not supported.")): + os_ops.read_binary(filename, 0, size=-1) + + os_ops.remove_file(filename) + return + def test_get_file_size( self, os_ops_descr: OsOpsDescr,