Merge pull request '2020.12.6' (#3) from 2020.12.6 into 2020.12

Reviewed-on: http://git.sh-edraft.de/sh-edraft.de/sh_common_py_lib/pulls/3
This commit is contained in:
Sven Heidemann 2020-11-26 10:48:12 +01:00
commit 5faac76c32
48 changed files with 846 additions and 312 deletions

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.source_code sh_edraft.coding
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.source_code
""" """
__title__ = 'sh_edraft.source_code' __title__ = 'sh_edraft.coding'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.source_code.model sh_edraft.coding.base
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.source_code.model
""" """
__title__ = 'sh_edraft.source_code.model' __title__ = 'sh_edraft.coding.base'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'

View File

@ -1,7 +1,7 @@
from typing import Optional from typing import Optional
from sh_edraft.source_code.model.version_enum import VersionEnum from sh_edraft.coding.model.version_enum import VersionEnum
from sh_edraft.configuration.model.configuration_model_base import ConfigurationModelBase from sh_edraft.configuration.base.configuration_model_base import ConfigurationModelBase
class Version(ConfigurationModelBase): class Version(ConfigurationModelBase):

View File

@ -20,7 +20,6 @@ __version__ = '2020.12.5'
from collections import namedtuple from collections import namedtuple
# imports: # imports:
from .application_host import ApplicationHost
VersionInfo = namedtuple('VersionInfo', 'major minor micro') VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major=2020, minor=12, micro=5) version_info = VersionInfo(major=2020, minor=12, micro=5)

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.configuration.model sh_edraft.configuration.base
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.configuration.model
""" """
__title__ = 'sh_edraft.configuration.model' __title__ = 'sh_edraft.configuration.base'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'

View File

@ -0,0 +1,23 @@
from abc import abstractmethod
from collections import Callable
from sh_edraft.configuration.base.configuration_model_base import ConfigurationModelBase
from sh_edraft.service.base.service_base import ServiceBase
class ConfigurationBase(ServiceBase):
@abstractmethod
def __init__(self):
ServiceBase.__init__(self)
self._config: dict[type, object] = {}
@property
@abstractmethod
def config(self) -> dict[type, object]: pass
@abstractmethod
def add_config_by_type(self, key_type: type, value: object): pass
@abstractmethod
def get_config_by_type(self, search_type: ConfigurationModelBase) -> Callable[ConfigurationModelBase]: pass

View File

@ -0,0 +1,27 @@
from collections import Callable
from sh_edraft.configuration.base.configuration_model_base import ConfigurationModelBase
from sh_edraft.configuration.base.configuration_base import ConfigurationBase
class Configuration(ConfigurationBase):
def __init__(self):
super().__init__()
@property
def config(self):
return self._config
def create(self): pass
def add_config_by_type(self, key_type: type, value: object):
self._config[key_type] = value
def get_config_by_type(self, search_type: type) -> Callable[ConfigurationModelBase]:
if search_type not in self._config:
raise Exception(f'Config model by type {search_type} not found')
for config_model in self._config:
if config_model == search_type:
return self._config[config_model]

View File

@ -0,0 +1,3 @@
# imports:
from .application_host import ApplicationHost

View File

@ -1,14 +1,16 @@
from datetime import datetime from datetime import datetime
from sh_edraft.service import ServiceProvider from sh_edraft.hosting.base.application_host_base import ApplicationHostBase
from sh_edraft.service.service_provider import ServiceProvider
class ApplicationHost: class ApplicationHost(ApplicationHostBase):
def __init__(self): def __init__(self):
ApplicationHostBase.__init__(self)
self._services = ServiceProvider() self._services = ServiceProvider()
self._end_time: datetime = datetime.now()
self._start_time: datetime = datetime.now() self._start_time: datetime = datetime.now()
self._end_time: datetime = datetime.now()
@property @property
def services(self): def services(self):
@ -19,7 +21,7 @@ class ApplicationHost:
return self._end_time return self._end_time
@end_time.setter @end_time.setter
def end_time(self, end_time: datetime) -> None: def end_time(self, end_time: datetime):
self._end_time = end_time self._end_time = end_time
@property @property
@ -27,7 +29,7 @@ class ApplicationHost:
return self._start_time return self._start_time
@start_time.setter @start_time.setter
def start_time(self, start_time: datetime) -> None: def start_time(self, start_time: datetime):
self._start_time = start_time self._start_time = start_time
@property @property

View File

@ -0,0 +1,2 @@
# imports:
from .application_host_base import ApplicationHostBase

View File

@ -0,0 +1,22 @@
from abc import ABC, abstractmethod
from datetime import datetime
class ApplicationHostBase(ABC):
@abstractmethod
def __init__(self): pass
@property
@abstractmethod
def start_time(self) -> datetime: pass
@start_time.setter
def start_time(self, start_time: datetime): pass
@property
@abstractmethod
def end_time(self): pass
@end_time.setter
def end_time(self, end_time: datetime): pass

View File

View File

@ -1,14 +1,20 @@
from abc import abstractmethod from abc import abstractmethod
from sh_edraft.service.base import ServiceBase from sh_edraft.hosting.application_host import ApplicationHost
from sh_edraft.logging.model.logging_settings import LoggingSettings
from sh_edraft.service.base.service_base import ServiceBase
from sh_edraft.time.model.time_format_settings import TimeFormatSettings
class LoggerBase(ServiceBase): class LoggerBase(ServiceBase):
@abstractmethod @abstractmethod
def __init__(self): def __init__(self, logging_settings: LoggingSettings, time_format: TimeFormatSettings):
ServiceBase.__init__(self) ServiceBase.__init__(self)
self._log_settings: LoggingSettings = logging_settings
self._time_format_settings: TimeFormatSettings = time_format
@abstractmethod @abstractmethod
def header(self, string: str): pass def header(self, string: str): pass

View File

