sh-edraft.de/sh_cpl
1
1
Fork 0
Code Issues 4 Pull Requests 1 Actions Packages Projects 1 Releases 12 Wiki Activity

Improved testing and service providing

Browse Source
This commit is contained in:
Sven Heidemann 2020-11-25 21:41:45 +01:00
parent cd9d4d3c3d
commit b501fef27d
20 changed files with 340 additions and 169 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/sh_edraft/configuration/application_host.py
View File

@ -1,11 +1,13 @@
from datetime import datetime
from sh_edraft.service import ServiceProvider
from sh_edraft.configuration.model.application_host_base import ApplicationHostBase
from sh_edraft.service.service_provider import ServiceProvider
class ApplicationHost:
class ApplicationHost(ApplicationHostBase):
def __init__(self):
ApplicationHostBase.__init__(self)
self._services = ServiceProvider()
self._end_time: datetime = datetime.now()
self._start_time: datetime = datetime.now()

27
src/sh_edraft/configuration/configuration.py Normal file
View File

@ -0,0 +1,27 @@
from collections import Callable
from sh_edraft.configuration.model.configuration_model_base import ConfigurationModelBase
from sh_edraft.configuration.model.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]

8
src/sh_edraft/configuration/model/application_host_base.py Normal file
View File

@ -0,0 +1,8 @@
from abc import ABC, abstractmethod
from datetime import datetime
class ApplicationHostBase(ABC):
@abstractmethod
def __init__(self): pass

23
src/sh_edraft/configuration/model/configuration_base.py Normal file
View File

@ -0,0 +1,23 @@
from abc import abstractmethod
from collections import Callable
from sh_edraft.configuration.model.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): 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

12
src/sh_edraft/logging/base/logger_base.py
View File

@ -1,8 +1,7 @@
from abc import abstractmethod
from typing import Optional
from sh_edraft.configuration.application_host import ApplicationHost
from sh_edraft.logging.model.log_settings import LoggingSettings
from sh_edraft.configuration 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
@ -10,12 +9,11 @@ from sh_edraft.time.model.time_format_settings import TimeFormatSettings
class LoggerBase(ServiceBase):
@abstractmethod
def __init__(self):
def __init__(self, logging_settings: LoggingSettings, time_format: TimeFormatSettings):
ServiceBase.__init__(self)
self._log_settings: Optional[LoggingSettings] = None
self._time_format_settings: Optional[TimeFormatSettings] = None
self._app_host: Optional[ApplicationHost] = None
self._log_settings: LoggingSettings = logging_settings
self._time_format_settings: TimeFormatSettings = time_format
@abstractmethod
def header(self, string: str): pass

21
src/sh_edraft/logging/logger.py
View File

@ -1,28 +1,23 @@
import datetime
import os
import traceback
from collections import Callable
from string import Template
from typing import Optional
from sh_edraft.configuration import ApplicationHost
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.logging_level import LoggingLevel
from sh_edraft.time.model import TimeFormatSettings
from sh_edraft.utils.console import Console
class Logger(LoggerBase):
def __init__(self):
LoggerBase.__init__(self)
def __init__(self, logging_settings: LoggingSettings, time_format: TimeFormatSettings, app_host: ApplicationHost):
LoggerBase.__init__(self, logging_settings, time_format)
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._app_host: ApplicationHost = app_host
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),

4
src/sh_edraft/logging/model/__init__.py
View File

@ -20,8 +20,8 @@ __version__ = '2020.12.5'
from collections import namedtuple
# imports:
from .log_level import LoggingLevel
from .log_settings import LoggingSettings
from .logging_level import LoggingLevel
from .logging_settings import LoggingSettings
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major=2020, minor=12, micro=5)

0
src/sh_edraft/logging/model/log_level.py → src/sh_edraft/logging/model/logging_level.py
View File

4
src/sh_edraft/logging/model/log_settings.py → src/sh_edraft/logging/model/logging_settings.py
View File

@ -2,9 +2,9 @@ import traceback
from typing import Optional
from sh_edraft.configuration.model.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.logging.model.log_level import LoggingLevel
from sh_edraft.logging.model.logging_level import LoggingLevel
class LoggingSettings(ConfigurationModelBase):

0
src/sh_edraft/logging/model/log_settings_name.py → src/sh_edraft/logging/model/logging_settings_name.py
View File

