diff --git a/src/implementation/v00/configuration_base.py b/src/implementation/v00/configuration_base.py index c242980..aa65d04 100644 --- a/src/implementation/v00/configuration_base.py +++ b/src/implementation/v00/configuration_base.py @@ -22,13 +22,11 @@ from ...core.model import ConfigurationData as PgCfgModel__ConfigurationData from ...core.model import ObjectData as PgCfgModel__ObjectData -from ...core.model import FileLineElementData as PgCfgModel__FileLineElementData from ...core.model import CommentData as PgCfgModel__CommentData from ...core.model import OptionData as PgCfgModel__OptionData from ...core.model import IncludeData as PgCfgModel__IncludeData from ...core.model import FileLineData as PgCfgModel__FileLineData from ...core.model import FileData as PgCfgModel__FileData -from ...core.model import FileStatus as PgCfgModel__FileStatus from ...core.handlers import OptionHandlerToPrepareSetValue as PgCfgModel__OptionHandlerToPrepareSetValue from ...core.handlers import OptionHandlerCtxToPrepareSetValue as PgCfgModel__OptionHandlerCtxPrepareToSetValue @@ -45,25 +43,18 @@ from ...core.handlers import OptionHandlerToSetValueItem as PgCfgModel__OptionHandlerToSetValueItem from ...core.handlers import OptionHandlerCtxToSetValueItem as PgCfgModel__OptionHandlerCtxToSetValueItem from ...core.handlers import OptionHandlerToWrite as PgCfgModel__OptionHandlerToWrite -from ...core.handlers import OptionHandlerCtxToWrite as PgCfgModel__OptionHandlerCtxToWrite from ...core.handlers import ConfigurationDataHandler as PgCfgModel__DataHandler from ...core.controller_utils import DataControllerUtils as PgCfgModel__DataControllerUtils from ...core.data_verificator import DataVerificator from ...core.helpers import Helpers -from ...core.write_utils import WriteUtils -from ...core.read_utils import ReadUtils -from ...core.read_utils import LineReader as ReadUtils__LineReader from ...core.raise_error import RaiseError 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 datetime # ////////////////////////////////////////////////////////////////////////////// # class PostgresConfigurationComment_Base @@ -3633,1203 +3624,6 @@ def Helper__CreateOption( ) -# ////////////////////////////////////////////////////////////////////////////// -# class PostgresConfigurationReader_Base - - -class PostgresConfigurationReader_Base: - @staticmethod - def LoadConfigurationFile( - cfg: PostgresConfiguration_Base, - filePath: str, - ) -> PostgresConfigurationTopLevelFile_Base: - assert cfg is not None - - assert isinstance(cfg, PostgresConfiguration_Base) - assert type(filePath) is str - assert filePath != "" - - existFileDatas: typing.Dict[str, PgCfgModel__FileData] = dict() - - for fileName in cfg.m_Data.m_AllFilesByName.keys(): - assert type(fileName) is str - assert fileName != "" - - indexData = cfg.m_Data.m_AllFilesByName[fileName] - assert indexData is not None - - if type(indexData) is PgCfgModel__FileData: - fileData: PgCfgModel__FileData = indexData - assert type(fileData.m_Path) is str - assert fileData.m_Path not in existFileDatas.keys() - existFileDatas[fileData.m_Path] = fileData - continue - - if type(indexData) is list: - for fileData in indexData: - assert type(fileData) is PgCfgModel__FileData - assert type(fileData.m_Path) is str - assert fileData.m_Path not in existFileDatas.keys() - existFileDatas[fileData.m_Path] = fileData - continue - continue - - BugCheckError.UnkFileObjectDataType( - fileName, - type(indexData), - ) - - # ---------------------------------------------------------------- - filePath_n = Helpers.NormalizeFilePath( - cfg.m_Data.OsOps, - cfg.m_Data.m_DataDir, - filePath, - ) - assert type(filePath_n) is str - - if filePath_n in existFileDatas: - file = PostgresConfigurationFactory_Base.GetObject( - cfg, - existFileDatas[filePath_n], - ) - assert type(file) is PostgresConfigurationTopLevelFile_Base - return file - - # ---------------------------------------------------------------- - rootFile = cfg.AddTopLevelFile(filePath) - assert type(rootFile) is PostgresConfigurationTopLevelFile_Base - assert type(rootFile.m_FileData) is PgCfgModel__FileData - assert type(rootFile.m_FileData.m_Lines) is list - assert rootFile.m_FileData.m_Status == PgCfgModel__FileStatus.IS_NEW - assert rootFile.m_FileData.m_LastModifiedTimestamp is None - assert len(rootFile.m_FileData.m_Lines) == 0 - - queuedFileDatas: typing.Set[PgCfgModel__FileData] = set() - - queuedFileDatas.add(rootFile.m_FileData) - - while len(queuedFileDatas) != 0: - currentFileData = queuedFileDatas.pop() - assert isinstance(currentFileData, PgCfgModel__FileData) - - # load content - # process content - - with cfg.m_Data.OsOps.OpenFileToRead(currentFileData.m_Path) as f: - assert isinstance(f, ConfigurationOsFile) - currentFile = PostgresConfigurationFactory_Base.GetObject( - cfg, - currentFileData, - ) - assert isinstance(currentFile, PostgresConfigurationFile_Base) - __class__.Helper__LoadFileContent( - currentFile, - f, - ) # raise - - lastMDate = f.GetModificationTS() - assert type(lastMDate) is datetime.datetime - - currentFileData.m_LastModifiedTimestamp = lastMDate - currentFileData.m_Status = PgCfgModel__FileStatus.EXISTS - - assert currentFileData.m_Path not in existFileDatas.keys() - existFileDatas[currentFileData.m_Path] = currentFileData - - # enumerate all the includes - for fileLineData in currentFileData.m_Lines: - assert type(fileLineData) is PgCfgModel__FileLineData - assert type(fileLineData.m_Items) is list - - for fileLineItem in fileLineData.m_Items: - assert type(fileLineItem) is PgCfgModel__FileLineData.tagItem - - fileLineElementData = fileLineItem.m_Element - - if type(fileLineElementData) is PgCfgModel__CommentData: - continue - - if type(fileLineElementData) is PgCfgModel__OptionData: - continue - - if type(fileLineElementData) is PgCfgModel__IncludeData: - # look at existFileDatas - includeData: PgCfgModel__IncludeData = fileLineElementData - assert type(includeData.m_File) is PgCfgModel__FileData - assert type(includeData.m_File.m_Path) is str - - if includeData.m_File.m_Path in existFileDatas: - continue # it is an old file - - assert ( - includeData.m_File.m_Status == PgCfgModel__FileStatus.IS_NEW - ) - assert includeData.m_File.m_LastModifiedTimestamp is None - - if includeData.m_File in queuedFileDatas: - continue # already in queue - - queuedFileDatas.add(includeData.m_File) - continue - - BugCheckError.UnkObjectDataType(type(fileLineElementData)) - - assert len(queuedFileDatas) == 0 - assert isinstance(rootFile, PostgresConfigurationTopLevelFile_Base) - - return rootFile - - # -------------------------------------------------------------------- - @staticmethod - def LoadFileContent( - file: PostgresConfigurationFile_Base, - fileContent: ConfigurationFileReader, - ) -> None: - assert isinstance(file, PostgresConfigurationFile_Base) - assert isinstance(fileContent, ConfigurationFileReader) - - return __class__.Helper__LoadFileContent( - file, - fileContent, - ) - - # Helper methods ----------------------------------------------------- - @staticmethod - def Helper__LoadFileContent( - file: PostgresConfigurationFile_Base, - fileContent: ConfigurationFileReader, - ) -> None: - assert isinstance(file, PostgresConfigurationFile_Base) - assert isinstance(fileContent, ConfigurationFileReader) - - lineReader = ReadUtils__LineReader() - - while True: - lineData = fileContent.ReadLine() - - if not lineData: - assert lineData is None - break - - assert type(lineData) is str - - lineReader.SetData(lineData) - - __class__.Helper__ProcessLineData( - file, - lineReader, - ) - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__ProcessLineData( - file: PostgresConfigurationFile_Base, - lineReader: ReadUtils__LineReader, - ) -> None: - assert isinstance(file, PostgresConfigurationFile_Base) - assert type(lineReader) is ReadUtils__LineReader - - fileLine = file.AddEmptyLine() - assert type(fileLine) is PostgresConfigurationFileLine_Base - - ch: typing.Optional[str] - - while True: - # skeep spaces - while True: - ch = lineReader.ReadSymbol() - if ch is None: - return - - if ReadUtils.IsSpace(ch): - continue - - break - - assert not ReadUtils.IsSpace(ch) - - if ReadUtils.IsEOL(ch): - return - - if ch == "#": - return __class__.Helper__ProcessLineData__Comment( - fileLine, - lineReader, - ) - - if len(fileLine) == 0 and ReadUtils.IsValidSeqCh1(ch): - # Read sequence - sequenceOffset = lineReader.GetColOffset() - sequence = ch - - while True: - ch = lineReader.ReadSymbol() - - if not ch: - assert ch is None - break - - assert type(ch) is str - - if ReadUtils.IsValidSeqCh2(ch): - sequence += ch - continue - - if ch == ".": - sequence += ch - continue - - break - - if sequence.lower() == "include": - if ch is not None: - lineReader.StepBack() - - __class__.Helper__ProcessLineData__Include( - fileLine, - lineReader, - sequenceOffset, - ) - else: - if ch is not None: - lineReader.StepBack() - - __class__.Helper__ProcessLineData__Option( - fileLine, - lineReader, - sequenceOffset, - sequence, - ) - continue - - RaiseError.CfgReader__UnexpectedSymbol( - lineReader.GetLineNum(), - lineReader.GetColNum(), - ch, - ) - - # -------------------------------------------------------------------- - @staticmethod - def Helper__ProcessLineData__Comment( - fileLine: PostgresConfigurationFileLine_Base, - lineReader: ReadUtils__LineReader, - ) -> None: - assert type(fileLine) is PostgresConfigurationFileLine_Base - assert type(lineReader) is ReadUtils__LineReader - - commentText = "" - commentOffset = lineReader.GetColOffset() - - while True: - ch = lineReader.ReadSymbol() - - if not ch: - assert ch is None - break - - assert type(ch) is str - - if ReadUtils.IsEOL(ch): - break - commentText += ch - - fileLine.AddComment( - commentText, - commentOffset, - ) - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__ProcessLineData__Include( - fileLine: PostgresConfigurationFileLine_Base, - lineReader: ReadUtils__LineReader, - includeOffset: int, - ) -> None: - assert type(fileLine) is PostgresConfigurationFileLine_Base - assert type(lineReader) is ReadUtils__LineReader - assert type(includeOffset) is int - assert includeOffset >= 0 - - # find first quote - while True: - ch = lineReader.ReadSymbol() - - if ch is None or ReadUtils.IsEOL(ch) or ch == "#": - RaiseError.CfgReader__IncludeWithoutPath( - lineReader.GetLineNum(), - ) - - if ReadUtils.IsSpace(ch): - continue - - if ch == "'": - break - - RaiseError.CfgReader__UnexpectedSymbol( - lineReader.GetLineNum(), - lineReader.GetColNum(), - ch, - ) - - # OK. First quote is found! - - # Let's process this quoted string - filePath = "" - - while True: - ch = lineReader.ReadSymbol() - - if ch is None or ReadUtils.IsEOL(ch): - RaiseError.CfgReader__EndOfIncludePathIsNotFound( - lineReader.GetLineNum(), - ) - break - - if ch == "'": - ch2 = lineReader.ReadSymbol() - - if ch2 is None: - break - - if ReadUtils.IsEOL(ch2): - lineReader.StepBack() - break - - if ch2 == "'": - filePath += ch - continue - - break - - if ch == "\\": - ch = lineReader.ReadSymbol() - - if ch is None or ReadUtils.IsEOL(ch): - RaiseError.CfgReader__IncompletedEscapeInInclude( - lineReader.GetLineNum(), - ) - - if ch == "b": - filePath += "\b" - elif ch == "f": - filePath += "\f" - elif ch == "n": - filePath += "\n" - elif ch == "r": - filePath += "\r" - elif ch == "t": - filePath += "\t" - elif ch == "'": - filePath += "'" - elif ch == '"': - filePath += '"' - elif ch >= "0" and ch <= "7": # octNumber - octVal = 0 - - octVal_n = 1 - - while True: - d = int(ord(ch) - ord("0")) - assert d >= 0 and d <= 7 - - octVal = (octVal * 8) + d - - if octVal_n == 3: - break - - ch = lineReader.ReadSymbol() - - if ch is None: - break - - if not (ch >= "0" and ch <= "7"): - lineReader.StepBack() - break - - octVal_n += 1 - continue - - filePath += chr(octVal) - else: - RaiseError.CfgReader__UnknownEscapedSymbolInInclude( - lineReader.GetLineNum(), - lineReader.GetColNum(), - ch, - ) - - continue - - filePath += ch - continue - - assert type(filePath) is str - - if len(filePath) == 0: - RaiseError.CfgReader__IncludeHasEmptyPath( - lineReader.GetLineNum(), - ) - - fileLine.AddInclude( - filePath, - includeOffset, - ) - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__ProcessLineData__Option( - fileLine: PostgresConfigurationFileLine_Base, - lineReader: ReadUtils__LineReader, - optionOffset: int, - optionName: str, - ) -> None: - assert type(fileLine) is PostgresConfigurationFileLine_Base - assert type(lineReader) is ReadUtils__LineReader - assert type(optionOffset) is int - assert type(optionName) is str - assert optionName != "" - - # skeep spaces - spaceIsDetected = False - - while True: - ch = lineReader.ReadSymbol() - if ch is None or ReadUtils.IsEOL(ch) or ch == "#": - RaiseError.CfgReader__OptionWithoutValue( - optionName, - lineReader.GetLineNum(), - ) - - if ReadUtils.IsSpace(ch): - spaceIsDetected = True - continue - - if ch == "=": - while True: - ch = lineReader.ReadSymbol() - - if ch is None or ReadUtils.IsEOL(ch) or ch == "#": - RaiseError.CfgReader__OptionWithoutValue( - optionName, - lineReader.GetLineNum(), - ) - - if ReadUtils.IsSpace(ch): - continue - - break - break - - if spaceIsDetected: - break - - RaiseError.CfgReader__UnexpectedSymbol( - lineReader.GetLineNum(), - lineReader.GetColNum(), - ch, - ) - - # ch is the first symbol of option value. - - if ch == "'": - __class__.Helper__ProcessLineData__Option__Quoted( - fileLine, - lineReader, - optionOffset, - optionName, - ) - else: - lineReader.StepBack() - - __class__.Helper__ProcessLineData__Option__Generic( - fileLine, - lineReader, - optionOffset, - optionName, - ) - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__ProcessLineData__Option__Quoted( - fileLine: PostgresConfigurationFileLine_Base, - lineReader: ReadUtils__LineReader, - optionOffset: int, - optionName: str, - ) -> None: - assert type(fileLine) is PostgresConfigurationFileLine_Base - assert type(lineReader) is ReadUtils__LineReader - assert type(optionOffset) is int - assert type(optionName) is str - assert optionName != "" - - optionValue = "" - - while True: - ch = lineReader.ReadSymbol() - - if ch is None or ReadUtils.IsEOL(ch): - RaiseError.CfgReader__EndQuotedOptionValueIsNotFound( - optionName, - lineReader.GetLineNum(), - ) - break - - if ch == "'": - ch2 = lineReader.ReadSymbol() - - if ch2 is None: - break - - if ReadUtils.IsEOL(ch2): - lineReader.StepBack() - break - - if ch2 == "'": - optionValue += ch - continue - - break - - if ch == "\\": - ch = lineReader.ReadSymbol() - - if ch is None or ReadUtils.IsEOL(ch): - RaiseError.CfgReader__IncompletedEscapeInQuotedOptionValue( - optionName, - lineReader.GetLineNum(), - ) - - if ch == "b": - optionValue += "\b" - elif ch == "f": - optionValue += "\f" - elif ch == "n": - optionValue += "\n" - elif ch == "r": - optionValue += "\r" - elif ch == "t": - optionValue += "\t" - elif ch == "'": - optionValue += "'" - elif ch == '"': - optionValue += '"' - elif ch >= "0" and ch <= "7": # octNumber - octVal = 0 - - octVal_n = 1 - - while True: - d = int(ord(ch) - ord("0")) - assert d >= 0 and d <= 7 - - octVal = (octVal * 8) + d - - if octVal_n == 3: - break - - ch = lineReader.ReadSymbol() - - if ch is None: - break - - if not (ch >= "0" and ch <= "7"): - lineReader.StepBack() - break - - octVal_n += 1 - continue - - optionValue += chr(octVal) - else: - RaiseError.CfgReader__UnknownEscapedSymbolInQuotedOptionValue( - optionName, - lineReader.GetLineNum(), - lineReader.GetColNum(), - ch, - ) - - continue - - optionValue += ch - continue - - assert type(optionValue) is str - - fileLine.AddOption( - optionName, - optionValue, - optionOffset, - ) - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__ProcessLineData__Option__Generic( - fileLine: PostgresConfigurationFileLine_Base, - lineReader: ReadUtils__LineReader, - optionOffset: int, - optionName: str, - ) -> None: - assert type(fileLine) is PostgresConfigurationFileLine_Base - assert type(lineReader) is ReadUtils__LineReader - assert type(optionOffset) is int - assert type(optionName) is str - assert optionName != "" - - optionValue = "" - - while True: - ch = lineReader.ReadSymbol() - - if not ch: - assert ch is None - break - - assert type(ch) is str - - if ch == "#" or ReadUtils.IsEOL(ch): - lineReader.StepBack() - break - - optionValue += ch - continue - - optionValue = optionValue.strip() - - assert type(optionValue) is str - assert optionValue != "" - - fileLine.AddOption( - optionName, - optionValue, - optionOffset, - ) - return - - -# ////////////////////////////////////////////////////////////////////////////// -# class PostgresConfigurationWriterFileCtx_Base - - -class PostgresConfigurationWriterFileCtx_Base: - FileData: PgCfgModel__FileData - Content: typing.Optional[str] - File: typing.Optional[ConfigurationOsFile] - - # TODO: We can use filelock (it is a separated lock file) - # to provide an exclusive access to our File - - # -------------------------------------------------------------------- - def __init__(self, fileData: PgCfgModel__FileData): - assert fileData is not None - assert type(fileData) is PgCfgModel__FileData - - self.FileData = fileData - self.Content = None - self.File = None - return - - -# ////////////////////////////////////////////////////////////////////////////// -# class PostgresConfigurationWriterCtx_Base - - -class PostgresConfigurationWriterCtx_Base: - Cfg: PostgresConfiguration_Base - - AllFiles: list[PostgresConfigurationWriterFileCtx_Base] - NewFiles: list[PostgresConfigurationWriterFileCtx_Base] - UpdFiles: list[PostgresConfigurationWriterFileCtx_Base] - - # -------------------------------------------------------------------- - def __init__(self, cfg: PostgresConfiguration_Base): - assert cfg is not None - assert isinstance(cfg, PostgresConfiguration_Base) - - self.Cfg = cfg - - self.AllFiles = list() - self.NewFiles = list() - self.UpdFiles = list() - - assert type(self.AllFiles) is list - assert type(self.NewFiles) is list - assert type(self.UpdFiles) is list - return - - # -------------------------------------------------------------------- - def Init(self) -> None: - assert type(self.AllFiles) is list - assert type(self.NewFiles) is list - assert type(self.UpdFiles) is list - - self.AllFiles.clear() - self.NewFiles.clear() - self.UpdFiles.clear() - return - - -# ////////////////////////////////////////////////////////////////////////////// -# class PostgresConfigurationWriter_Base - - -class PostgresConfigurationWriter_Base: - @staticmethod - def MakeFileDataContent( - ctx: PostgresConfigurationWriterCtx_Base, - fileData: PgCfgModel__FileData, - ) -> str: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert type(fileData) is PgCfgModel__FileData - - return __class__.Helper__MakeFileDataContent( - ctx, - fileData, - ) - - # -------------------------------------------------------------------- - @staticmethod - def DoWork( - ctx: PostgresConfigurationWriterCtx_Base, - ) -> None: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert isinstance(ctx.Cfg, PostgresConfiguration_Base) - assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData - - return __class__.Helper__DoWork(ctx) - - # Helper Methods ----------------------------------------------------- - @staticmethod - def Helper__DoWork( - ctx: PostgresConfigurationWriterCtx_Base, - ) -> None: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert isinstance(ctx.Cfg, PostgresConfiguration_Base) - assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData - - # 0. - ctx.Init() - - # 1. - __class__.Helper__DoWork__Stage01__CreateFileContexts(ctx) - - # 2. - __class__.Helper__DoWork__Stage02__MakeFileDataContents(ctx) - - # 3. - __class__.Helper__DoWork__Stage03__OpenUpdFilesToWrite(ctx) - - # 4. - __class__.Helper__DoWork__Stage04__OpenNewFilesToWrite(ctx) - - # 5. - __class__.Helper__DoWork__Stage05__WriteContents(ctx) - - # OK. Go HOME! - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__DoWork__Stage01__CreateFileContexts( - ctx: PostgresConfigurationWriterCtx_Base, - ) -> None: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert isinstance(ctx.Cfg, PostgresConfiguration_Base) - assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData - - for fileData in ctx.Cfg.m_Data.m_AllFilesByName.values(): - assert fileData is not None - assert type(fileData) is PgCfgModel__FileData - - fileCtx = PostgresConfigurationWriterFileCtx_Base(fileData) - - assert fileCtx.FileData is fileData - assert fileCtx.Content is None - assert fileCtx.File is None - - assert fileCtx not in ctx.AllFiles - assert fileCtx not in ctx.NewFiles - assert fileCtx not in ctx.UpdFiles - - ctx.AllFiles.append(fileCtx) - - if fileData.m_Status == PgCfgModel__FileStatus.IS_NEW: - ctx.NewFiles.append(fileCtx) - elif fileData.m_Status == PgCfgModel__FileStatus.EXISTS: - ctx.UpdFiles.append(fileCtx) - else: - BugCheckError.UnkFileDataStatus( - fileData.m_Path, - fileData.m_Status, - ) - continue - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__DoWork__Stage02__MakeFileDataContents( - ctx: PostgresConfigurationWriterCtx_Base, - ) -> None: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert isinstance(ctx.Cfg, PostgresConfiguration_Base) - assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData - - for fileCtx in ctx.AllFiles: - assert fileCtx is not None - assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base - assert type(fileCtx.FileData) is PgCfgModel__FileData - assert fileCtx.Content is None - - fileCtx.Content = __class__.Helper__MakeFileDataContent( - ctx, - fileCtx.FileData, - ) - continue - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__DoWork__Stage03__OpenUpdFilesToWrite( - ctx: PostgresConfigurationWriterCtx_Base, - ) -> None: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert isinstance(ctx.Cfg, PostgresConfiguration_Base) - assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData - - for fileCtx in ctx.UpdFiles: - assert fileCtx is not None - assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base - assert type(fileCtx.FileData) is PgCfgModel__FileData - - assert fileCtx.FileData.m_Status == PgCfgModel__FileStatus.EXISTS - assert fileCtx.FileData.m_LastModifiedTimestamp is not None - assert type(fileCtx.FileData.m_LastModifiedTimestamp) is datetime.datetime - assert fileCtx.File is None - - # Let's open an exist file to read and write without truncation - fileCtx.File = ctx.Cfg.m_Data.OsOps.OpenFileToWrite( - fileCtx.FileData.m_Path - ) # raise - - assert fileCtx.File is not None - assert isinstance(fileCtx.File, ConfigurationOsFile) - - lastMDate = fileCtx.File.GetModificationTS() - assert type(lastMDate) is datetime.datetime - - if fileCtx.FileData.m_LastModifiedTimestamp != lastMDate: - RaiseError.FileWasModifiedExternally( - fileCtx.FileData.m_Path, - fileCtx.FileData.m_LastModifiedTimestamp, - lastMDate, - ) - - # OK! - continue - - # OK, Go HOME! - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__DoWork__Stage04__OpenNewFilesToWrite( - ctx: PostgresConfigurationWriterCtx_Base, - ) -> None: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert isinstance(ctx.Cfg, PostgresConfiguration_Base) - assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData - assert type(ctx.NewFiles) is list - - iFile = 0 - - try: - for iFile in range(len(ctx.NewFiles)): - fileCtx = ctx.NewFiles[iFile] - - assert fileCtx is not None - assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base - assert type(fileCtx.FileData) is PgCfgModel__FileData - - assert fileCtx.FileData.m_Status == PgCfgModel__FileStatus.IS_NEW - assert fileCtx.File is None - - fileCtx.File = ctx.Cfg.m_Data.OsOps.CreateFile( - fileCtx.FileData.m_Path - ) # raise - - assert fileCtx.File is not None - assert isinstance(fileCtx.File, ConfigurationOsFile) - - # OK - continue - - except: # Rollback new files - assert type(iFile) is int - assert iFile >= 0 - assert iFile <= len(ctx.NewFiles) - - for iFile2 in range(iFile): - fileCtx = ctx.NewFiles[iFile2] - - assert fileCtx is not None - assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base - assert type(fileCtx.FileData) is PgCfgModel__FileData - - assert fileCtx.FileData.m_Status == PgCfgModel__FileStatus.IS_NEW - assert fileCtx.File is not None - assert isinstance(fileCtx.File, ConfigurationOsFile) - - assert not fileCtx.File.IsClosed - - filePath = fileCtx.File.Name - assert filePath is not None - assert type(filePath) is str - assert filePath == fileCtx.FileData.m_Path - - fileCtx.File.Close() # raise - - ctx.Cfg.m_Data.OsOps.Remove(filePath) # raise - continue - - raise - - # OK, Go HOME! - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__DoWork__Stage05__WriteContents( - ctx: PostgresConfigurationWriterCtx_Base, - ) -> None: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - - for iFile in range(len(ctx.AllFiles)): - fileCtx = ctx.AllFiles[iFile] - - assert fileCtx is not None - assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base - assert type(fileCtx.FileData) is PgCfgModel__FileData - - assert fileCtx.File is not None - assert isinstance(fileCtx.File, ConfigurationOsFile) - - assert fileCtx.Content is not None - assert type(fileCtx.Content) is str - - fileCtx.File.Overwrite(fileCtx.Content) - - lastMDate = fileCtx.File.GetModificationTS() - assert type(lastMDate) is datetime.datetime - - fileCtx.File.Close() - - fileCtx.FileData.m_LastModifiedTimestamp = lastMDate - fileCtx.FileData.m_Status = PgCfgModel__FileStatus.EXISTS - continue - return - - # -------------------------------------------------------------------- - @staticmethod - def Helper__MakeFileDataContent( - ctx: PostgresConfigurationWriterCtx_Base, - fileData: PgCfgModel__FileData, - ) -> str: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert type(fileData) is PgCfgModel__FileData - - fileContent = "" - - for fileLineData in fileData.m_Lines: - assert type(fileLineData) is PgCfgModel__FileLineData - lineContent = __class__.Helper__FileLineToString( - ctx, - fileLineData, - ) - assert type(lineContent) is str - fileContent += lineContent - fileContent += "\n" - continue - - return fileContent - - # -------------------------------------------------------------------- - @staticmethod - def Helper__FileLineToString( - ctx: PostgresConfigurationWriterCtx_Base, - fileLineData: PgCfgModel__FileLineData, - ) -> str: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert type(fileLineData) is PgCfgModel__FileLineData - - fileLineItemCount = len(fileLineData.m_Items) - - if fileLineItemCount == 0: - return "" - - lineContent = "" - - for lineItem in fileLineData.m_Items: - assert type(lineItem) is PgCfgModel__FileLineData.tagItem - assert lineItem.m_Element is not None - assert isinstance(lineItem.m_Element, PgCfgModel__FileLineElementData) - assert ( - lineItem.m_Element.m_Offset is None - or type(lineItem.m_Element.m_Offset) is int - ) - assert ( - lineItem.m_Element.m_Offset is None or lineItem.m_Element.m_Offset >= 0 - ) - - itemContent = __class__.Helper__ElementToString( - ctx, - lineItem.m_Element, - ) - assert type(itemContent) is str - - lineContent = __class__.Helper__AppendItemToLine( - lineContent, - lineItem.m_Element.m_Offset, - itemContent, - ) - assert type(lineContent) is str - continue - - assert type(lineContent) is str - return lineContent - - # -------------------------------------------------------------------- - @staticmethod - def Helper__AppendItemToLine( - lineContent: str, - offset: typing.Optional[int], - text: str, - ) -> str: - assert type(lineContent) is str - assert offset is None or type(offset) is int - assert type(text) is str - assert offset is None or offset >= 0 - - if text == "": - return lineContent - - lineContentLen = len(lineContent) - - assert type(lineContentLen) is int - assert lineContentLen >= 0 - - if offset is not None and lineContentLen < offset: - lineContent += " " * (offset - lineContentLen) - elif lineContentLen > 0: - lineContent += " " - - lineContent += text - return lineContent - - # -------------------------------------------------------------------- - @staticmethod - def Helper__ElementToString( - ctx: PostgresConfigurationWriterCtx_Base, - elementData: PgCfgModel__FileLineElementData, - ) -> str: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert elementData is not None - assert isinstance(elementData, PgCfgModel__FileLineElementData) - - if type(elementData) is PgCfgModel__OptionData: - return __class__.Helper__OptionToString( - ctx, - elementData, - ) - - if type(elementData) is PgCfgModel__CommentData: - return __class__.Helper__CommentToString( - ctx, - elementData, - ) - - if type(elementData) is PgCfgModel__IncludeData: - return __class__.Helper__IncludeToString( - ctx, - elementData, - ) - - BugCheckError.UnkObjectDataType(type(elementData)) - - # -------------------------------------------------------------------- - @staticmethod - def Helper__OptionToString( - ctx: PostgresConfigurationWriterCtx_Base, - optionData: PgCfgModel__OptionData, - ) -> str: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert ctx.Cfg is not None - assert isinstance(ctx.Cfg, PostgresConfiguration_Base) - assert type(optionData) is PgCfgModel__OptionData - assert type(optionData.m_Name) is str - assert optionData.m_Name != "" - assert optionData.m_Value is not None - - writeHandler = ctx.Cfg.Internal__GetOptionHandlerToWrite(optionData.m_Name) - - assert writeHandler is not None - assert isinstance(writeHandler, PgCfgModel__OptionHandlerToWrite) - - writeHandlerCtx = PgCfgModel__OptionHandlerCtxToWrite( - ctx.Cfg, - optionData.m_Name, - optionData.m_Value, - ) - - optValueAsText = writeHandler.OptionValueToString(writeHandlerCtx) - - assert type(optValueAsText) is str - assert optValueAsText != "" - - result = optionData.m_Name + " = " + optValueAsText - return result - - # -------------------------------------------------------------------- - @staticmethod - def Helper__CommentToString( - ctx: PostgresConfigurationWriterCtx_Base, - commentData: PgCfgModel__CommentData, - ) -> str: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert ctx.Cfg is not None - assert isinstance(ctx.Cfg, PostgresConfiguration_Base) - assert type(commentData) is PgCfgModel__CommentData - assert commentData.m_Text is not None - assert type(commentData.m_Text) is str - - assert DataVerificator.IsValidCommentText(commentData.m_Text) - - result = "#" + commentData.m_Text - return result - - # -------------------------------------------------------------------- - @staticmethod - def Helper__IncludeToString( - ctx: PostgresConfigurationWriterCtx_Base, - includeData: PgCfgModel__IncludeData, - ) -> str: - assert type(ctx) is PostgresConfigurationWriterCtx_Base - assert ctx.Cfg is not None - assert isinstance(ctx.Cfg, PostgresConfiguration_Base) - assert type(includeData) is PgCfgModel__IncludeData - assert includeData.m_Path is not None - assert type(includeData.m_Path) is str - assert includeData.m_Path != "" - - result = "include " + WriteUtils.Pack_Str(includeData.m_Path) - - assert type(result) is str - assert result != "" - return result - - # ////////////////////////////////////////////////////////////////////////////// # PostgresConfigurationController__Base diff --git a/src/implementation/v00/configuration_base_r.py b/src/implementation/v00/configuration_base_r.py new file mode 100644 index 0000000..b4aea37 --- /dev/null +++ b/src/implementation/v00/configuration_base_r.py @@ -0,0 +1,710 @@ +# ////////////////////////////////////////////////////////////////////////////// +# Postgres Pro. PostgreSQL Configuration Python Library. + +from __future__ import annotations + +# fmt: off +from .configuration_base import PostgresConfiguration_Base +from .configuration_base import PostgresConfigurationFileLine_Base +from .configuration_base import PostgresConfigurationFile_Base +from .configuration_base import PostgresConfigurationTopLevelFile_Base +from .configuration_base import PostgresConfigurationFactory_Base + +from ...core.model import CommentData as PgCfgModel__CommentData +from ...core.model import OptionData as PgCfgModel__OptionData +from ...core.model import IncludeData as PgCfgModel__IncludeData +from ...core.model import FileLineData as PgCfgModel__FileLineData +from ...core.model import FileData as PgCfgModel__FileData +from ...core.model import FileStatus as PgCfgModel__FileStatus + +from ...core.helpers import Helpers +from ...core.read_utils import ReadUtils +from ...core.read_utils import LineReader as ReadUtils__LineReader +from ...core.raise_error import RaiseError +from ...core.bugcheck_error import BugCheckError +# fmt: on + +from ...os.abstract.configuration_os_ops import ConfigurationFileReader +from ...os.abstract.configuration_os_ops import ConfigurationOsFile + +import typing +import datetime + +# ////////////////////////////////////////////////////////////////////////////// +# class PostgresConfigurationReader_Base + + +class PostgresConfigurationReader_Base: + @staticmethod + def LoadConfigurationFile( + cfg: PostgresConfiguration_Base, + filePath: str, + ) -> PostgresConfigurationTopLevelFile_Base: + assert cfg is not None + + assert isinstance(cfg, PostgresConfiguration_Base) + assert type(filePath) is str + assert filePath != "" + + existFileDatas: typing.Dict[str, PgCfgModel__FileData] = dict() + + for fileName in cfg.m_Data.m_AllFilesByName.keys(): + assert type(fileName) is str + assert fileName != "" + + indexData = cfg.m_Data.m_AllFilesByName[fileName] + assert indexData is not None + + if type(indexData) is PgCfgModel__FileData: + fileData: PgCfgModel__FileData = indexData + assert type(fileData.m_Path) is str + assert fileData.m_Path not in existFileDatas.keys() + existFileDatas[fileData.m_Path] = fileData + continue + + if type(indexData) is list: + for fileData in indexData: + assert type(fileData) is PgCfgModel__FileData + assert type(fileData.m_Path) is str + assert fileData.m_Path not in existFileDatas.keys() + existFileDatas[fileData.m_Path] = fileData + continue + continue + + BugCheckError.UnkFileObjectDataType( + fileName, + type(indexData), + ) + + # ---------------------------------------------------------------- + filePath_n = Helpers.NormalizeFilePath( + cfg.m_Data.OsOps, + cfg.m_Data.m_DataDir, + filePath, + ) + assert type(filePath_n) is str + + if filePath_n in existFileDatas: + file = PostgresConfigurationFactory_Base.GetObject( + cfg, + existFileDatas[filePath_n], + ) + assert type(file) is PostgresConfigurationTopLevelFile_Base + return file + + # ---------------------------------------------------------------- + rootFile = cfg.AddTopLevelFile(filePath) + assert type(rootFile) is PostgresConfigurationTopLevelFile_Base + assert type(rootFile.m_FileData) is PgCfgModel__FileData + assert type(rootFile.m_FileData.m_Lines) is list + assert rootFile.m_FileData.m_Status == PgCfgModel__FileStatus.IS_NEW + assert rootFile.m_FileData.m_LastModifiedTimestamp is None + assert len(rootFile.m_FileData.m_Lines) == 0 + + queuedFileDatas: typing.Set[PgCfgModel__FileData] = set() + + queuedFileDatas.add(rootFile.m_FileData) + + while len(queuedFileDatas) != 0: + currentFileData = queuedFileDatas.pop() + assert isinstance(currentFileData, PgCfgModel__FileData) + + # load content + # process content + + with cfg.m_Data.OsOps.OpenFileToRead(currentFileData.m_Path) as f: + assert isinstance(f, ConfigurationOsFile) + currentFile = PostgresConfigurationFactory_Base.GetObject( + cfg, + currentFileData, + ) + assert isinstance(currentFile, PostgresConfigurationFile_Base) + __class__.Helper__LoadFileContent( + currentFile, + f, + ) # raise + + lastMDate = f.GetModificationTS() + assert type(lastMDate) is datetime.datetime + + currentFileData.m_LastModifiedTimestamp = lastMDate + currentFileData.m_Status = PgCfgModel__FileStatus.EXISTS + + assert currentFileData.m_Path not in existFileDatas.keys() + existFileDatas[currentFileData.m_Path] = currentFileData + + # enumerate all the includes + for fileLineData in currentFileData.m_Lines: + assert type(fileLineData) is PgCfgModel__FileLineData + assert type(fileLineData.m_Items) is list + + for fileLineItem in fileLineData.m_Items: + assert type(fileLineItem) is PgCfgModel__FileLineData.tagItem + + fileLineElementData = fileLineItem.m_Element + + if type(fileLineElementData) is PgCfgModel__CommentData: + continue + + if type(fileLineElementData) is PgCfgModel__OptionData: + continue + + if type(fileLineElementData) is PgCfgModel__IncludeData: + # look at existFileDatas + includeData: PgCfgModel__IncludeData = fileLineElementData + assert type(includeData.m_File) is PgCfgModel__FileData + assert type(includeData.m_File.m_Path) is str + + if includeData.m_File.m_Path in existFileDatas: + continue # it is an old file + + assert ( + includeData.m_File.m_Status == PgCfgModel__FileStatus.IS_NEW + ) + assert includeData.m_File.m_LastModifiedTimestamp is None + + if includeData.m_File in queuedFileDatas: + continue # already in queue + + queuedFileDatas.add(includeData.m_File) + continue + + BugCheckError.UnkObjectDataType(type(fileLineElementData)) + + assert len(queuedFileDatas) == 0 + assert isinstance(rootFile, PostgresConfigurationTopLevelFile_Base) + + return rootFile + + # -------------------------------------------------------------------- + @staticmethod + def LoadFileContent( + file: PostgresConfigurationFile_Base, + fileContent: ConfigurationFileReader, + ) -> None: + assert isinstance(file, PostgresConfigurationFile_Base) + assert isinstance(fileContent, ConfigurationFileReader) + + return __class__.Helper__LoadFileContent( + file, + fileContent, + ) + + # Helper methods ----------------------------------------------------- + @staticmethod + def Helper__LoadFileContent( + file: PostgresConfigurationFile_Base, + fileContent: ConfigurationFileReader, + ) -> None: + assert isinstance(file, PostgresConfigurationFile_Base) + assert isinstance(fileContent, ConfigurationFileReader) + + lineReader = ReadUtils__LineReader() + + while True: + lineData = fileContent.ReadLine() + + if not lineData: + assert lineData is None + break + + assert type(lineData) is str + + lineReader.SetData(lineData) + + __class__.Helper__ProcessLineData( + file, + lineReader, + ) + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ProcessLineData( + file: PostgresConfigurationFile_Base, + lineReader: ReadUtils__LineReader, + ) -> None: + assert isinstance(file, PostgresConfigurationFile_Base) + assert type(lineReader) is ReadUtils__LineReader + + fileLine = file.AddEmptyLine() + assert type(fileLine) is PostgresConfigurationFileLine_Base + + ch: typing.Optional[str] + + while True: + # skeep spaces + while True: + ch = lineReader.ReadSymbol() + if ch is None: + return + + if ReadUtils.IsSpace(ch): + continue + + break + + assert not ReadUtils.IsSpace(ch) + + if ReadUtils.IsEOL(ch): + return + + if ch == "#": + return __class__.Helper__ProcessLineData__Comment( + fileLine, + lineReader, + ) + + if len(fileLine) == 0 and ReadUtils.IsValidSeqCh1(ch): + # Read sequence + sequenceOffset = lineReader.GetColOffset() + sequence = ch + + while True: + ch = lineReader.ReadSymbol() + + if not ch: + assert ch is None + break + + assert type(ch) is str + + if ReadUtils.IsValidSeqCh2(ch): + sequence += ch + continue + + if ch == ".": + sequence += ch + continue + + break + + if sequence.lower() == "include": + if ch is not None: + lineReader.StepBack() + + __class__.Helper__ProcessLineData__Include( + fileLine, + lineReader, + sequenceOffset, + ) + else: + if ch is not None: + lineReader.StepBack() + + __class__.Helper__ProcessLineData__Option( + fileLine, + lineReader, + sequenceOffset, + sequence, + ) + continue + + RaiseError.CfgReader__UnexpectedSymbol( + lineReader.GetLineNum(), + lineReader.GetColNum(), + ch, + ) + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ProcessLineData__Comment( + fileLine: PostgresConfigurationFileLine_Base, + lineReader: ReadUtils__LineReader, + ) -> None: + assert type(fileLine) is PostgresConfigurationFileLine_Base + assert type(lineReader) is ReadUtils__LineReader + + commentText = "" + commentOffset = lineReader.GetColOffset() + + while True: + ch = lineReader.ReadSymbol() + + if not ch: + assert ch is None + break + + assert type(ch) is str + + if ReadUtils.IsEOL(ch): + break + commentText += ch + + fileLine.AddComment( + commentText, + commentOffset, + ) + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ProcessLineData__Include( + fileLine: PostgresConfigurationFileLine_Base, + lineReader: ReadUtils__LineReader, + includeOffset: int, + ) -> None: + assert type(fileLine) is PostgresConfigurationFileLine_Base + assert type(lineReader) is ReadUtils__LineReader + assert type(includeOffset) is int + assert includeOffset >= 0 + + # find first quote + while True: + ch = lineReader.ReadSymbol() + + if ch is None or ReadUtils.IsEOL(ch) or ch == "#": + RaiseError.CfgReader__IncludeWithoutPath( + lineReader.GetLineNum(), + ) + + if ReadUtils.IsSpace(ch): + continue + + if ch == "'": + break + + RaiseError.CfgReader__UnexpectedSymbol( + lineReader.GetLineNum(), + lineReader.GetColNum(), + ch, + ) + + # OK. First quote is found! + + # Let's process this quoted string + filePath = "" + + while True: + ch = lineReader.ReadSymbol() + + if ch is None or ReadUtils.IsEOL(ch): + RaiseError.CfgReader__EndOfIncludePathIsNotFound( + lineReader.GetLineNum(), + ) + break + + if ch == "'": + ch2 = lineReader.ReadSymbol() + + if ch2 is None: + break + + if ReadUtils.IsEOL(ch2): + lineReader.StepBack() + break + + if ch2 == "'": + filePath += ch + continue + + break + + if ch == "\\": + ch = lineReader.ReadSymbol() + + if ch is None or ReadUtils.IsEOL(ch): + RaiseError.CfgReader__IncompletedEscapeInInclude( + lineReader.GetLineNum(), + ) + + if ch == "b": + filePath += "\b" + elif ch == "f": + filePath += "\f" + elif ch == "n": + filePath += "\n" + elif ch == "r": + filePath += "\r" + elif ch == "t": + filePath += "\t" + elif ch == "'": + filePath += "'" + elif ch == '"': + filePath += '"' + elif ch >= "0" and ch <= "7": # octNumber + octVal = 0 + + octVal_n = 1 + + while True: + d = int(ord(ch) - ord("0")) + assert d >= 0 and d <= 7 + + octVal = (octVal * 8) + d + + if octVal_n == 3: + break + + ch = lineReader.ReadSymbol() + + if ch is None: + break + + if not (ch >= "0" and ch <= "7"): + lineReader.StepBack() + break + + octVal_n += 1 + continue + + filePath += chr(octVal) + else: + RaiseError.CfgReader__UnknownEscapedSymbolInInclude( + lineReader.GetLineNum(), + lineReader.GetColNum(), + ch, + ) + + continue + + filePath += ch + continue + + assert type(filePath) is str + + if len(filePath) == 0: + RaiseError.CfgReader__IncludeHasEmptyPath( + lineReader.GetLineNum(), + ) + + fileLine.AddInclude( + filePath, + includeOffset, + ) + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ProcessLineData__Option( + fileLine: PostgresConfigurationFileLine_Base, + lineReader: ReadUtils__LineReader, + optionOffset: int, + optionName: str, + ) -> None: + assert type(fileLine) is PostgresConfigurationFileLine_Base + assert type(lineReader) is ReadUtils__LineReader + assert type(optionOffset) is int + assert type(optionName) is str + assert optionName != "" + + # skeep spaces + spaceIsDetected = False + + while True: + ch = lineReader.ReadSymbol() + if ch is None or ReadUtils.IsEOL(ch) or ch == "#": + RaiseError.CfgReader__OptionWithoutValue( + optionName, + lineReader.GetLineNum(), + ) + + if ReadUtils.IsSpace(ch): + spaceIsDetected = True + continue + + if ch == "=": + while True: + ch = lineReader.ReadSymbol() + + if ch is None or ReadUtils.IsEOL(ch) or ch == "#": + RaiseError.CfgReader__OptionWithoutValue( + optionName, + lineReader.GetLineNum(), + ) + + if ReadUtils.IsSpace(ch): + continue + + break + break + + if spaceIsDetected: + break + + RaiseError.CfgReader__UnexpectedSymbol( + lineReader.GetLineNum(), + lineReader.GetColNum(), + ch, + ) + + # ch is the first symbol of option value. + + if ch == "'": + __class__.Helper__ProcessLineData__Option__Quoted( + fileLine, + lineReader, + optionOffset, + optionName, + ) + else: + lineReader.StepBack() + + __class__.Helper__ProcessLineData__Option__Generic( + fileLine, + lineReader, + optionOffset, + optionName, + ) + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ProcessLineData__Option__Quoted( + fileLine: PostgresConfigurationFileLine_Base, + lineReader: ReadUtils__LineReader, + optionOffset: int, + optionName: str, + ) -> None: + assert type(fileLine) is PostgresConfigurationFileLine_Base + assert type(lineReader) is ReadUtils__LineReader + assert type(optionOffset) is int + assert type(optionName) is str + assert optionName != "" + + optionValue = "" + + while True: + ch = lineReader.ReadSymbol() + + if ch is None or ReadUtils.IsEOL(ch): + RaiseError.CfgReader__EndQuotedOptionValueIsNotFound( + optionName, + lineReader.GetLineNum(), + ) + break + + if ch == "'": + ch2 = lineReader.ReadSymbol() + + if ch2 is None: + break + + if ReadUtils.IsEOL(ch2): + lineReader.StepBack() + break + + if ch2 == "'": + optionValue += ch + continue + + break + + if ch == "\\": + ch = lineReader.ReadSymbol() + + if ch is None or ReadUtils.IsEOL(ch): + RaiseError.CfgReader__IncompletedEscapeInQuotedOptionValue( + optionName, + lineReader.GetLineNum(), + ) + + if ch == "b": + optionValue += "\b" + elif ch == "f": + optionValue += "\f" + elif ch == "n": + optionValue += "\n" + elif ch == "r": + optionValue += "\r" + elif ch == "t": + optionValue += "\t" + elif ch == "'": + optionValue += "'" + elif ch == '"': + optionValue += '"' + elif ch >= "0" and ch <= "7": # octNumber + octVal = 0 + + octVal_n = 1 + + while True: + d = int(ord(ch) - ord("0")) + assert d >= 0 and d <= 7 + + octVal = (octVal * 8) + d + + if octVal_n == 3: + break + + ch = lineReader.ReadSymbol() + + if ch is None: + break + + if not (ch >= "0" and ch <= "7"): + lineReader.StepBack() + break + + octVal_n += 1 + continue + + optionValue += chr(octVal) + else: + RaiseError.CfgReader__UnknownEscapedSymbolInQuotedOptionValue( + optionName, + lineReader.GetLineNum(), + lineReader.GetColNum(), + ch, + ) + + continue + + optionValue += ch + continue + + assert type(optionValue) is str + + fileLine.AddOption( + optionName, + optionValue, + optionOffset, + ) + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ProcessLineData__Option__Generic( + fileLine: PostgresConfigurationFileLine_Base, + lineReader: ReadUtils__LineReader, + optionOffset: int, + optionName: str, + ) -> None: + assert type(fileLine) is PostgresConfigurationFileLine_Base + assert type(lineReader) is ReadUtils__LineReader + assert type(optionOffset) is int + assert type(optionName) is str + assert optionName != "" + + optionValue = "" + + while True: + ch = lineReader.ReadSymbol() + + if not ch: + assert ch is None + break + + assert type(ch) is str + + if ch == "#" or ReadUtils.IsEOL(ch): + lineReader.StepBack() + break + + optionValue += ch + continue + + optionValue = optionValue.strip() + + assert type(optionValue) is str + assert optionValue != "" + + fileLine.AddOption( + optionName, + optionValue, + optionOffset, + ) + return + + +# ////////////////////////////////////////////////////////////////////////////// diff --git a/src/implementation/v00/configuration_base_w.py b/src/implementation/v00/configuration_base_w.py new file mode 100644 index 0000000..f73f4fe --- /dev/null +++ b/src/implementation/v00/configuration_base_w.py @@ -0,0 +1,552 @@ +# ////////////////////////////////////////////////////////////////////////////// +# Postgres Pro. PostgreSQL Configuration Python Library. + +from __future__ import annotations + +# fmt: off +from .configuration_base import PostgresConfiguration_Base + +from ...core.model import ConfigurationData as PgCfgModel__ConfigurationData +from ...core.model import FileLineElementData as PgCfgModel__FileLineElementData +from ...core.model import CommentData as PgCfgModel__CommentData +from ...core.model import OptionData as PgCfgModel__OptionData +from ...core.model import IncludeData as PgCfgModel__IncludeData +from ...core.model import FileLineData as PgCfgModel__FileLineData +from ...core.model import FileData as PgCfgModel__FileData +from ...core.model import FileStatus as PgCfgModel__FileStatus + +from ...core.handlers import OptionHandlerToWrite as PgCfgModel__OptionHandlerToWrite +from ...core.handlers import OptionHandlerCtxToWrite as PgCfgModel__OptionHandlerCtxToWrite + +from ...core.data_verificator import DataVerificator +from ...core.write_utils import WriteUtils +from ...core.raise_error import RaiseError +from ...core.bugcheck_error import BugCheckError +# fmt: on + +from ...os.abstract.configuration_os_ops import ConfigurationOsFile + +import typing +import datetime + +# ////////////////////////////////////////////////////////////////////////////// +# class PostgresConfigurationWriterFileCtx_Base + + +class PostgresConfigurationWriterFileCtx_Base: + FileData: PgCfgModel__FileData + Content: typing.Optional[str] + File: typing.Optional[ConfigurationOsFile] + + # TODO: We can use filelock (it is a separated lock file) + # to provide an exclusive access to our File + + # -------------------------------------------------------------------- + def __init__(self, fileData: PgCfgModel__FileData): + assert fileData is not None + assert type(fileData) is PgCfgModel__FileData + + self.FileData = fileData + self.Content = None + self.File = None + return + + +# ////////////////////////////////////////////////////////////////////////////// +# class PostgresConfigurationWriterCtx_Base + + +class PostgresConfigurationWriterCtx_Base: + Cfg: PostgresConfiguration_Base + + AllFiles: list[PostgresConfigurationWriterFileCtx_Base] + NewFiles: list[PostgresConfigurationWriterFileCtx_Base] + UpdFiles: list[PostgresConfigurationWriterFileCtx_Base] + + # -------------------------------------------------------------------- + def __init__(self, cfg: PostgresConfiguration_Base): + assert cfg is not None + assert isinstance(cfg, PostgresConfiguration_Base) + + self.Cfg = cfg + + self.AllFiles = list() + self.NewFiles = list() + self.UpdFiles = list() + + assert type(self.AllFiles) is list + assert type(self.NewFiles) is list + assert type(self.UpdFiles) is list + return + + # -------------------------------------------------------------------- + def Init(self) -> None: + assert type(self.AllFiles) is list + assert type(self.NewFiles) is list + assert type(self.UpdFiles) is list + + self.AllFiles.clear() + self.NewFiles.clear() + self.UpdFiles.clear() + return + + +# ////////////////////////////////////////////////////////////////////////////// +# class PostgresConfigurationWriter_Base + + +class PostgresConfigurationWriter_Base: + @staticmethod + def MakeFileDataContent( + ctx: PostgresConfigurationWriterCtx_Base, + fileData: PgCfgModel__FileData, + ) -> str: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert type(fileData) is PgCfgModel__FileData + + return __class__.Helper__MakeFileDataContent( + ctx, + fileData, + ) + + # -------------------------------------------------------------------- + @staticmethod + def DoWork( + ctx: PostgresConfigurationWriterCtx_Base, + ) -> None: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert isinstance(ctx.Cfg, PostgresConfiguration_Base) + assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData + + return __class__.Helper__DoWork(ctx) + + # Helper Methods ----------------------------------------------------- + @staticmethod + def Helper__DoWork( + ctx: PostgresConfigurationWriterCtx_Base, + ) -> None: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert isinstance(ctx.Cfg, PostgresConfiguration_Base) + assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData + + # 0. + ctx.Init() + + # 1. + __class__.Helper__DoWork__Stage01__CreateFileContexts(ctx) + + # 2. + __class__.Helper__DoWork__Stage02__MakeFileDataContents(ctx) + + # 3. + __class__.Helper__DoWork__Stage03__OpenUpdFilesToWrite(ctx) + + # 4. + __class__.Helper__DoWork__Stage04__OpenNewFilesToWrite(ctx) + + # 5. + __class__.Helper__DoWork__Stage05__WriteContents(ctx) + + # OK. Go HOME! + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__DoWork__Stage01__CreateFileContexts( + ctx: PostgresConfigurationWriterCtx_Base, + ) -> None: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert isinstance(ctx.Cfg, PostgresConfiguration_Base) + assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData + + for fileData in ctx.Cfg.m_Data.m_AllFilesByName.values(): + assert fileData is not None + assert type(fileData) is PgCfgModel__FileData + + fileCtx = PostgresConfigurationWriterFileCtx_Base(fileData) + + assert fileCtx.FileData is fileData + assert fileCtx.Content is None + assert fileCtx.File is None + + assert fileCtx not in ctx.AllFiles + assert fileCtx not in ctx.NewFiles + assert fileCtx not in ctx.UpdFiles + + ctx.AllFiles.append(fileCtx) + + if fileData.m_Status == PgCfgModel__FileStatus.IS_NEW: + ctx.NewFiles.append(fileCtx) + elif fileData.m_Status == PgCfgModel__FileStatus.EXISTS: + ctx.UpdFiles.append(fileCtx) + else: + BugCheckError.UnkFileDataStatus( + fileData.m_Path, + fileData.m_Status, + ) + continue + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__DoWork__Stage02__MakeFileDataContents( + ctx: PostgresConfigurationWriterCtx_Base, + ) -> None: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert isinstance(ctx.Cfg, PostgresConfiguration_Base) + assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData + + for fileCtx in ctx.AllFiles: + assert fileCtx is not None + assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base + assert type(fileCtx.FileData) is PgCfgModel__FileData + assert fileCtx.Content is None + + fileCtx.Content = __class__.Helper__MakeFileDataContent( + ctx, + fileCtx.FileData, + ) + continue + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__DoWork__Stage03__OpenUpdFilesToWrite( + ctx: PostgresConfigurationWriterCtx_Base, + ) -> None: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert isinstance(ctx.Cfg, PostgresConfiguration_Base) + assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData + + for fileCtx in ctx.UpdFiles: + assert fileCtx is not None + assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base + assert type(fileCtx.FileData) is PgCfgModel__FileData + + assert fileCtx.FileData.m_Status == PgCfgModel__FileStatus.EXISTS + assert fileCtx.FileData.m_LastModifiedTimestamp is not None + assert type(fileCtx.FileData.m_LastModifiedTimestamp) is datetime.datetime + assert fileCtx.File is None + + # Let's open an exist file to read and write without truncation + fileCtx.File = ctx.Cfg.m_Data.OsOps.OpenFileToWrite( + fileCtx.FileData.m_Path + ) # raise + + assert fileCtx.File is not None + assert isinstance(fileCtx.File, ConfigurationOsFile) + + lastMDate = fileCtx.File.GetModificationTS() + assert type(lastMDate) is datetime.datetime + + if fileCtx.FileData.m_LastModifiedTimestamp != lastMDate: + RaiseError.FileWasModifiedExternally( + fileCtx.FileData.m_Path, + fileCtx.FileData.m_LastModifiedTimestamp, + lastMDate, + ) + + # OK! + continue + + # OK, Go HOME! + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__DoWork__Stage04__OpenNewFilesToWrite( + ctx: PostgresConfigurationWriterCtx_Base, + ) -> None: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert isinstance(ctx.Cfg, PostgresConfiguration_Base) + assert type(ctx.Cfg.m_Data) is PgCfgModel__ConfigurationData + assert type(ctx.NewFiles) is list + + iFile = 0 + + try: + for iFile in range(len(ctx.NewFiles)): + fileCtx = ctx.NewFiles[iFile] + + assert fileCtx is not None + assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base + assert type(fileCtx.FileData) is PgCfgModel__FileData + + assert fileCtx.FileData.m_Status == PgCfgModel__FileStatus.IS_NEW + assert fileCtx.File is None + + fileCtx.File = ctx.Cfg.m_Data.OsOps.CreateFile( + fileCtx.FileData.m_Path + ) # raise + + assert fileCtx.File is not None + assert isinstance(fileCtx.File, ConfigurationOsFile) + + # OK + continue + + except: # Rollback new files + assert type(iFile) is int + assert iFile >= 0 + assert iFile <= len(ctx.NewFiles) + + for iFile2 in range(iFile): + fileCtx = ctx.NewFiles[iFile2] + + assert fileCtx is not None + assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base + assert type(fileCtx.FileData) is PgCfgModel__FileData + + assert fileCtx.FileData.m_Status == PgCfgModel__FileStatus.IS_NEW + assert fileCtx.File is not None + assert isinstance(fileCtx.File, ConfigurationOsFile) + + assert not fileCtx.File.IsClosed + + filePath = fileCtx.File.Name + assert filePath is not None + assert type(filePath) is str + assert filePath == fileCtx.FileData.m_Path + + fileCtx.File.Close() # raise + + ctx.Cfg.m_Data.OsOps.Remove(filePath) # raise + continue + + raise + + # OK, Go HOME! + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__DoWork__Stage05__WriteContents( + ctx: PostgresConfigurationWriterCtx_Base, + ) -> None: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + + for iFile in range(len(ctx.AllFiles)): + fileCtx = ctx.AllFiles[iFile] + + assert fileCtx is not None + assert type(fileCtx) is PostgresConfigurationWriterFileCtx_Base + assert type(fileCtx.FileData) is PgCfgModel__FileData + + assert fileCtx.File is not None + assert isinstance(fileCtx.File, ConfigurationOsFile) + + assert fileCtx.Content is not None + assert type(fileCtx.Content) is str + + fileCtx.File.Overwrite(fileCtx.Content) + + lastMDate = fileCtx.File.GetModificationTS() + assert type(lastMDate) is datetime.datetime + + fileCtx.File.Close() + + fileCtx.FileData.m_LastModifiedTimestamp = lastMDate + fileCtx.FileData.m_Status = PgCfgModel__FileStatus.EXISTS + continue + return + + # -------------------------------------------------------------------- + @staticmethod + def Helper__MakeFileDataContent( + ctx: PostgresConfigurationWriterCtx_Base, + fileData: PgCfgModel__FileData, + ) -> str: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert type(fileData) is PgCfgModel__FileData + + fileContent = "" + + for fileLineData in fileData.m_Lines: + assert type(fileLineData) is PgCfgModel__FileLineData + lineContent = __class__.Helper__FileLineToString( + ctx, + fileLineData, + ) + assert type(lineContent) is str + fileContent += lineContent + fileContent += "\n" + continue + + return fileContent + + # -------------------------------------------------------------------- + @staticmethod + def Helper__FileLineToString( + ctx: PostgresConfigurationWriterCtx_Base, + fileLineData: PgCfgModel__FileLineData, + ) -> str: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert type(fileLineData) is PgCfgModel__FileLineData + + fileLineItemCount = len(fileLineData.m_Items) + + if fileLineItemCount == 0: + return "" + + lineContent = "" + + for lineItem in fileLineData.m_Items: + assert type(lineItem) is PgCfgModel__FileLineData.tagItem + assert lineItem.m_Element is not None + assert isinstance(lineItem.m_Element, PgCfgModel__FileLineElementData) + assert ( + lineItem.m_Element.m_Offset is None + or type(lineItem.m_Element.m_Offset) is int + ) + assert ( + lineItem.m_Element.m_Offset is None or lineItem.m_Element.m_Offset >= 0 + ) + + itemContent = __class__.Helper__ElementToString( + ctx, + lineItem.m_Element, + ) + assert type(itemContent) is str + + lineContent = __class__.Helper__AppendItemToLine( + lineContent, + lineItem.m_Element.m_Offset, + itemContent, + ) + assert type(lineContent) is str + continue + + assert type(lineContent) is str + return lineContent + + # -------------------------------------------------------------------- + @staticmethod + def Helper__AppendItemToLine( + lineContent: str, + offset: typing.Optional[int], + text: str, + ) -> str: + assert type(lineContent) is str + assert offset is None or type(offset) is int + assert type(text) is str + assert offset is None or offset >= 0 + + if text == "": + return lineContent + + lineContentLen = len(lineContent) + + assert type(lineContentLen) is int + assert lineContentLen >= 0 + + if offset is not None and lineContentLen < offset: + lineContent += " " * (offset - lineContentLen) + elif lineContentLen > 0: + lineContent += " " + + lineContent += text + return lineContent + + # -------------------------------------------------------------------- + @staticmethod + def Helper__ElementToString( + ctx: PostgresConfigurationWriterCtx_Base, + elementData: PgCfgModel__FileLineElementData, + ) -> str: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert elementData is not None + assert isinstance(elementData, PgCfgModel__FileLineElementData) + + if type(elementData) is PgCfgModel__OptionData: + return __class__.Helper__OptionToString( + ctx, + elementData, + ) + + if type(elementData) is PgCfgModel__CommentData: + return __class__.Helper__CommentToString( + ctx, + elementData, + ) + + if type(elementData) is PgCfgModel__IncludeData: + return __class__.Helper__IncludeToString( + ctx, + elementData, + ) + + BugCheckError.UnkObjectDataType(type(elementData)) + + # -------------------------------------------------------------------- + @staticmethod + def Helper__OptionToString( + ctx: PostgresConfigurationWriterCtx_Base, + optionData: PgCfgModel__OptionData, + ) -> str: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert ctx.Cfg is not None + assert isinstance(ctx.Cfg, PostgresConfiguration_Base) + assert type(optionData) is PgCfgModel__OptionData + assert type(optionData.m_Name) is str + assert optionData.m_Name != "" + assert optionData.m_Value is not None + + writeHandler = ctx.Cfg.Internal__GetOptionHandlerToWrite(optionData.m_Name) + + assert writeHandler is not None + assert isinstance(writeHandler, PgCfgModel__OptionHandlerToWrite) + + writeHandlerCtx = PgCfgModel__OptionHandlerCtxToWrite( + ctx.Cfg, + optionData.m_Name, + optionData.m_Value, + ) + + optValueAsText = writeHandler.OptionValueToString(writeHandlerCtx) + + assert type(optValueAsText) is str + assert optValueAsText != "" + + result = optionData.m_Name + " = " + optValueAsText + return result + + # -------------------------------------------------------------------- + @staticmethod + def Helper__CommentToString( + ctx: PostgresConfigurationWriterCtx_Base, + commentData: PgCfgModel__CommentData, + ) -> str: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert ctx.Cfg is not None + assert isinstance(ctx.Cfg, PostgresConfiguration_Base) + assert type(commentData) is PgCfgModel__CommentData + assert commentData.m_Text is not None + assert type(commentData.m_Text) is str + + assert DataVerificator.IsValidCommentText(commentData.m_Text) + + result = "#" + commentData.m_Text + return result + + # -------------------------------------------------------------------- + @staticmethod + def Helper__IncludeToString( + ctx: PostgresConfigurationWriterCtx_Base, + includeData: PgCfgModel__IncludeData, + ) -> str: + assert type(ctx) is PostgresConfigurationWriterCtx_Base + assert ctx.Cfg is not None + assert isinstance(ctx.Cfg, PostgresConfiguration_Base) + assert type(includeData) is PgCfgModel__IncludeData + assert includeData.m_Path is not None + assert type(includeData.m_Path) is str + assert includeData.m_Path != "" + + result = "include " + WriteUtils.Pack_Str(includeData.m_Path) + + assert type(result) is str + assert result != "" + return result + + +# ////////////////////////////////////////////////////////////////////////////// diff --git a/src/implementation/v00/configuration_std.py b/src/implementation/v00/configuration_std.py index fc433b5..a69a021 100644 --- a/src/implementation/v00/configuration_std.py +++ b/src/implementation/v00/configuration_std.py @@ -4,9 +4,6 @@ from __future__ import annotations from .configuration_base import PostgresConfiguration_Base -from .configuration_base import PostgresConfigurationReader_Base -from .configuration_base import PostgresConfigurationWriter_Base -from .configuration_base import PostgresConfigurationWriterCtx_Base from .configuration_base import PgCfgModel__OptionHandlerToPrepareSetValue from .configuration_base import PgCfgModel__OptionHandlerToPrepareSetValueItem from .configuration_base import PgCfgModel__OptionHandlerToPrepareGetValue @@ -16,6 +13,11 @@ from .configuration_base import PgCfgModel__OptionHandlerToSetValueItem from .configuration_base import PgCfgModel__OptionHandlerToWrite +from .configuration_base_r import PostgresConfigurationReader_Base + +from .configuration_base_w import PostgresConfigurationWriter_Base +from .configuration_base_w import PostgresConfigurationWriterCtx_Base + from ...os.abstract.configuration_os_ops import ConfigurationOsOps from ...os.local.configuration_os_ops import SingleInstance as LocalCfgOsOps diff --git a/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationReader_Base/LoadConfigurationFile/test_set001__common.py b/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationReader_Base/LoadConfigurationFile/test_set001__common.py index 780fa18..44d7907 100644 --- a/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationReader_Base/LoadConfigurationFile/test_set001__common.py +++ b/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationReader_Base/LoadConfigurationFile/test_set001__common.py @@ -3,17 +3,18 @@ # fmt: off from src.implementation.v00.configuration_std import PostgresConfiguration_Std as PgCfg_Std +from src.implementation.v00.configuration_std import PostgresConfigurationReader_Base as PgCfg_Reader_Base from src.implementation.v00.configuration_base import PostgresConfigurationFile_Base as PgCfg_File_Base from src.implementation.v00.configuration_base import PostgresConfigurationTopLevelFile_Base as PgCfg_TopLevelFile_Base -from src.implementation.v00.configuration_base import PostgresConfigurationReader_Base as PgCfg_Reader_Base from src.implementation.v00.configuration_base import PostgresConfigurationFile as PgCfg_File from src.implementation.v00.configuration_base import PgCfgModel__CommentData from src.implementation.v00.configuration_base import PgCfgModel__OptionData from src.implementation.v00.configuration_base import PgCfgModel__FileData -from src.implementation.v00.configuration_base import PgCfgModel__FileStatus + +from src.implementation.v00.configuration_base_r import PgCfgModel__FileStatus from .......TestServices import TestServices # fmt: on diff --git a/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationReader_Base/LoadFileDataContent/test_set001__common.py b/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationReader_Base/LoadFileDataContent/test_set001__common.py index 181d32c..ceca29a 100644 --- a/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationReader_Base/LoadFileDataContent/test_set001__common.py +++ b/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationReader_Base/LoadFileDataContent/test_set001__common.py @@ -3,10 +3,10 @@ # fmt: off from src.implementation.v00.configuration_std import PostgresConfiguration_Std as PgCfg_Std +from src.implementation.v00.configuration_std import PostgresConfigurationReader_Base as PgCfg_Reader_Base from src.implementation.v00.configuration_base import PostgresConfigurationFile_Base as PgCfg_File_Base from src.implementation.v00.configuration_base import PostgresConfigurationTopLevelFile_Base as PgCfg_TopLevelFile_Base -from src.implementation.v00.configuration_base import PostgresConfigurationReader_Base as PgCfg_Reader_Base from src.implementation.v00.configuration_base import PgCfgModel__FileData from src.implementation.v00.configuration_base import PgCfgModel__CommentData diff --git a/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationWriter_Base/DoWork/test_set001__work.py b/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationWriter_Base/DoWork/test_set001__work.py index 7ceb2e9..33d9e4b 100644 --- a/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationWriter_Base/DoWork/test_set001__work.py +++ b/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationWriter_Base/DoWork/test_set001__work.py @@ -3,16 +3,16 @@ # fmt: off from src.implementation.v00.configuration_std import PostgresConfiguration_Std as PgCfg_Std +from src.implementation.v00.configuration_std import PostgresConfigurationWriter_Base as PgCfg_Writer_Base +from src.implementation.v00.configuration_std import PostgresConfigurationWriterCtx_Base as PgCfg_WriterCtx_Base from src.implementation.v00.configuration_base import PostgresConfigurationFile_Base as PgCfg_File_Base from src.implementation.v00.configuration_base import PostgresConfigurationTopLevelFile_Base as PgCfg_TopLevelFile_Base -from src.implementation.v00.configuration_base import PostgresConfigurationWriter_Base as PgCfg_Writer_Base -from src.implementation.v00.configuration_base import PostgresConfigurationWriterCtx_Base as PgCfg_WriterCtx_Base from src.implementation.v00.configuration_base import PostgresConfigurationFile as PgCfg_File from src.implementation.v00.configuration_base import PgCfgModel__FileData -from src.implementation.v00.configuration_base import PgCfgModel__FileStatus +from src.implementation.v00.configuration_base_r import PgCfgModel__FileStatus from .......TestServices import TestServices # fmt: on diff --git a/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationWriter_Base/test_set001__common.py b/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationWriter_Base/test_set001__common.py index 7dbc598..d13e9c9 100644 --- a/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationWriter_Base/test_set001__common.py +++ b/tests/implementation/v00/configuration_std/Objects/PostgresConfigurationWriter_Base/test_set001__common.py @@ -4,8 +4,8 @@ # fmt: off from src.implementation.v00.configuration_std import PostgresConfiguration_Std as PgCfg_Std -from src.implementation.v00.configuration_base import PostgresConfigurationWriter_Base as PgCfg_Writer_Base -from src.implementation.v00.configuration_base import PostgresConfigurationWriterCtx_Base as PgCfg_WriterCtx_Base +from src.implementation.v00.configuration_std import PostgresConfigurationWriter_Base as PgCfg_Writer_Base +from src.implementation.v00.configuration_std import PostgresConfigurationWriterCtx_Base as PgCfg_WriterCtx_Base from src.implementation.v00.configuration_base import PgCfgModel__FileData diff --git a/tests/implementation/v00/configuration_std/Options/STD/generic/bool_option/test_set001__common.py b/tests/implementation/v00/configuration_std/Options/STD/generic/bool_option/test_set001__common.py index 281e18e..7e3da23 100644 --- a/tests/implementation/v00/configuration_std/Options/STD/generic/bool_option/test_set001__common.py +++ b/tests/implementation/v00/configuration_std/Options/STD/generic/bool_option/test_set001__common.py @@ -3,8 +3,7 @@ # fmt: off from src.implementation.v00.configuration_std import PostgresConfiguration_Std as PgCfg_Std - -from src.implementation.v00.configuration_base import PostgresConfigurationReader_Base as PgCfg_Reader_Base +from src.implementation.v00.configuration_std import PostgresConfigurationReader_Base as PgCfg_Reader_Base from src.implementation.v00.configuration_base import PgCfgModel__OptionData diff --git a/tests/implementation/v00/configuration_std/Options/STD/listen_addresses/test_set001__common.py b/tests/implementation/v00/configuration_std/Options/STD/listen_addresses/test_set001__common.py index 0fc7710..6e47945 100644 --- a/tests/implementation/v00/configuration_std/Options/STD/listen_addresses/test_set001__common.py +++ b/tests/implementation/v00/configuration_std/Options/STD/listen_addresses/test_set001__common.py @@ -3,9 +3,8 @@ # fmt: off from src.implementation.v00.configuration_std import PostgresConfiguration_Std as PgCfg_Std - -from src.implementation.v00.configuration_base import PostgresConfigurationWriter_Base as PgCfg_Writer_Base -from src.implementation.v00.configuration_base import PostgresConfigurationWriterCtx_Base as PgCfg_WriterCtx_Base +from src.implementation.v00.configuration_std import PostgresConfigurationWriter_Base as PgCfg_Writer_Base +from src.implementation.v00.configuration_std import PostgresConfigurationWriterCtx_Base as PgCfg_WriterCtx_Base from .......TestServices import TestServices # fmt: on