@ -1,35 +1,23 @@
import datetime import datetime
import os import os
import traceback import traceback
from collections import Callable
from string import Template from string import Template
from typing import Optional
from sh_edraft.configuration.application_host import ApplicationHost from sh_edraft.hosting.application_host import ApplicationHost
from sh_edraft.logging.base.logger_base import LoggerBase from sh_edraft.logging.base.logger_base import LoggerBase
from sh_edraft.logging.model.log_level import LoggingLevel from sh_edraft.logging.model import LoggingSettings
from sh_edraft.logging.model.log_settings import LoggingSettings from sh_edraft.logging.model.logging_level import LoggingLevel
from sh_edraft.time.model.time_format_settings import TimeFormatSettings from sh_edraft.time.model import TimeFormatSettings
from sh_edraft.utils.console import Console from sh_edraft.utils.console import Console
class Logger(LoggerBase): class Logger(LoggerBase):
def __init__(self): def __init__(self, logging_settings: LoggingSettings, time_format: TimeFormatSettings, app_host: ApplicationHost):
LoggerBase.__init__(self) LoggerBase.__init__(self, logging_settings, time_format)
self._log_settings: Optional[LoggingSettings] = None self._app_host: ApplicationHost = app_host
self._time_format_settings: Optional[TimeFormatSettings] = None
self._app_host: Optional[ApplicationHost] = None
self._log: Optional[str] = None
self._path: Optional[str] = None
self._level: Optional[LoggingLevel] = None
self._console: Optional[LoggingLevel] = None
def init(self, args: tuple):
self._log_settings = args[0]
self._time_format_settings = args[1]
self._app_host = args[2]
self._log = Template(self._log_settings.filename).substitute( self._log = Template(self._log_settings.filename).substitute(
date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format), date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format),

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.logging.model sh_edraft.logging.base
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.logging.model
""" """
__title__ = 'sh_edraft.logging.model' __title__ = 'sh_edraft.logging.base'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'
@ -20,8 +20,8 @@ __version__ = '2020.12.5'
from collections import namedtuple from collections import namedtuple
# imports: # imports:
from .log_level import LoggingLevel from .logging_level import LoggingLevel
from .log_settings import LoggingSettings from .logging_settings import LoggingSettings
VersionInfo = namedtuple('VersionInfo', 'major minor micro') VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major=2020, minor=12, micro=5) version_info = VersionInfo(major=2020, minor=12, micro=5)

View File

@ -1,10 +1,10 @@
import traceback import traceback
from typing import Optional from typing import Optional
from sh_edraft.configuration.model.configuration_model_base import ConfigurationModelBase from sh_edraft.configuration.base.configuration_model_base import ConfigurationModelBase
from sh_edraft.logging.model.log_settings_name import LogSettingsName from sh_edraft.logging.model.logging_settings_name import LogSettingsName
from sh_edraft.utils.console import Console from sh_edraft.utils.console import Console
from sh_edraft.logging.model.log_level import LoggingLevel from sh_edraft.logging.model.logging_level import LoggingLevel
class LoggingSettings(ConfigurationModelBase): class LoggingSettings(ConfigurationModelBase):

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.messenger sh_edraft.messaging
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.messenger
""" """
__title__ = 'sh_edraft.messenger' __title__ = 'sh_edraft.messaging'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'

View File

@ -1,21 +0,0 @@
from abc import abstractmethod
from sh_edraft.service.base import ServiceBase
class PublisherBase(ServiceBase):
@abstractmethod
def __init__(self):
ServiceBase.__init__(self)
@property
@abstractmethod
def source_path(self) -> str: pass
@property
@abstractmethod
def dist_path(self) -> str: pass
@abstractmethod
def publish(self) -> str: pass

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.publish sh_edraft.publishing
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.publish
""" """
__title__ = 'sh_edraft.publish' __title__ = 'sh_edraft.publishing'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.publish.base sh_edraft.publishing.base
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.publish.base
""" """
__title__ = 'sh_edraft.publish.base' __title__ = 'sh_edraft.publishing.base'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'

View File

@ -0,0 +1,32 @@
from abc import abstractmethod
from sh_edraft.logging.base.logger_base import LoggerBase
from sh_edraft.publishing.model.publish_settings_model import PublishSettingsModel
from sh_edraft.service.base.service_base import ServiceBase
class PublisherBase(ServiceBase):
@abstractmethod
def __init__(self, logger: LoggerBase, publish_settings: PublishSettingsModel):
ServiceBase.__init__(self)
self._logger: LoggerBase = logger
self._publish_settings: PublishSettingsModel = publish_settings
@property
@abstractmethod
def source_path(self) -> str: pass
@property
@abstractmethod
def dist_path(self) -> str: pass
@abstractmethod
def include(self, path: str): pass
@abstractmethod
def exclude(self, path: str): pass
@abstractmethod
def publish(self) -> str: pass

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.publish.model sh_edraft.publishing.base
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.publish.model
""" """
__title__ = 'sh_edraft.publish.model' __title__ = 'sh_edraft.publishing.base'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'

View File

@ -0,0 +1,83 @@
import traceback
from typing import Optional
from sh_edraft.configuration.base.configuration_model_base import ConfigurationModelBase
from sh_edraft.publishing.model import Template
from sh_edraft.publishing.model.publish_settings_name import PublishSettingsName
from sh_edraft.utils import Console
class PublishSettingsModel(ConfigurationModelBase):
def __init__(self):
ConfigurationModelBase.__init__(self)
self._source_path: Optional[str] = None
self._dist_path: Optional[str] = None
self._templates: Optional[list[Template]] = None
self._included_files: Optional[list[str]] = None
self._excluded_files: Optional[list[str]] = None
self._template_ending: Optional[str] = None
@property
def source_path(self) -> str:
return self._source_path
@source_path.setter
def source_path(self, source_path: str):
self._source_path = source_path
@property
def dist_path(self) -> str:
return self._dist_path
@dist_path.setter
def dist_path(self, dist_path: str):
self._dist_path = dist_path
@property
def templates(self) -> list[Template]:
return self._templates
@templates.setter
def templates(self, templates: list[Template]):
self._templates = templates
@property
def included_files(self) -> list[str]:
return self._included_files
@included_files.setter
def included_files(self, included_files: list[str]):
self._included_files = included_files
@property
def excluded_files(self) -> list[str]:
return self._excluded_files
@excluded_files.setter
def excluded_files(self, excluded_files: list[str]):
self._excluded_files = excluded_files
@property
def template_ending(self) -> str:
return self._template_ending
@template_ending.setter
def template_ending(self, template_ending: str):
self._template_ending = template_ending
def from_dict(self, settings: dict):
try:
self._source_path = settings[PublishSettingsName.source_path.value]
self._dist_path = settings[PublishSettingsName.dist_path.value]
self._templates = settings[PublishSettingsName.templates.value]
self._included_files = settings[PublishSettingsName.included_files.value]
self._excluded_files = settings[PublishSettingsName.excluded_files.value]
self._template_ending = settings[PublishSettingsName.template_ending.value]
except Exception as e:
Console.write_line(
f'[ ERROR ] [ {__name__} ]: Reading error in {PublishSettingsName.publish.value} settings', 'red')
Console.write_line(f'[ EXCEPTION ] [ {__name__} ]: {e} -> {traceback.format_exc()}', 'red')