20
src/sh_edraft/publishing/base/publisher_base.py
View File

@ -1,24 +1,18 @@
from abc import abstractmethod
from typing import Optional
from sh_edraft.logging.base.logger_base import LoggerBase
from sh_edraft.publishing.model.template import Template
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):
def __init__(self, logger: LoggerBase, publish_settings: PublishSettingsModel):
ServiceBase.__init__(self)
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._logger: LoggerBase = logger
self._publish_settings: PublishSettingsModel = publish_settings
@property
@abstractmethod
@ -28,5 +22,11 @@ class PublisherBase(ServiceBase):
@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

83
src/sh_edraft/publishing/model/publish_settings_model.py Normal file
View File

@ -0,0 +1,83 @@
import traceback
from typing import Optional
from sh_edraft.configuration.model 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')

12
src/sh_edraft/publishing/model/publish_settings_name.py Normal file
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'

66
src/sh_edraft/publishing/publisher.py
View File

@ -4,26 +4,22 @@ from string import Template as stringTemplate
from sh_edraft.logging.base.logger_base import LoggerBase
from sh_edraft.publishing.base.publisher_base import PublisherBase
from sh_edraft.publishing.model.publish_settings_model import PublishSettingsModel
from sh_edraft.publishing.model.template import Template
class Publisher(PublisherBase):
def __init__(self):
super().__init__()
self._included_files: list[str] = []
self._excluded_files: list[str] = []
self._template_ending = '_template.txt'
def __init__(self, logger: LoggerBase, publish_settings: PublishSettingsModel):
super().__init__(logger, publish_settings)
@property
def source_path(self) -> str:
return self._source_path
return self._publish_settings.source_path
@property
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:
self._logger.trace(__name__, f'Started {__name__}._get_template_output')
@ -54,16 +50,16 @@ class Publisher(PublisherBase):
def _read_source_path(self):
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:
if file.endswith('.py') or file in self._included_files:
self._included_files.append(os.path.join(r, file))
if file.endswith('.py') or file in self._publish_settings.included_files:
self._publish_settings.included_files.append(os.path.join(r, file))
self._logger.trace(__name__, f'Stopped {__name__}._read_source_path')
def _read_templates(self):
self._logger.trace(__name__, f'Started {__name__}._read_templates')
for t in self._settings:
for t in self._publish_settings.templates:
output_template: str = ''
if not os.path.isfile(t.template_path):
self._logger.fatal(__name__, f'Template not found: {t.template_path}')
@ -78,17 +74,17 @@ class Publisher(PublisherBase):
def _create_dist_path(self):
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:
shutil.rmtree(self._dist_path)
self._logger.info(__name__, f'Deleted {self._dist_path}')
shutil.rmtree(self._publish_settings.dist_path)
self._logger.info(__name__, f'Deleted {self._publish_settings.dist_path}')
except Exception as 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:
os.makedirs(self._dist_path)
self._logger.debug(__name__, f'Created directories: {self._dist_path}')
os.makedirs(self._publish_settings.dist_path)
self._logger.debug(__name__, f'Created directories: {self._publish_settings.dist_path}')
self._logger.info(__name__, f'Created dist directory')
except Exception as e:
self._logger.fatal(__name__, f'Cannot create dist directory', e)
@ -109,9 +105,9 @@ class Publisher(PublisherBase):
def _write_templates(self):
self._logger.trace(__name__, f'Started {__name__}._write_templates')
for template in self._settings:
for file in self._included_files:
if os.path.basename(file) == '__init__.py' and file not in self._excluded_files:
for template in self._publish_settings.templates:
for file in self._publish_settings.included_files:
if os.path.basename(file) == '__init__.py' and file not in self._publish_settings.excluded_files:
template_name = template.name
if template.name == '*' or template.name == '':
template_name = self._get_template_name_from_dirs(file)
@ -164,16 +160,16 @@ class Publisher(PublisherBase):
def _copy_all_included_files(self):
self._logger.trace(__name__, f'Started {__name__}._copy_all_included_files')
dist_path = self._dist_path
if self._dist_path.endswith('/'):
dist_path = self._publish_settings.dist_path
if self._publish_settings.dist_path.endswith('/'):
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
if file in self._excluded_files:
if file in self._publish_settings.excluded_files:
is_file_excluded = True
else:
for excluded in self._excluded_files:
for excluded in self._publish_settings.excluded_files:
if file.__contains__(excluded):
is_file_excluded = True
@ -206,26 +202,18 @@ class Publisher(PublisherBase):
def include(self, path: str):
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')
def exclude(self, path: str):
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')
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):
self._logger.trace(__name__, f'Started {__name__}.create')
if not self._dist_path.endswith('/'):
self._dist_path += '/'
if not self._publish_settings.dist_path.endswith('/'):
self._publish_settings.dist_path += '/'
self._read_source_path()
self._read_templates()

