OsOperations::read_binary supports 'size' argument (#117)

Declarations:
    def read_binary(
        self,
        filename: str,
        offset: int,
        size: typing.Optional[int] = None,
    ) -> bytes:

Tests are added.

Author: dmitry-lipetsk

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
 

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:

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)

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,