View File

@ -0,0 +1,12 @@
from enum import Enum
class PublishSettingsName(Enum):
publish = 'Publish'
source_path = 'SourcePath'
dist_path = 'DistPath'
templates = 'Templates'
included_files = 'IncludedFiles'
excluded_files = 'ExcludedFiles'
template_ending = 'TemplateEnding'

View File

@ -1,8 +1,8 @@
from typing import Optional from typing import Optional
from sh_edraft.source_code.model.version import Version from sh_edraft.coding.model.version import Version
from sh_edraft.configuration.model import ConfigurationModelBase from sh_edraft.configuration.base.configuration_model_base import ConfigurationModelBase
from sh_edraft.publish.model.template_enum import TemplateEnum from sh_edraft.publishing.model.template_enum import TemplateEnum
class Template(ConfigurationModelBase): class Template(ConfigurationModelBase):

View File

@ -1,35 +1,25 @@
import os import os
import shutil import shutil
from string import Template as stringTemplate from string import Template as stringTemplate
from typing import Optional
from sh_edraft.logging.base.logger_base import LoggerBase from sh_edraft.logging.base.logger_base import LoggerBase
from sh_edraft.publish.base.publisher_base import PublisherBase from sh_edraft.publishing.base.publisher_base import PublisherBase
from sh_edraft.publish.model.template import Template from sh_edraft.publishing.model.publish_settings_model import PublishSettingsModel
from sh_edraft.publishing.model.template import Template
class Publisher(PublisherBase): class Publisher(PublisherBase):
def __init__(self): def __init__(self, logger: LoggerBase, publish_settings: PublishSettingsModel):
super().__init__() super().__init__(logger, publish_settings)
self._logger: Optional[LoggerBase] = None
self._source_path: Optional[str] = None
self._dist_path: Optional[str] = None
self._settings: Optional[list[Template]] = None
self._included_files: list[str] = []
self._excluded_files: list[str] = []
self._template_ending = '_template.txt'
@property @property
def source_path(self) -> str: def source_path(self) -> str:
return self._source_path return self._publish_settings.source_path
@property @property
def dist_path(self): def dist_path(self):
return self._dist_path return self._publish_settings.dist_path
def _get_template_output(self, t: Template, name: str, imports: str) -> str: def _get_template_output(self, t: Template, name: str, imports: str) -> str:
self._logger.trace(__name__, f'Started {__name__}._get_template_output') self._logger.trace(__name__, f'Started {__name__}._get_template_output')
@ -60,41 +50,41 @@ class Publisher(PublisherBase):
def _read_source_path(self): def _read_source_path(self):
self._logger.trace(__name__, f'Started {__name__}._read_source_path') self._logger.trace(__name__, f'Started {__name__}._read_source_path')
for r, d, f in os.walk(self._source_path): for r, d, f in os.walk(self._publish_settings.source_path):
for file in f: for file in f:
if file.endswith('.py') or file in self._included_files: if file.endswith('.py') or file in self._publish_settings.included_files:
self._included_files.append(os.path.join(r, file)) self._publish_settings.included_files.append(os.path.join(r, file))
self._logger.trace(__name__, f'Stopped {__name__}._read_source_path') self._logger.trace(__name__, f'Stopped {__name__}._read_source_path')
def _read_templates(self): def _read_templates(self):
self._logger.trace(__name__, f'Started {__name__}._read_templates') self._logger.trace(__name__, f'Started {__name__}._read_templates')
for t in self._settings: for t in self._publish_settings.templates:
output_template: str = '' output_template: str = ''
if not os.path.isfile(t.template_path): if not os.path.isfile(t.template_path):
raise Exception(f'Template not found: {t.template_path}') self._logger.fatal(__name__, f'Template not found: {t.template_path}')
with open(t.template_path) as template: with open(t.template_path) as template:
t.file_content = template.read() t.file_content = template.read()
template.close() template.close()
if t.file_content == '': if t.file_content == '':
raise Exception(f'Template is empty: {t.template_path}') self._logger.fatal(__name__, f'Template is empty: {t.template_path}')
self._logger.trace(__name__, f'Stopped {__name__}._read_templates') self._logger.trace(__name__, f'Stopped {__name__}._read_templates')
def _create_dist_path(self): def _create_dist_path(self):
self._logger.trace(__name__, f'Started {__name__}._create_dist_path') self._logger.trace(__name__, f'Started {__name__}._create_dist_path')
if os.path.isdir(self._dist_path): if os.path.isdir(self._publish_settings.dist_path):
try: try:
shutil.rmtree(self._dist_path) shutil.rmtree(self._publish_settings.dist_path)
self._logger.info(__name__, f'Deleted {self._dist_path}') self._logger.info(__name__, f'Deleted {self._publish_settings.dist_path}')
except Exception as e: except Exception as e:
self._logger.fatal(__name__, f'Cannot delete old dist directory', e) self._logger.fatal(__name__, f'Cannot delete old dist directory', e)
if not os.path.isdir(self._dist_path): if not os.path.isdir(self._publish_settings.dist_path):
try: try:
os.makedirs(self._dist_path) os.makedirs(self._publish_settings.dist_path)
self._logger.debug(__name__, f'Created directories: {self._dist_path}') self._logger.debug(__name__, f'Created directories: {self._publish_settings.dist_path}')
self._logger.info(__name__, f'Created dist directory') self._logger.info(__name__, f'Created dist directory')
except Exception as e: except Exception as e:
self._logger.fatal(__name__, f'Cannot create dist directory', e) self._logger.fatal(__name__, f'Cannot create dist directory', e)
@ -115,11 +105,11 @@ class Publisher(PublisherBase):
def _write_templates(self): def _write_templates(self):
self._logger.trace(__name__, f'Started {__name__}._write_templates') self._logger.trace(__name__, f'Started {__name__}._write_templates')
for template in self._settings: for template in self._publish_settings.templates:
for file in self._included_files: for file in self._publish_settings.included_files:
if os.path.basename(file) == '__init__.py' and file not in self._excluded_files: if os.path.basename(file) == '__init__.py' and file not in self._publish_settings.excluded_files:
template_name = template.name template_name = template.name
if template.name == '*' or template.name == '': if template.name == 'all' or template.name == '':
template_name = self._get_template_name_from_dirs(file) template_name = self._get_template_name_from_dirs(file)
else: else:
name = self._get_template_name_from_dirs(file) name = self._get_template_name_from_dirs(file)
@ -170,16 +160,16 @@ class Publisher(PublisherBase):
def _copy_all_included_files(self): def _copy_all_included_files(self):
self._logger.trace(__name__, f'Started {__name__}._copy_all_included_files') self._logger.trace(__name__, f'Started {__name__}._copy_all_included_files')
dist_path = self._dist_path dist_path = self._publish_settings.dist_path
if self._dist_path.endswith('/'): if self._publish_settings.dist_path.endswith('/'):
dist_path = dist_path[:len(dist_path) - 1] dist_path = dist_path[:len(dist_path) - 1]
for file in self._included_files: for file in self._publish_settings.included_files:
is_file_excluded = False is_file_excluded = False
if file in self._excluded_files: if file in self._publish_settings.excluded_files:
is_file_excluded = True is_file_excluded = True
else: else:
for excluded in self._excluded_files: for excluded in self._publish_settings.excluded_files:
if file.__contains__(excluded): if file.__contains__(excluded):
is_file_excluded = True is_file_excluded = True
@ -212,26 +202,18 @@ class Publisher(PublisherBase):
def include(self, path: str): def include(self, path: str):
self._logger.trace(__name__, f'Started {__name__}.include') self._logger.trace(__name__, f'Started {__name__}.include')
self._included_files.append(path) self._publish_settings.included_files.append(path)
self._logger.trace(__name__, f'Stopped {__name__}.include') self._logger.trace(__name__, f'Stopped {__name__}.include')
def exclude(self, path: str): def exclude(self, path: str):
self._logger.trace(__name__, f'Started {__name__}.exclude') self._logger.trace(__name__, f'Started {__name__}.exclude')
self._excluded_files.append(path) self._publish_settings.excluded_files.append(path)
self._logger.trace(__name__, f'Stopped {__name__}.exclude') self._logger.trace(__name__, f'Stopped {__name__}.exclude')
def init(self, args: tuple):
self._logger: LoggerBase = args[0]
self._source_path: str = args[1]
self._dist_path: str = args[2]
self._settings: list[Template] = args[3]
self._logger.header(f'{__name__}:')
def create(self): def create(self):
self._logger.trace(__name__, f'Started {__name__}.create') self._logger.trace(__name__, f'Started {__name__}.create')
if not self._dist_path.endswith('/'): if not self._publish_settings.dist_path.endswith('/'):
self._dist_path += '/' self._publish_settings.dist_path += '/'
self._read_source_path() self._read_source_path()
self._read_templates() self._read_templates()

