Compare commits
17 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2a248bd249 | ||
|
|
c559a6bafb | ||
|
|
19d7e9b5ed | ||
|
|
3e5a5accf7 | ||
|
|
424c91945a | ||
|
|
c657dc564e | ||
|
|
208f002284 | ||
|
|
084ecc01e0 | ||
|
|
08cb994d8d | ||
|
|
67f1a6688d | ||
|
|
efb7968e93 | ||
|
|
fe7c7db004 | ||
|
|
79d1ccae9a | ||
|
|
6e69af4aa8 | ||
|
|
d500b7d473 | ||
|
|
ef599a1aad | ||
|
|
2d197134f1 |
27
README.md
27
README.md
@@ -12,11 +12,12 @@ This is a pip package that can be installed into any project and covers the foll
|
||||
|
||||
## Current list
|
||||
|
||||
- config_handling: simple INI config file data loader with check/convert/etc
|
||||
- csv_handling: csv dict writer helper
|
||||
- debug_handling: various debug helpers like data dumper, timer, utilization, etc
|
||||
- file_handling: crc handling for file content and file names, progress bar
|
||||
- json_handling: jmespath support and json date support
|
||||
- list_dict_handling: list and dictionary handling support (search, fingerprinting, etc)
|
||||
- iterator_handling: list and dictionary handling support (search, fingerprinting, etc)
|
||||
- logging_handling: extend log and also error message handling
|
||||
- requests_handling: requests wrapper for better calls with auth headers
|
||||
- script_handling: pid lock file handling, abort timer
|
||||
@@ -71,18 +72,34 @@ uv run pytest --cov=corelibs
|
||||
|
||||
In the test-run folder usage and run tests are located
|
||||
|
||||
#### Progress
|
||||
|
||||
```sh
|
||||
uv run test-run/progress/progress_test.py
|
||||
```
|
||||
|
||||
#### Double byte string format
|
||||
|
||||
```sh
|
||||
uv run test-run/double_byte_string_format/double_byte_string_format.py
|
||||
```
|
||||
|
||||
#### Strings helpers
|
||||
|
||||
```sh
|
||||
uv run test-run/timestamp_strings/timestamp_strings.py
|
||||
```
|
||||
|
||||
```sh
|
||||
uv run test-run/string_handling/string_helpers.py
|
||||
```
|
||||
|
||||
#### Log
|
||||
|
||||
```sh
|
||||
uv run test-run/logging_handling/log.py
|
||||
```
|
||||
|
||||
## How to install in another project
|
||||
|
||||
This will also add the index entry
|
||||
@@ -93,13 +110,7 @@ uv add corelibs --index egra-gitea=https://git.egplusww.jp/api/packages/PyPI/pyp
|
||||
|
||||
## Python venv setup
|
||||
|
||||
In the folder where the script will be located
|
||||
|
||||
```sh
|
||||
uv venv --python 3.13
|
||||
```
|
||||
|
||||
Install all neded dependencies
|
||||
After clone, run the command below to install all dependenciss
|
||||
|
||||
```sh
|
||||
uv sync
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# MARK: Project info
|
||||
[project]
|
||||
name = "corelibs"
|
||||
version = "0.8.0"
|
||||
version = "0.11.0"
|
||||
description = "Collection of utils for Python scripts"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.13"
|
||||
|
||||
507
src/corelibs/config_handling/settings_loader.py
Normal file
507
src/corelibs/config_handling/settings_loader.py
Normal file
@@ -0,0 +1,507 @@
|
||||
"""
|
||||
Load settings file for a certain group
|
||||
Check data for existing and valid
|
||||
Additional check for override settings as arguments
|
||||
"""
|
||||
|
||||
import re
|
||||
import sys
|
||||
import configparser
|
||||
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.config_handling.settings_loader_handling.settings_loader_check import SettingsLoaderCheck
|
||||
|
||||
|
||||
class SettingsLoader:
|
||||
"""
|
||||
Settings Loader with Argument parser
|
||||
"""
|
||||
|
||||
# split char
|
||||
DEFAULT_ELEMENT_SPLIT_CHAR: str = ','
|
||||
|
||||
CONVERT_TO_LIST: list[str] = ['str', 'int', 'float', 'bool', 'auto']
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
args: dict[str, Any],
|
||||
config_file: Path,
|
||||
log: 'Log | None' = None,
|
||||
always_print: bool = False
|
||||
) -> None:
|
||||
"""
|
||||
init the Settings loader
|
||||
|
||||
Args:
|
||||
args (dict): Script Arguments
|
||||
config_file (Path): config file including path
|
||||
log (Log | None): Lop class, if set errors are written to this
|
||||
always_print (bool): Set to true to always print errors, even if Log is available
|
||||
element_split_char (str): Split character, default is ','
|
||||
|
||||
Raises:
|
||||
ValueError: _description_
|
||||
"""
|
||||
self.args = args
|
||||
self.config_file = config_file
|
||||
self.log = log
|
||||
self.always_print = always_print
|
||||
# entries that have to be split
|
||||
self.entry_split_char: dict[str, str] = {}
|
||||
self.entry_convert: dict[str, str] = {}
|
||||
# config parser
|
||||
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
|
||||
|
||||
def load_settings(self, config_id: str, config_validate: dict[str, list[str]]) -> dict[str, str]:
|
||||
"""
|
||||
neutral settings loader
|
||||
|
||||
The settings values on the right side are seen as a list if they have "," inside (see ELEMENT SPLIT CHAR)
|
||||
but only if the "check:list." is set
|
||||
|
||||
for the allowe entries set, each set is "key => checks", check set is "check type:settings"
|
||||
key: the key name in the settings file
|
||||
check: check set with the following allowed entries on the left side for type
|
||||
- mandatory: must be set as "mandatory:yes", if the key entry is missing or empty throws error
|
||||
- check: see __check_settings for the settings currently available
|
||||
- matching: a | list of entries where the value has to match too
|
||||
- in: the right side is another KEY value from the settings where this value must be inside
|
||||
- split: character to split entries, if set check:list+ must be set if checks are needed
|
||||
- convert: convert to int, float -> if element is number convert, else leave as is
|
||||
|
||||
Args:
|
||||
config_id (str): what block to load
|
||||
config_allowed (list[str]): list of allowed entries sets
|
||||
|
||||
Returns:
|
||||
dict[str, str]: key = value list
|
||||
"""
|
||||
settings: dict[str, dict[str, Any]] = {
|
||||
config_id: {},
|
||||
}
|
||||
if self.config_parser is not None:
|
||||
try:
|
||||
# load all data as is, validation is done afterwards
|
||||
settings[config_id] = dict(self.config_parser[config_id])
|
||||
for key, checks in config_validate.items():
|
||||
skip = True
|
||||
split_char = self.DEFAULT_ELEMENT_SPLIT_CHAR
|
||||
# if one is set as list in check -> do not skip, but add to list
|
||||
for check in checks:
|
||||
if check.startswith("convert:"):
|
||||
try:
|
||||
[_, convert_to] = check.split(":")
|
||||
if convert_to not in self.CONVERT_TO_LIST:
|
||||
self.__print(
|
||||
f"[!] In [{config_id}] the convert type is invalid {check}: {convert_to}",
|
||||
'CRITICAL',
|
||||
raise_exception=True
|
||||
)
|
||||
except ValueError as e:
|
||||
self.__print(
|
||||
f"[!] In [{config_id}] the convert type setup for entry failed: {check}: {e}",
|
||||
'CRITICAL',
|
||||
raise_exception=True
|
||||
)
|
||||
sys.exit(1)
|
||||
# split char, also check to not set it twice, first one only
|
||||
if check.startswith("split:") and not self.entry_split_char.get(key):
|
||||
try:
|
||||
[_, split_char] = check.split(":")
|
||||
if len(split_char) == 0:
|
||||
self.__print(
|
||||
(
|
||||
f"[*] In [{config_id}] the [{key}] split char character is empty, "
|
||||
f"fallback to: {self.DEFAULT_ELEMENT_SPLIT_CHAR}"
|
||||
),
|
||||
"WARNING"
|
||||
)
|
||||
split_char = self.DEFAULT_ELEMENT_SPLIT_CHAR
|
||||
self.entry_split_char[key] = split_char
|
||||
skip = False
|
||||
except ValueError as e:
|
||||
self.__print(
|
||||
f"[!] In [{config_id}] the split character setup for entry failed: {check}: {e}",
|
||||
'CRITICAL',
|
||||
raise_exception=True
|
||||
)
|
||||
sys.exit(1)
|
||||
if skip:
|
||||
continue
|
||||
settings[config_id][key] = [
|
||||
__value.replace(" ", "")
|
||||
for __value in settings[config_id][key].split(split_char)
|
||||
]
|
||||
except KeyError as e:
|
||||
self.__print(
|
||||
f"[!] Cannot read [{config_id}] block in the {self.config_file}: {e}",
|
||||
'CRITICAL', raise_exception=True
|
||||
)
|
||||
sys.exit(1)
|
||||
else:
|
||||
# ignore error if arguments are set
|
||||
if not self.__check_arguments(config_validate, True):
|
||||
self.__print(f"[!] Cannot find file: {self.config_file}", 'CRITICAL', raise_exception=True)
|
||||
sys.exit(1)
|
||||
else:
|
||||
# base set
|
||||
settings[config_id] = {}
|
||||
# make sure all are set
|
||||
# if we have arguments set, this override config settings
|
||||
error: bool = False
|
||||
for entry, validate in config_validate.items():
|
||||
# if we have command line option set, this one overrides config
|
||||
if self.__get_arg(entry):
|
||||
self.__print(f"[*] Command line option override for: {entry}", 'WARNING')
|
||||
settings[config_id][entry] = self.args.get(entry)
|
||||
# validate checks
|
||||
for check in validate:
|
||||
# CHECKS
|
||||
# - mandatory
|
||||
# - check: regex check (see SettingsLoaderCheck class for entries)
|
||||
# - matching: entry in given list
|
||||
# - in: entry in other setting entry list
|
||||
# - length: for string length
|
||||
# - range: for int/float range check
|
||||
# mandatory check
|
||||
if check == "mandatory:yes" and not settings[config_id].get(entry):
|
||||
error = True
|
||||
self.__print(f"[!] Missing content entry for: {entry}", 'ERROR')
|
||||
# skip if empty none
|
||||
if settings[config_id].get(entry) is None:
|
||||
continue
|
||||
if check.startswith("check:"):
|
||||
# replace the check and run normal checks
|
||||
settings[config_id][entry] = self.__check_settings(
|
||||
check, entry, settings[config_id][entry]
|
||||
)
|
||||
elif check.startswith("matching:"):
|
||||
checks = check.replace("matching:", "").split("|")
|
||||
if __result := is_list_in_list(convert_to_list(settings[config_id][entry]), list(checks)):
|
||||
error = True
|
||||
self.__print(f"[!] [{entry}] '{__result}' not matching {checks}", 'ERROR')
|
||||
elif check.startswith("in:"):
|
||||
check = check.replace("in:", "")
|
||||
# skip if check does not exist, and set error
|
||||
if settings[config_id].get(check) is None:
|
||||
error = True
|
||||
self.__print(f"[!] [{entry}] '{check}' target does not exist", 'ERROR')
|
||||
continue
|
||||
# entry must be in check entry
|
||||
# in for list, else equal with convert to string
|
||||
if (
|
||||
__result := is_list_in_list(
|
||||
convert_to_list(settings[config_id][entry]),
|
||||
__checks := convert_to_list(settings[config_id][check])
|
||||
)
|
||||
):
|
||||
self.__print(f"[!] [{entry}] '{__result}' must be in the '{__checks}' values list", 'ERROR')
|
||||
error = True
|
||||
elif check.startswith('length:'):
|
||||
check = check.replace("length:", "")
|
||||
# length can be: n, n-, n-m, -m
|
||||
# as: equal, >= >=< =<
|
||||
self.__build_from_to_equal(entry, check)
|
||||
if not self.__length_range_validate(
|
||||
entry,
|
||||
'length',
|
||||
cast(list[str], convert_to_list(settings[config_id][entry])),
|
||||
self.__build_from_to_equal(entry, check, convert_to_int=True)
|
||||
):
|
||||
error = True
|
||||
elif check.startswith('range:'):
|
||||
check = check.replace("range:", "")
|
||||
if not self.__length_range_validate(
|
||||
entry,
|
||||
'range',
|
||||
cast(list[str], convert_to_list(settings[config_id][entry])),
|
||||
self.__build_from_to_equal(entry, check)
|
||||
):
|
||||
error = True
|
||||
if error is True:
|
||||
self.__print("[!] Missing or incorrect settings data. Cannot proceed", 'CRITICAL', raise_exception=True)
|
||||
sys.exit(1)
|
||||
# Convert input
|
||||
for [entry, convert_type] in self.entry_convert:
|
||||
if convert_type in ["int", "any"] and is_int(settings[config_id][entry]):
|
||||
settings[config_id][entry] = int(settings[config_id][entry])
|
||||
elif convert_type in ["float", "any"] and is_float(settings[config_id][entry]):
|
||||
settings[config_id][entry] = float(settings[config_id][entry])
|
||||
elif convert_type in ["bool", "any"] and (
|
||||
settings[config_id][entry] == "true" or
|
||||
settings[config_id][entry] == "True" or
|
||||
settings[config_id][entry] == "false" or
|
||||
settings[config_id][entry] == "False"
|
||||
):
|
||||
try:
|
||||
settings[config_id][entry] = str_to_bool(settings[config_id][entry])
|
||||
except ValueError:
|
||||
self.__print(
|
||||
f"[!] Could not convert to boolean for '{entry}': {settings[config_id][entry]}",
|
||||
'ERROR'
|
||||
)
|
||||
# string is always string
|
||||
|
||||
return settings[config_id]
|
||||
|
||||
def __build_from_to_equal(
|
||||
self, entry: str, check: str, convert_to_int: bool = False
|
||||
) -> Tuple[float | None, float | None, float | None]:
|
||||
"""
|
||||
split out the "n-m" part to get the to/from/equal
|
||||
|
||||
Arguments:
|
||||
entry {str} -- _description_
|
||||
check {str} -- _description_
|
||||
|
||||
Returns:
|
||||
Tuple[float | None, float | None, float | None] -- _description_
|
||||
|
||||
Throws:
|
||||
ValueError if range/length entries are not float
|
||||
"""
|
||||
__from = None
|
||||
__to = None
|
||||
__equal = None
|
||||
try:
|
||||
[__from, __to] = check.split('-')
|
||||
if (__from and not is_float(__from)) or (__to and not is_float(__to)):
|
||||
self.__print(
|
||||
f"[{entry}] Check value for length is not in: {check}",
|
||||
'CRITICAL', raise_exception=True
|
||||
)
|
||||
sys.exit(1)
|
||||
if len(__from) == 0:
|
||||
__from = None
|
||||
if len(__to) == 0:
|
||||
__to = None
|
||||
except ValueError:
|
||||
if not is_float(__equal := check):
|
||||
self.__print(
|
||||
f"[{entry}] Check value for length is not a valid integer: {check}",
|
||||
'CRITICAL', raise_exception=True
|
||||
)
|
||||
sys.exit(1)
|
||||
if len(__equal) == 0:
|
||||
__equal = None
|
||||
# makre sure this is all int or None
|
||||
if __from is not None:
|
||||
__from = int(__from) if convert_to_int else float(__from)
|
||||
if __to is not None:
|
||||
__to = int(__to) if convert_to_int else float(__to)
|
||||
if __equal is not None:
|
||||
__equal = int(__equal) if convert_to_int else float(__equal)
|
||||
return (
|
||||
__from,
|
||||
__to,
|
||||
__equal
|
||||
)
|
||||
|
||||
def __length_range_validate(
|
||||
self,
|
||||
entry: str,
|
||||
check_type: str,
|
||||
values: Sequence[str | int | float],
|
||||
check: Tuple[float | None, float | None, float | None],
|
||||
) -> bool:
|
||||
(__from, __to, __equal) = check
|
||||
valid = True
|
||||
for value_raw in convert_to_list(values):
|
||||
value = 0
|
||||
error_mark = ''
|
||||
if check_type == 'length':
|
||||
error_mark = 'length'
|
||||
value = len(str(value_raw))
|
||||
elif check_type == 'range':
|
||||
error_mark = 'range'
|
||||
value = float(str(value_raw))
|
||||
if __equal is not None and value != __equal:
|
||||
self.__print(f"[!] [{entry}] '{value_raw}' {error_mark} does not match {__equal}", 'ERROR')
|
||||
valid = False
|
||||
continue
|
||||
if __from is not None and __to is None and value < __from:
|
||||
self.__print(f"[!] [{entry}] '{value_raw}' {error_mark} smaller than minimum {__from}", 'ERROR')
|
||||
valid = False
|
||||
continue
|
||||
if __from is None and __to is not None and value > __to:
|
||||
self.__print(f"[!] [{entry}] '{value_raw}' {error_mark} larger than maximum {__to}", 'ERROR')
|
||||
valid = False
|
||||
continue
|
||||
if __from is not None and __to is not None and (
|
||||
value < __from or value > __to
|
||||
):
|
||||
self.__print(
|
||||
f"[!] [{entry}] '{value_raw}' {error_mark} outside valid range {__from} to {__to}",
|
||||
'ERROR'
|
||||
)
|
||||
valid = False
|
||||
continue
|
||||
return valid
|
||||
|
||||
def __load_config_file(self) -> configparser.ConfigParser | None:
|
||||
"""
|
||||
load and parse the config file
|
||||
if not loadable return None
|
||||
"""
|
||||
config = configparser.ConfigParser()
|
||||
if self.config_file.is_file():
|
||||
config.read(self.config_file)
|
||||
return config
|
||||
return None
|
||||
|
||||
def __clean_invalid_setting(
|
||||
self,
|
||||
entry: str,
|
||||
validate: str,
|
||||
value: str,
|
||||
regex: str,
|
||||
regex_clean: str,
|
||||
replace: str = "",
|
||||
print_error: bool = True,
|
||||
) -> str:
|
||||
"""
|
||||
check is a string is invalid, print optional error message and clean up string
|
||||
|
||||
Args:
|
||||
entry (str): what entry key
|
||||
validate (str): validate type
|
||||
value (str): the value to check against
|
||||
regex (str): regex used for checking as r'...'
|
||||
regex_clean (str): regex used for cleaning as r'...'
|
||||
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):
|
||||
self.__print(
|
||||
f"[!] Invalid content for '{entry}' with check '{validate}' and data: {value}",
|
||||
'ERROR', print_error
|
||||
)
|
||||
# clean up
|
||||
return clean.sub(replace, value)
|
||||
# else return as is
|
||||
return value
|
||||
|
||||
def __check_settings(
|
||||
self,
|
||||
check: str, entry: str, setting_value: list[str] | str
|
||||
) -> list[str] | str:
|
||||
"""
|
||||
check each setting valid
|
||||
The settings are defined in the SettingsLoaderCheck class
|
||||
|
||||
Args:
|
||||
check (str): What check to run
|
||||
entry (str): Variable name, just for information message
|
||||
setting_value (list[str | int] | str | int): settings value data
|
||||
entry_split_char (str | None): split char, for list check
|
||||
|
||||
Returns:
|
||||
list[str | int] |111 str | int: cleaned up settings value data
|
||||
"""
|
||||
check = check.replace("check:", "")
|
||||
# get the check settings
|
||||
__check_settings = SettingsLoaderCheck.CHECK_SETTINGS.get(check)
|
||||
if __check_settings is None:
|
||||
self.__print(
|
||||
f"[{entry}] Cannot get SettingsLoaderCheck.CHECK_SETTINGS for {check}",
|
||||
'CRITICAL', raise_exception=True
|
||||
)
|
||||
sys.exit(1)
|
||||
# either removes or replaces invalid characters in the list
|
||||
if isinstance(setting_value, list):
|
||||
# clean up invalid characters
|
||||
# loop over result and keep only filled (strip empty)
|
||||
setting_value = [e for e in [
|
||||
self.__clean_invalid_setting(
|
||||
entry, check, str(__entry),
|
||||
__check_settings['regex'], __check_settings['regex_clean'], __check_settings['replace']
|
||||
)
|
||||
for __entry in setting_value
|
||||
] if e]
|
||||
else:
|
||||
setting_value = self.__clean_invalid_setting(
|
||||
entry, check, str(setting_value),
|
||||
__check_settings['regex'], __check_settings['regex_clean'], __check_settings['replace']
|
||||
)
|
||||
# else:
|
||||
# self.__print(f"[!] Unkown type to check", 'ERROR)
|
||||
# return data
|
||||
return setting_value
|
||||
|
||||
def __check_arguments(self, arguments: dict[str, list[str]], all_set: bool = False) -> bool:
|
||||
"""
|
||||
check if ast least one argument is set
|
||||
|
||||
Args:
|
||||
arguments (list[str]): _description_
|
||||
|
||||
Returns:
|
||||
bool: _description_
|
||||
"""
|
||||
count_set = 0
|
||||
count_arguments = 0
|
||||
has_argument = False
|
||||
for argument, validate in arguments.items():
|
||||
# if argument is mandatory add to count, if not mandatory set has "has" to skip error
|
||||
mandatory = any(entry == "mandatory:yes" for entry in validate)
|
||||
if not mandatory:
|
||||
has_argument = True
|
||||
continue
|
||||
count_arguments += 1
|
||||
if self.__get_arg(argument):
|
||||
has_argument = True
|
||||
count_set += 1
|
||||
# for all set, True only if all are set
|
||||
if all_set is True:
|
||||
has_argument = count_set == count_arguments
|
||||
|
||||
return has_argument
|
||||
|
||||
def __get_arg(self, entry: str) -> Any:
|
||||
"""
|
||||
check if an argument entry xists, if None -> returns None else value of argument
|
||||
|
||||
Arguments:
|
||||
entry {str} -- _description_
|
||||
|
||||
Returns:
|
||||
Any -- _description_
|
||||
"""
|
||||
if self.args.get(entry) is None:
|
||||
return None
|
||||
return self.args.get(entry)
|
||||
|
||||
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
|
||||
|
||||
Arguments:
|
||||
msg {str} -- _description_
|
||||
level {str} -- _description_
|
||||
|
||||
Keyword Arguments:
|
||||
print_error {bool} -- _description_ (default: {True})
|
||||
"""
|
||||
if self.log is not None:
|
||||
if not Log.validate_log_level(level):
|
||||
level = 'ERROR'
|
||||
self.log.logger.log(Log.get_log_level_int(level), msg)
|
||||
if self.log is None or self.always_print:
|
||||
if print_error:
|
||||
print(msg)
|
||||
if raise_exception:
|
||||
raise ValueError(msg)
|
||||
|
||||
|
||||
# __END__
|
||||
@@ -0,0 +1,44 @@
|
||||
"""
|
||||
Class of checks that can be run on value entries
|
||||
"""
|
||||
|
||||
from typing import TypedDict
|
||||
|
||||
|
||||
class SettingsLoaderCheckValue(TypedDict):
|
||||
"""Settings check entries"""
|
||||
regex: str
|
||||
regex_clean: str
|
||||
replace: str
|
||||
|
||||
|
||||
class SettingsLoaderCheck:
|
||||
"""
|
||||
check:<NAME> or check:list+<NAME>
|
||||
"""
|
||||
CHECK_SETTINGS: dict[str, SettingsLoaderCheckValue] = {
|
||||
"int": {
|
||||
"regex": r"^[0-9]+$",
|
||||
"regex_clean": r"[^0-9]",
|
||||
"replace": ""
|
||||
},
|
||||
"string.alphanumeric": {
|
||||
"regex": r"^[a-zA-Z0-9]+$",
|
||||
"regex_clean": r"[^a-zA-Z0-9]",
|
||||
"replace": ""
|
||||
},
|
||||
"string.alphanumeric.lower.dash": {
|
||||
"regex": r"^[a-z0-9-]+$",
|
||||
"regex_clean": r"[^a-z0-9-]",
|
||||
"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": "_"
|
||||
},
|
||||
}
|
||||
|
||||
# __END__
|
||||
0
src/corelibs/iterator_handling/__init__.py
Normal file
0
src/corelibs/iterator_handling/__init__.py
Normal file
47
src/corelibs/iterator_handling/list_helpers.py
Normal file
47
src/corelibs/iterator_handling/list_helpers.py
Normal file
@@ -0,0 +1,47 @@
|
||||
"""
|
||||
List type helpers
|
||||
"""
|
||||
|
||||
from typing import Any, Sequence
|
||||
|
||||
|
||||
def convert_to_list(
|
||||
entry: str | int | float | bool | Sequence[str | int | float | bool | Sequence[Any]]
|
||||
) -> Sequence[str | int | float | bool | Sequence[Any]]:
|
||||
"""
|
||||
Convert any of the non list values (except dictionary) to a list
|
||||
|
||||
Arguments:
|
||||
entry {str | int | float | bool | list[str | int | float | bool]} -- _description_
|
||||
|
||||
Returns:
|
||||
list[str | int | float | bool] -- _description_
|
||||
"""
|
||||
if isinstance(entry, list):
|
||||
return entry
|
||||
return [entry]
|
||||
|
||||
|
||||
def is_list_in_list(
|
||||
list_a: Sequence[str | int | float | bool | Sequence[Any]],
|
||||
list_b: Sequence[str | int | float | bool | Sequence[Any]]
|
||||
) -> Sequence[str | int | float | bool | Sequence[Any]]:
|
||||
"""
|
||||
Return entries from list_a that are not in list_b
|
||||
Type safe compare
|
||||
|
||||
Arguments:
|
||||
list_a {list[Any]} -- _description_
|
||||
list_b {list[Any]} -- _description_
|
||||
|
||||
Returns:
|
||||
list[Any] -- _description_
|
||||
"""
|
||||
# Create sets of (value, type) tuples
|
||||
set_a = set((item, type(item)) for item in list_a)
|
||||
set_b = set((item, type(item)) for item in list_b)
|
||||
|
||||
# Get the difference and extract just the values
|
||||
return [item for item, _ in set_a - set_b]
|
||||
|
||||
# __END__
|
||||
@@ -1,11 +1,71 @@
|
||||
"""
|
||||
A log handler wrapper
|
||||
if log_settings['log_queue'] is set to multiprocessing.Queue it will launch with listeners
|
||||
attach "init_worker_logging" with the set log_queue
|
||||
"""
|
||||
|
||||
import re
|
||||
import logging.handlers
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import Mapping
|
||||
from typing import Mapping, 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
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from multiprocessing import Queue
|
||||
|
||||
|
||||
class LogSettings(TypedDict):
|
||||
"""
|
||||
log settings
|
||||
|
||||
Arguments:
|
||||
TypedDict {_type_} -- _description_
|
||||
"""
|
||||
log_level_console: str
|
||||
log_level_file: str
|
||||
console_enabled: bool
|
||||
console_color_output_enabled: bool
|
||||
add_start_info: bool
|
||||
add_end_info: bool
|
||||
log_queue: 'Queue[str] | None'
|
||||
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
"""
|
||||
set the color highlight
|
||||
|
||||
Arguments:
|
||||
record {logging.LogRecord} -- _description_
|
||||
|
||||
Returns:
|
||||
str -- _description_
|
||||
"""
|
||||
# Add color to levelname for console output
|
||||
reset = Colors.reset
|
||||
color = self.COLORS.get(record.levelname, reset)
|
||||
# only highlight level for basic
|
||||
if record.levelname in ['DEBUG', 'INFO']:
|
||||
record.levelname = f"{color}{record.levelname}{reset}"
|
||||
return super().format(record)
|
||||
# highlight whole line
|
||||
message = super().format(record)
|
||||
return f"{color}{message}{reset}"
|
||||
|
||||
|
||||
class Log:
|
||||
@@ -13,57 +73,146 @@ class Log:
|
||||
logger setup
|
||||
"""
|
||||
|
||||
EXCEPTION: int = 60
|
||||
# spacer lenght characters and the character
|
||||
SPACER_CHAR: str = '='
|
||||
SPACER_LENGTH: int = 32
|
||||
# default logging level
|
||||
DEFAULT_LOG_LEVEL: str = '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,
|
||||
}
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
log_path: Path,
|
||||
log_name: str,
|
||||
log_level_console: str = 'WARNING',
|
||||
log_level_file: str = 'DEBUG',
|
||||
add_start_info: bool = True
|
||||
log_settings: dict[str, 'str | bool | None | Queue[str]'] | LogSettings | None = None,
|
||||
):
|
||||
logging.addLevelName(Log.EXCEPTION, 'EXCEPTION')
|
||||
if not log_name.endswith('.log'):
|
||||
# add new level for EXCEPTION
|
||||
logging.addLevelName(LoggingLevel.EXCEPTION.value, 'EXCEPTION')
|
||||
# parse the logging settings
|
||||
self.log_settings = self.__parse_log_settings(log_settings)
|
||||
# if path, set log name with .log
|
||||
# if log name with .log, strip .log for naming
|
||||
if log_path.is_dir():
|
||||
__log_file_name = re.sub(r'[^a-zA-Z0-9]', '', log_name)
|
||||
if not log_name.endswith('.log'):
|
||||
log_path = log_path.joinpath(Path(__log_file_name).with_suffix('.log'))
|
||||
else:
|
||||
log_path = log_path.joinpath(__log_file_name)
|
||||
elif not log_path.suffix == '.log':
|
||||
# add .log if the path is a file but without .log
|
||||
log_path = log_path.with_suffix('.log')
|
||||
# overall logger settings
|
||||
self.logger = logging.getLogger(log_name)
|
||||
# set maximum logging level for all logging output
|
||||
self.logger.setLevel(logging.DEBUG)
|
||||
# stip .log from the log name if set
|
||||
if not log_name.endswith('.log'):
|
||||
log_name = Path(log_name).stem
|
||||
# general log name
|
||||
self.log_name = log_name
|
||||
|
||||
# self.handlers = []
|
||||
# console logger
|
||||
self.__console_handler(log_level_console)
|
||||
# file logger
|
||||
self.__file_handler(log_level_file, log_path)
|
||||
self.log_queue: 'Queue[str] | None' = None
|
||||
self.listener: logging.handlers.QueueListener | None = None
|
||||
|
||||
# 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)
|
||||
]
|
||||
if self.log_settings['console_enabled']:
|
||||
# console
|
||||
self.handlers.append(self.__console_handler(self.log_settings['log_level_console']))
|
||||
# init listener if we have a log_queue set
|
||||
self.__init_listener(self.log_settings['log_queue'])
|
||||
|
||||
# overall logger start
|
||||
self.__init_log(log_name)
|
||||
# if requests set a start log
|
||||
if add_start_info is True:
|
||||
if self.log_settings['add_start_info'] is True:
|
||||
self.break_line('START')
|
||||
|
||||
def __del__(self):
|
||||
"""
|
||||
Call when class is destroyed, make sure the listender is closed or else we throw a thread error
|
||||
"""
|
||||
if self.log_settings['add_end_info']:
|
||||
self.break_line('END')
|
||||
self.stop_listener()
|
||||
|
||||
def __parse_log_settings(
|
||||
self,
|
||||
log_settings: dict[str, 'str | bool | None | Queue[str]'] | LogSettings | None
|
||||
) -> LogSettings:
|
||||
# skip with defaul it not set
|
||||
if log_settings is None:
|
||||
return self.DEFAULT_LOG_SETTINGS
|
||||
# check entries
|
||||
default_log_settings = self.DEFAULT_LOG_SETTINGS
|
||||
# check log levels
|
||||
for __log_entry in ['log_level_console', 'log_level_file']:
|
||||
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(
|
||||
__log_entry, self.DEFAULT_LOG_LEVEL
|
||||
)
|
||||
default_log_settings[__log_entry] = str(_log_level)
|
||||
# check bool
|
||||
for __log_entry in [
|
||||
"console_enabled",
|
||||
"console_color_output_enabled",
|
||||
"add_start_info",
|
||||
"add_end_info",
|
||||
]:
|
||||
if log_settings.get(__log_entry) is None:
|
||||
continue
|
||||
if not isinstance(__setting := log_settings.get(__log_entry, ''), bool):
|
||||
__setting = self.DEFAULT_LOG_SETTINGS.get(__log_entry, True)
|
||||
default_log_settings[__log_entry] = __setting
|
||||
# check log queue
|
||||
__setting = log_settings.get('log_queue', self.DEFAULT_LOG_SETTINGS['log_queue'])
|
||||
if __setting is not None:
|
||||
__setting = cast('Queue[str]', __setting)
|
||||
default_log_settings['log_queue'] = __setting
|
||||
return default_log_settings
|
||||
|
||||
def __filter_exceptions(self, record: logging.LogRecord) -> bool:
|
||||
return record.levelname != "EXCEPTION"
|
||||
|
||||
def __console_handler(self, log_level_console: str = 'WARNING'):
|
||||
def __console_handler(self, log_level_console: str = 'WARNING') -> logging.StreamHandler[TextIO]:
|
||||
# console logger
|
||||
if not isinstance(getattr(logging, log_level_console.upper(), None), int):
|
||||
log_level_console = 'WARNING'
|
||||
console_handler = logging.StreamHandler()
|
||||
formatter_console = logging.Formatter(
|
||||
(
|
||||
'[%(asctime)s.%(msecs)03d] '
|
||||
'[%(filename)s:%(funcName)s:%(lineno)d] '
|
||||
'<%(levelname)s> '
|
||||
'%(message)s'
|
||||
),
|
||||
datefmt="%Y-%m-%d %H:%M:%S",
|
||||
# format layouts
|
||||
format_string = (
|
||||
'[%(asctime)s.%(msecs)03d] '
|
||||
'[%(name)s] '
|
||||
'[%(filename)s:%(funcName)s:%(lineno)d] '
|
||||
'<%(levelname)s> '
|
||||
'%(message)s'
|
||||
)
|
||||
format_date = "%Y-%m-%d %H:%M:%S"
|
||||
# color or not
|
||||
if self.log_settings['console_color_output_enabled']:
|
||||
formatter_console = CustomConsoleFormatter(format_string, datefmt=format_date)
|
||||
else:
|
||||
formatter_console = logging.Formatter(format_string, datefmt=format_date)
|
||||
console_handler.setLevel(log_level_console)
|
||||
# do not show exceptions logs on console
|
||||
console_handler.addFilter(self.__filter_exceptions)
|
||||
console_handler.setFormatter(formatter_console)
|
||||
self.logger.addHandler(console_handler)
|
||||
return console_handler
|
||||
|
||||
def __file_handler(self, log_level_file: str, log_path: Path) -> None:
|
||||
def __file_handler(self, log_level_file: str, log_path: Path) -> logging.handlers.TimedRotatingFileHandler:
|
||||
# file logger
|
||||
if not isinstance(getattr(logging, log_level_file.upper(), None), int):
|
||||
log_level_file = 'DEBUG'
|
||||
@@ -75,17 +224,84 @@ class Log:
|
||||
)
|
||||
formatter_file_handler = logging.Formatter(
|
||||
(
|
||||
# time stamp
|
||||
'[%(asctime)s.%(msecs)03d] '
|
||||
'[%(name)s:%(process)d] '
|
||||
# log name
|
||||
'[%(name)s] '
|
||||
# filename + pid
|
||||
'[%(filename)s:%(process)d] '
|
||||
# path + func + line number
|
||||
'[%(pathname)s:%(funcName)s:%(lineno)d] '
|
||||
# error level
|
||||
'<%(levelname)s> '
|
||||
# message
|
||||
'%(message)s'
|
||||
),
|
||||
datefmt="%Y-%m-%dT%H:%M:%S",
|
||||
)
|
||||
file_handler.setLevel(log_level_file)
|
||||
file_handler.setFormatter(formatter_file_handler)
|
||||
self.logger.addHandler(file_handler)
|
||||
return file_handler
|
||||
|
||||
def __init_listener(self, log_queue: 'Queue[str] | None' = None):
|
||||
"""
|
||||
If we have a Queue option start the logging queue
|
||||
|
||||
Keyword Arguments:
|
||||
log_queue {Queue[str] | None} -- _description_ (default: {None})
|
||||
"""
|
||||
if log_queue is None:
|
||||
return
|
||||
self.log_queue = log_queue
|
||||
self.listener = logging.handlers.QueueListener(
|
||||
self.log_queue,
|
||||
*self.handlers,
|
||||
respect_handler_level=True
|
||||
)
|
||||
self.listener.start()
|
||||
|
||||
def __init_log(self, log_name: str) -> None:
|
||||
"""
|
||||
Initialize the main loggger
|
||||
"""
|
||||
queue_handler: logging.handlers.QueueHandler | None = None
|
||||
if self.log_queue is not None:
|
||||
queue_handler = logging.handlers.QueueHandler(self.log_queue)
|
||||
# overall logger settings
|
||||
self.logger = logging.getLogger(log_name)
|
||||
# add all the handlers
|
||||
if queue_handler is None:
|
||||
for handler in self.handlers:
|
||||
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)
|
||||
|
||||
@staticmethod
|
||||
def init_worker_logging(log_queue: 'Queue[str]') -> logging.Logger:
|
||||
"""
|
||||
This initalizes a logger that can be used in pool/thread queue calls
|
||||
"""
|
||||
queue_handler = logging.handlers.QueueHandler(log_queue)
|
||||
# getLogger call MUST be WITHOUT and logger name
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(logging.DEBUG)
|
||||
root_logger.handlers.clear()
|
||||
root_logger.addHandler(queue_handler)
|
||||
|
||||
# for debug only
|
||||
root_logger.debug('[LOGGER] Init log: %s - %s', log_queue, root_logger.handlers)
|
||||
|
||||
return root_logger
|
||||
|
||||
def stop_listener(self):
|
||||
"""
|
||||
stop the listener
|
||||
"""
|
||||
if self.listener is not None:
|
||||
self.listener.stop()
|
||||
|
||||
def break_line(self, info: str = "BREAK"):
|
||||
"""
|
||||
@@ -94,7 +310,7 @@ class Log:
|
||||
Keyword Arguments:
|
||||
info {str} -- _description_ (default: {"BREAK"})
|
||||
"""
|
||||
self.logger.info("[%s] ================================>", info)
|
||||
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:
|
||||
"""
|
||||
@@ -105,18 +321,40 @@ class Log:
|
||||
*args (object): arguments for msg
|
||||
extra: Mapping[str, object] | None: extra arguments for the formatting if needed
|
||||
"""
|
||||
self.logger.log(Log.EXCEPTION, msg, *args, exc_info=True, extra=extra)
|
||||
self.logger.log(LoggingLevel.EXCEPTION.value, msg, *args, exc_info=True, extra=extra)
|
||||
|
||||
def validate_log_level(self, log_level: str) -> bool:
|
||||
@staticmethod
|
||||
def validate_log_level(log_level: Any) -> bool:
|
||||
"""
|
||||
if the log level is invalid, will erturn false
|
||||
|
||||
Args:
|
||||
log_level (str): _description_
|
||||
log_level (Any): _description_
|
||||
|
||||
Returns:
|
||||
bool: _description_
|
||||
"""
|
||||
return isinstance(getattr(logging, log_level.upper(), None), int)
|
||||
try:
|
||||
_ = LoggingLevel.from_any(log_level).value
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def get_log_level_int(log_level: Any) -> int:
|
||||
"""
|
||||
Return log level as INT
|
||||
If invalid return set level in default log level
|
||||
|
||||
Arguments:
|
||||
log_level {Any} -- _description_
|
||||
|
||||
Returns:
|
||||
int -- _description_
|
||||
"""
|
||||
try:
|
||||
return LoggingLevel.from_any(log_level).value
|
||||
except ValueError:
|
||||
return LoggingLevel.from_string(Log.DEFAULT_LOG_LEVEL).value
|
||||
|
||||
# __END__
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
"""
|
||||
All logging levels
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class LoggingLevel(Enum):
|
||||
"""
|
||||
Log class levels
|
||||
"""
|
||||
NOTSET = logging.NOTSET # 0
|
||||
DEBUG = logging.DEBUG # 10
|
||||
INFO = logging.INFO # 20
|
||||
WARNING = logging.WARNING # 30
|
||||
ERROR = logging.ERROR # 40
|
||||
CRITICAL = logging.CRITICAL # 50
|
||||
EXCEPTION = 60 # 60 (manualy set)
|
||||
# Alternative names
|
||||
WARN = logging.WARN # 30 (alias for WARNING)
|
||||
FATAL = logging.FATAL # 50 (alias for CRITICAL)
|
||||
|
||||
# Optional: Add string representation for better readability
|
||||
@classmethod
|
||||
def from_string(cls, level_str: str):
|
||||
"""Convert string to LogLevel enum"""
|
||||
try:
|
||||
return cls[level_str.upper()]
|
||||
except KeyError as e:
|
||||
raise ValueError(f"Invalid log level: {level_str}") from e
|
||||
|
||||
@classmethod
|
||||
def from_int(cls, level_int: int):
|
||||
"""Convert integer to LogLevel enum"""
|
||||
try:
|
||||
return cls(level_int)
|
||||
except ValueError as e:
|
||||
raise ValueError(f"Invalid log level: {level_int}") from e
|
||||
|
||||
@classmethod
|
||||
def from_any(cls, level_any: Any):
|
||||
"""Convert either str or int"""
|
||||
if isinstance(level_any, int):
|
||||
return cls.from_int(level_any)
|
||||
return cls.from_string(level_any)
|
||||
|
||||
def to_logging_level(self):
|
||||
"""Convert to logging module level"""
|
||||
return self.value
|
||||
|
||||
def to_lower_case(self):
|
||||
"""return loser case"""
|
||||
return self.name.lower()
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
def includes(self, level: 'LoggingLevel'):
|
||||
"""
|
||||
if given level is included in set level
|
||||
eg: INFO set, ERROR is included in INFO because INFO level would print ERROR
|
||||
"""
|
||||
return self.value <= level.value
|
||||
|
||||
def is_higher_than(self, level: 'LoggingLevel'):
|
||||
"""if given value is higher than set"""
|
||||
return self.value > level.value
|
||||
|
||||
def is_lower_than(self, level: 'LoggingLevel'):
|
||||
"""if given value is lower than set"""
|
||||
return self.value > level.value
|
||||
|
||||
# __END__
|
||||
@@ -2,6 +2,7 @@
|
||||
String helpers
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
from decimal import Decimal, getcontext
|
||||
from textwrap import shorten
|
||||
|
||||
@@ -101,4 +102,62 @@ 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__
|
||||
|
||||
@@ -10,40 +10,73 @@ class Colors:
|
||||
"""
|
||||
ANSI colors defined
|
||||
"""
|
||||
# General sets
|
||||
bold = '\033[1m'
|
||||
underline = '\033[4m'
|
||||
end = '\033[0m'
|
||||
reset = '\033[0m'
|
||||
# General sets, these should not be accessd
|
||||
__BOLD = '\033[1m'
|
||||
__UNDERLINE = '\033[4m'
|
||||
__END = '\033[0m'
|
||||
__RESET = '\033[0m'
|
||||
# Define ANSI color codes as class attributes
|
||||
black = "\033[30m"
|
||||
red = "\033[31m"
|
||||
green = "\033[32m"
|
||||
yellow = "\033[33m"
|
||||
blue = "\033[34m"
|
||||
magenta = "\033[35m"
|
||||
cyan = "\033[36m"
|
||||
white = "\033[37m"
|
||||
__BLACK = "\033[30m"
|
||||
__RED = "\033[31m"
|
||||
__GREEN = "\033[32m"
|
||||
__YELLOW = "\033[33m"
|
||||
__BLUE = "\033[34m"
|
||||
__MAGENTA = "\033[35m"
|
||||
__CYAN = "\033[36m"
|
||||
__WHITE = "\033[37m"
|
||||
|
||||
# Define bold/bright versions of the colors
|
||||
black_bold = "\033[1;30m"
|
||||
red_bold = "\033[1;31m"
|
||||
green_bold = "\033[1;32m"
|
||||
yellow_bold = "\033[1;33m"
|
||||
blue_bold = "\033[1;34m"
|
||||
magenta_bold = "\033[1;35m"
|
||||
cyan_bold = "\033[1;36m"
|
||||
white_bold = "\033[1;37m"
|
||||
__BLACK_BOLD = "\033[1;30m"
|
||||
__RED_BOLD = "\033[1;31m"
|
||||
__GREEN_BOLD = "\033[1;32m"
|
||||
__YELLOW_BOLD = "\033[1;33m"
|
||||
__BLUE_BOLD = "\033[1;34m"
|
||||
__MAGENTA_BOLD = "\033[1;35m"
|
||||
__CYAN_BOLD = "\033[1;36m"
|
||||
__WHITE_BOLD = "\033[1;37m"
|
||||
|
||||
# BRIGHT, alternative
|
||||
black_bright = '\033[90m'
|
||||
red_bright = '\033[91m'
|
||||
green_bright = '\033[92m'
|
||||
yellow_bright = '\033[93m'
|
||||
blue_bright = '\033[94m'
|
||||
magenta_bright = '\033[95m'
|
||||
cyan_bright = '\033[96m'
|
||||
white_bright = '\033[97m'
|
||||
__BLACK_BRIGHT = '\033[90m'
|
||||
__RED_BRIGHT = '\033[91m'
|
||||
__GREEN_BRIGHT = '\033[92m'
|
||||
__YELLOW_BRIGHT = '\033[93m'
|
||||
__BLUE_BRIGHT = '\033[94m'
|
||||
__MAGENTA_BRIGHT = '\033[95m'
|
||||
__CYAN_BRIGHT = '\033[96m'
|
||||
__WHITE_BRIGHT = '\033[97m'
|
||||
|
||||
# set access vars
|
||||
bold = __BOLD
|
||||
underline = __UNDERLINE
|
||||
end = __END
|
||||
reset = __RESET
|
||||
# normal
|
||||
black = __BLACK
|
||||
red = __RED
|
||||
green = __GREEN
|
||||
yellow = __YELLOW
|
||||
blue = __BLUE
|
||||
magenta = __MAGENTA
|
||||
cyan = __CYAN
|
||||
white = __WHITE
|
||||
# bold
|
||||
black_bold = __BLACK_BOLD
|
||||
red_bold = __RED_BOLD
|
||||
green_bold = __GREEN_BOLD
|
||||
yellow_bold = __YELLOW_BOLD
|
||||
blue_bold = __BLUE_BOLD
|
||||
magenta_bold = __MAGENTA_BOLD
|
||||
cyan_bold = __CYAN_BOLD
|
||||
white_bold = __WHITE_BOLD
|
||||
# bright
|
||||
black_bright = __BLACK_BRIGHT
|
||||
red_bright = __RED_BRIGHT
|
||||
green_bright = __GREEN_BRIGHT
|
||||
yellow_bright = __YELLOW_BRIGHT
|
||||
blue_bright = __BLUE_BRIGHT
|
||||
magenta_bright = __MAGENTA_BRIGHT
|
||||
cyan_bright = __CYAN_BRIGHT
|
||||
white_bright = __WHITE_BRIGHT
|
||||
|
||||
@staticmethod
|
||||
def disable():
|
||||
@@ -81,4 +114,43 @@ class Colors:
|
||||
Colors.magenta_bright = ''
|
||||
Colors.cyan_bright = ''
|
||||
Colors.white_bright = ''
|
||||
|
||||
@staticmethod
|
||||
def reset_colors():
|
||||
"""
|
||||
reset colors to the original ones
|
||||
"""
|
||||
# set access vars
|
||||
Colors.bold = Colors.__BOLD
|
||||
Colors.underline = Colors.__UNDERLINE
|
||||
Colors.end = Colors.__END
|
||||
Colors.reset = Colors.__RESET
|
||||
# normal
|
||||
Colors.black = Colors.__BLACK
|
||||
Colors.red = Colors.__RED
|
||||
Colors.green = Colors.__GREEN
|
||||
Colors.yellow = Colors.__YELLOW
|
||||
Colors.blue = Colors.__BLUE
|
||||
Colors.magenta = Colors.__MAGENTA
|
||||
Colors.cyan = Colors.__CYAN
|
||||
Colors.white = Colors.__WHITE
|
||||
# bold
|
||||
Colors.black_bold = Colors.__BLACK_BOLD
|
||||
Colors.red_bold = Colors.__RED_BOLD
|
||||
Colors.green_bold = Colors.__GREEN_BOLD
|
||||
Colors.yellow_bold = Colors.__YELLOW_BOLD
|
||||
Colors.blue_bold = Colors.__BLUE_BOLD
|
||||
Colors.magenta_bold = Colors.__MAGENTA_BOLD
|
||||
Colors.cyan_bold = Colors.__CYAN_BOLD
|
||||
Colors.white_bold = Colors.__WHITE_BOLD
|
||||
# bright
|
||||
Colors.black_bright = Colors.__BLACK_BRIGHT
|
||||
Colors.red_bright = Colors.__RED_BRIGHT
|
||||
Colors.green_bright = Colors.__GREEN_BRIGHT
|
||||
Colors.yellow_bright = Colors.__YELLOW_BRIGHT
|
||||
Colors.blue_bright = Colors.__BLUE_BRIGHT
|
||||
Colors.magenta_bright = Colors.__MAGENTA_BRIGHT
|
||||
Colors.cyan_bright = Colors.__CYAN_BRIGHT
|
||||
Colors.white_bright = Colors.__WHITE_BRIGHT
|
||||
|
||||
# __END__
|
||||
|
||||
@@ -11,11 +11,11 @@ class TimestampStrings:
|
||||
set default time stamps
|
||||
"""
|
||||
|
||||
time_zone: str = 'Asia/Tokyo'
|
||||
TIME_ZONE: str = 'Asia/Tokyo'
|
||||
|
||||
def __init__(self, time_zone: str | None = None):
|
||||
self.timestamp_now = datetime.now()
|
||||
self.time_zone = time_zone if time_zone is not None else __class__.time_zone
|
||||
self.time_zone = time_zone if time_zone is not None else self.TIME_ZONE
|
||||
try:
|
||||
self.timestamp_now_tz = datetime.now(ZoneInfo(self.time_zone))
|
||||
except ZoneInfoNotFoundError as e:
|
||||
|
||||
23
test-run/config_handling/config/settings.ini
Normal file
23
test-run/config_handling/config/settings.ini
Normal file
@@ -0,0 +1,23 @@
|
||||
[TestA]
|
||||
foo=bar
|
||||
foobar=1
|
||||
some_match=foo
|
||||
some_match_list=foo,bar
|
||||
test_list=a,b,c,d f, g h
|
||||
other_list=a|b|c|d|
|
||||
third_list=xy|ab|df|fg
|
||||
str_length=foobar
|
||||
int_range=20
|
||||
#
|
||||
match_target=foo
|
||||
match_target_list=foo,bar,baz
|
||||
#
|
||||
match_source_a=foo
|
||||
match_source_b=foo
|
||||
; match_source_c=foo
|
||||
match_source_list=foo,bar
|
||||
|
||||
[TestB]
|
||||
element_a=Static energy
|
||||
element_b=123.5
|
||||
element_c=True
|
||||
2
test-run/config_handling/log/.gitignore
vendored
Normal file
2
test-run/config_handling/log/.gitignore
vendored
Normal file
@@ -0,0 +1,2 @@
|
||||
*
|
||||
!.gitignore
|
||||
77
test-run/config_handling/settings_loader.py
Normal file
77
test-run/config_handling/settings_loader.py
Normal file
@@ -0,0 +1,77 @@
|
||||
"""
|
||||
Settings loader test
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from corelibs.iterator_handling.dump_data import dump_data
|
||||
from corelibs.logging_handling.log import Log
|
||||
from corelibs.config_handling.settings_loader import SettingsLoader
|
||||
|
||||
SCRIPT_PATH: Path = Path(__file__).resolve().parent
|
||||
ROOT_PATH: Path = SCRIPT_PATH
|
||||
CONFIG_DIR: Path = Path("config")
|
||||
CONFIG_FILE: str = "settings.ini"
|
||||
|
||||
|
||||
def main():
|
||||
"""
|
||||
Main run
|
||||
"""
|
||||
|
||||
# for log testing
|
||||
script_path: Path = Path(__file__).resolve().parent
|
||||
log = Log(
|
||||
log_path=script_path.joinpath('log', 'settings_loader.log'),
|
||||
log_name="Settings Loader",
|
||||
log_settings={
|
||||
"log_level_console": 'DEBUG',
|
||||
"log_level_file": 'DEBUG',
|
||||
}
|
||||
)
|
||||
log.logger.info('Settings loader')
|
||||
|
||||
sl = SettingsLoader(
|
||||
{
|
||||
'foo': 'OVERLOAD'
|
||||
},
|
||||
ROOT_PATH.joinpath(CONFIG_DIR, CONFIG_FILE),
|
||||
log=log
|
||||
)
|
||||
try:
|
||||
config_load = 'TestA'
|
||||
config_data = sl.load_settings(
|
||||
config_load,
|
||||
{
|
||||
"foo": ["mandatory:yes"],
|
||||
"foobar": ["check:int"],
|
||||
"some_match": ["matching:foo|bar"],
|
||||
"some_match_list": ["split:,", "matching:foo|bar"],
|
||||
"test_list": [
|
||||
"check:string.alphanumeric",
|
||||
"split:,"
|
||||
],
|
||||
"other_list": ["split:|"],
|
||||
"third_list": [
|
||||
"split:|",
|
||||
"check:string.alphanumeric"
|
||||
],
|
||||
"str_length": [
|
||||
"length:2-10"
|
||||
],
|
||||
"int_range": [
|
||||
"range:2-50"
|
||||
],
|
||||
"match_target": ["matching:foo"],
|
||||
"match_target_list": ["split:,", "matching:foo|bar|baz",],
|
||||
"match_source_a": ["in:match_target"],
|
||||
"match_source_b": ["in:match_target_list"],
|
||||
"match_source_list": ["split:,", "in:match_target_list"],
|
||||
}
|
||||
)
|
||||
print(f"Load: {config_load} -> {dump_data(config_data)}")
|
||||
except ValueError as e:
|
||||
print(f"Could not load settings: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
29
test-run/iterator_handling/list_helpers.py
Normal file
29
test-run/iterator_handling/list_helpers.py
Normal file
@@ -0,0 +1,29 @@
|
||||
"""
|
||||
test list helpers
|
||||
"""
|
||||
|
||||
from corelibs.iterator_handling.list_helpers import is_list_in_list, convert_to_list
|
||||
|
||||
|
||||
def __test_is_list_in_list_a():
|
||||
list_a = [1, "hello", 3.14, True, "world"]
|
||||
list_b = ["hello", True, 42]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
print(f"RESULT: {result}")
|
||||
|
||||
|
||||
def __convert_list():
|
||||
source = "hello"
|
||||
result = convert_to_list(source)
|
||||
print(f"IN: {source} -> {result}")
|
||||
|
||||
|
||||
def main():
|
||||
__test_is_list_in_list_a()
|
||||
__convert_list()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
# __END__
|
||||
73
test-run/logging_handling/log.py
Normal file
73
test-run/logging_handling/log.py
Normal file
@@ -0,0 +1,73 @@
|
||||
"""
|
||||
Log logging_handling.log testing
|
||||
"""
|
||||
|
||||
# import atexit
|
||||
from pathlib import Path
|
||||
# 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():
|
||||
"""
|
||||
Log testing
|
||||
"""
|
||||
script_path: Path = Path(__file__).resolve().parent
|
||||
log = Log(
|
||||
log_path=script_path.joinpath('log', 'test.log'),
|
||||
log_name="Test Log",
|
||||
log_settings={
|
||||
"log_level_console": 'DEBUG',
|
||||
"log_level_file": 'DEBUG',
|
||||
# "console_color_output_enabled": False,
|
||||
}
|
||||
)
|
||||
|
||||
log.logger.debug('[NORMAL] Debug test: %s', log.logger.name)
|
||||
log.logger.info('[NORMAL] Info test: %s', log.logger.name)
|
||||
log.logger.warning('[NORMAL] Warning test: %s', log.logger.name)
|
||||
log.logger.error('[NORMAL] Error test: %s', log.logger.name)
|
||||
log.logger.critical('[NORMAL] Critical test: %s', log.logger.name)
|
||||
log.exception('[NORMAL] Exception test: %s', log.logger.name)
|
||||
|
||||
bad_level = 'WRONG'
|
||||
if not Log.validate_log_level(bad_level):
|
||||
print(f"Invalid level: {bad_level}")
|
||||
good_level = 'WARNING'
|
||||
if Log.validate_log_level(good_level):
|
||||
print(f"Valid level: {good_level}")
|
||||
|
||||
print(f"ERROR is to_logging_level(): {LoggingLevel.ERROR.to_logging_level()}")
|
||||
print(f"ERROR is to_lower_case(): {LoggingLevel.ERROR.to_lower_case()}")
|
||||
print(f"ERROR is: {LoggingLevel.ERROR}")
|
||||
print(f"ERROR is value: {LoggingLevel.ERROR.value}")
|
||||
print(f"ERROR is name: {LoggingLevel.ERROR.name}")
|
||||
print(f"ERROR is from_string(lower): {LoggingLevel.from_string('ERROR')}")
|
||||
print(f"ERROR is from_string(upper): {LoggingLevel.from_string('ERROR')}")
|
||||
print(f"ERROR is from_int: {LoggingLevel.from_int(40)}")
|
||||
print(f"ERROR is from_any(text lower): {LoggingLevel.from_any('ERROR')}")
|
||||
print(f"ERROR is from_any(text upper): {LoggingLevel.from_any('ERROR')}")
|
||||
print(f"ERROR is from_any(int): {LoggingLevel.from_any(40)}")
|
||||
print(f"INFO <= ERROR: {LoggingLevel.INFO.includes(LoggingLevel.ERROR)}")
|
||||
print(f"INFO > ERROR: {LoggingLevel.INFO.is_higher_than(LoggingLevel.ERROR)}")
|
||||
print(f"INFO < ERROR: {LoggingLevel.INFO.is_lower_than(LoggingLevel.ERROR)}")
|
||||
print(f"INFO < ERROR: {LoggingLevel.INFO.is_lower_than(LoggingLevel.ERROR)}")
|
||||
|
||||
try:
|
||||
print(f"INVALID is A: {LoggingLevel.from_string('INVALID')}")
|
||||
except ValueError as e:
|
||||
print(f"* ERROR: {e}")
|
||||
|
||||
try:
|
||||
__test = 5 / 0
|
||||
print(f"Divied: {__test}")
|
||||
except ZeroDivisionError as e:
|
||||
log.logger.critical("Divison through zero: %s", e)
|
||||
log.exception("Divison through zero")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
# __END__
|
||||
2
test-run/logging_handling/log/.gitignore
vendored
Normal file
2
test-run/logging_handling/log/.gitignore
vendored
Normal file
@@ -0,0 +1,2 @@
|
||||
*
|
||||
!.gitignore
|
||||
75
test-run/logging_handling/log_pool.py
Normal file
75
test-run/logging_handling/log_pool.py
Normal file
@@ -0,0 +1,75 @@
|
||||
"""
|
||||
Pool Queue log handling
|
||||
Thread Queue log handling
|
||||
"""
|
||||
|
||||
import random
|
||||
import time
|
||||
from multiprocessing import Queue
|
||||
import concurrent.futures
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from corelibs.logging_handling.log import Log
|
||||
|
||||
|
||||
def work_function(log_name: str, worker_id: int, data: list[int]) -> int:
|
||||
"""
|
||||
simulate worker
|
||||
|
||||
Arguments:
|
||||
worker_id {int} -- _description_
|
||||
data {list[int]} -- _description_
|
||||
|
||||
Returns:
|
||||
int -- _description_
|
||||
"""
|
||||
log = logging.getLogger(f'{log_name}-WorkerFn-{worker_id}')
|
||||
log.info('Starting worker: %s', worker_id)
|
||||
time.sleep(random.uniform(1, 3))
|
||||
result = sum(data) * worker_id
|
||||
return result
|
||||
|
||||
|
||||
def main():
|
||||
"""
|
||||
Queue log tester
|
||||
"""
|
||||
print("[START] Queue logger test")
|
||||
log_queue: 'Queue[str]' = Queue()
|
||||
script_path: Path = Path(__file__).resolve().parent
|
||||
log = Log(
|
||||
log_path=script_path.joinpath('log', 'test.log'),
|
||||
log_name="Test Log",
|
||||
log_settings={
|
||||
"log_level_console": 'INFO',
|
||||
"log_level_file": 'INFO',
|
||||
"log_queue": log_queue,
|
||||
}
|
||||
)
|
||||
log.logger.info('Pool Fork logging test')
|
||||
max_forks = 2
|
||||
data_sets = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
|
||||
with concurrent.futures.ProcessPoolExecutor(
|
||||
max_workers=max_forks,
|
||||
initializer=Log.init_worker_logging,
|
||||
initargs=(log_queue,)
|
||||
) as executor:
|
||||
log.logger.info('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('Workders started')
|
||||
|
||||
for future in concurrent.futures.as_completed(futures):
|
||||
log.logger.info('Processing result: %s', future.result())
|
||||
print(f"Processing result: {future.result()}")
|
||||
|
||||
log.logger.info('[END] Queue logger test')
|
||||
log.stop_listener()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
# __END__
|
||||
42
test-run/logging_handling/log_queue.py
Normal file
42
test-run/logging_handling/log_queue.py
Normal file
@@ -0,0 +1,42 @@
|
||||
"""
|
||||
Log logging_handling.log testing
|
||||
"""
|
||||
|
||||
# import atexit
|
||||
from pathlib import Path
|
||||
from multiprocessing import Queue
|
||||
# this is for testing only
|
||||
from corelibs.logging_handling.log import Log
|
||||
|
||||
|
||||
def main():
|
||||
"""
|
||||
Log testing
|
||||
"""
|
||||
script_path: Path = Path(__file__).resolve().parent
|
||||
|
||||
log_queue: 'Queue[str]' = Queue()
|
||||
log_q = Log(
|
||||
log_path=script_path.joinpath('log', 'test_queue.log'),
|
||||
log_name="Test Log",
|
||||
log_settings={
|
||||
"log_level_console": 'WARNING',
|
||||
"log_level_file": 'ERROR',
|
||||
"log_queue": log_queue
|
||||
# "console_color_output_enabled": False,
|
||||
}
|
||||
)
|
||||
|
||||
log_q.logger.debug('[QUEUE] Debug test: %s', log_q.logger.name)
|
||||
log_q.logger.info('[QUEUE] Info test: %s', log_q.logger.name)
|
||||
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.stop_listener()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
# __END__
|
||||
31
test-run/logging_handling/log_queue_legacy.py
Normal file
31
test-run/logging_handling/log_queue_legacy.py
Normal file
@@ -0,0 +1,31 @@
|
||||
"""
|
||||
Log logging_handling.log testing
|
||||
"""
|
||||
|
||||
# import atexit
|
||||
from pathlib import Path
|
||||
from multiprocessing import Queue
|
||||
# this is for testing only
|
||||
from queue_logger.log_queue import QueueLogger
|
||||
|
||||
|
||||
def main():
|
||||
"""
|
||||
Log testing
|
||||
"""
|
||||
script_path: Path = Path(__file__).resolve().parent
|
||||
|
||||
log_queue: 'Queue[str]' = Queue()
|
||||
log_q_legacy = QueueLogger(
|
||||
log_file=script_path.joinpath('log', 'test_queue_legacy.log'),
|
||||
log_name="Test Log Queue",
|
||||
log_queue=log_queue
|
||||
)
|
||||
log_q_legacy.mlog.info('Log test: %s', 'Queue Legacy')
|
||||
# log_q.stop_listener()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
# __END__
|
||||
96
test-run/logging_handling/queue_logger/log_queue.py
Normal file
96
test-run/logging_handling/queue_logger/log_queue.py
Normal file
@@ -0,0 +1,96 @@
|
||||
"""
|
||||
test queue logger interface
|
||||
NOTE: this has all moved to the default log interface
|
||||
"""
|
||||
|
||||
import logging
|
||||
import logging.handlers
|
||||
from pathlib import Path
|
||||
from multiprocessing import Queue
|
||||
|
||||
|
||||
class QueueLogger:
|
||||
"""
|
||||
Queue logger
|
||||
"""
|
||||
|
||||
def __init__(self, log_file: Path, log_name: str, log_queue: 'Queue[str] | None' = None):
|
||||
self.log_file = log_file
|
||||
self.log_name = log_name
|
||||
self.handlers = self.setup_logging()
|
||||
self.log_queue: 'Queue[str]' = log_queue if log_queue is not None else Queue()
|
||||
self.listener = logging.handlers.QueueListener(self.log_queue, *self.handlers)
|
||||
self.listener.start()
|
||||
|
||||
self.mlog: logging.Logger = self.main_log(log_name)
|
||||
|
||||
def __del__(self):
|
||||
self.mlog.info("[%s] ================================>", "END")
|
||||
self.listener.stop()
|
||||
|
||||
def setup_logging(self):
|
||||
"""
|
||||
setup basic logging
|
||||
"""
|
||||
|
||||
# Create formatters
|
||||
file_formatter = logging.Formatter(
|
||||
'%(asctime)s - %(name)s - %(levelname)s - [PID:%(process)d] [%(filename)s:%(lineno)d] - %(message)s'
|
||||
)
|
||||
|
||||
console_formatter = logging.Formatter(
|
||||
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
|
||||
# Create handlers
|
||||
file_handler = logging.FileHandler(self.log_file)
|
||||
file_handler.setFormatter(file_formatter)
|
||||
file_handler.setLevel(logging.DEBUG)
|
||||
|
||||
console_handler = logging.StreamHandler()
|
||||
console_handler.setFormatter(console_formatter)
|
||||
console_handler.setLevel(logging.DEBUG)
|
||||
|
||||
return [file_handler, console_handler]
|
||||
|
||||
def main_log(self, log_name: str) -> logging.Logger:
|
||||
"""
|
||||
main logger
|
||||
|
||||
Arguments:
|
||||
log_name {str} -- _description_
|
||||
|
||||
Returns:
|
||||
logging.Logger -- _description_
|
||||
"""
|
||||
mlog_handler = logging.handlers.QueueHandler(self.log_queue)
|
||||
mlog = logging.getLogger(f'{log_name}-MainProcess')
|
||||
mlog.addHandler(mlog_handler)
|
||||
mlog.setLevel(logging.DEBUG)
|
||||
return mlog
|
||||
|
||||
@staticmethod
|
||||
def init_worker_logging(log_queue: 'Queue[str]', log_name: str, ):
|
||||
"""
|
||||
Initialize logging for worker processes
|
||||
"""
|
||||
|
||||
# Create QueueHandler
|
||||
queue_handler = logging.handlers.QueueHandler(log_queue)
|
||||
|
||||
# Setup root logger for this process
|
||||
# NOTE: This must be EMPTY or new SINGLE NEW logger is created, we need one for EACH fork
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(logging.DEBUG)
|
||||
root_logger.handlers.clear()
|
||||
root_logger.addHandler(queue_handler)
|
||||
|
||||
root_logger.info('[LOGGER] Init log: %s - %s', log_queue, log_name)
|
||||
|
||||
return root_logger
|
||||
|
||||
def stop_listener(self):
|
||||
"""
|
||||
stop the listener
|
||||
"""
|
||||
self.listener.stop()
|
||||
@@ -52,7 +52,7 @@ def __sh_format_number():
|
||||
print(f"Format {number} ({precision}) -> {result}")
|
||||
|
||||
|
||||
def _sh_colors():
|
||||
def __sh_colors():
|
||||
for color in [
|
||||
"black",
|
||||
"red",
|
||||
@@ -79,7 +79,7 @@ def main():
|
||||
"""
|
||||
__sh_shorten_string()
|
||||
__sh_format_number()
|
||||
_sh_colors()
|
||||
__sh_colors()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
0
tests/integration/__init__.py
Normal file
0
tests/integration/__init__.py
Normal file
300
tests/unit/iterator_handling/test_list_helpers.py
Normal file
300
tests/unit/iterator_handling/test_list_helpers.py
Normal file
@@ -0,0 +1,300 @@
|
||||
"""
|
||||
iterator_handling.list_helepr tests
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
import pytest
|
||||
from corelibs.iterator_handling.list_helpers import convert_to_list, is_list_in_list
|
||||
|
||||
|
||||
class TestConvertToList:
|
||||
"""Test cases for convert_to_list function"""
|
||||
|
||||
def test_string_input(self):
|
||||
"""Test with string inputs"""
|
||||
assert convert_to_list("hello") == ["hello"]
|
||||
assert convert_to_list("") == [""]
|
||||
assert convert_to_list("123") == ["123"]
|
||||
assert convert_to_list("true") == ["true"]
|
||||
|
||||
def test_integer_input(self):
|
||||
"""Test with integer inputs"""
|
||||
assert convert_to_list(42) == [42]
|
||||
assert convert_to_list(0) == [0]
|
||||
assert convert_to_list(-10) == [-10]
|
||||
assert convert_to_list(999999) == [999999]
|
||||
|
||||
def test_float_input(self):
|
||||
"""Test with float inputs"""
|
||||
assert convert_to_list(3.14) == [3.14]
|
||||
assert convert_to_list(0.0) == [0.0]
|
||||
assert convert_to_list(-2.5) == [-2.5]
|
||||
assert convert_to_list(1.0) == [1.0]
|
||||
|
||||
def test_boolean_input(self):
|
||||
"""Test with boolean inputs"""
|
||||
assert convert_to_list(True) == [True]
|
||||
assert convert_to_list(False) == [False]
|
||||
|
||||
def test_list_input_unchanged(self):
|
||||
"""Test that list inputs are returned unchanged"""
|
||||
# String lists
|
||||
str_list = ["a", "b", "c"]
|
||||
assert convert_to_list(str_list) == str_list
|
||||
assert convert_to_list(str_list) is str_list # Same object reference
|
||||
|
||||
# Integer lists
|
||||
int_list = [1, 2, 3]
|
||||
assert convert_to_list(int_list) == int_list
|
||||
assert convert_to_list(int_list) is int_list
|
||||
|
||||
# Float lists
|
||||
float_list = [1.1, 2.2, 3.3]
|
||||
assert convert_to_list(float_list) == float_list
|
||||
assert convert_to_list(float_list) is float_list
|
||||
|
||||
# Boolean lists
|
||||
bool_list = [True, False, True]
|
||||
assert convert_to_list(bool_list) == bool_list
|
||||
assert convert_to_list(bool_list) is bool_list
|
||||
|
||||
# Mixed lists
|
||||
mixed_list = [1, "hello", 3.14, True]
|
||||
assert convert_to_list(mixed_list) == mixed_list
|
||||
assert convert_to_list(mixed_list) is mixed_list
|
||||
|
||||
# Empty list
|
||||
empty_list: list[int] = []
|
||||
assert convert_to_list(empty_list) == empty_list
|
||||
assert convert_to_list(empty_list) is empty_list
|
||||
|
||||
def test_nested_lists(self):
|
||||
"""Test with nested lists (should still return the same list)"""
|
||||
nested_list: list[list[int]] = [[1, 2], [3, 4]]
|
||||
assert convert_to_list(nested_list) == nested_list
|
||||
assert convert_to_list(nested_list) is nested_list
|
||||
|
||||
def test_single_element_lists(self):
|
||||
"""Test with single element lists"""
|
||||
single_str = ["hello"]
|
||||
assert convert_to_list(single_str) == single_str
|
||||
assert convert_to_list(single_str) is single_str
|
||||
|
||||
single_int = [42]
|
||||
assert convert_to_list(single_int) == single_int
|
||||
assert convert_to_list(single_int) is single_int
|
||||
|
||||
|
||||
class TestIsListInList:
|
||||
"""Test cases for is_list_in_list function"""
|
||||
|
||||
def test_string_lists(self):
|
||||
"""Test with string lists"""
|
||||
list_a = ["a", "b", "c", "d"]
|
||||
list_b = ["b", "d", "e"]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert set(result) == {"a", "c"}
|
||||
assert isinstance(result, list)
|
||||
|
||||
def test_integer_lists(self):
|
||||
"""Test with integer lists"""
|
||||
list_a = [1, 2, 3, 4, 5]
|
||||
list_b = [2, 4, 6]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert set(result) == {1, 3, 5}
|
||||
assert isinstance(result, list)
|
||||
|
||||
def test_float_lists(self):
|
||||
"""Test with float lists"""
|
||||
list_a = [1.1, 2.2, 3.3, 4.4]
|
||||
list_b = [2.2, 4.4, 5.5]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert set(result) == {1.1, 3.3}
|
||||
assert isinstance(result, list)
|
||||
|
||||
def test_boolean_lists(self):
|
||||
"""Test with boolean lists"""
|
||||
list_a = [True, False, True]
|
||||
list_b = [True]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert set(result) == {False}
|
||||
assert isinstance(result, list)
|
||||
|
||||
def test_mixed_type_lists(self):
|
||||
"""Test with mixed type lists"""
|
||||
list_a = [1, "hello", 3.14, True, "world"]
|
||||
list_b = ["hello", True, 42]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert set(result) == {1, 3.14, "world"}
|
||||
assert isinstance(result, list)
|
||||
|
||||
def test_empty_lists(self):
|
||||
"""Test with empty lists"""
|
||||
# Empty list_a
|
||||
assert is_list_in_list([], [1, 2, 3]) == []
|
||||
|
||||
# Empty list_b
|
||||
list_a = [1, 2, 3]
|
||||
result = is_list_in_list(list_a, [])
|
||||
assert set(result) == {1, 2, 3}
|
||||
|
||||
# Both empty
|
||||
assert is_list_in_list([], []) == []
|
||||
|
||||
def test_no_common_elements(self):
|
||||
"""Test when lists have no common elements"""
|
||||
list_a = [1, 2, 3]
|
||||
list_b = [4, 5, 6]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert set(result) == {1, 2, 3}
|
||||
|
||||
def test_all_elements_common(self):
|
||||
"""Test when all elements in list_a are in list_b"""
|
||||
list_a = [1, 2, 3]
|
||||
list_b = [1, 2, 3, 4, 5]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert result == []
|
||||
|
||||
def test_identical_lists(self):
|
||||
"""Test with identical lists"""
|
||||
list_a = [1, 2, 3]
|
||||
list_b = [1, 2, 3]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert result == []
|
||||
|
||||
def test_duplicate_elements(self):
|
||||
"""Test with duplicate elements in lists"""
|
||||
list_a = [1, 2, 2, 3, 3, 3]
|
||||
list_b = [2, 4]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
# Should return unique elements only (set behavior)
|
||||
assert set(result) == {1, 3}
|
||||
assert isinstance(result, list)
|
||||
|
||||
def test_list_b_larger_than_list_a(self):
|
||||
"""Test when list_b is larger than list_a"""
|
||||
list_a = [1, 2]
|
||||
list_b = [2, 3, 4, 5, 6, 7, 8]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert set(result) == {1}
|
||||
|
||||
def test_order_independence(self):
|
||||
"""Test that order doesn't matter due to set operations"""
|
||||
list_a = [3, 1, 4, 1, 5]
|
||||
list_b = [1, 2, 6]
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert set(result) == {3, 4, 5}
|
||||
|
||||
|
||||
# Parametrized tests for more comprehensive coverage
|
||||
class TestParametrized:
|
||||
"""Parametrized tests for better coverage"""
|
||||
|
||||
@pytest.mark.parametrize("input_value,expected", [
|
||||
("hello", ["hello"]),
|
||||
(42, [42]),
|
||||
(3.14, [3.14]),
|
||||
(True, [True]),
|
||||
(False, [False]),
|
||||
("", [""]),
|
||||
(0, [0]),
|
||||
(0.0, [0.0]),
|
||||
(-1, [-1]),
|
||||
(-2.5, [-2.5]),
|
||||
])
|
||||
def test_convert_to_list_parametrized(self, input_value: Any, expected: Any):
|
||||
"""Test convert_to_list with various single values"""
|
||||
assert convert_to_list(input_value) == expected
|
||||
|
||||
@pytest.mark.parametrize("input_list", [
|
||||
[1, 2, 3],
|
||||
["a", "b", "c"],
|
||||
[1.1, 2.2, 3.3],
|
||||
[True, False],
|
||||
[1, "hello", 3.14, True],
|
||||
[],
|
||||
[42],
|
||||
[[1, 2], [3, 4]],
|
||||
])
|
||||
def test_convert_to_list_with_lists_parametrized(self, input_list: Any):
|
||||
"""Test convert_to_list with various list inputs"""
|
||||
result = convert_to_list(input_list)
|
||||
assert result == input_list
|
||||
assert result is input_list # Same object reference
|
||||
|
||||
@pytest.mark.parametrize("list_a,list_b,expected_set", [
|
||||
([1, 2, 3], [2], {1, 3}),
|
||||
(["a", "b", "c"], ["b", "d"], {"a", "c"}),
|
||||
([1, 2, 3], [4, 5, 6], {1, 2, 3}),
|
||||
([1, 2, 3], [1, 2, 3], set()),
|
||||
([], [1, 2, 3], set()),
|
||||
([1, 2, 3], [], {1, 2, 3}),
|
||||
([True, False], [True], {False}),
|
||||
([1.1, 2.2, 3.3], [2.2], {1.1, 3.3}),
|
||||
])
|
||||
def test_is_list_in_list_parametrized(self, list_a: list[Any], list_b: list[Any], expected_set: Any):
|
||||
"""Test is_list_in_list with various input combinations"""
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
assert set(result) == expected_set
|
||||
assert isinstance(result, list)
|
||||
|
||||
|
||||
# Edge cases and special scenarios
|
||||
class TestEdgeCases:
|
||||
"""Test edge cases and special scenarios"""
|
||||
|
||||
def test_convert_to_list_with_none_like_values(self):
|
||||
"""Test convert_to_list with None-like values (if function supports them)"""
|
||||
# Note: Based on type hints, None is not supported, but testing behavior
|
||||
# This test might need to be adjusted based on actual function behavior
|
||||
pass
|
||||
|
||||
def test_is_list_in_list_preserves_type_distinctions(self):
|
||||
"""Test that different types are treated as different"""
|
||||
list_a = [1, "1", 1.0, True]
|
||||
list_b = [1] # Only integer 1
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
|
||||
# Note: This test depends on how Python's set handles type equality
|
||||
# 1, 1.0, and True are considered equal in sets
|
||||
# "1" is different from 1
|
||||
# expected_items = {"1"} # String "1" should remain
|
||||
assert "1" in result
|
||||
assert isinstance(result, list)
|
||||
|
||||
def test_large_lists(self):
|
||||
"""Test with large lists"""
|
||||
large_list_a = list(range(1000))
|
||||
large_list_b = list(range(500, 1500))
|
||||
result = is_list_in_list(large_list_a, large_list_b)
|
||||
expected = list(range(500)) # 0 to 499
|
||||
assert set(result) == set(expected)
|
||||
|
||||
def test_memory_efficiency(self):
|
||||
"""Test that convert_to_list doesn't create unnecessary copies"""
|
||||
original_list = [1, 2, 3, 4, 5]
|
||||
result = convert_to_list(original_list)
|
||||
|
||||
# Should be the same object, not a copy
|
||||
assert result is original_list
|
||||
|
||||
# Modifying the original should affect the result
|
||||
original_list.append(6)
|
||||
assert 6 in result
|
||||
|
||||
|
||||
# Performance tests (optional)
|
||||
class TestPerformance:
|
||||
"""Performance-related tests"""
|
||||
|
||||
def test_is_list_in_list_with_duplicates_performance(self):
|
||||
"""Test that function handles duplicates efficiently"""
|
||||
# List with many duplicates
|
||||
list_a = [1, 2, 3] * 100 # 300 elements, many duplicates
|
||||
list_b = [2] * 50 # 50 elements, all the same
|
||||
|
||||
result = is_list_in_list(list_a, list_b)
|
||||
|
||||
# Should still work correctly despite duplicates
|
||||
assert set(result) == {1, 3}
|
||||
assert isinstance(result, list)
|
||||
@@ -2,9 +2,12 @@
|
||||
PyTest: string_handling/string_helpers
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from textwrap import shorten
|
||||
from corelibs.string_handling.string_helpers import shorten_string, left_fill, format_number
|
||||
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
|
||||
)
|
||||
|
||||
|
||||
class TestShortenString:
|
||||
@@ -234,4 +237,238 @@ 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__
|
||||
|
||||
@@ -82,7 +82,7 @@ class TestTimestampStrings:
|
||||
|
||||
def test_class_default_timezone(self):
|
||||
"""Test that class default timezone is correctly set"""
|
||||
assert TimestampStrings.time_zone == 'Asia/Tokyo'
|
||||
assert TimestampStrings.TIME_ZONE == 'Asia/Tokyo'
|
||||
|
||||
def test_none_timezone_uses_default(self):
|
||||
"""Test that passing None for timezone uses class default"""
|
||||
|
||||
Reference in New Issue
Block a user