3
src/sh_edraft/service/base/service_base.py
View File

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

18
src/sh_edraft/service/base/service_provider_base.py
View File

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

85
src/sh_edraft/service/service_provider.py
View File

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

21
src/tests/logging/logger.py
View File

@ -16,7 +16,6 @@ class LoggerTest(unittest.TestCase):
def setUp(self):
self._app_host = ApplicationHost()
self._services = self._app_host.services
self._services.init(())
self._services.create()
self._log_settings = LoggingSettings()
@ -35,8 +34,6 @@ class LoggerTest(unittest.TestCase):
"DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
})
self._services.add_singleton(Logger, self._log_settings, self._time_format_settings, self._app_host)
def tearDown(self):
if os.path.isdir(self._log_settings.path):
shutil.rmtree(self._log_settings.path)
@ -48,9 +45,7 @@ class LoggerTest(unittest.TestCase):
def test_create(self):
print(f'{__name__}.test_create:')
logger: Logger = self._services.get_service(LoggerBase)
self.assertIsNotNone(logger)
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
self.assertTrue(os.path.isdir(self._log_settings.path))
@ -62,7 +57,7 @@ class LoggerTest(unittest.TestCase):
def test_header(self):
print(f'{__name__}.test_header:')
logger: Logger = self._services.get_service(LoggerBase)
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.header('HeaderTest:')
@ -84,7 +79,7 @@ class LoggerTest(unittest.TestCase):
def test_trace(self):
print(f'{__name__}.test_trace:')
logger: Logger = self._services.get_service(LoggerBase)
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.trace(__name__, f'{__name__}.test_trace:')
@ -106,7 +101,7 @@ class LoggerTest(unittest.TestCase):
def test_debug(self):
print(f'{__name__}.test_debug:')
logger: Logger = self._services.get_service(LoggerBase)
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.debug(__name__, f'{__name__}.test_debug:')
@ -128,7 +123,7 @@ class LoggerTest(unittest.TestCase):
def test_info(self):
print(f'{__name__}.test_info:')
logger: Logger = self._services.get_service(LoggerBase)
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.info(__name__, f'{__name__}.test_info:')
@ -150,7 +145,7 @@ class LoggerTest(unittest.TestCase):
def test_warn(self):
print(f'{__name__}.test_warn:')
logger: Logger = self._services.get_service(LoggerBase)
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.warn(__name__, f'{__name__}.test_warn:')
@ -172,7 +167,7 @@ class LoggerTest(unittest.TestCase):
def test_error(self):
print(f'{__name__}.test_error:')
logger: Logger = self._services.get_service(LoggerBase)
logger = Logger(self._log_settings, self._time_format_settings, self._app_host)
logger.create()
logger.error(__name__, f'{__name__}.test_error:')
@ -194,7 +189,7 @@ class LoggerTest(unittest.TestCase):
def test_fatal(self):
print(f'{__name__}.test_fatal:')
logger: Logger = self._services.get_service(LoggerBase)
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:')

31
src/tests/publishing/publisher.py
View File

@ -9,6 +9,7 @@ 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.source_code.model import Version
from sh_edraft.time.model import TimeFormatSettings
@ -33,7 +34,7 @@ class PublisherTest(unittest.TestCase):
})
self._version = Version(2020, 12, 5).to_dict()
self._templates = [
templates = [
Template(
'../../publish_templates/*_template.txt',
'*',
@ -62,29 +63,33 @@ class PublisherTest(unittest.TestCase):
)
]
self._source = '../'
self._dist = '../../dist'
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._services = self._app_host.services
self._services.init(())
self._services.create()
self._services.add_singleton(Logger, self._log_settings, self._time_format_settings, self._app_host)
logger: Logger = self._services.get_service(LoggerBase)
logger.create()
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):
self._services.add_transient(Publisher, self._services.get_service(LoggerBase), self._source, self._dist, self._templates)
publisher: Publisher = self._services.get_service(PublisherBase)
publisher: Publisher = Publisher(self._logger, self._publish_settings_model)
self.assertIsNotNone(publisher)
publisher.create()
self.assertTrue(os.path.isdir(self._dist))
self.assertTrue(os.path.isdir(self._dist_path))