View File

@ -6,8 +6,5 @@ class ServiceBase(ABC):
@abstractmethod @abstractmethod
def __init__(self): pass def __init__(self): pass
@abstractmethod
def init(self, args: tuple): pass
@abstractmethod @abstractmethod
def create(self): pass def create(self): pass

View File

@ -3,7 +3,6 @@ from collections import Callable
from typing import Type from typing import Type
from sh_edraft.service.base.service_base import ServiceBase from sh_edraft.service.base.service_base import ServiceBase
from sh_edraft.service.model.provide_state import ProvideState
class ServiceProviderBase(ServiceBase): class ServiceProviderBase(ServiceBase):
@ -11,18 +10,23 @@ class ServiceProviderBase(ServiceBase):
@abstractmethod @abstractmethod
def __init__(self): def __init__(self):
ServiceBase.__init__(self) ServiceBase.__init__(self)
self._transient_services: list[ProvideState] = []
self._scoped_services: list[ProvideState] = [] self._transient_services: dict[Type[ServiceBase], Type[ServiceBase]] = {}
self._singleton_services: list[ServiceBase] = [] self._scoped_services: dict[Type[ServiceBase], Type[ServiceBase]] = {}
self._singleton_services: dict[Type[ServiceBase], ServiceBase] = {}
@property
@abstractmethod
def config(self): pass
@abstractmethod @abstractmethod
def add_transient(self, service: Type[ServiceBase], *args): pass def add_transient(self, service_type: Type[ServiceBase], service: Type[ServiceBase]): pass
@abstractmethod @abstractmethod
def add_scoped(self, service: Type[ServiceBase], *args): pass def add_scoped(self, service_type: Type[ServiceBase], service: Type[ServiceBase]): pass
@abstractmethod @abstractmethod
def add_singleton(self, service: Type[ServiceBase], *args): pass def add_singleton(self, service_type: Type[ServiceBase], service: ServiceBase): pass
@abstractmethod @abstractmethod
def get_service(self, instance_type: Type[ServiceBase]) -> Callable[ServiceBase]: pass def get_service(self, instance_type: Type[ServiceBase]) -> Callable[ServiceBase]: pass

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.service.model sh_edraft.service.base
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.service.model
""" """
__title__ = 'sh_edraft.service.model' __title__ = 'sh_edraft.service.base'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'

View File

