OsOperations::get_file_size(self, filename) is added

It is a function to get a size of file.

Author: dmitry-lipetsk

testgres/operations/local_ops.py

@@ -325,6 +325,11 @@ def isfile(self, remote_file):
     def isdir(self, dirname):
         return os.path.isdir(dirname)
 
+    def get_file_size(self, filename):
+        assert filename is not None
+        assert type(filename) == str  # noqa: E721
+        return os.path.getsize(filename)
+
     def remove_file(self, filename):
         return os.remove(filename)
 

testgres/operations/os_ops.py

@@ -107,6 +107,9 @@ def read_binary(self, filename, start_pos):
     def isfile(self, remote_file):
         raise NotImplementedError()
 
+    def get_file_size(self, filename):
+        raise NotImplementedError()
+
     # Processes control
     def kill(self, pid, signal):
         # Kill the process

testgres/operations/remote_ops.py

@@ -360,6 +360,70 @@ def isdir(self, dirname):
         response = self.exec_command(cmd)
         return response.strip() == b"True"
 
+    def get_file_size(self, filename):
+        C_ERR_SRC = "RemoteOpertions::get_file_size"
+
+        assert filename is not None
+        assert type(filename) == str  # noqa: E721
+        cmd = "du -b " + __class__._escape_path(filename)
+
+        s = self.exec_command(cmd, encoding=get_default_encoding())
+        assert type(s) == str  # noqa: E721
+
+        if len(s) == 0:
+            raise Exception(
+                "[BUG CHECK] Can't get size of file [{2}]. Remote operation returned an empty string. Check point [{0}][{1}].".format(
+                    C_ERR_SRC,
+                    "#001",
+                    filename
+                )
+            )
+
+        i = 0
+
+        while i < len(s) and s[i].isdigit():
+            assert s[i] >= '0'
+            assert s[i] <= '9'
+            i += 1
+
+        if i == 0:
+            raise Exception(
+                "[BUG CHECK] Can't get size of file [{2}]. Remote operation returned a bad formatted string. Check point [{0}][{1}].".format(
+                    C_ERR_SRC,
+                    "#002",
+                    filename
+                )
+            )
+
+        if i == len(s):
+            raise Exception(
+                "[BUG CHECK] Can't get size of file [{2}]. Remote operation returned a bad formatted string. Check point [{0}][{1}].".format(
+                    C_ERR_SRC,
+                    "#003",
+                    filename
+                )
+            )
+
+        if not s[i].isspace():
+            raise Exception(
+                "[BUG CHECK] Can't get size of file [{2}]. Remote operation returned a bad formatted string. Check point [{0}][{1}].".format(
+                    C_ERR_SRC,
+                    "#004",
+                    filename
+                )
+            )
+
+        r = 0
+
+        for i2 in range(0, i):
+            ch = s[i2]
+            assert ch >= '0'
+            assert ch <= '9'
+            # Here is needed to check overflow or that it is a human-valid result?
+            r = (r * 10) + ord(ch) - ord('0')
+
+        return r
+
     def remove_file(self, filename):
         cmd = "rm {}".format(filename)
         return self.exec_command(cmd)

tests/test_local.py

@@ -97,3 +97,24 @@ def test_read_binary__spec__unk_file(self):
 
         with pytest.raises(FileNotFoundError, match=re.escape("[Errno 2] No such file or directory: '/dummy'")):
             self.operations.read_binary("/dummy", 0)
+
+    def test_get_file_size(self):
+        """
+        Test LocalOperations::get_file_size.
+        """
+        filename = __file__  # current file
+
+        sz0 = os.path.getsize(filename)
+        assert type(sz0) == int  # noqa: E721
+
+        sz1 = self.operations.get_file_size(filename)
+        assert type(sz1) == int  # noqa: E721
+        assert sz1 == sz0
+
+    def test_get_file_size__unk_file(self):
+        """
+        Test LocalOperations::get_file_size.
+        """
+
+        with pytest.raises(FileNotFoundError, match=re.escape("[Errno 2] No such file or directory: '/dummy'")):
+            self.operations.get_file_size("/dummy")

tests/test_remote.py

@@ -224,6 +224,27 @@ def test_read_binary__spec__unk_file(self):
         with pytest.raises(ExecUtilException, match=re.escape("tail: cannot open '/dummy' for reading: No such file or directory")):
             self.operations.read_binary("/dummy", 0)
 
+    def test_get_file_size(self):
+        """
+        Test LocalOperations::get_file_size.
+        """
+        filename = __file__  # current file
+
+        sz0 = os.path.getsize(filename)
+        assert type(sz0) == int  # noqa: E721
+
+        sz1 = self.operations.get_file_size(filename)
+        assert type(sz1) == int  # noqa: E721
+        assert sz1 == sz0
+
+    def test_get_file_size__unk_file(self):
+        """
+        Test LocalOperations::get_file_size.
+        """
+
+        with pytest.raises(ExecUtilException, match=re.escape("du: cannot access '/dummy': No such file or directory")):
+            self.operations.get_file_size("/dummy")
+
     def test_touch(self):
         """
         Test touch for creating a new file or updating access and modification times of an existing file.