65
src/tests/service_providing/service_provider.py
View File

@ -1,11 +1,13 @@
import unittest
from sh_edraft.configuration import ApplicationHost
from sh_edraft.configuration.model.application_host_base import ApplicationHostBase
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
@ -16,7 +18,6 @@ class ServiceProviderTest(unittest.TestCase):
def setUp(self):
self._app_host = ApplicationHost()
self._services = self._app_host.services
self._services.init(())
self._services.create()
self._log_settings = LoggingSettings()
@ -26,6 +27,7 @@ class ServiceProviderTest(unittest.TestCase):
"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({
@ -34,48 +36,69 @@ class ServiceProviderTest(unittest.TestCase):
"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 _check_logger_requirements(self):
self.assertIsNotNone(self._log_settings)
self.assertIsNotNone(self._time_format_settings)
def _add_logger(self):
self._services.add_singleton(Logger, self._log_settings, self._time_format_settings, self._app_host)
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.init(())
provider.create()
self.assertIsNotNone(provider)
def test_add_singleton(self):
print(f'{__name__}.test_add_singleton:')
self._check_general_requirements()
self._check_logger_requirements()
self._services.add_singleton(Logger, self._log_settings, self._time_format_settings, self._app_host)
self._services.add_singleton(LoggerBase, Logger)
self.assertGreater(len(self._services._singleton_services), 0)
found = False
for service in self._services._singleton_services:
if isinstance(service, Logger) and isinstance(service, LoggerBase) and isinstance(service, ServiceBase):
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._check_logger_requirements()
self._services.add_singleton(Logger, self._log_settings, self._time_format_settings, self._app_host)
self._services.add_singleton(LoggerBase, Logger)
logger: Logger = self._services.get_service(LoggerBase)
self.assertIsNotNone(logger)
self.assertTrue(isinstance(logger, Logger))
@ -89,18 +112,16 @@ class ServiceProviderTest(unittest.TestCase):
def test_add_scoped(self):
print(f'{__name__}.test_add_scoped:')
self._check_general_requirements()
self._check_logger_requirements()
self._add_logger()
self._services.add_singleton(Logger, self._log_settings, self._time_format_settings, self._app_host)
self._services.add_scoped(Publisher, self._services.get_service(LoggerBase), '../', '../../dist', [])
self._services.add_scoped(PublisherBase, Publisher)
def test_get_scoped(self):
print(f'{__name__}.test_get_scoped:')
self._check_general_requirements()
self._check_logger_requirements()
self._add_logger()
self._services.add_scoped(Publisher, self._services.get_service(LoggerBase), '../', '../../dist', [])
self._services.add_scoped(PublisherBase, Publisher)
publisher: Publisher = self._services.get_service(PublisherBase)
self.assertIsNotNone(publisher)
self.assertTrue(isinstance(publisher, Publisher))
@ -114,10 +135,9 @@ class ServiceProviderTest(unittest.TestCase):
def test_add_transient(self):
print(f'{__name__}.test_add_transient:')
self._check_general_requirements()
self._check_logger_requirements()
self._add_logger()
self._services.add_transient(Publisher, self._services.get_service(LoggerBase), '../', '../../dist', [])
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))
@ -125,10 +145,9 @@ class ServiceProviderTest(unittest.TestCase):
def test_get_transient(self):
print(f'{__name__}.test_get_transient:')
self._check_general_requirements()
self._check_logger_requirements()
self._add_logger()
self._services.add_singleton(Logger, self._log_settings, self._time_format_settings, self._app_host)
self._services.add_transient(Publisher, self._services.get_service(LoggerBase), '../', '../../dist', [])
self._services.add_transient(PublisherBase, Publisher)
publisher: Publisher = self._services.get_service(PublisherBase)
self.assertIsNotNone(publisher)
self.assertTrue(isinstance(publisher, Publisher))