@ -1,69 +1,83 @@
from collections import Callable from collections import Callable
from inspect import signature, Parameter
from typing import Type from typing import Type
from termcolor import colored from sh_edraft.configuration.configuration import Configuration
from sh_edraft.hosting.base.application_host_base import ApplicationHostBase
from sh_edraft.configuration.base.configuration_model_base import ConfigurationModelBase
from sh_edraft.service.base.service_provider_base import ServiceProviderBase from sh_edraft.service.base.service_provider_base import ServiceProviderBase
from sh_edraft.service.base.service_base import ServiceBase from sh_edraft.service.base.service_base import ServiceBase
from sh_edraft.service.model.provide_state import ProvideState
class ServiceProvider(ServiceProviderBase): class ServiceProvider(ServiceProviderBase):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self._config = Configuration()
def init(self, args: tuple): pass @property
def config(self):
return self._config
def create(self): pass def create(self): pass
@staticmethod def _create_instance(self, service: Callable[ServiceBase]) -> ServiceBase:
def _create_instance(service: type[ServiceBase], args: tuple) -> ServiceBase: sig = signature(service.__init__)
instance = service() params = []
try: for param in sig.parameters.items():
instance.init(args) parameter = param[1]
return instance if parameter.name != 'self' and parameter.annotation != Parameter.empty:
except Exception as e: if issubclass(parameter.annotation, ServiceBase):
print(colored(f'Argument error\n{e}', 'red')) params.append(self.get_service(parameter.annotation))
def add_transient(self, service: Type[ServiceBase], *args): elif issubclass(parameter.annotation, ConfigurationModelBase) or issubclass(parameter.annotation, ApplicationHostBase):
self._transient_services.append(ProvideState(service, args)) params.append(self._config.get_config_by_type(parameter.annotation))
def add_scoped(self, service: Type[ServiceBase], *args): return service(*params)
self._transient_services.append(ProvideState(service, args)) # try:
# instance.init(args)
# return instance
# except Exception as e:
# print(colored(f'Argument error\n{e}', 'red'))
def add_singleton(self, service: Type[ServiceBase], *args): def add_transient(self, service_type: Type[ServiceBase], service: Type[ServiceBase]):
self._transient_services[service_type] = service
def add_scoped(self, service_type: Type[ServiceBase], service: Type[ServiceBase]):
self._scoped_services[service_type] = service
def add_singleton(self, service_type: Type[ServiceBase], service: ServiceBase):
for known_service in self._singleton_services: for known_service in self._singleton_services:
if type(known_service) == type(service): if type(known_service) == type(service_type):
raise Exception(f'Service from type {type(service)} already exists') raise Exception(f'Service with type {type(service_type)} already exists')
self._singleton_services.append(self._create_instance(service, args)) self._singleton_services[service_type] = self._create_instance(service)
def get_service(self, instance_type: Type[ServiceBase]) -> Callable[ServiceBase]: def get_service(self, instance_type: Type[ServiceBase]) -> Callable[ServiceBase]:
for state in self._transient_services: for service in self._transient_services:
if isinstance(state.service, type(instance_type)): if service == instance_type and isinstance(self._transient_services[service], type(instance_type)):
return self._create_instance(state.service, state.args) return self._create_instance(self._transient_services[service])
for state in self._scoped_services: for service in self._scoped_services:
if isinstance(state.service, type(instance_type)): if service == instance_type and isinstance(self._scoped_services[service], type(instance_type)):
return self._create_instance(state.service, state.args) return self._create_instance(self._scoped_services[service])
for service in self._singleton_services: for service in self._singleton_services:
if isinstance(service, instance_type): if service == instance_type and isinstance(self._singleton_services[service], instance_type):
return service return self._singleton_services[service]
def remove_service(self, instance_type: type): def remove_service(self, instance_type: Type[ServiceBase]):
for state in self._transient_services: for service in self._transient_services:
if isinstance(state.service, type(instance_type)): if isinstance(service, type(instance_type)):
self._transient_services.remove(state) del self._transient_services[service]
return return
for state in self._scoped_services: for service in self._scoped_services:
if isinstance(state.service, type(instance_type)): if isinstance(service, type(instance_type)):
self._scoped_services.remove(state) del self._scoped_services[service]
return return
for service in self._singleton_services: for service in self._singleton_services:
if isinstance(service, instance_type): if isinstance(service, instance_type):
self._singleton_services.remove(service) del self._singleton_services[service]
return return

View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
sh_edraft.time.model sh_edraft.time.base
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
@ -11,7 +11,7 @@ sh_edraft.time.model
""" """
__title__ = 'sh_edraft.time.model' __title__ = 'sh_edraft.time.base'
__author__ = 'Sven Heidemann' __author__ = 'Sven Heidemann'
__license__ = 'MIT' __license__ = 'MIT'
__copyright__ = 'Copyright (c) 2020 sh-edraft.de' __copyright__ = 'Copyright (c) 2020 sh-edraft.de'

View File

@ -1,7 +1,7 @@
import traceback import traceback
from typing import Optional from typing import Optional
from sh_edraft.configuration.model import ConfigurationModelBase from sh_edraft.configuration.base.configuration_model_base import ConfigurationModelBase
from sh_edraft.time.model.time_format_settings_names import TimeFormatSettingsNames from sh_edraft.time.model.time_format_settings_names import TimeFormatSettingsNames
from sh_edraft.utils.console import Console from sh_edraft.utils.console import Console

View File

@ -1,50 +0,0 @@
import os
from string import Template
from sh_edraft.configuration import ApplicationHost
from sh_edraft.logging.base.logger_base import LoggerBase
from sh_edraft.logging.logger import Logger
from sh_edraft.logging.model.log_settings import LoggingSettings
from sh_edraft.time.model.time_format_settings import TimeFormatSettings
class LoggerTest:
@staticmethod
def start(app_host: ApplicationHost):
services = app_host.services
log_settings = LoggingSettings()
log_settings.from_dict({
"Path": "logs/",
"Filename": "log_$start_time.log",
"ConsoleLogLevel": "TRACE",
"FileLogLevel": "TRACE"
})
time_format_settings = TimeFormatSettings()
time_format_settings.from_dict({
"DateFormat": "%Y-%m-%d",
"TimeFormat": "%H:%M:%S",
"DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
"DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
})
services.add_singleton(Logger, log_settings, time_format_settings, app_host)
logger: Logger = services.get_service(LoggerBase)
if logger is None:
raise Exception(f'{__name__}: Service is None')
logger.create()
logger.info(__name__, 'test')
if not os.path.isdir(log_settings.path):
raise Exception(f'{__name__}: Log path was not created')
log_file = Template(log_settings.filename).substitute(
date_time_now=app_host.date_time_now.strftime(time_format_settings.date_time_format),
start_time=app_host.start_time.strftime(time_format_settings.date_time_log_format)
)
if not os.path.isfile(log_settings.path + log_file):
raise Exception(f'{__name__}: Log file was not created')

