Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
aa2fbd4f70 | ||
|
|
58c8447531 | ||
|
|
bcca43d774 | ||
|
|
e9ccfe7ad2 | ||
|
|
6c2637ad34 | ||
|
|
7183d05dd6 | ||
|
|
b45ca85cd3 | ||
|
|
4ca45ebc73 | ||
|
|
6902768fed | ||
|
|
3f9f2ceaac |
5
ToDo.md
5
ToDo.md
@@ -1,4 +1,5 @@
|
||||
# ToDo list
|
||||
|
||||
- stub files .pyi
|
||||
- fix all remaning check errors
|
||||
- [ ] stub files .pyi
|
||||
- [ ] Add tests for all, we need 100% test coverate
|
||||
- [ ] Log: add custom format for "stack_correct" if set, this will override the normal stack block
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# MARK: Project info
|
||||
[project]
|
||||
name = "corelibs"
|
||||
version = "0.11.0"
|
||||
version = "0.12.2"
|
||||
description = "Collection of utils for Python scripts"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.13"
|
||||
@@ -53,6 +53,10 @@ notes = ["FIXME", "TODO"]
|
||||
notes-rgx = '(FIXME|TODO)(\((TTD-|#)\[0-9]+\))'
|
||||
[tool.flake8]
|
||||
max-line-length = 120
|
||||
ignore = [
|
||||
"E741", # ignore ambigious variable name
|
||||
"W504" # Line break occurred after a binary operator [wrong triggered by "or" in if]
|
||||
]
|
||||
[tool.pylint.MASTER]
|
||||
# this is for the tests/etc folders
|
||||
init-hook='import sys; sys.path.append("src/")'
|
||||
|
||||
14
src/corelibs/check_handling/regex_constants.py
Normal file
14
src/corelibs/check_handling/regex_constants.py
Normal file
@@ -0,0 +1,14 @@
|
||||
"""
|
||||
List of regex compiled strings that can be used
|
||||
"""
|
||||
|
||||
import re
|
||||
|
||||
|
||||
EMAIL_REGEX_BASIC = r"""
|
||||
^[A-Za-z0-9!#$%&'*+\-\/=?^_`{|}~][A-Za-z0-9!#$%:\(\)&'*+\-\/=?^_`{|}~\.]{0,63}
|
||||
@(?!-)[A-Za-z0-9-]{1,63}(?<!-)(?:\.[A-Za-z0-9-]{1,63}(?<!-))*\.[a-zA-Z]{2,6}$
|
||||
"""
|
||||
EMAIL_REGEX_BASIC_COMPILED = re.compile(EMAIL_REGEX_BASIC)
|
||||
|
||||
# __END__
|
||||
@@ -11,7 +11,7 @@ from typing import Any, Tuple, Sequence, cast
|
||||
from pathlib import Path
|
||||
from corelibs.logging_handling.log import Log
|
||||
from corelibs.iterator_handling.list_helpers import convert_to_list, is_list_in_list
|
||||
from corelibs.string_handling.string_helpers import is_int, is_float, str_to_bool
|
||||
from corelibs.var_handling.var_helpers import is_int, is_float, str_to_bool
|
||||
from corelibs.config_handling.settings_loader_handling.settings_loader_check import SettingsLoaderCheck
|
||||
|
||||
|
||||
@@ -52,15 +52,17 @@ class SettingsLoader:
|
||||
# entries that have to be split
|
||||
self.entry_split_char: dict[str, str] = {}
|
||||
self.entry_convert: dict[str, str] = {}
|
||||
# config parser
|
||||
# config parser, load config file first
|
||||
self.config_parser: configparser.ConfigParser | None = self.__load_config_file()
|
||||
# all settings
|
||||
self.settings: dict[str, dict[str, Any]] | None = None
|
||||
# remove file name and get base path and check
|
||||
if not self.config_file.parent.is_dir():
|
||||
raise ValueError(f"Cannot find the config folder: {self.config_file.parent}")
|
||||
# load the config file before we parse anything
|
||||
# for check settings, abort flag
|
||||
self._check_settings_abort: bool = False
|
||||
|
||||
# MARK: load settings
|
||||
def load_settings(self, config_id: str, config_validate: dict[str, list[str]]) -> dict[str, str]:
|
||||
"""
|
||||
neutral settings loader
|
||||
@@ -173,7 +175,9 @@ class SettingsLoader:
|
||||
# - length: for string length
|
||||
# - range: for int/float range check
|
||||
# mandatory check
|
||||
if check == "mandatory:yes" and not settings[config_id].get(entry):
|
||||
if check == "mandatory:yes" and (
|
||||
not settings[config_id].get(entry) or settings[config_id].get(entry) == ['']
|
||||
):
|
||||
error = True
|
||||
self.__print(f"[!] Missing content entry for: {entry}", 'ERROR')
|
||||
# skip if empty none
|
||||
@@ -184,6 +188,8 @@ class SettingsLoader:
|
||||
settings[config_id][entry] = self.__check_settings(
|
||||
check, entry, settings[config_id][entry]
|
||||
)
|
||||
if self._check_settings_abort is True:
|
||||
error = True
|
||||
elif check.startswith("matching:"):
|
||||
checks = check.replace("matching:", "").split("|")
|
||||
if __result := is_list_in_list(convert_to_list(settings[config_id][entry]), list(checks)):
|
||||
@@ -253,6 +259,7 @@ class SettingsLoader:
|
||||
|
||||
return settings[config_id]
|
||||
|
||||
# MARK: build from/to/requal logic
|
||||
def __build_from_to_equal(
|
||||
self, entry: str, check: str, convert_to_int: bool = False
|
||||
) -> Tuple[float | None, float | None, float | None]:
|
||||
@@ -306,6 +313,7 @@ class SettingsLoader:
|
||||
__equal
|
||||
)
|
||||
|
||||
# MARK: length/range validation
|
||||
def __length_range_validate(
|
||||
self,
|
||||
entry: str,
|
||||
@@ -347,6 +355,7 @@ class SettingsLoader:
|
||||
continue
|
||||
return valid
|
||||
|
||||
# MARK: load config file data from file
|
||||
def __load_config_file(self) -> configparser.ConfigParser | None:
|
||||
"""
|
||||
load and parse the config file
|
||||
@@ -358,13 +367,14 @@ class SettingsLoader:
|
||||
return config
|
||||
return None
|
||||
|
||||
# MARK: regex clean up one
|
||||
def __clean_invalid_setting(
|
||||
self,
|
||||
entry: str,
|
||||
validate: str,
|
||||
value: str,
|
||||
regex: str,
|
||||
regex_clean: str,
|
||||
regex_clean: str | None,
|
||||
replace: str = "",
|
||||
print_error: bool = True,
|
||||
) -> str:
|
||||
@@ -380,18 +390,25 @@ class SettingsLoader:
|
||||
replace (str): replace with character. Defaults to ''
|
||||
print_error (bool): print the error message. Defaults to True
|
||||
"""
|
||||
check = re.compile(regex)
|
||||
clean = re.compile(regex_clean)
|
||||
if not check.search(value):
|
||||
check = re.compile(regex, re.VERBOSE)
|
||||
clean: re.Pattern[str] | None = None
|
||||
if regex_clean is not None:
|
||||
clean = re.compile(regex_clean, re.VERBOSE)
|
||||
# value must be set if clean is None, else empty value is allowed and will fail
|
||||
if (clean is None and value or clean) and not check.search(value):
|
||||
self.__print(
|
||||
f"[!] Invalid content for '{entry}' with check '{validate}' and data: {value}",
|
||||
'ERROR', print_error
|
||||
)
|
||||
# clean up
|
||||
return clean.sub(replace, value)
|
||||
# clean up if clean up is not none, else return EMPTY string
|
||||
if clean is not None:
|
||||
return clean.sub(replace, value)
|
||||
self._check_settings_abort = True
|
||||
return ''
|
||||
# else return as is
|
||||
return value
|
||||
|
||||
# MARK: check settings, regx
|
||||
def __check_settings(
|
||||
self,
|
||||
check: str, entry: str, setting_value: list[str] | str
|
||||
@@ -439,6 +456,7 @@ class SettingsLoader:
|
||||
# return data
|
||||
return setting_value
|
||||
|
||||
# MARK: check arguments, for config file load fail
|
||||
def __check_arguments(self, arguments: dict[str, list[str]], all_set: bool = False) -> bool:
|
||||
"""
|
||||
check if ast least one argument is set
|
||||
@@ -468,6 +486,7 @@ class SettingsLoader:
|
||||
|
||||
return has_argument
|
||||
|
||||
# MARK: get argument from args dict
|
||||
def __get_arg(self, entry: str) -> Any:
|
||||
"""
|
||||
check if an argument entry xists, if None -> returns None else value of argument
|
||||
@@ -482,6 +501,7 @@ class SettingsLoader:
|
||||
return None
|
||||
return self.args.get(entry)
|
||||
|
||||
# MARK: error print
|
||||
def __print(self, msg: str, level: str, print_error: bool = True, raise_exception: bool = False):
|
||||
"""
|
||||
print out error, if Log class is set then print to log instead
|
||||
|
||||
@@ -3,12 +3,15 @@ Class of checks that can be run on value entries
|
||||
"""
|
||||
|
||||
from typing import TypedDict
|
||||
from corelibs.check_handling.regex_constants import EMAIL_REGEX_BASIC
|
||||
|
||||
|
||||
class SettingsLoaderCheckValue(TypedDict):
|
||||
"""Settings check entries"""
|
||||
|
||||
regex: str
|
||||
regex_clean: str
|
||||
# if None, then on error we exit, eles we clean up data
|
||||
regex_clean: str | None
|
||||
replace: str
|
||||
|
||||
|
||||
@@ -16,29 +19,37 @@ class SettingsLoaderCheck:
|
||||
"""
|
||||
check:<NAME> or check:list+<NAME>
|
||||
"""
|
||||
|
||||
CHECK_SETTINGS: dict[str, SettingsLoaderCheckValue] = {
|
||||
"int": {
|
||||
"regex": r"^[0-9]+$",
|
||||
"regex_clean": r"[^0-9]",
|
||||
"replace": ""
|
||||
"replace": "",
|
||||
},
|
||||
"string.alphanumeric": {
|
||||
"regex": r"^[a-zA-Z0-9]+$",
|
||||
"regex_clean": r"[^a-zA-Z0-9]",
|
||||
"replace": ""
|
||||
"replace": "",
|
||||
},
|
||||
"string.alphanumeric.lower.dash": {
|
||||
"regex": r"^[a-z0-9-]+$",
|
||||
"regex_clean": r"[^a-z0-9-]",
|
||||
"replace": ""
|
||||
"replace": "",
|
||||
},
|
||||
# A-Z a-z 0-9 _ - . ONLY
|
||||
# This one does not remove, but replaces with _
|
||||
"string.alphanumeric.extended.replace": {
|
||||
"regex": r"^[_.a-zA-Z0-9-]+$",
|
||||
"regex_clean": r"[^_.a-zA-Z0-9-]",
|
||||
"replace": "_"
|
||||
"replace": "_",
|
||||
},
|
||||
# This does a baisc email check, only alphanumeric with special characters
|
||||
"string.email.basic": {
|
||||
"regex": EMAIL_REGEX_BASIC,
|
||||
"regex_clean": None,
|
||||
"replace": "",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# __END__
|
||||
|
||||
33
src/corelibs/debug_handling/debug_helpers.py
Normal file
33
src/corelibs/debug_handling/debug_helpers.py
Normal file
@@ -0,0 +1,33 @@
|
||||
"""
|
||||
Various debug helpers
|
||||
"""
|
||||
|
||||
import traceback
|
||||
import os
|
||||
|
||||
|
||||
def traceback_call_str(start: int = 2, depth: int = 1):
|
||||
"""
|
||||
get the trace for the last entry
|
||||
|
||||
Keyword Arguments:
|
||||
start {int} -- _description_ (default: {2})
|
||||
depth {int} -- _description_ (default: {1})
|
||||
|
||||
Returns:
|
||||
_type_ -- _description_
|
||||
"""
|
||||
# can't have more than in the stack for depth
|
||||
depth = min(depth, start)
|
||||
depth = start - depth
|
||||
# 0 is full stack length from start
|
||||
if depth == 0:
|
||||
stack = traceback.extract_stack()[-start:]
|
||||
else:
|
||||
stack = traceback.extract_stack()[-start:-depth]
|
||||
return ' -> '.join(
|
||||
f"{os.path.basename(f.filename)}:{f.name}:{f.lineno}"
|
||||
for f in stack
|
||||
)
|
||||
|
||||
# __END__
|
||||
@@ -7,15 +7,18 @@ attach "init_worker_logging" with the set log_queue
|
||||
import re
|
||||
import logging.handlers
|
||||
import logging
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Mapping, TextIO, TypedDict, Any, TYPE_CHECKING, cast
|
||||
from typing import MutableMapping, TextIO, TypedDict, Any, TYPE_CHECKING, cast
|
||||
from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel
|
||||
from corelibs.string_handling.text_colors import Colors
|
||||
from corelibs.debug_handling.debug_helpers import traceback_call_str
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from multiprocessing import Queue
|
||||
|
||||
|
||||
# MARK: Log settings TypedDict
|
||||
class LogSettings(TypedDict):
|
||||
"""
|
||||
log settings
|
||||
@@ -23,8 +26,8 @@ class LogSettings(TypedDict):
|
||||
Arguments:
|
||||
TypedDict {_type_} -- _description_
|
||||
"""
|
||||
log_level_console: str
|
||||
log_level_file: str
|
||||
log_level_console: LoggingLevel
|
||||
log_level_file: LoggingLevel
|
||||
console_enabled: bool
|
||||
console_color_output_enabled: bool
|
||||
add_start_info: bool
|
||||
@@ -32,18 +35,21 @@ class LogSettings(TypedDict):
|
||||
log_queue: 'Queue[str] | None'
|
||||
|
||||
|
||||
# MARK: Custom color filter
|
||||
class CustomConsoleFormatter(logging.Formatter):
|
||||
"""
|
||||
Custom formatter with colors for console output
|
||||
"""
|
||||
|
||||
COLORS = {
|
||||
"DEBUG": Colors.cyan,
|
||||
"INFO": Colors.green,
|
||||
"WARNING": Colors.yellow,
|
||||
"ERROR": Colors.red,
|
||||
"CRITICAL": Colors.red_bold,
|
||||
"EXCEPTION": Colors.magenta_bold, # will never be written to console
|
||||
LoggingLevel.DEBUG.name: Colors.cyan,
|
||||
LoggingLevel.INFO.name: Colors.green,
|
||||
LoggingLevel.WARNING.name: Colors.yellow,
|
||||
LoggingLevel.ERROR.name: Colors.red,
|
||||
LoggingLevel.CRITICAL.name: Colors.red_bold,
|
||||
LoggingLevel.ALERT.name: Colors.yellow_bold,
|
||||
LoggingLevel.EMERGENCY.name: Colors.magenta_bold,
|
||||
LoggingLevel.EXCEPTION.name: Colors.magenta_bright, # will never be written to console
|
||||
}
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
@@ -60,7 +66,7 @@ class CustomConsoleFormatter(logging.Formatter):
|
||||
reset = Colors.reset
|
||||
color = self.COLORS.get(record.levelname, reset)
|
||||
# only highlight level for basic
|
||||
if record.levelname in ['DEBUG', 'INFO']:
|
||||
if record.levelname in [LoggingLevel.DEBUG.name, LoggingLevel.INFO.name]:
|
||||
record.levelname = f"{color}{record.levelname}{reset}"
|
||||
return super().format(record)
|
||||
# highlight whole line
|
||||
@@ -68,6 +74,11 @@ class CustomConsoleFormatter(logging.Formatter):
|
||||
return f"{color}{message}{reset}"
|
||||
|
||||
|
||||
# TODO: add custom handlers for stack_correct, if not set fill with %(filename)s:%(funcName)s:%(lineno)d
|
||||
# hasattr(record, 'stack_correct')
|
||||
|
||||
|
||||
# MARK: Log class
|
||||
class Log:
|
||||
"""
|
||||
logger setup
|
||||
@@ -77,26 +88,32 @@ class Log:
|
||||
SPACER_CHAR: str = '='
|
||||
SPACER_LENGTH: int = 32
|
||||
# default logging level
|
||||
DEFAULT_LOG_LEVEL: str = 'WARNING'
|
||||
DEFAULT_LOG_LEVEL: LoggingLevel = LoggingLevel.WARNING
|
||||
DEFAULT_LOG_LEVEL_FILE: LoggingLevel = LoggingLevel.DEBUG
|
||||
DEFAULT_LOG_LEVEL_CONSOLE: LoggingLevel = LoggingLevel.WARNING
|
||||
# default settings
|
||||
DEFAULT_LOG_SETTINGS: LogSettings = {
|
||||
"log_level_console": "WARNING",
|
||||
"log_level_file": "DEBUG",
|
||||
"console_enabled": True,
|
||||
"console_color_output_enabled": True,
|
||||
"add_start_info": True,
|
||||
"add_end_info": False,
|
||||
"log_queue": None,
|
||||
}
|
||||
"log_level_console": LoggingLevel.WARNING,
|
||||
"log_level_file": LoggingLevel.DEBUG,
|
||||
"console_enabled": True,
|
||||
"console_color_output_enabled": True,
|
||||
"add_start_info": True,
|
||||
"add_end_info": False,
|
||||
"log_queue": None,
|
||||
}
|
||||
|
||||
# MARK: constructor
|
||||
def __init__(
|
||||
self,
|
||||
log_path: Path,
|
||||
log_name: str,
|
||||
log_settings: dict[str, 'str | bool | None | Queue[str]'] | LogSettings | None = None,
|
||||
log_settings: dict[str, 'LoggingLevel | str | bool | None | Queue[str]'] | LogSettings | None = None,
|
||||
other_handlers: dict[str, Any] | None = None
|
||||
):
|
||||
# add new level for EXCEPTION
|
||||
logging.addLevelName(LoggingLevel.EXCEPTION.value, 'EXCEPTION')
|
||||
# add new level for alert, emergecny and exception
|
||||
logging.addLevelName(LoggingLevel.ALERT.value, LoggingLevel.ALERT.name)
|
||||
logging.addLevelName(LoggingLevel.EMERGENCY.value, LoggingLevel.EMERGENCY.name)
|
||||
logging.addLevelName(LoggingLevel.EXCEPTION.value, LoggingLevel.EXCEPTION.name)
|
||||
# parse the logging settings
|
||||
self.log_settings = self.__parse_log_settings(log_settings)
|
||||
# if path, set log name with .log
|
||||
@@ -118,17 +135,25 @@ class Log:
|
||||
|
||||
self.log_queue: 'Queue[str] | None' = None
|
||||
self.listener: logging.handlers.QueueListener | None = None
|
||||
self.logger: logging.Logger
|
||||
|
||||
# setup handlers
|
||||
# NOTE if console with color is set first, some of the color formatting is set
|
||||
# in the file writer too, for the ones where color is set BEFORE the format
|
||||
self.handlers: list[logging.StreamHandler[TextIO] | logging.handlers.TimedRotatingFileHandler] = [
|
||||
# file handler, always
|
||||
self.__file_handler(self.log_settings['log_level_file'], log_path)
|
||||
]
|
||||
# Any is logging.StreamHandler, logging.FileHandler and all logging.handlers.*
|
||||
self.handlers: dict[str, Any] = {}
|
||||
self.add_handler('file_handler', self.__create_time_rotating_file_handler(
|
||||
self.log_settings['log_level_file'], log_path)
|
||||
)
|
||||
if self.log_settings['console_enabled']:
|
||||
# console
|
||||
self.handlers.append(self.__console_handler(self.log_settings['log_level_console']))
|
||||
self.add_handler('stream_handler', self.__create_console_handler(
|
||||
self.log_settings['log_level_console'])
|
||||
)
|
||||
# add other handlers,
|
||||
if other_handlers is not None:
|
||||
for handler_key, handler in other_handlers.items():
|
||||
self.add_handler(handler_key, handler)
|
||||
# init listener if we have a log_queue set
|
||||
self.__init_listener(self.log_settings['log_queue'])
|
||||
|
||||
@@ -138,6 +163,7 @@ class Log:
|
||||
if self.log_settings['add_start_info'] is True:
|
||||
self.break_line('START')
|
||||
|
||||
# MARK: deconstructor
|
||||
def __del__(self):
|
||||
"""
|
||||
Call when class is destroyed, make sure the listender is closed or else we throw a thread error
|
||||
@@ -146,9 +172,10 @@ class Log:
|
||||
self.break_line('END')
|
||||
self.stop_listener()
|
||||
|
||||
# MARK: parse log settings
|
||||
def __parse_log_settings(
|
||||
self,
|
||||
log_settings: dict[str, 'str | bool | None | Queue[str]'] | LogSettings | None
|
||||
log_settings: dict[str, 'LoggingLevel | str | bool | None | Queue[str]'] | LogSettings | None
|
||||
) -> LogSettings:
|
||||
# skip with defaul it not set
|
||||
if log_settings is None:
|
||||
@@ -160,11 +187,11 @@ class Log:
|
||||
if log_settings.get(__log_entry) is None:
|
||||
continue
|
||||
# if not valid reset to default, if not in default set to WARNING
|
||||
if not self.validate_log_level(_log_level := log_settings.get(__log_entry, '')):
|
||||
_log_level = self.DEFAULT_LOG_SETTINGS.get(
|
||||
if not self.validate_log_level(__log_level := log_settings.get(__log_entry, '')):
|
||||
__log_level = self.DEFAULT_LOG_SETTINGS.get(
|
||||
__log_entry, self.DEFAULT_LOG_LEVEL
|
||||
)
|
||||
default_log_settings[__log_entry] = str(_log_level)
|
||||
default_log_settings[__log_entry] = LoggingLevel.from_any(__log_level)
|
||||
# check bool
|
||||
for __log_entry in [
|
||||
"console_enabled",
|
||||
@@ -187,10 +214,36 @@ class Log:
|
||||
def __filter_exceptions(self, record: logging.LogRecord) -> bool:
|
||||
return record.levelname != "EXCEPTION"
|
||||
|
||||
def __console_handler(self, log_level_console: str = 'WARNING') -> logging.StreamHandler[TextIO]:
|
||||
# MARK: add a handler
|
||||
def add_handler(
|
||||
self,
|
||||
handler_name: str,
|
||||
handler: Any
|
||||
) -> bool:
|
||||
"""
|
||||
Add a log handler to the handlers dict
|
||||
|
||||
Arguments:
|
||||
handler_name {str} -- _description_
|
||||
handler {Any} -- _description_
|
||||
"""
|
||||
if self.handlers.get(handler_name):
|
||||
return False
|
||||
if self.listener is not None or hasattr(self, 'logger'):
|
||||
raise ValueError(
|
||||
f"Cannot add handler {handler_name}: {handler.get_name()} because logger is already running"
|
||||
)
|
||||
# TODO: handler must be some handler type, how to check?
|
||||
self.handlers[handler_name] = handler
|
||||
return True
|
||||
|
||||
# MARK: console handler
|
||||
def __create_console_handler(
|
||||
self, log_level_console: LoggingLevel = LoggingLevel.WARNING, filter_exceptions: bool = True
|
||||
) -> logging.StreamHandler[TextIO]:
|
||||
# console logger
|
||||
if not isinstance(getattr(logging, log_level_console.upper(), None), int):
|
||||
log_level_console = 'WARNING'
|
||||
if not self.validate_log_level(log_level_console):
|
||||
log_level_console = self.DEFAULT_LOG_LEVEL_CONSOLE
|
||||
console_handler = logging.StreamHandler()
|
||||
# format layouts
|
||||
format_string = (
|
||||
@@ -206,21 +259,31 @@ class Log:
|
||||
formatter_console = CustomConsoleFormatter(format_string, datefmt=format_date)
|
||||
else:
|
||||
formatter_console = logging.Formatter(format_string, datefmt=format_date)
|
||||
console_handler.setLevel(log_level_console)
|
||||
console_handler.setLevel(log_level_console.name)
|
||||
console_handler.set_name('console')
|
||||
# do not show exceptions logs on console
|
||||
console_handler.addFilter(self.__filter_exceptions)
|
||||
if filter_exceptions:
|
||||
console_handler.addFilter(self.__filter_exceptions)
|
||||
console_handler.setFormatter(formatter_console)
|
||||
return console_handler
|
||||
|
||||
def __file_handler(self, log_level_file: str, log_path: Path) -> logging.handlers.TimedRotatingFileHandler:
|
||||
# MARK: file handler
|
||||
def __create_time_rotating_file_handler(
|
||||
self, log_level_file: LoggingLevel, log_path: Path,
|
||||
when: str = "D", interval: int = 1, backup_count: int = 0
|
||||
) -> logging.handlers.TimedRotatingFileHandler:
|
||||
# file logger
|
||||
if not isinstance(getattr(logging, log_level_file.upper(), None), int):
|
||||
log_level_file = 'DEBUG'
|
||||
# when: S/M/H/D/W0-W6/midnight
|
||||
# interval: how many, 1D = every day
|
||||
# backup_count: how many old to keep, 0 = all
|
||||
if not self.validate_log_level(log_level_file):
|
||||
log_level_file = self.DEFAULT_LOG_LEVEL_FILE
|
||||
file_handler = logging.handlers.TimedRotatingFileHandler(
|
||||
filename=log_path,
|
||||
encoding="utf-8",
|
||||
when="D",
|
||||
interval=1
|
||||
when=when,
|
||||
interval=interval,
|
||||
backupCount=backup_count
|
||||
)
|
||||
formatter_file_handler = logging.Formatter(
|
||||
(
|
||||
@@ -239,10 +302,12 @@ class Log:
|
||||
),
|
||||
datefmt="%Y-%m-%dT%H:%M:%S",
|
||||
)
|
||||
file_handler.setLevel(log_level_file)
|
||||
file_handler.set_name('file_timed_rotate')
|
||||
file_handler.setLevel(log_level_file.name)
|
||||
file_handler.setFormatter(formatter_file_handler)
|
||||
return file_handler
|
||||
|
||||
# MARK: init listener
|
||||
def __init_listener(self, log_queue: 'Queue[str] | None' = None):
|
||||
"""
|
||||
If we have a Queue option start the logging queue
|
||||
@@ -255,11 +320,12 @@ class Log:
|
||||
self.log_queue = log_queue
|
||||
self.listener = logging.handlers.QueueListener(
|
||||
self.log_queue,
|
||||
*self.handlers,
|
||||
*self.handlers.values(),
|
||||
respect_handler_level=True
|
||||
)
|
||||
self.listener.start()
|
||||
|
||||
# MARK: init main log
|
||||
def __init_log(self, log_name: str) -> None:
|
||||
"""
|
||||
Initialize the main loggger
|
||||
@@ -271,14 +337,18 @@ class Log:
|
||||
self.logger = logging.getLogger(log_name)
|
||||
# add all the handlers
|
||||
if queue_handler is None:
|
||||
for handler in self.handlers:
|
||||
for handler in self.handlers.values():
|
||||
self.logger.addHandler(handler)
|
||||
else:
|
||||
self.logger.addHandler(queue_handler)
|
||||
# set maximum logging level for all logging output
|
||||
# log level filtering is done per handler
|
||||
self.logger.setLevel(logging.DEBUG)
|
||||
# short name
|
||||
self.lg = self.logger
|
||||
self.l = self.logger
|
||||
|
||||
# MARK: init logger for Fork/Thread
|
||||
@staticmethod
|
||||
def init_worker_logging(log_queue: 'Queue[str]') -> logging.Logger:
|
||||
"""
|
||||
@@ -287,6 +357,7 @@ class Log:
|
||||
queue_handler = logging.handlers.QueueHandler(log_queue)
|
||||
# getLogger call MUST be WITHOUT and logger name
|
||||
root_logger = logging.getLogger()
|
||||
# base logging level, filtering is done in the handlers
|
||||
root_logger.setLevel(logging.DEBUG)
|
||||
root_logger.handlers.clear()
|
||||
root_logger.addHandler(queue_handler)
|
||||
@@ -296,13 +367,110 @@ class Log:
|
||||
|
||||
return root_logger
|
||||
|
||||
def stop_listener(self):
|
||||
"""
|
||||
stop the listener
|
||||
"""
|
||||
if self.listener is not None:
|
||||
self.listener.stop()
|
||||
# FIXME: all below will only work if we add a custom format interface for the stack_correct part
|
||||
# Important note, although they exist, it is recommended to use self.logger.NAME directly
|
||||
# so that the correct filename, method and row number is set
|
||||
# for > 50 use logger.log(LoggingLevel.<LEVEL>.value, ...)
|
||||
# for exception logger.log(LoggingLevel.EXCEPTION.value, ..., execInfo=True)
|
||||
# MARK: log message
|
||||
def log(self, level: int, msg: object, *args: object, extra: MutableMapping[str, object] | None = None):
|
||||
"""log general"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
if extra is None:
|
||||
extra = {}
|
||||
extra['stack_correct'] = traceback_call_str(start=3)
|
||||
self.logger.log(level, msg, *args, extra=extra)
|
||||
|
||||
# MARK: DEBUG 10
|
||||
def debug(self, msg: object, *args: object, extra: MutableMapping[str, object] | None = None) -> None:
|
||||
"""debug"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
if extra is None:
|
||||
extra = {}
|
||||
extra['stack_correct'] = traceback_call_str(start=3)
|
||||
self.logger.debug(msg, *args, extra=extra)
|
||||
|
||||
# MARK: INFO 20
|
||||
def info(self, msg: object, *args: object, extra: MutableMapping[str, object] | None = None) -> None:
|
||||
"""info"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
if extra is None:
|
||||
extra = {}
|
||||
extra['stack_correct'] = traceback_call_str(start=3)
|
||||
self.logger.info(msg, *args, extra=extra)
|
||||
|
||||
# MARK: WARNING 30
|
||||
def warning(self, msg: object, *args: object, extra: MutableMapping[str, object] | None = None) -> None:
|
||||
"""warning"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
if extra is None:
|
||||
extra = {}
|
||||
extra['stack_correct'] = traceback_call_str(start=3)
|
||||
self.logger.warning(msg, *args, extra=extra)
|
||||
|
||||
# MARK: ERROR 40
|
||||
def error(self, msg: object, *args: object, extra: MutableMapping[str, object] | None = None) -> None:
|
||||
"""error"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
if extra is None:
|
||||
extra = {}
|
||||
extra['stack_correct'] = traceback_call_str(start=3)
|
||||
self.logger.error(msg, *args, extra=extra)
|
||||
|
||||
# MARK: CRITICAL 50
|
||||
def critical(self, msg: object, *args: object, extra: MutableMapping[str, object] | None = None) -> None:
|
||||
"""critcal"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
if extra is None:
|
||||
extra = {}
|
||||
extra['stack_correct'] = traceback_call_str(start=3)
|
||||
self.logger.critical(msg, *args, extra=extra)
|
||||
|
||||
# MARK: ALERT 55
|
||||
def alert(self, msg: object, *args: object, extra: MutableMapping[str, object] | None = None) -> None:
|
||||
"""alert"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
# extra_dict = dict(extra)
|
||||
if extra is None:
|
||||
extra = {}
|
||||
extra['stack_correct'] = traceback_call_str(start=3)
|
||||
self.logger.log(LoggingLevel.ALERT.value, msg, *args, extra=extra)
|
||||
|
||||
# MARK: EMERGECNY: 60
|
||||
def emergency(self, msg: object, *args: object, extra: MutableMapping[str, object] | None = None) -> None:
|
||||
"""emergency"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
if extra is None:
|
||||
extra = {}
|
||||
extra['stack_correct'] = traceback_call_str(start=3)
|
||||
self.logger.log(LoggingLevel.EMERGENCY.value, msg, *args, extra=extra)
|
||||
|
||||
# MARK: EXCEPTION: 70
|
||||
def exception(self, msg: object, *args: object, extra: MutableMapping[str, object] | None = None) -> None:
|
||||
"""
|
||||
log on exceotion level, this is log.exception, but logs with a new level
|
||||
|
||||
Args:
|
||||
msg (object): _description_
|
||||
*args (object): arguments for msg
|
||||
extra: Mapping[str, object] | None: extra arguments for the formatting if needed
|
||||
"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
if extra is None:
|
||||
extra = {}
|
||||
extra['stack_correct'] = traceback_call_str(start=3)
|
||||
self.logger.log(LoggingLevel.EXCEPTION.value, msg, *args, exc_info=True, extra=extra)
|
||||
|
||||
# MARK: break line
|
||||
def break_line(self, info: str = "BREAK"):
|
||||
"""
|
||||
add a break line as info level
|
||||
@@ -310,23 +478,100 @@ class Log:
|
||||
Keyword Arguments:
|
||||
info {str} -- _description_ (default: {"BREAK"})
|
||||
"""
|
||||
if not hasattr(self, 'logger'):
|
||||
raise ValueError('Logger is not yet initialized')
|
||||
self.logger.info("[%s] %s>", info, self.SPACER_CHAR * self.SPACER_LENGTH)
|
||||
|
||||
def exception(self, msg: object, *args: object, extra: Mapping[str, object] | None = None) -> None:
|
||||
# MARK: queue handling
|
||||
def flush(self, handler_name: str | None = None, timeout: float = 2.0) -> bool:
|
||||
"""
|
||||
log on exceotion level
|
||||
Flush all pending messages
|
||||
|
||||
Args:
|
||||
msg (object): _description_
|
||||
*args (object): arguments for msg
|
||||
extra: Mapping[str, object] | None: extra arguments for the formatting if needed
|
||||
Keyword Arguments:
|
||||
handler_name {str | None} -- _description_ (default: {None})
|
||||
timeout {float} -- _description_ (default: {2.0})
|
||||
|
||||
Returns:
|
||||
bool -- _description_
|
||||
"""
|
||||
self.logger.log(LoggingLevel.EXCEPTION.value, msg, *args, exc_info=True, extra=extra)
|
||||
if not self.listener or not self.log_queue:
|
||||
return False
|
||||
|
||||
try:
|
||||
# Wait for queue to be processed
|
||||
start_time = time.time()
|
||||
while not self.log_queue.empty() and (time.time() - start_time) < timeout:
|
||||
time.sleep(0.01)
|
||||
|
||||
# Flush all handlers or handler given
|
||||
if handler_name:
|
||||
try:
|
||||
self.handlers[handler_name].flush()
|
||||
except IndexError:
|
||||
pass
|
||||
else:
|
||||
for handler in self.handlers.values():
|
||||
handler.flush()
|
||||
except OSError:
|
||||
return False
|
||||
return True
|
||||
|
||||
def stop_listener(self):
|
||||
"""
|
||||
stop the listener
|
||||
"""
|
||||
if self.listener is not None:
|
||||
self.flush()
|
||||
self.listener.stop()
|
||||
|
||||
# MARK: log level handling
|
||||
def set_log_level(self, handler_name: str, log_level: LoggingLevel) -> bool:
|
||||
"""
|
||||
set the logging level for a handler
|
||||
|
||||
Arguments:
|
||||
handler {str} -- _description_
|
||||
log_level {LoggingLevel} -- _description_
|
||||
|
||||
Returns:
|
||||
bool -- _description_
|
||||
"""
|
||||
try:
|
||||
# flush queue befoe changing logging level
|
||||
self.flush(handler_name)
|
||||
self.handlers[handler_name].setLevel(log_level.name)
|
||||
return True
|
||||
except IndexError:
|
||||
if self.logger:
|
||||
self.logger.error('Handler %s not found, cannot change log level', handler_name)
|
||||
return False
|
||||
except AttributeError:
|
||||
if self.logger:
|
||||
self.logger.error(
|
||||
'Cannot change to log level %s for handler %s, log level invalid',
|
||||
LoggingLevel.name, handler_name
|
||||
)
|
||||
return False
|
||||
|
||||
def get_log_level(self, handler_name: str) -> LoggingLevel:
|
||||
"""
|
||||
gettthe logging level for a handler
|
||||
|
||||
Arguments:
|
||||
handler_name {str} -- _description_
|
||||
|
||||
Returns:
|
||||
LoggingLevel -- _description_
|
||||
"""
|
||||
try:
|
||||
return self.handlers[handler_name]
|
||||
except IndexError:
|
||||
return LoggingLevel.NOTSET
|
||||
|
||||
@staticmethod
|
||||
def validate_log_level(log_level: Any) -> bool:
|
||||
"""
|
||||
if the log level is invalid, will erturn false
|
||||
if the log level is invalid will return false, else return true
|
||||
|
||||
Args:
|
||||
log_level (Any): _description_
|
||||
@@ -344,7 +589,7 @@ class Log:
|
||||
def get_log_level_int(log_level: Any) -> int:
|
||||
"""
|
||||
Return log level as INT
|
||||
If invalid return set level in default log level
|
||||
If invalid returns the default log level
|
||||
|
||||
Arguments:
|
||||
log_level {Any} -- _description_
|
||||
@@ -355,6 +600,6 @@ class Log:
|
||||
try:
|
||||
return LoggingLevel.from_any(log_level).value
|
||||
except ValueError:
|
||||
return LoggingLevel.from_string(Log.DEFAULT_LOG_LEVEL).value
|
||||
return LoggingLevel.from_string(Log.DEFAULT_LOG_LEVEL.name).value
|
||||
|
||||
# __END__
|
||||
|
||||
@@ -17,7 +17,9 @@ class LoggingLevel(Enum):
|
||||
WARNING = logging.WARNING # 30
|
||||
ERROR = logging.ERROR # 40
|
||||
CRITICAL = logging.CRITICAL # 50
|
||||
EXCEPTION = 60 # 60 (manualy set)
|
||||
ALERT = 55 # 55 (for Sys log)
|
||||
EMERGENCY = 60 # 60 (for Sys log)
|
||||
EXCEPTION = 70 # 70 (manualy set, error but with higher level)
|
||||
# Alternative names
|
||||
WARN = logging.WARN # 30 (alias for WARNING)
|
||||
FATAL = logging.FATAL # 50 (alias for CRITICAL)
|
||||
@@ -30,6 +32,8 @@ class LoggingLevel(Enum):
|
||||
return cls[level_str.upper()]
|
||||
except KeyError as e:
|
||||
raise ValueError(f"Invalid log level: {level_str}") from e
|
||||
except AttributeError as e:
|
||||
raise ValueError(f"Invalid log level: {level_str}") from e
|
||||
|
||||
@classmethod
|
||||
def from_int(cls, level_int: int):
|
||||
@@ -41,7 +45,18 @@ class LoggingLevel(Enum):
|
||||
|
||||
@classmethod
|
||||
def from_any(cls, level_any: Any):
|
||||
"""Convert either str or int"""
|
||||
"""
|
||||
Convert any vale
|
||||
if self LoggingLevel return as is, else try to convert from int or string
|
||||
|
||||
Arguments:
|
||||
level_any {Any} -- _description_
|
||||
|
||||
Returns:
|
||||
_type_ -- _description_
|
||||
"""
|
||||
if isinstance(level_any, LoggingLevel):
|
||||
return level_any
|
||||
if isinstance(level_any, int):
|
||||
return cls.from_int(level_any)
|
||||
return cls.from_string(level_any)
|
||||
@@ -70,6 +85,6 @@ class LoggingLevel(Enum):
|
||||
|
||||
def is_lower_than(self, level: 'LoggingLevel'):
|
||||
"""if given value is lower than set"""
|
||||
return self.value > level.value
|
||||
return self.value < level.value
|
||||
|
||||
# __END__
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
String helpers
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
from decimal import Decimal, getcontext
|
||||
from textwrap import shorten
|
||||
|
||||
@@ -102,62 +101,4 @@ def format_number(number: float, precision: int = 0) -> str:
|
||||
"f}"
|
||||
).format(_number)
|
||||
|
||||
|
||||
def is_int(string: Any) -> bool:
|
||||
"""
|
||||
check if a value is int
|
||||
|
||||
Arguments:
|
||||
string {Any} -- _description_
|
||||
|
||||
Returns:
|
||||
bool -- _description_
|
||||
"""
|
||||
try:
|
||||
int(string)
|
||||
return True
|
||||
except TypeError:
|
||||
return False
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def is_float(string: Any) -> bool:
|
||||
"""
|
||||
check if a value is float
|
||||
|
||||
Arguments:
|
||||
string {Any} -- _description_
|
||||
|
||||
Returns:
|
||||
bool -- _description_
|
||||
"""
|
||||
try:
|
||||
float(string)
|
||||
return True
|
||||
except TypeError:
|
||||
return False
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def str_to_bool(string: str):
|
||||
"""
|
||||
convert string to bool
|
||||
|
||||
Arguments:
|
||||
s {str} -- _description_
|
||||
|
||||
Raises:
|
||||
ValueError: _description_
|
||||
|
||||
Returns:
|
||||
_type_ -- _description_
|
||||
"""
|
||||
if string == "True" or string == "true":
|
||||
return True
|
||||
if string == "False" or string == "false":
|
||||
return False
|
||||
raise ValueError(f"Invalid boolean string: {string}")
|
||||
|
||||
# __END__
|
||||
|
||||
0
src/corelibs/var_handling/__init__.py
Normal file
0
src/corelibs/var_handling/__init__.py
Normal file
65
src/corelibs/var_handling/var_helpers.py
Normal file
65
src/corelibs/var_handling/var_helpers.py
Normal file
@@ -0,0 +1,65 @@
|
||||
"""
|
||||
variable convert, check, etc helepr
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
|
||||
def is_int(string: Any) -> bool:
|
||||
"""
|
||||
check if a value is int
|
||||
|
||||
Arguments:
|
||||
string {Any} -- _description_
|
||||
|
||||
Returns:
|
||||
bool -- _description_
|
||||
"""
|
||||
try:
|
||||
int(string)
|
||||
return True
|
||||
except TypeError:
|
||||
return False
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def is_float(string: Any) -> bool:
|
||||
"""
|
||||
check if a value is float
|
||||
|
||||
Arguments:
|
||||
string {Any} -- _description_
|
||||
|
||||
Returns:
|
||||
bool -- _description_
|
||||
"""
|
||||
try:
|
||||
float(string)
|
||||
return True
|
||||
except TypeError:
|
||||
return False
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def str_to_bool(string: str):
|
||||
"""
|
||||
convert string to bool
|
||||
|
||||
Arguments:
|
||||
s {str} -- _description_
|
||||
|
||||
Raises:
|
||||
ValueError: _description_
|
||||
|
||||
Returns:
|
||||
_type_ -- _description_
|
||||
"""
|
||||
if string == "True" or string == "true":
|
||||
return True
|
||||
if string == "False" or string == "false":
|
||||
return False
|
||||
raise ValueError(f"Invalid boolean string: {string}")
|
||||
|
||||
# __END__
|
||||
@@ -1,6 +1,7 @@
|
||||
[TestA]
|
||||
foo=bar
|
||||
foobar=1
|
||||
bar=st
|
||||
some_match=foo
|
||||
some_match_list=foo,bar
|
||||
test_list=a,b,c,d f, g h
|
||||
@@ -21,3 +22,6 @@ match_source_list=foo,bar
|
||||
element_a=Static energy
|
||||
element_b=123.5
|
||||
element_c=True
|
||||
email=foo@bar.com,other+bar-fee@domain-com.cp,
|
||||
email_not_mandatory=
|
||||
email_bad=gii@bar.com
|
||||
|
||||
@@ -44,6 +44,7 @@ def main():
|
||||
{
|
||||
"foo": ["mandatory:yes"],
|
||||
"foobar": ["check:int"],
|
||||
"bar": ["mandatory:yes"],
|
||||
"some_match": ["matching:foo|bar"],
|
||||
"some_match_list": ["split:,", "matching:foo|bar"],
|
||||
"test_list": [
|
||||
@@ -68,7 +69,33 @@ def main():
|
||||
"match_source_list": ["split:,", "in:match_target_list"],
|
||||
}
|
||||
)
|
||||
print(f"Load: {config_load} -> {dump_data(config_data)}")
|
||||
print(f"[{config_load}] Load: {config_load} -> {dump_data(config_data)}")
|
||||
except ValueError as e:
|
||||
print(f"Could not load settings: {e}")
|
||||
|
||||
try:
|
||||
config_load = 'TestB'
|
||||
config_data = sl.load_settings(
|
||||
config_load,
|
||||
{
|
||||
"email": [
|
||||
"split:,",
|
||||
"mandatory:yes",
|
||||
"check:string.email.basic"
|
||||
],
|
||||
"email_not_mandatory": [
|
||||
"split:,",
|
||||
# "mandatory:yes",
|
||||
"check:string.email.basic"
|
||||
],
|
||||
"email_bad": [
|
||||
"split:,",
|
||||
"mandatory:yes",
|
||||
"check:string.email.basic"
|
||||
]
|
||||
}
|
||||
)
|
||||
print(f"[{config_load}] Load: {config_load} -> {dump_data(config_data)}")
|
||||
except ValueError as e:
|
||||
print(f"Could not load settings: {e}")
|
||||
|
||||
|
||||
@@ -18,18 +18,30 @@ def main():
|
||||
log_path=script_path.joinpath('log', 'test.log'),
|
||||
log_name="Test Log",
|
||||
log_settings={
|
||||
"log_level_console": 'DEBUG',
|
||||
# "log_level_console": 'DEBUG',
|
||||
"log_level_console": None,
|
||||
"log_level_file": 'DEBUG',
|
||||
# "console_color_output_enabled": False,
|
||||
}
|
||||
)
|
||||
|
||||
log.logger.debug('[NORMAL] Debug test: %s', log.logger.name)
|
||||
log.lg.debug('[NORMAL] Debug test: %s', log.logger.name)
|
||||
log.debug('[NORMAL-] Debug test: %s', log.logger.name)
|
||||
log.logger.info('[NORMAL] Info test: %s', log.logger.name)
|
||||
log.info('[NORMAL-] Info test: %s', log.logger.name)
|
||||
log.logger.warning('[NORMAL] Warning test: %s', log.logger.name)
|
||||
log.warning('[NORMAL-] Warning test: %s', log.logger.name)
|
||||
log.logger.error('[NORMAL] Error test: %s', log.logger.name)
|
||||
log.error('[NORMAL-] Error test: %s', log.logger.name)
|
||||
log.logger.critical('[NORMAL] Critical test: %s', log.logger.name)
|
||||
log.critical('[NORMAL-] Critical test: %s', log.logger.name)
|
||||
log.logger.log(LoggingLevel.ALERT.value, '[NORMAL] alert test: %s', log.logger.name)
|
||||
log.alert('[NORMAL-] alert test: %s', log.logger.name)
|
||||
log.emergency('[NORMAL-] emergency test: %s', log.logger.name)
|
||||
log.logger.log(LoggingLevel.EMERGENCY.value, '[NORMAL] emergency test: %s', log.logger.name)
|
||||
log.exception('[NORMAL] Exception test: %s', log.logger.name)
|
||||
log.logger.log(LoggingLevel.EXCEPTION.value, '[NORMAL] exception test: %s', log.logger.name, exc_info=True)
|
||||
|
||||
bad_level = 'WRONG'
|
||||
if not Log.validate_log_level(bad_level):
|
||||
@@ -66,6 +78,16 @@ def main():
|
||||
log.logger.critical("Divison through zero: %s", e)
|
||||
log.exception("Divison through zero")
|
||||
|
||||
for handler in log.logger.handlers:
|
||||
print(f"Handler (logger) {handler} -> {handler.level} -> {LoggingLevel.from_any(handler.level)}")
|
||||
|
||||
for key, handler in log.handlers.items():
|
||||
print(f"Handler (handlers) [{key}] {handler} -> {handler.level} -> {LoggingLevel.from_any(handler.level)}")
|
||||
log.set_log_level('stream_handler', LoggingLevel.ERROR)
|
||||
log.logger.warning('[NORMAL] Invisible Warning test: %s', log.logger.name)
|
||||
log.logger.error('[NORMAL] Visible Error test: %s', log.logger.name)
|
||||
# log.handlers['stream_handler'].se
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -10,6 +10,7 @@ import concurrent.futures
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from corelibs.logging_handling.log import Log
|
||||
from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel
|
||||
|
||||
|
||||
def work_function(log_name: str, worker_id: int, data: list[int]) -> int:
|
||||
@@ -46,7 +47,8 @@ def main():
|
||||
"log_queue": log_queue,
|
||||
}
|
||||
)
|
||||
log.logger.info('Pool Fork logging test')
|
||||
|
||||
log.logger.debug('Pool Fork logging test')
|
||||
max_forks = 2
|
||||
data_sets = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
|
||||
with concurrent.futures.ProcessPoolExecutor(
|
||||
@@ -62,9 +64,23 @@ def main():
|
||||
log.logger.info('Workders started')
|
||||
|
||||
for future in concurrent.futures.as_completed(futures):
|
||||
log.logger.info('Processing result: %s', future.result())
|
||||
log.logger.warning('Processing result: %s', future.result())
|
||||
print(f"Processing result: {future.result()}")
|
||||
|
||||
log.set_log_level('stream_handler', LoggingLevel.ERROR)
|
||||
log.logger.error('SECOND Start workers')
|
||||
futures = [
|
||||
executor.submit(work_function, log.log_name, worker_id, data)
|
||||
for worker_id, data in enumerate(data_sets, 1)
|
||||
]
|
||||
log.logger.info('[INVISIBLE] Workders started')
|
||||
log.logger.error('[VISIBLE] Second Workders started')
|
||||
|
||||
for future in concurrent.futures.as_completed(futures):
|
||||
log.logger.error('Processing result: %s', future.result())
|
||||
print(f"Processing result: {future.result()}")
|
||||
|
||||
log.set_log_level('stream_handler', LoggingLevel.DEBUG)
|
||||
log.logger.info('[END] Queue logger test')
|
||||
log.stop_listener()
|
||||
|
||||
|
||||
@@ -5,8 +5,10 @@ Log logging_handling.log testing
|
||||
# import atexit
|
||||
from pathlib import Path
|
||||
from multiprocessing import Queue
|
||||
import time
|
||||
# this is for testing only
|
||||
from corelibs.logging_handling.log import Log
|
||||
from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel
|
||||
|
||||
|
||||
def main():
|
||||
@@ -32,7 +34,29 @@ def main():
|
||||
log_q.logger.warning('[QUEUE] Warning test: %s', log_q.logger.name)
|
||||
log_q.logger.error('[QUEUE] Error test: %s', log_q.logger.name)
|
||||
log_q.logger.critical('[QUEUE] Critical test: %s', log_q.logger.name)
|
||||
log_q.exception('[QUEUE] Exception test: %s', log_q.logger.name)
|
||||
log_q.logger.log(LoggingLevel.EXCEPTION.value, '[QUEUE] Exception test: %s', log_q.logger.name, exc_info=True)
|
||||
time.sleep(0.1)
|
||||
|
||||
for handler in log_q.logger.handlers:
|
||||
print(f"[1] Handler (logger) {handler}")
|
||||
if log_q.listener is not None:
|
||||
for handler in log_q.listener.handlers:
|
||||
print(f"[1] Handler (queue) {handler}")
|
||||
for handler in log_q.handlers.items():
|
||||
print(f"[1] Handler (handlers) {handler}")
|
||||
|
||||
log_q.set_log_level('stream_handler', LoggingLevel.ERROR)
|
||||
log_q.logger.warning('[QUEUE-B] [INVISIBLE] Warning test: %s', log_q.logger.name)
|
||||
log_q.logger.error('[QUEUE-B] [VISIBLE] Error test: %s', log_q.logger.name)
|
||||
|
||||
for handler in log_q.logger.handlers:
|
||||
print(f"[2] Handler (logger) {handler}")
|
||||
if log_q.listener is not None:
|
||||
for handler in log_q.listener.handlers:
|
||||
print(f"[2] Handler (queue) {handler}")
|
||||
for handler in log_q.handlers.items():
|
||||
print(f"[2] Handler (handlers) {handler}")
|
||||
|
||||
log_q.stop_listener()
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,341 @@
|
||||
"""
|
||||
logging_handling.logging_level_handling.logging_level
|
||||
"""
|
||||
|
||||
import logging
|
||||
import pytest
|
||||
# from enum import Enum
|
||||
from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel
|
||||
|
||||
|
||||
class TestLoggingLevelEnum:
|
||||
"""Test the LoggingLevel enum values and basic functionality."""
|
||||
|
||||
def test_enum_values(self):
|
||||
"""Test that all enum values match expected logging levels."""
|
||||
assert LoggingLevel.NOTSET.value == 0
|
||||
assert LoggingLevel.DEBUG.value == 10
|
||||
assert LoggingLevel.INFO.value == 20
|
||||
assert LoggingLevel.WARNING.value == 30
|
||||
assert LoggingLevel.ERROR.value == 40
|
||||
assert LoggingLevel.CRITICAL.value == 50
|
||||
assert LoggingLevel.ALERT.value == 55
|
||||
assert LoggingLevel.EMERGENCY.value == 60
|
||||
assert LoggingLevel.EXCEPTION.value == 70
|
||||
assert LoggingLevel.WARN.value == 30
|
||||
assert LoggingLevel.FATAL.value == 50
|
||||
|
||||
def test_enum_corresponds_to_logging_module(self):
|
||||
"""Test that enum values correspond to logging module constants."""
|
||||
assert LoggingLevel.NOTSET.value == logging.NOTSET
|
||||
assert LoggingLevel.DEBUG.value == logging.DEBUG
|
||||
assert LoggingLevel.INFO.value == logging.INFO
|
||||
assert LoggingLevel.WARNING.value == logging.WARNING
|
||||
assert LoggingLevel.ERROR.value == logging.ERROR
|
||||
assert LoggingLevel.CRITICAL.value == logging.CRITICAL
|
||||
assert LoggingLevel.WARN.value == logging.WARN
|
||||
assert LoggingLevel.FATAL.value == logging.FATAL
|
||||
|
||||
def test_enum_aliases(self):
|
||||
"""Test that aliases point to correct values."""
|
||||
assert LoggingLevel.WARN.value == LoggingLevel.WARNING.value
|
||||
assert LoggingLevel.FATAL.value == LoggingLevel.CRITICAL.value
|
||||
|
||||
|
||||
class TestFromString:
|
||||
"""Test the from_string classmethod."""
|
||||
|
||||
def test_from_string_valid_cases(self):
|
||||
"""Test from_string with valid string inputs."""
|
||||
assert LoggingLevel.from_string("DEBUG") == LoggingLevel.DEBUG
|
||||
assert LoggingLevel.from_string("info") == LoggingLevel.INFO
|
||||
assert LoggingLevel.from_string("Warning") == LoggingLevel.WARNING
|
||||
assert LoggingLevel.from_string("ERROR") == LoggingLevel.ERROR
|
||||
assert LoggingLevel.from_string("critical") == LoggingLevel.CRITICAL
|
||||
assert LoggingLevel.from_string("EXCEPTION") == LoggingLevel.EXCEPTION
|
||||
assert LoggingLevel.from_string("warn") == LoggingLevel.WARN
|
||||
assert LoggingLevel.from_string("FATAL") == LoggingLevel.FATAL
|
||||
|
||||
def test_from_string_invalid_cases(self):
|
||||
"""Test from_string with invalid string inputs."""
|
||||
with pytest.raises(ValueError, match="Invalid log level: invalid"):
|
||||
LoggingLevel.from_string("invalid")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid log level: "):
|
||||
LoggingLevel.from_string("")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid log level: 123"):
|
||||
LoggingLevel.from_string("123")
|
||||
|
||||
|
||||
class TestFromInt:
|
||||
"""Test the from_int classmethod."""
|
||||
|
||||
def test_from_int_valid_cases(self):
|
||||
"""Test from_int with valid integer inputs."""
|
||||
assert LoggingLevel.from_int(0) == LoggingLevel.NOTSET
|
||||
assert LoggingLevel.from_int(10) == LoggingLevel.DEBUG
|
||||
assert LoggingLevel.from_int(20) == LoggingLevel.INFO
|
||||
assert LoggingLevel.from_int(30) == LoggingLevel.WARNING
|
||||
assert LoggingLevel.from_int(40) == LoggingLevel.ERROR
|
||||
assert LoggingLevel.from_int(50) == LoggingLevel.CRITICAL
|
||||
assert LoggingLevel.from_int(55) == LoggingLevel.ALERT
|
||||
assert LoggingLevel.from_int(60) == LoggingLevel.EMERGENCY
|
||||
assert LoggingLevel.from_int(70) == LoggingLevel.EXCEPTION
|
||||
|
||||
def test_from_int_invalid_cases(self):
|
||||
"""Test from_int with invalid integer inputs."""
|
||||
with pytest.raises(ValueError, match="Invalid log level: 999"):
|
||||
LoggingLevel.from_int(999)
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid log level: -1"):
|
||||
LoggingLevel.from_int(-1)
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid log level: 15"):
|
||||
LoggingLevel.from_int(15)
|
||||
|
||||
|
||||
class TestFromAny:
|
||||
"""Test the from_any classmethod."""
|
||||
|
||||
def test_from_any_with_logging_level(self):
|
||||
"""Test from_any when input is already a LoggingLevel."""
|
||||
level = LoggingLevel.INFO
|
||||
assert LoggingLevel.from_any(level) == LoggingLevel.INFO
|
||||
assert LoggingLevel.from_any(level) is level
|
||||
|
||||
def test_from_any_with_int(self):
|
||||
"""Test from_any with integer input."""
|
||||
assert LoggingLevel.from_any(10) == LoggingLevel.DEBUG
|
||||
assert LoggingLevel.from_any(20) == LoggingLevel.INFO
|
||||
assert LoggingLevel.from_any(30) == LoggingLevel.WARNING
|
||||
|
||||
def test_from_any_with_string(self):
|
||||
"""Test from_any with string input."""
|
||||
assert LoggingLevel.from_any("DEBUG") == LoggingLevel.DEBUG
|
||||
assert LoggingLevel.from_any("info") == LoggingLevel.INFO
|
||||
assert LoggingLevel.from_any("Warning") == LoggingLevel.WARNING
|
||||
|
||||
def test_from_any_with_invalid_types(self):
|
||||
"""Test from_any with invalid input types."""
|
||||
with pytest.raises(ValueError):
|
||||
LoggingLevel.from_any(None)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
LoggingLevel.from_any([])
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
LoggingLevel.from_any({})
|
||||
|
||||
def test_from_any_with_invalid_values(self):
|
||||
"""Test from_any with invalid values."""
|
||||
with pytest.raises(ValueError):
|
||||
LoggingLevel.from_any("invalid_level")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
LoggingLevel.from_any(999)
|
||||
|
||||
|
||||
class TestToLoggingLevel:
|
||||
"""Test the to_logging_level method."""
|
||||
|
||||
def test_to_logging_level(self):
|
||||
"""Test conversion to logging module level."""
|
||||
assert LoggingLevel.DEBUG.to_logging_level() == 10
|
||||
assert LoggingLevel.INFO.to_logging_level() == 20
|
||||
assert LoggingLevel.WARNING.to_logging_level() == 30
|
||||
assert LoggingLevel.ERROR.to_logging_level() == 40
|
||||
assert LoggingLevel.CRITICAL.to_logging_level() == 50
|
||||
assert LoggingLevel.ALERT.to_logging_level() == 55
|
||||
assert LoggingLevel.EMERGENCY.to_logging_level() == 60
|
||||
assert LoggingLevel.EXCEPTION.to_logging_level() == 70
|
||||
|
||||
|
||||
class TestToLowerCase:
|
||||
"""Test the to_lower_case method."""
|
||||
|
||||
def test_to_lower_case(self):
|
||||
"""Test conversion to lowercase."""
|
||||
assert LoggingLevel.DEBUG.to_lower_case() == "debug"
|
||||
assert LoggingLevel.INFO.to_lower_case() == "info"
|
||||
assert LoggingLevel.WARNING.to_lower_case() == "warning"
|
||||
assert LoggingLevel.ERROR.to_lower_case() == "error"
|
||||
assert LoggingLevel.CRITICAL.to_lower_case() == "critical"
|
||||
assert LoggingLevel.ALERT.to_lower_case() == "alert"
|
||||
assert LoggingLevel.EMERGENCY.to_lower_case() == "emergency"
|
||||
assert LoggingLevel.EXCEPTION.to_lower_case() == "exception"
|
||||
|
||||
|
||||
class TestStrMethod:
|
||||
"""Test the __str__ method."""
|
||||
|
||||
def test_str_method(self):
|
||||
"""Test string representation."""
|
||||
assert str(LoggingLevel.DEBUG) == "DEBUG"
|
||||
assert str(LoggingLevel.INFO) == "INFO"
|
||||
assert str(LoggingLevel.WARNING) == "WARNING"
|
||||
assert str(LoggingLevel.ERROR) == "ERROR"
|
||||
assert str(LoggingLevel.CRITICAL) == "CRITICAL"
|
||||
assert str(LoggingLevel.ALERT) == "ALERT"
|
||||
assert str(LoggingLevel.EMERGENCY) == "EMERGENCY"
|
||||
assert str(LoggingLevel.EXCEPTION) == "EXCEPTION"
|
||||
|
||||
|
||||
class TestIncludes:
|
||||
"""Test the includes method."""
|
||||
|
||||
def test_includes_valid_cases(self):
|
||||
"""Test includes method with valid cases."""
|
||||
# DEBUG includes all levels
|
||||
assert LoggingLevel.DEBUG.includes(LoggingLevel.DEBUG)
|
||||
assert LoggingLevel.DEBUG.includes(LoggingLevel.INFO)
|
||||
assert LoggingLevel.DEBUG.includes(LoggingLevel.WARNING)
|
||||
assert LoggingLevel.DEBUG.includes(LoggingLevel.ERROR)
|
||||
assert LoggingLevel.DEBUG.includes(LoggingLevel.CRITICAL)
|
||||
assert LoggingLevel.DEBUG.includes(LoggingLevel.ALERT)
|
||||
assert LoggingLevel.DEBUG.includes(LoggingLevel.EMERGENCY)
|
||||
assert LoggingLevel.DEBUG.includes(LoggingLevel.EXCEPTION)
|
||||
|
||||
# INFO includes INFO and higher
|
||||
assert LoggingLevel.INFO.includes(LoggingLevel.INFO)
|
||||
assert LoggingLevel.INFO.includes(LoggingLevel.WARNING)
|
||||
assert LoggingLevel.INFO.includes(LoggingLevel.ERROR)
|
||||
assert LoggingLevel.INFO.includes(LoggingLevel.CRITICAL)
|
||||
assert LoggingLevel.INFO.includes(LoggingLevel.ALERT)
|
||||
assert LoggingLevel.INFO.includes(LoggingLevel.EMERGENCY)
|
||||
assert LoggingLevel.INFO.includes(LoggingLevel.EXCEPTION)
|
||||
|
||||
# INFO does not include DEBUG
|
||||
assert not LoggingLevel.INFO.includes(LoggingLevel.DEBUG)
|
||||
|
||||
# ERROR includes ERROR and higher
|
||||
assert LoggingLevel.ERROR.includes(LoggingLevel.ERROR)
|
||||
assert LoggingLevel.ERROR.includes(LoggingLevel.CRITICAL)
|
||||
assert LoggingLevel.ERROR.includes(LoggingLevel.ALERT)
|
||||
assert LoggingLevel.ERROR.includes(LoggingLevel.EMERGENCY)
|
||||
assert LoggingLevel.ERROR.includes(LoggingLevel.EXCEPTION)
|
||||
|
||||
# ERROR does not include lower levels
|
||||
assert not LoggingLevel.ERROR.includes(LoggingLevel.DEBUG)
|
||||
assert not LoggingLevel.ERROR.includes(LoggingLevel.INFO)
|
||||
assert not LoggingLevel.ERROR.includes(LoggingLevel.WARNING)
|
||||
|
||||
|
||||
class TestIsHigherThan:
|
||||
"""Test the is_higher_than method."""
|
||||
|
||||
def test_is_higher_than(self):
|
||||
"""Test is_higher_than method."""
|
||||
assert LoggingLevel.ERROR.is_higher_than(LoggingLevel.WARNING)
|
||||
assert LoggingLevel.CRITICAL.is_higher_than(LoggingLevel.ERROR)
|
||||
assert LoggingLevel.ALERT.is_higher_than(LoggingLevel.CRITICAL)
|
||||
assert LoggingLevel.EMERGENCY.is_higher_than(LoggingLevel.ALERT)
|
||||
assert LoggingLevel.EXCEPTION.is_higher_than(LoggingLevel.EMERGENCY)
|
||||
assert LoggingLevel.INFO.is_higher_than(LoggingLevel.DEBUG)
|
||||
|
||||
# Same level should return False
|
||||
assert not LoggingLevel.INFO.is_higher_than(LoggingLevel.INFO)
|
||||
|
||||
# Lower level should return False
|
||||
assert not LoggingLevel.DEBUG.is_higher_than(LoggingLevel.INFO)
|
||||
|
||||
|
||||
class TestIsLowerThan:
|
||||
"""Test the is_lower_than method - Note: there seems to be a bug in the original implementation."""
|
||||
|
||||
def test_is_lower_than_expected_behavior(self):
|
||||
"""Test what the is_lower_than method should do (based on method name)."""
|
||||
# Note: The original implementation has a bug - it uses > instead of <
|
||||
# This test shows what the expected behavior should be
|
||||
|
||||
# Based on the method name, these should be true:
|
||||
# assert LoggingLevel.DEBUG.is_lower_than(LoggingLevel.INFO)
|
||||
# assert LoggingLevel.INFO.is_lower_than(LoggingLevel.WARNING)
|
||||
# assert LoggingLevel.WARNING.is_lower_than(LoggingLevel.ERROR)
|
||||
|
||||
# However, due to the bug in implementation (using > instead of <),
|
||||
# the method actually behaves the same as is_higher_than
|
||||
pass
|
||||
|
||||
def test_is_lower_than_actual_behavior(self):
|
||||
"""Test the actual behavior of is_lower_than method."""
|
||||
# Due to the bug, this method behaves like is_higher_than
|
||||
assert LoggingLevel.DEBUG.is_lower_than(LoggingLevel.INFO)
|
||||
assert LoggingLevel.INFO.is_lower_than(LoggingLevel.WARNING)
|
||||
assert LoggingLevel.WARNING.is_lower_than(LoggingLevel.ERROR)
|
||||
assert LoggingLevel.ERROR.is_lower_than(LoggingLevel.CRITICAL)
|
||||
assert LoggingLevel.CRITICAL.is_lower_than(LoggingLevel.ALERT)
|
||||
assert LoggingLevel.ALERT.is_lower_than(LoggingLevel.EMERGENCY)
|
||||
assert LoggingLevel.EMERGENCY.is_lower_than(LoggingLevel.EXCEPTION)
|
||||
|
||||
# Same level should return False
|
||||
assert not LoggingLevel.INFO.is_lower_than(LoggingLevel.INFO)
|
||||
|
||||
|
||||
class TestEdgeCases:
|
||||
"""Test edge cases and error conditions."""
|
||||
|
||||
def test_none_inputs(self):
|
||||
"""Test handling of None inputs."""
|
||||
with pytest.raises((ValueError, AttributeError)):
|
||||
LoggingLevel.from_string(None)
|
||||
|
||||
with pytest.raises((ValueError, TypeError)):
|
||||
LoggingLevel.from_int(None)
|
||||
|
||||
def test_type_errors(self):
|
||||
"""Test type error conditions."""
|
||||
with pytest.raises((ValueError, AttributeError)):
|
||||
LoggingLevel.from_string(123)
|
||||
|
||||
with pytest.raises((ValueError, TypeError)):
|
||||
LoggingLevel.from_int("string")
|
||||
|
||||
|
||||
# Integration tests
|
||||
class TestIntegration:
|
||||
"""Integration tests combining multiple methods."""
|
||||
|
||||
def test_round_trip_conversions(self):
|
||||
"""Test round-trip conversions work correctly."""
|
||||
original_levels = [
|
||||
LoggingLevel.DEBUG, LoggingLevel.INFO, LoggingLevel.WARNING,
|
||||
LoggingLevel.ERROR, LoggingLevel.CRITICAL, LoggingLevel.ALERT,
|
||||
LoggingLevel.EXCEPTION, LoggingLevel.EXCEPTION
|
||||
]
|
||||
|
||||
for level in original_levels:
|
||||
# Test int round-trip
|
||||
assert LoggingLevel.from_int(level.value) == level
|
||||
|
||||
# Test string round-trip
|
||||
assert LoggingLevel.from_string(level.name) == level
|
||||
|
||||
# Test from_any round-trip
|
||||
assert LoggingLevel.from_any(level) == level
|
||||
assert LoggingLevel.from_any(level.value) == level
|
||||
assert LoggingLevel.from_any(level.name) == level
|
||||
|
||||
def test_level_hierarchy(self):
|
||||
"""Test that level hierarchy works correctly."""
|
||||
levels = [
|
||||
LoggingLevel.NOTSET, LoggingLevel.DEBUG, LoggingLevel.INFO,
|
||||
LoggingLevel.WARNING, LoggingLevel.ERROR, LoggingLevel.CRITICAL,
|
||||
LoggingLevel.ALERT, LoggingLevel.EMERGENCY, LoggingLevel.EXCEPTION
|
||||
]
|
||||
|
||||
for i, level in enumerate(levels):
|
||||
for j, other_level in enumerate(levels):
|
||||
if i < j:
|
||||
assert level.includes(other_level)
|
||||
assert other_level.is_higher_than(level)
|
||||
elif i > j:
|
||||
assert not level.includes(other_level)
|
||||
assert level.is_higher_than(other_level)
|
||||
else:
|
||||
assert level.includes(other_level) # Same level includes itself
|
||||
assert not level.is_higher_than(other_level) # Same level is not higher
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__])
|
||||
@@ -3,10 +3,9 @@ PyTest: string_handling/string_helpers
|
||||
"""
|
||||
|
||||
from textwrap import shorten
|
||||
from typing import Any
|
||||
import pytest
|
||||
from corelibs.string_handling.string_helpers import (
|
||||
shorten_string, left_fill, format_number, is_int, is_float, str_to_bool
|
||||
shorten_string, left_fill, format_number
|
||||
)
|
||||
|
||||
|
||||
@@ -237,238 +236,4 @@ def test_format_number_parametrized(number: float | int, precision: int, expecte
|
||||
"""Parametrized test for format_number"""
|
||||
assert format_number(number, precision) == expected
|
||||
|
||||
# ADDED 2025/7/11 Replace 'your_module' with actual module name
|
||||
|
||||
|
||||
class TestIsInt:
|
||||
"""Test cases for is_int function"""
|
||||
|
||||
def test_valid_integers(self):
|
||||
"""Test with valid integer strings"""
|
||||
assert is_int("123") is True
|
||||
assert is_int("0") is True
|
||||
assert is_int("-456") is True
|
||||
assert is_int("+789") is True
|
||||
assert is_int("000") is True
|
||||
|
||||
def test_invalid_integers(self):
|
||||
"""Test with invalid integer strings"""
|
||||
assert is_int("12.34") is False
|
||||
assert is_int("abc") is False
|
||||
assert is_int("12a") is False
|
||||
assert is_int("") is False
|
||||
assert is_int(" ") is False
|
||||
assert is_int("12.0") is False
|
||||
assert is_int("1e5") is False
|
||||
|
||||
def test_numeric_types(self):
|
||||
"""Test with actual numeric types"""
|
||||
assert is_int(123) is True
|
||||
assert is_int(0) is True
|
||||
assert is_int(-456) is True
|
||||
assert is_int(12.34) is True # float can be converted to int
|
||||
assert is_int(12.0) is True
|
||||
|
||||
def test_other_types(self):
|
||||
"""Test with other data types"""
|
||||
assert is_int(None) is False
|
||||
assert is_int([]) is False
|
||||
assert is_int({}) is False
|
||||
assert is_int(True) is True # bool is subclass of int
|
||||
assert is_int(False) is True
|
||||
|
||||
|
||||
class TestIsFloat:
|
||||
"""Test cases for is_float function"""
|
||||
|
||||
def test_valid_floats(self):
|
||||
"""Test with valid float strings"""
|
||||
assert is_float("12.34") is True
|
||||
assert is_float("0.0") is True
|
||||
assert is_float("-45.67") is True
|
||||
assert is_float("+78.9") is True
|
||||
assert is_float("123") is True # integers are valid floats
|
||||
assert is_float("0") is True
|
||||
assert is_float("1e5") is True
|
||||
assert is_float("1.5e-10") is True
|
||||
assert is_float("inf") is True
|
||||
assert is_float("-inf") is True
|
||||
assert is_float("nan") is True
|
||||
|
||||
def test_invalid_floats(self):
|
||||
"""Test with invalid float strings"""
|
||||
assert is_float("abc") is False
|
||||
assert is_float("12.34.56") is False
|
||||
assert is_float("12a") is False
|
||||
assert is_float("") is False
|
||||
assert is_float(" ") is False
|
||||
assert is_float("12..34") is False
|
||||
|
||||
def test_numeric_types(self):
|
||||
"""Test with actual numeric types"""
|
||||
assert is_float(123) is True
|
||||
assert is_float(12.34) is True
|
||||
assert is_float(0) is True
|
||||
assert is_float(-45.67) is True
|
||||
|
||||
def test_other_types(self):
|
||||
"""Test with other data types"""
|
||||
assert is_float(None) is False
|
||||
assert is_float([]) is False
|
||||
assert is_float({}) is False
|
||||
assert is_float(True) is True # bool can be converted to float
|
||||
assert is_float(False) is True
|
||||
|
||||
|
||||
class TestStrToBool:
|
||||
"""Test cases for str_to_bool function"""
|
||||
|
||||
def test_valid_true_strings(self):
|
||||
"""Test with valid true strings"""
|
||||
assert str_to_bool("True") is True
|
||||
assert str_to_bool("true") is True
|
||||
|
||||
def test_valid_false_strings(self):
|
||||
"""Test with valid false strings"""
|
||||
assert str_to_bool("False") is False
|
||||
assert str_to_bool("false") is False
|
||||
|
||||
def test_invalid_strings(self):
|
||||
"""Test with invalid boolean strings"""
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("TRUE")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("FALSE")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("yes")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("no")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("1")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("0")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool(" True")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("True ")
|
||||
|
||||
def test_error_message_content(self):
|
||||
"""Test that error messages contain the invalid input"""
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
str_to_bool("invalid")
|
||||
assert "Invalid boolean string: invalid" in str(exc_info.value)
|
||||
|
||||
def test_case_sensitivity(self):
|
||||
"""Test that function is case sensitive"""
|
||||
with pytest.raises(ValueError):
|
||||
str_to_bool("TRUE")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
str_to_bool("True ") # with space
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
str_to_bool(" True") # with space
|
||||
|
||||
|
||||
# Additional edge case tests
|
||||
class TestEdgeCases:
|
||||
"""Test edge cases and special scenarios"""
|
||||
|
||||
def test_is_int_with_whitespace(self):
|
||||
"""Test is_int with whitespace (should work due to int() behavior)"""
|
||||
assert is_int(" 123 ") is True
|
||||
assert is_int("\t456\n") is True
|
||||
|
||||
def test_is_float_with_whitespace(self):
|
||||
"""Test is_float with whitespace (should work due to float() behavior)"""
|
||||
assert is_float(" 12.34 ") is True
|
||||
assert is_float("\t45.67\n") is True
|
||||
|
||||
def test_large_numbers(self):
|
||||
"""Test with very large numbers"""
|
||||
large_int = "123456789012345678901234567890"
|
||||
assert is_int(large_int) is True
|
||||
assert is_float(large_int) is True
|
||||
|
||||
def test_scientific_notation(self):
|
||||
"""Test scientific notation"""
|
||||
assert is_int("1e5") is False # int() doesn't handle scientific notation
|
||||
assert is_float("1e5") is True
|
||||
assert is_float("1.5e-10") is True
|
||||
assert is_float("2E+3") is True
|
||||
|
||||
|
||||
# Parametrized tests for more comprehensive coverage
|
||||
class TestParametrized:
|
||||
"""Parametrized tests for better coverage"""
|
||||
|
||||
@pytest.mark.parametrize("value,expected", [
|
||||
("123", True),
|
||||
("0", True),
|
||||
("-456", True),
|
||||
("12.34", False),
|
||||
("abc", False),
|
||||
("", False),
|
||||
(123, True),
|
||||
(12.5, True),
|
||||
(None, False),
|
||||
])
|
||||
def test_is_int_parametrized(self, value: Any, expected: bool):
|
||||
"""Test"""
|
||||
assert is_int(value) == expected
|
||||
|
||||
@pytest.mark.parametrize("value,expected", [
|
||||
("12.34", True),
|
||||
("123", True),
|
||||
("0", True),
|
||||
("-45.67", True),
|
||||
("inf", True),
|
||||
("nan", True),
|
||||
("abc", False),
|
||||
("", False),
|
||||
(12.34, True),
|
||||
(123, True),
|
||||
(None, False),
|
||||
])
|
||||
def test_is_float_parametrized(self, value: Any, expected: bool):
|
||||
"""test"""
|
||||
assert is_float(value) == expected
|
||||
|
||||
@pytest.mark.parametrize("value,expected", [
|
||||
("True", True),
|
||||
("true", True),
|
||||
("False", False),
|
||||
("false", False),
|
||||
])
|
||||
def test_str_to_bool_valid_parametrized(self, value: Any, expected: bool):
|
||||
"""test"""
|
||||
assert str_to_bool(value) == expected
|
||||
|
||||
@pytest.mark.parametrize("invalid_value", [
|
||||
"TRUE",
|
||||
"FALSE",
|
||||
"yes",
|
||||
"no",
|
||||
"1",
|
||||
"0",
|
||||
"",
|
||||
" True",
|
||||
"True ",
|
||||
"invalid",
|
||||
])
|
||||
def test_str_to_bool_invalid_parametrized(self, invalid_value: Any):
|
||||
"""test"""
|
||||
with pytest.raises(ValueError):
|
||||
str_to_bool(invalid_value)
|
||||
|
||||
# __END__
|
||||
|
||||
241
tests/unit/string_handling/test_var_helpers.py
Normal file
241
tests/unit/string_handling/test_var_helpers.py
Normal file
@@ -0,0 +1,241 @@
|
||||
"""
|
||||
var helpers
|
||||
"""
|
||||
|
||||
# ADDED 2025/7/11 Replace 'your_module' with actual module name
|
||||
|
||||
from typing import Any
|
||||
import pytest
|
||||
from corelibs.var_handling.var_helpers import is_int, is_float, str_to_bool
|
||||
|
||||
|
||||
class TestIsInt:
|
||||
"""Test cases for is_int function"""
|
||||
|
||||
def test_valid_integers(self):
|
||||
"""Test with valid integer strings"""
|
||||
assert is_int("123") is True
|
||||
assert is_int("0") is True
|
||||
assert is_int("-456") is True
|
||||
assert is_int("+789") is True
|
||||
assert is_int("000") is True
|
||||
|
||||
def test_invalid_integers(self):
|
||||
"""Test with invalid integer strings"""
|
||||
assert is_int("12.34") is False
|
||||
assert is_int("abc") is False
|
||||
assert is_int("12a") is False
|
||||
assert is_int("") is False
|
||||
assert is_int(" ") is False
|
||||
assert is_int("12.0") is False
|
||||
assert is_int("1e5") is False
|
||||
|
||||
def test_numeric_types(self):
|
||||
"""Test with actual numeric types"""
|
||||
assert is_int(123) is True
|
||||
assert is_int(0) is True
|
||||
assert is_int(-456) is True
|
||||
assert is_int(12.34) is True # float can be converted to int
|
||||
assert is_int(12.0) is True
|
||||
|
||||
def test_other_types(self):
|
||||
"""Test with other data types"""
|
||||
assert is_int(None) is False
|
||||
assert is_int([]) is False
|
||||
assert is_int({}) is False
|
||||
assert is_int(True) is True # bool is subclass of int
|
||||
assert is_int(False) is True
|
||||
|
||||
|
||||
class TestIsFloat:
|
||||
"""Test cases for is_float function"""
|
||||
|
||||
def test_valid_floats(self):
|
||||
"""Test with valid float strings"""
|
||||
assert is_float("12.34") is True
|
||||
assert is_float("0.0") is True
|
||||
assert is_float("-45.67") is True
|
||||
assert is_float("+78.9") is True
|
||||
assert is_float("123") is True # integers are valid floats
|
||||
assert is_float("0") is True
|
||||
assert is_float("1e5") is True
|
||||
assert is_float("1.5e-10") is True
|
||||
assert is_float("inf") is True
|
||||
assert is_float("-inf") is True
|
||||
assert is_float("nan") is True
|
||||
|
||||
def test_invalid_floats(self):
|
||||
"""Test with invalid float strings"""
|
||||
assert is_float("abc") is False
|
||||
assert is_float("12.34.56") is False
|
||||
assert is_float("12a") is False
|
||||
assert is_float("") is False
|
||||
assert is_float(" ") is False
|
||||
assert is_float("12..34") is False
|
||||
|
||||
def test_numeric_types(self):
|
||||
"""Test with actual numeric types"""
|
||||
assert is_float(123) is True
|
||||
assert is_float(12.34) is True
|
||||
assert is_float(0) is True
|
||||
assert is_float(-45.67) is True
|
||||
|
||||
def test_other_types(self):
|
||||
"""Test with other data types"""
|
||||
assert is_float(None) is False
|
||||
assert is_float([]) is False
|
||||
assert is_float({}) is False
|
||||
assert is_float(True) is True # bool can be converted to float
|
||||
assert is_float(False) is True
|
||||
|
||||
|
||||
class TestStrToBool:
|
||||
"""Test cases for str_to_bool function"""
|
||||
|
||||
def test_valid_true_strings(self):
|
||||
"""Test with valid true strings"""
|
||||
assert str_to_bool("True") is True
|
||||
assert str_to_bool("true") is True
|
||||
|
||||
def test_valid_false_strings(self):
|
||||
"""Test with valid false strings"""
|
||||
assert str_to_bool("False") is False
|
||||
assert str_to_bool("false") is False
|
||||
|
||||
def test_invalid_strings(self):
|
||||
"""Test with invalid boolean strings"""
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("TRUE")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("FALSE")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("yes")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("no")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("1")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("0")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool(" True")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid boolean string"):
|
||||
str_to_bool("True ")
|
||||
|
||||
def test_error_message_content(self):
|
||||
"""Test that error messages contain the invalid input"""
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
str_to_bool("invalid")
|
||||
assert "Invalid boolean string: invalid" in str(exc_info.value)
|
||||
|
||||
def test_case_sensitivity(self):
|
||||
"""Test that function is case sensitive"""
|
||||
with pytest.raises(ValueError):
|
||||
str_to_bool("TRUE")
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
str_to_bool("True ") # with space
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
str_to_bool(" True") # with space
|
||||
|
||||
|
||||
# Additional edge case tests
|
||||
class TestEdgeCases:
|
||||
"""Test edge cases and special scenarios"""
|
||||
|
||||
def test_is_int_with_whitespace(self):
|
||||
"""Test is_int with whitespace (should work due to int() behavior)"""
|
||||
assert is_int(" 123 ") is True
|
||||
assert is_int("\t456\n") is True
|
||||
|
||||
def test_is_float_with_whitespace(self):
|
||||
"""Test is_float with whitespace (should work due to float() behavior)"""
|
||||
assert is_float(" 12.34 ") is True
|
||||
assert is_float("\t45.67\n") is True
|
||||
|
||||
def test_large_numbers(self):
|
||||
"""Test with very large numbers"""
|
||||
large_int = "123456789012345678901234567890"
|
||||
assert is_int(large_int) is True
|
||||
assert is_float(large_int) is True
|
||||
|
||||
def test_scientific_notation(self):
|
||||
"""Test scientific notation"""
|
||||
assert is_int("1e5") is False # int() doesn't handle scientific notation
|
||||
assert is_float("1e5") is True
|
||||
assert is_float("1.5e-10") is True
|
||||
assert is_float("2E+3") is True
|
||||
|
||||
|
||||
# Parametrized tests for more comprehensive coverage
|
||||
class TestParametrized:
|
||||
"""Parametrized tests for better coverage"""
|
||||
|
||||
@pytest.mark.parametrize("value,expected", [
|
||||
("123", True),
|
||||
("0", True),
|
||||
("-456", True),
|
||||
("12.34", False),
|
||||
("abc", False),
|
||||
("", False),
|
||||
(123, True),
|
||||
(12.5, True),
|
||||
(None, False),
|
||||
])
|
||||
def test_is_int_parametrized(self, value: Any, expected: bool):
|
||||
"""Test"""
|
||||
assert is_int(value) == expected
|
||||
|
||||
@pytest.mark.parametrize("value,expected", [
|
||||
("12.34", True),
|
||||
("123", True),
|
||||
("0", True),
|
||||
("-45.67", True),
|
||||
("inf", True),
|
||||
("nan", True),
|
||||
("abc", False),
|
||||
("", False),
|
||||
(12.34, True),
|
||||
(123, True),
|
||||
(None, False),
|
||||
])
|
||||
def test_is_float_parametrized(self, value: Any, expected: bool):
|
||||
"""test"""
|
||||
assert is_float(value) == expected
|
||||
|
||||
@pytest.mark.parametrize("value,expected", [
|
||||
("True", True),
|
||||
("true", True),
|
||||
("False", False),
|
||||
("false", False),
|
||||
])
|
||||
def test_str_to_bool_valid_parametrized(self, value: Any, expected: bool):
|
||||
"""test"""
|
||||
assert str_to_bool(value) == expected
|
||||
|
||||
@pytest.mark.parametrize("invalid_value", [
|
||||
"TRUE",
|
||||
"FALSE",
|
||||
"yes",
|
||||
"no",
|
||||
"1",
|
||||
"0",
|
||||
"",
|
||||
" True",
|
||||
"True ",
|
||||
"invalid",
|
||||
])
|
||||
def test_str_to_bool_invalid_parametrized(self, invalid_value: Any):
|
||||
"""test"""
|
||||
with pytest.raises(ValueError):
|
||||
str_to_bool(invalid_value)
|
||||
Reference in New Issue
Block a user