ConfigurationOsFile is added

Author: d.kovalenko

src/implementation/v00/configuration_base.py

@@ -58,11 +58,11 @@
 from ...core.bugcheck_error import BugCheckError
 # fmt: on
 
+from ...os.abstract.configuration_os_ops import ConfigurationFileReader
+from ...os.abstract.configuration_os_ops import ConfigurationOsFile
 from ...os.abstract.configuration_os_ops import ConfigurationOsOps
 
 import typing
-import os
-import io
 import datetime
 
 # //////////////////////////////////////////////////////////////////////////////
@@ -3223,17 +3223,14 @@ def LoadConfigurationFile(
             # load content
             # process content
 
-            with open(currentFileData.m_Path) as f:
-                assert isinstance(f, io.TextIOBase)
+            with cfg.m_Data.OsOps.OpenFileToRead(currentFileData.m_Path) as f:
+                assert isinstance(f, ConfigurationOsFile)
                 currentFile = PostgresConfigurationFactory_Base.GetObject(
                     cfg, currentFileData
                 )
                 __class__.Helper__LoadFileContent(currentFile, f)  # raise
 
-                fd = f.fileno()
-                assert type(fd) == int
-
-                lastMDate = datetime.datetime.fromtimestamp(os.path.getmtime(fd))
+                lastMDate = f.GetModificationTS()
                 assert type(lastMDate) == datetime.datetime
 
                 currentFileData.m_LastModifiedTimestamp = lastMDate
@@ -3289,23 +3286,23 @@ def LoadConfigurationFile(
 
     # --------------------------------------------------------------------
     def LoadFileContent(
-        file: PostgresConfigurationFile_Base, fileContent: io.TextIOBase
+        file: PostgresConfigurationFile_Base, fileContent: ConfigurationFileReader
     ) -> None:
         assert isinstance(file, PostgresConfigurationFile_Base)
-        assert isinstance(fileContent, io.TextIOBase)
+        assert isinstance(fileContent, ConfigurationFileReader)
 
         return __class__.Helper__LoadFileContent(file, fileContent)
 
     # Helper methods -----------------------------------------------------
     def Helper__LoadFileContent(
-        file: PostgresConfigurationFile_Base, fileContent: io.TextIOBase
+        file: PostgresConfigurationFile_Base, fileContent: ConfigurationFileReader
     ) -> None:
         assert isinstance(file, PostgresConfigurationFile_Base)
-        assert isinstance(fileContent, io.TextIOBase)
+        assert isinstance(fileContent, ConfigurationFileReader)
 
         lineReader = ReadUtils__LineReader()
 
-        while lineData := fileContent.readline():
+        while lineData := fileContent.ReadLine():
             assert type(lineData) == str
 
             lineReader.SetData(lineData)
@@ -3727,7 +3724,7 @@ def Helper__ProcessLineData__Option__Generic(
 class PostgresConfigurationWriterFileCtx_Base:
     FileData: PgCfgModel__FileData
     Content: typing.Optional[str]
-    File: typing.Optional[io.TextIOWrapper]
+    File: typing.Optional[ConfigurationOsFile]
 
     # TODO: We can use filelock (it is a separated lock file)
     # to provide an exclusive access to our File
@@ -3891,15 +3888,12 @@ def Helper__DoWork__Stage03__OpenUpdFilesToWrite(
             assert fileCtx.File is None
 
             # Let's open an exist file to read and write without truncation
-            fileCtx.File = open(fileCtx.FileData.m_Path, "r+")  # raise
+            fileCtx.File = ctx.Cfg.m_Data.OsOps.OpenFileToWrite(fileCtx.FileData.m_Path)  # raise
 
             assert fileCtx.File is not None
-            assert isinstance(fileCtx.File, io.TextIOWrapper)
-
-            fd = fileCtx.File.fileno()
-            assert type(fd) == int
+            assert isinstance(fileCtx.File, ConfigurationOsFile)
 
-            lastMDate = datetime.datetime.fromtimestamp(os.path.getmtime(fd))
+            lastMDate = fileCtx.File.GetModificationTS()
             assert type(lastMDate) == datetime.datetime
 
             if fileCtx.FileData.m_LastModifiedTimestamp != lastMDate:
@@ -3937,10 +3931,10 @@ def Helper__DoWork__Stage04__OpenNewFilesToWrite(
                 assert fileCtx.FileData.m_Status == PgCfgModel__FileStatus.IS_NEW
                 assert fileCtx.File is None
 
-                fileCtx.File = open(fileCtx.FileData.m_Path, "x")  # raise
+                fileCtx.File = ctx.Cfg.m_Data.OsOps.CreateFile(fileCtx.FileData.m_Path)  # raise                
 
                 assert fileCtx.File is not None
-                assert isinstance(fileCtx.File, io.TextIOWrapper)
+                assert isinstance(fileCtx.File, ConfigurationOsFile)
 
                 # OK
                 continue
@@ -3959,16 +3953,16 @@ def Helper__DoWork__Stage04__OpenNewFilesToWrite(
 
                 assert fileCtx.FileData.m_Status == PgCfgModel__FileStatus.IS_NEW
                 assert fileCtx.File is not None
-                assert isinstance(fileCtx.File, io.TextIOWrapper)
+                assert isinstance(fileCtx.File, ConfigurationOsFile)
 
-                assert not fileCtx.File.closed
+                assert not fileCtx.File.IsClosed
 
-                filePath = fileCtx.File.name
+                filePath = fileCtx.File.Name
                 assert filePath is not None
                 assert type(filePath) == str
                 assert filePath == fileCtx.FileData.m_Path
 
-                fileCtx.File.close()  # raise
+                fileCtx.File.Close()  # raise
 
                 ctx.Cfg.m_Data.OsOps.Remove(filePath)  # raise
                 continue
@@ -3992,22 +3986,17 @@ def Helper__DoWork__Stage05__WriteContents(
             assert type(fileCtx.FileData) == PgCfgModel__FileData
 
             assert fileCtx.File is not None
-            assert isinstance(fileCtx.File, io.TextIOWrapper)
+            assert isinstance(fileCtx.File, ConfigurationOsFile)
 
             assert fileCtx.Content is not None
             assert type(fileCtx.Content) == str
 
-            fileCtx.File.write(fileCtx.Content)
-            fileCtx.File.truncate()
-            fileCtx.File.flush()
-
-            fd = fileCtx.File.fileno()
-            assert type(fd) == int
+            fileCtx.File.Overwrite(fileCtx.Content)
 
-            lastMDate = datetime.datetime.fromtimestamp(os.path.getmtime(fd))
+            lastMDate = fileCtx.File.GetModificationTS()
             assert type(lastMDate) == datetime.datetime
 
-            fileCtx.File.close()
+            fileCtx.File.Close()
 
             fileCtx.FileData.m_LastModifiedTimestamp = lastMDate
             fileCtx.FileData.m_Status = PgCfgModel__FileStatus.EXISTS

src/os/abstract/configuration_os_ops.py

@@ -5,6 +5,45 @@
 
 from ...core.raise_error import RaiseError
 
+import datetime
+
+
+# //////////////////////////////////////////////////////////////////////////////
+# class ConfigurationFileReader
+
+
+class ConfigurationFileReader:
+    def ReadLine(self) -> str:
+        RaiseError.MethodIsNotImplemented(__class__, "ReadLine")
+
+
+# //////////////////////////////////////////////////////////////////////////////
+# class ConfigurationOsFile
+
+
+class ConfigurationOsFile(ConfigurationFileReader):
+    def __init__(self):
+        pass
+
+    @property
+    def Name(self) -> str:
+        RaiseError.MethodIsNotImplemented(__class__, "get_Name")
+
+    @property
+    def IsClosed(self) -> bool:
+        RaiseError.MethodIsNotImplemented(__class__, "get_IsClosed")
+
+    def Overwrite(self, text: str) -> None:
+        assert type(text) == str  # noqa: E721
+        RaiseError.MethodIsNotImplemented(__class__, "Write")
+
+    def Close(self):
+        RaiseError.MethodIsNotImplemented(__class__, "Close")
+
+    def GetModificationTS(self) -> datetime.datetime:
+        RaiseError.MethodIsNotImplemented(__class__, "GetModificationTS")
+
+
 # //////////////////////////////////////////////////////////////////////////////
 # class ConfigurationOsOps
 
@@ -43,5 +82,16 @@ def Remove(self, a: str) -> str:
         assert type(a) == str  # noqa: E721
         RaiseError.MethodIsNotImplemented(__class__, "Remove")
 
+    def OpenFileToRead(self, filePath: str) -> ConfigurationOsFile:
+        assert type(filePath) == str  # noqa: E721
+        RaiseError.MethodIsNotImplemented(__class__, "OpenFileToRead")
+
+    def OpenFileToWrite(self, filePath: str) -> ConfigurationOsFile:
+        assert type(filePath) == str  # noqa: E721
+        RaiseError.MethodIsNotImplemented(__class__, "OpenFileToWrite")
+
+    def CreateFile(self, filePath: str) -> ConfigurationOsFile:
+        assert type(filePath) == str  # noqa: E721
+        RaiseError.MethodIsNotImplemented(__class__, "CreateFile")
 
 # //////////////////////////////////////////////////////////////////////////////

src/os/local/configuration_os_ops.py

@@ -6,6 +6,82 @@
 from ..abstract import configuration_os_ops as abstract
 
 import os
+import io
+import datetime
+
+# //////////////////////////////////////////////////////////////////////////////
+# class ConfigurationOsFile
+
+
+class ConfigurationOsFile(abstract.ConfigurationOsFile):
+    m_file: io.TextIOWrapper
+
+    # --------------------------------------------------------------------
+    def __init__(self, file: io.TextIOWrapper):
+        assert isinstance(file, io.TextIOWrapper)
+
+        super().__init__()
+
+        self.m_file = file
+
+    # --------------------------------------------------------------------
+    def __enter__(self) -> ConfigurationOsFile:
+        assert isinstance(self.m_file, io.TextIOWrapper)
+        return self
+
+    # --------------------------------------------------------------------
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.m_file is not None:
+            self.Close()
+
+        assert self.m_file is None
+        return
+
+    # --------------------------------------------------------------------
+    @property
+    def Name(self) -> str:
+        assert isinstance(self.m_file, io.TextIOWrapper)
+        return self.m_file.name
+
+    # --------------------------------------------------------------------
+    @property
+    def IsClosed(self) -> bool:
+        return self.m_file is None
+
+    # --------------------------------------------------------------------
+    def ReadLine(self) -> str:
+        assert isinstance(self.m_file, io.TextIOWrapper)
+        return self.m_file.readline()
+
+    # --------------------------------------------------------------------
+    def Overwrite(self, text: str) -> None:
+        assert type(text) == str  # noqa: E721
+        assert isinstance(self.m_file, io.TextIOWrapper)
+        self.m_file.seek(0)
+        self.m_file.write(text)
+        self.m_file.truncate()
+        self.m_file.flush()
+
+    # --------------------------------------------------------------------
+    def Close(self):
+        assert isinstance(self.m_file, io.TextIOWrapper)
+        f = self.m_file
+        self.m_file = None
+        f.close()
+        # Returning False (or None) re-raises the exception
+        return False
+
+    # --------------------------------------------------------------------
+    def GetModificationTS(self) -> datetime.datetime:
+        assert isinstance(self.m_file, io.TextIOWrapper)
+
+        fd = self.m_file.fileno()
+        assert type(fd) == int  # noqa: E721
+
+        lastMDate = datetime.datetime.fromtimestamp(os.path.getmtime(fd))
+        assert type(lastMDate) == datetime.datetime  # noqa: E721
+        return lastMDate
+
 
 # //////////////////////////////////////////////////////////////////////////////
 # class ConfigurationOsOps
@@ -45,6 +121,21 @@ def Remove(self, a: str) -> str:
         assert type(a) == str  # noqa: E721
         os.remove(a)
 
+    def OpenFileToRead(self, filePath: str) -> ConfigurationOsFile:
+        assert type(filePath) == str  # noqa: E721
+        f = open(filePath)
+        return ConfigurationOsFile(f)
+
+    def OpenFileToWrite(self, filePath: str) -> ConfigurationOsFile:
+        assert type(filePath) == str  # noqa: E721
+        f = open(filePath, mode="r+")
+        return ConfigurationOsFile(f)
+
+    def CreateFile(self, filePath: str) -> ConfigurationOsFile:
+        assert type(filePath) == str  # noqa: E721
+        f = open(filePath, mode="x")
+        return ConfigurationOsFile(f)
+
 
 # //////////////////////////////////////////////////////////////////////////////
 

tests/CfgFileReader.py

@@ -0,0 +1,27 @@
+# /////////////////////////////////////////////////////////////////////////////
+# Postgres Pro. PostgreSQL Configuration Python Library. Tests.
+
+from ..src.os.abstract.configuration_os_ops import ConfigurationFileReader
+
+import io
+
+# //////////////////////////////////////////////////////////////////////////////
+# class CfgFileReader
+
+
+class CfgFileReader(ConfigurationFileReader):
+    m_file: io.StringIO
+
+    # --------------------------------------------------------------------
+    def __init__(self, text: str):
+        super().__init__()
+
+        self.m_file = io.StringIO(text)
+
+    # --------------------------------------------------------------------
+    def ReadLine(self) -> str:
+        assert type(self.m_file) == io.StringIO
+        return self.m_file.readline()
+
+
+# //////////////////////////////////////////////////////////////////////////////