View File

210
src/tests/logging/logger.py Normal file
View File

@ -0,0 +1,210 @@
import os
import shutil
import unittest
from datetime import datetime
from string import Template
from sh_edraft.hosting import ApplicationHost
from sh_edraft.logging import Logger
from sh_edraft.logging.model import LoggingSettings
from sh_edraft.time.model import TimeFormatSettings
class LoggerTest(unittest.TestCase):
def setUp(self):
self._app_host = ApplicationHost()
self._services = self._app_host.services
self._services.create()
self._log_settings = LoggingSettings()
self._log_settings.from_dict({
"Path": "logs/",
"Filename": "log_$start_time.log",
"ConsoleLogLevel": "TRACE",
"FileLogLevel": "TRACE"
})
self._time_format_settings = TimeFormatSettings()
self._time_format_settings.from_dict({
"DateFormat": "%Y-%m-%d",
"TimeFormat": "%H:%M:%S",
"DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
"DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
})
def tearDown(self):
if os.path.isdir(self._log_settings.path):
shutil.rmtree(self._log_settings.path)
def _check_general_requirements(self):
self.assertIsNotNone(self._services)
self.assertIsNotNone(self._log_settings)
self.assertIsNotNone(self._time_format_settings)
def test_create(self):
print(f'{__name__}.test_create:')
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
self.assertTrue(os.path.isdir(self._log_settings.path))
log_file = Template(self._log_settings.filename).substitute(
date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format),
start_time=self._app_host.start_time.strftime(self._time_format_settings.date_time_log_format)
)
self.assertTrue(os.path.isfile(self._log_settings.path + log_file))
def test_header(self):
print(f'{__name__}.test_header:')
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.header('HeaderTest:')
log_file = Template(self._log_settings.filename).substitute(
date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format),
start_time=self._app_host.start_time.strftime(self._time_format_settings.date_time_log_format)
)
log_content = []
try:
with open(self._log_settings.path + log_file, "r") as log:
log_content = log.readlines()
log.close()
except Exception as e:
print('Cannot open log file', e)
self.assertGreater(len(log_content), 0)
self.assertEqual(log_content[len(log_content) - 1], 'HeaderTest:\n')
def test_trace(self):
print(f'{__name__}.test_trace:')
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.trace(__name__, f'{__name__}.test_trace:')
log_file = Template(self._log_settings.filename).substitute(
date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format),
start_time=self._app_host.start_time.strftime(self._time_format_settings.date_time_log_format)
)
log_content = []
try:
with open(self._log_settings.path + log_file, "r") as log:
log_content = log.readlines()
log.close()
except Exception as e:
print('Cannot open log file', e)
self.assertGreater(len(log_content), 0)
self.assertTrue(log_content[len(log_content) - 1].endswith(f'[ TRACE ] [ {__name__} ]: {__name__}.test_trace:\n'))
def test_debug(self):
print(f'{__name__}.test_debug:')
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.debug(__name__, f'{__name__}.test_debug:')
log_file = Template(self._log_settings.filename).substitute(
date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format),
start_time=self._app_host.start_time.strftime(self._time_format_settings.date_time_log_format)
)
log_content = []
try:
with open(self._log_settings.path + log_file, "r") as log:
log_content = log.readlines()
log.close()
except Exception as e:
print('Cannot open log file', e)
self.assertGreater(len(log_content), 0)
self.assertTrue(log_content[len(log_content) - 1].endswith(f'[ DEBUG ] [ {__name__} ]: {__name__}.test_debug:\n'))
def test_info(self):
print(f'{__name__}.test_info:')
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.info(__name__, f'{__name__}.test_info:')
log_file = Template(self._log_settings.filename).substitute(
date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format),
start_time=self._app_host.start_time.strftime(self._time_format_settings.date_time_log_format)
)
log_content = []
try:
with open(self._log_settings.path + log_file, "r") as log:
log_content = log.readlines()
log.close()
except Exception as e:
print('Cannot open log file', e)
self.assertGreater(len(log_content), 0)
self.assertTrue(log_content[len(log_content) - 1].endswith(f'[ INFO ] [ {__name__} ]: {__name__}.test_info:\n'))
def test_warn(self):
print(f'{__name__}.test_warn:')
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.warn(__name__, f'{__name__}.test_warn:')
log_file = Template(self._log_settings.filename).substitute(
date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format),
start_time=self._app_host.start_time.strftime(self._time_format_settings.date_time_log_format)
)
log_content = []
try:
with open(self._log_settings.path + log_file, "r") as log:
log_content = log.readlines()
log.close()
except Exception as e:
print('Cannot open log file', e)
self.assertGreater(len(log_content), 0)
self.assertTrue(log_content[len(log_content) - 1].endswith(f'[ WARN ] [ {__name__} ]: {__name__}.test_warn:\n'))
def test_error(self):
print(f'{__name__}.test_error:')
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.error(__name__, f'{__name__}.test_error:')
log_file = Template(self._log_settings.filename).substitute(
date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format),
start_time=self._app_host.start_time.strftime(self._time_format_settings.date_time_log_format)
)
log_content = []
try:
with open(self._log_settings.path + log_file, "r") as log:
log_content = log.readlines()
log.close()
except Exception as e:
print('Cannot open log file', e)
self.assertGreater(len(log_content), 0)
self.assertTrue(log_content[len(log_content) - 1].endswith(f'[ ERROR ] [ {__name__} ]: {__name__}.test_error:\n'))
def test_fatal(self):
print(f'{__name__}.test_fatal:')
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
with self.assertRaises(SystemExit):
logger.fatal(__name__, f'{__name__}.test_fatal:')
log_file = Template(self._log_settings.filename).substitute(
date_time_now=self._app_host.date_time_now.strftime(self._time_format_settings.date_time_format),
start_time=self._app_host.start_time.strftime(self._time_format_settings.date_time_log_format)
)
log_content = []
try:
with open(self._log_settings.path + log_file, "r") as log:
log_content = log.readlines()
log.close()
except Exception as e:
print('Cannot open log file', e)
self.assertGreater(len(log_content), 0)
self.assertTrue(log_content[len(log_content) - 1].endswith(f'[ ERROR ] [ {__name__} ]: {__name__}.test_fatal:\n'))

View File

@ -1,11 +1,11 @@
import os import os
from sh_edraft.logging.base.logger_base import LoggerBase from sh_edraft.logging.base.logger_base import LoggerBase
from sh_edraft.publish.base import PublisherBase from sh_edraft.publishing.base import PublisherBase
from sh_edraft.service.base import ServiceProviderBase from sh_edraft.service.base import ServiceProviderBase
from sh_edraft.source_code.model import Version from sh_edraft.coding.model import Version
from sh_edraft.publish import Publisher from sh_edraft.publishing import Publisher
from sh_edraft.publish.model import Template from sh_edraft.publishing.model import Template
class PublisherTest: class PublisherTest:

View File

View File

@ -0,0 +1,95 @@
import os
import shutil
import unittest
from sh_edraft.hosting import ApplicationHost
from sh_edraft.logging import Logger
from sh_edraft.logging.base import LoggerBase
from sh_edraft.logging.model import LoggingSettings
from sh_edraft.publishing import Publisher
from sh_edraft.publishing.base import PublisherBase
from sh_edraft.publishing.model import Template
from sh_edraft.publishing.model.publish_settings_model import PublishSettingsModel
from sh_edraft.coding.model import Version
from sh_edraft.time.model import TimeFormatSettings
class PublisherTest(unittest.TestCase):
def _config(self):
self._log_settings = LoggingSettings()
self._log_settings.from_dict({
"Path": "logs/",
"Filename": "log_$start_time.log",
"ConsoleLogLevel": "TRACE",
"FileLogLevel": "TRACE"
})
self._time_format_settings = TimeFormatSettings()
self._time_format_settings.from_dict({
"DateFormat": "%Y-%m-%d",
"TimeFormat": "%H:%M:%S",
"DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
"DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
})
self._version = Version(2020, 12, 5).to_dict()
templates = [
Template(
'../../publish_templates/all_template.txt',
'all',
'',
'',
'2020',
'sh-edraft.de',
'MIT',
', see LICENSE for more details.',
'',
'Sven Heidemann',
self._version
),
Template(
'../../publish_templates/all_template.txt',
'sh_edraft',
'common python library',
'Library to share common classes and models used at sh-edraft.de',
'2020',
'sh-edraft.de',
'MIT',
', see LICENSE for more details.',
'',
'Sven Heidemann',
self._version
)
]
self._source_path = '../'
self._dist_path = '../../dist'
self._publish_settings_model = PublishSettingsModel()
self._publish_settings_model.from_dict({
"SourcePath": self._source_path,
"DistPath": self._dist_path,
"Templates": templates,
"IncludedFiles": [],
"ExcludedFiles": [],
"TemplateEnding": "_template.txt",
})
def setUp(self):
self._config()
self._app_host = ApplicationHost()
self._logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
self._logger.create()
def tearDown(self):
if os.path.isdir(self._log_settings.path):
shutil.rmtree(self._log_settings.path)
def test_create(self):
publisher: Publisher = Publisher(self._logger, self._publish_settings_model)
self.assertIsNotNone(publisher)
publisher.create()
self.assertTrue(os.path.isdir(self._dist_path))

View File

@ -1,20 +0,0 @@
from sh_edraft.logging.base.logger_base import LoggerBase
from sh_edraft.publish import Publisher
from sh_edraft.publish.base import PublisherBase
from sh_edraft.service.base import ServiceProviderBase
class ServiceProviderTest:
@staticmethod
def start(services: ServiceProviderBase):
services.add_transient(Publisher, services.get_service(LoggerBase), '../', '../../dist', [])
publisher: Publisher = services.get_service(PublisherBase)
if publisher is None or publisher.source_path != '../' or publisher.dist_path != '../../dist':
raise Exception(f'{__name__}: Invalid value in {Publisher.__name__}')
services.remove_service(PublisherBase)
if services.get_service(PublisherBase) is not None:
raise Exception(f'{__name__}: Service {Publisher.__name__} was not removed')

View File

View File

@ -0,0 +1,158 @@
import unittest
from sh_edraft.hosting import ApplicationHost
from sh_edraft.logging import Logger
from sh_edraft.logging.base import LoggerBase
from sh_edraft.logging.model import LoggingSettings
from sh_edraft.publishing import Publisher
from sh_edraft.publishing.base import PublisherBase
from sh_edraft.publishing.model.publish_settings_model import PublishSettingsModel
from sh_edraft.service import ServiceProvider
from sh_edraft.service.base import ServiceBase
from sh_edraft.time.model import TimeFormatSettings
class ServiceProviderTest(unittest.TestCase):
def setUp(self):
self._app_host = ApplicationHost()
self._services = self._app_host.services
self._services.create()
self._log_settings = LoggingSettings()
self._log_settings.from_dict({
"Path": "logs/",
"Filename": "log_$start_time.log",
"ConsoleLogLevel": "TRACE",
"FileLogLevel": "TRACE"
})
self._services.config.add_config_by_type(LoggingSettings, self._log_settings)
self._time_format_settings = TimeFormatSettings()
self._time_format_settings.from_dict({
"DateFormat": "%Y-%m-%d",
"TimeFormat": "%H:%M:%S",
"DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
"DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
})
self._services.config.add_config_by_type(TimeFormatSettings, self._time_format_settings)
self._services.config.add_config_by_type(ApplicationHost, self._app_host)
self._publish_settings_model = PublishSettingsModel()
self._publish_settings_model.from_dict({
"SourcePath": "../",
"DistPath": "../../dist",
"Templates": [],
"IncludedFiles": [],
"ExcludedFiles": [],
"TemplateEnding": "_template.txt",
})
self._services.config.add_config_by_type(PublishSettingsModel, self._publish_settings_model)
def _check_general_requirements(self):
self.assertIsNotNone(self._services)
def _add_logger(self):
self._services.add_singleton(LoggerBase, Logger)
logger: Logger = self._services.get_service(LoggerBase)
logger.create()
def test_create(self):
print(f'{__name__}.test_create:')
provider = ServiceProvider()
self.assertIsNotNone(provider)
provider.create()
self.assertIsNotNone(provider)
def test_add_singleton(self):
print(f'{__name__}.test_add_singleton:')
self._check_general_requirements()
self._services.add_singleton(LoggerBase, Logger)
self.assertGreater(len(self._services._singleton_services), 0)
found = False
for service_type in self._services._singleton_services:
service = self._services._singleton_services[service_type]
if service_type == LoggerBase and (
isinstance(service, Logger) and isinstance(service, LoggerBase) and isinstance(service, ServiceBase)
):
if not found:
found = True
self.assertTrue(found)
found = False
for service_type in self._services._singleton_services:
service = self._services._singleton_services[service_type]
if service_type == PublisherBase and (
isinstance(service, Publisher) and isinstance(service, PublisherBase) and isinstance(service, ServiceBase)
):
if not found:
found = True
self.assertFalse(found)
def test_get_singleton(self):
print(f'{__name__}.test_get_singleton:')
self._check_general_requirements()
self._services.add_singleton(LoggerBase, Logger)
logger: Logger = self._services.get_service(LoggerBase)
self.assertIsNotNone(logger)
self.assertTrue(isinstance(logger, Logger))
self.assertTrue(isinstance(logger, LoggerBase))
self.assertTrue(isinstance(logger, ServiceBase))
self.assertEqual(logger._log_settings, self._log_settings)
self.assertEqual(logger._time_format_settings, self._time_format_settings)
self.assertEqual(logger._app_host, self._app_host)
def test_add_scoped(self):
print(f'{__name__}.test_add_scoped:')
self._check_general_requirements()
self._add_logger()
self._services.add_scoped(PublisherBase, Publisher)
def test_get_scoped(self):
print(f'{__name__}.test_get_scoped:')
self._check_general_requirements()
self._add_logger()
self._services.add_scoped(PublisherBase, Publisher)
publisher: Publisher = self._services.get_service(PublisherBase)
self.assertIsNotNone(publisher)
self.assertTrue(isinstance(publisher, Publisher))
self.assertTrue(isinstance(publisher, PublisherBase))
self.assertTrue(isinstance(publisher, ServiceBase))
self.assertTrue(isinstance(publisher._logger, Logger))
self.assertTrue(isinstance(publisher._logger, LoggerBase))
self.assertTrue(isinstance(publisher._logger, ServiceBase))
def test_add_transient(self):
print(f'{__name__}.test_add_transient:')
self._check_general_requirements()
self._add_logger()
self._services.add_transient(PublisherBase, Publisher)
self.assertGreater(len(self._services._transient_services), 0)
self.assertTrue(bool(isinstance(service, ServiceBase) for service in self._services._transient_services))
def test_get_transient(self):
print(f'{__name__}.test_get_transient:')
self._check_general_requirements()
self._add_logger()
self._services.add_transient(PublisherBase, Publisher)
publisher: Publisher = self._services.get_service(PublisherBase)
self.assertIsNotNone(publisher)
self.assertTrue(isinstance(publisher, Publisher))
self.assertTrue(isinstance(publisher, PublisherBase))
self.assertTrue(isinstance(publisher, ServiceBase))
self.assertTrue(isinstance(publisher._logger, Logger))
self.assertTrue(isinstance(publisher._logger, LoggerBase))
self.assertTrue(isinstance(publisher._logger, ServiceBase))

View File

@ -1,79 +1,45 @@
import os import unittest
import sys
import traceback
from termcolor import colored from tests.logging.logger import LoggerTest
from tests.publishing.publisher import PublisherTest
from sh_edraft.configuration import ApplicationHost from tests.service_providing.service_provider import ServiceProviderTest
from tests.logger import LoggerTest
from tests.publisher import PublisherTest
from tests.service_provider import ServiceProviderTest
class Tester: class Tester:
def __init__(self): def __init__(self):
self._app_host = ApplicationHost() self._suite = unittest.TestSuite()
self._error: bool = False def create(self):
# providing
self._suite.addTest(ServiceProviderTest('test_create'))
self._suite.addTest(ServiceProviderTest('test_add_singleton'))
self._suite.addTest(ServiceProviderTest('test_get_singleton'))
self._suite.addTest(ServiceProviderTest('test_add_scoped'))
self._suite.addTest(ServiceProviderTest('test_get_scoped'))
self._suite.addTest(ServiceProviderTest('test_add_transient'))
self._suite.addTest(ServiceProviderTest('test_get_transient'))
@staticmethod # logging
def disable_print(): self._suite.addTest(LoggerTest('test_create'))
sys.stdout = open(os.devnull, 'w') self._suite.addTest(LoggerTest('test_header'))
self._suite.addTest(LoggerTest('test_trace'))
self._suite.addTest(LoggerTest('test_debug'))
self._suite.addTest(LoggerTest('test_info'))
self._suite.addTest(LoggerTest('test_warn'))
self._suite.addTest(LoggerTest('test_error'))
self._suite.addTest(LoggerTest('test_fatal'))
@staticmethod # publishing
def enable_print(): self._suite.addTest(PublisherTest('test_create'))
sys.stdout = sys.__stdout__
def success(self, message: str):
self.enable_print()
print(colored(message, 'green'))
self.disable_print()
def failed(self, message: str):
self.enable_print()
print(colored(message, 'red'))
self.disable_print()
def exception(self):
self.enable_print()
print(colored(traceback.format_exc(), 'red'))
self.disable_print()
def create(self): pass
def start(self): def start(self):
if not self._error: runner = unittest.TextTestRunner()
try: runner.run(self._suite)
LoggerTest.start(self._app_host) # unittest.main()
self.success(f'{LoggerTest.__name__} test succeeded.')
except Exception as e:
self._error = True
self.failed(f'{LoggerTest.__name__} test failed!\n{e}')
self.exception()
if not self._error:
try:
ServiceProviderTest.start(self._app_host.services)
self.success(f'{ServiceProviderTest.__name__} test succeeded.')
except Exception as e:
self._error = True
self.failed(f'{ServiceProviderTest.__name__} test failed!\n{e}')
self.exception()
if not self._error:
try:
PublisherTest.start(self._app_host.services)
self.success(f'{PublisherTest.__name__} test succeeded.')
except Exception as e:
self._error = True
self.failed(f'{PublisherTest.__name__} test failed!\n{e}')
self.exception()
if __name__ == '__main__': if __name__ == '__main__':
tester = Tester() tester = Tester()
tester.create() tester.create()
tester.disable_print()
tester.start() tester.start()
tester.enable_print()