seb/2024-bsc-sebastian-lenzlinger
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Seperate out subcommands and utils; root logger setup begins in __init__ of iottb module

Browse Source
This commit is contained in:
Sebastian Lenzlinger 2024-06-27 23:01:43 +02:00
parent 18f80fc6fe
commit 00780b5ef4
9 changed files with 241 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
iottb/__init__.py
View File

@ -0,0 +1,8 @@
from iottb import definitions
import logging
logging.basicConfig(level=definitions.LOGLEVEL)
log_dir = definitions.LOGDIR
# Ensure logs dir exists before new handlers are registered in main.py
if not log_dir.is_dir():
log_dir.mkdir()

0
iottb/commands/__init__.py Normal file
View File

44
iottb/commands/initialize_testbed.py Normal file
View File

@ -0,0 +1,44 @@
import click
from pathlib import Path
import logging
from logging.handlers import RotatingFileHandler
import sys
from iottb.contexts import IottbConfig
from iottb.definitions import DB_NAME
logger = logging.getLogger(__name__)
@click.command()
@click.option('-d', '--dest', type=Path, help='Location to put (new) iottb database')
@click.option('--name', default=DB_NAME, type=str, help='Name of new database.')
@click.option('--update-default/--no-update-default', default=True, help='If new db should be set as the new default')
@click.pass_context
def init_db(ctx, dest, name, update_default):
logger.info('init-db invoked')
config = ctx.obj['CONFIG']
logger.debug(f'str(config)')
# Use the default path from config if dest is not provided
known_dbs = config.get_known_databases()
if name in known_dbs:
click.echo(f'A database {name} already exists.')
logger.info(f'Exiting...')
exit()
if not dest:
logger.info('No dest set, choosing default destination.')
dest = Path(config.default_path).parent
db_path = dest / name
logger.debug(f'Full path for db {db_path}')
# Create the directory if it doesn't exist
db_path.mkdir(parents=True, exist_ok=True)
logger.info(f"Created directory {db_path.parent} for the database")
# Update configuration
config.set_database_location(name, str(db_path))
if update_default:
config.set_default_database(name, str(db_path))
config.save_config()
logger.info(f"Updated configuration with database {name} at {db_path}")

43
iottb/contexts.py
View File

@ -6,12 +6,25 @@ logger = logging.getLogger(__name__)
DB_NAME = 'iottb.db'
class IottbConfig:
""" Class to handle operations on the testbed configuration.
TODO: Add instead of overwrite Database locations when initializing if a location with valid db
exists.
"""
@staticmethod
def warn():
logger.warning(f'DatabaseLocations are DatabaseLocationMap in the class {__name__}')
def __init__(self, cfg_file):
logger.info('Initializing Config object')
IottbConfig.warn()
self.cfg_file = cfg_file
self.default_database = None
self.default_path = None
self.database_locations = {}
self.DatabaseLocationMap = {}
self.load_config()
def create_default_config(self):
@ -19,14 +32,14 @@ class IottbConfig:
logger.info(f'Creating default config file at {self.cfg_file}')
self.default_database = DB_NAME
self.default_path = str(Path.home() / DB_NAME)
self.database_locations = {
self.DatabaseLocationMap = {
DB_NAME: self.default_path
}
defaults = {
'DefaultDatabase': self.default_database,
'DefaultDatabasePath': self.default_path,
'DatabaseLocations': self.database_locations
'DatabaseLocations': self.DatabaseLocationMap
}
try:
@ -39,21 +52,24 @@ class IottbConfig:
def load_config(self):
"""Loads or creates default configuration from given file path."""
logger.info('Loading configuration file')
if not self.cfg_file.is_file():
logger.info('Config file does not exist.')
self.create_default_config()
else:
logger.info('Config file exists, opening.')
with self.cfg_file.open('r') as config_file:
data = json.load(config_file)
self.default_database = data.get('DefaultDatabase')
self.default_path = data.get('DefaultDatabasePath')
self.database_locations = data.get('DatabaseLocations', {})
self.DatabaseLocationMap = data.get('DatabaseLocations', {})
def save_config(self):
"""Save the current configuration to the config file."""
data = {
'DefaultDatabase': self.default_database,
'DefaultDatabasePath': self.default_path,
'DatabaseLocations': self.database_locations
'DatabaseLocations': self.DatabaseLocationMap
}
try:
with self.cfg_file.open('w') as config_file:
@ -66,12 +82,23 @@ class IottbConfig:
"""Set the default database and its path."""
self.default_database = name
self.default_path = path
self.database_locations[name] = path
self.DatabaseLocationMap[name] = path
def get_database_location(self, name):
"""Get the location of a specific database."""
return self.database_locations.get(name)
return self.DatabaseLocationMap.get(name)
def set_database_location(self, name, path):
"""Set the location for a database."""
self.database_locations[name] = path
self.DatabaseLocationMap[name] = path
def get_known_databases(self):
"""Get the set of known databases"""
logger.info(f'Getting known databases.')
return self.DatabaseLocationMap.keys()
# TODO: Know issue:
class Database:
pass

24
iottb/definitions.py Normal file
View File

@ -0,0 +1,24 @@
import logging
from pathlib import Path
import click
APP_NAME = 'iottb'
DB_NAME = 'iottb.db'
CFG_FILE_PATH = str(Path(click.get_app_dir(APP_NAME)).joinpath('iottb.cfg'))
CONSOLE_LOG_FORMATS = {
0: '%(levelname)s - %(message)s',
1: '%(levelname)s - %(module)s - %(message)s',
2: '%(levelname)s - %(module)s - %(funcName)s - %(lineno)d - %(message)s'
}
LOGFILE_LOG_FORMAT = {
0: '%(levelname)s - %(asctime)s - %(module)s - %(message)s',
1: '%(levelname)s - %(asctime)s - %(module)s - %(funcName)s - %(message)s',
2: '%(levelname)s - %(asctime)s - %(module)s - %(funcName)s - %(lineno)d - %(message)s'
}
MAX_VERBOSITY = len(CONSOLE_LOG_FORMATS) - 1
assert len(LOGFILE_LOG_FORMAT) == len(CONSOLE_LOG_FORMATS), 'Log formats must be same size'
LOGLEVEL = logging.WARNING
LOGDIR = Path.cwd() / 'logs'

137
iottb/main.py
View File

@ -2,89 +2,29 @@ import json
import click
from pathlib import Path
import logging
from logging.handlers import RotatingFileHandler
import sys
##################################################
# Import package modules
#################################################
from iottb.utils.logger_config import setup_logging
from iottb import definitions
from iottb.contexts import IottbConfig
from iottb.commands.initialize_testbed import init_db
############################################################################
# Module shortcuts for global definitions
###########################################################################
APP_NAME = definitions.APP_NAME
DB_NAME = definitions.DB_NAME
CFG_FILE_PATH = definitions.CFG_FILE_PATH
# These are (possibly) redundant when defined in definitions.py
# keeping them here until refactored and tested
MAX_VERBOSITY = definitions.MAX_VERBOSITY
APP_NAME = 'iottb'
DB_NAME = 'iottb.db'
CFG_FILE_PATH = str(Path(click.get_app_dir(APP_NAME)).joinpath('iottb.cfg'))
CONSOLE_LOG_FORMATS = {
0: '%(levelname)s - %(message)s',
1: '%(levelname)s - %(module)s - %(message)s',
2: '%(levelname)s - %(asctime)s - %(module)s - %(funcName)s - %(lineno)d - %(message)s'
}
LOGFILE_LOG_FORMAT = {
0: '%(levelname)s - %(asctime)s - %(module)s - %(message)s',
1: '%(levelname)s - %(asctime)s - %(module)s - %(funcName)s - %(message)s',
2: '%(levelname)s - %(asctime)s - %(module)s - %(funcName)s - %(lineno)d - %(message)s'
}
MAX_VERBOSITY = len(CONSOLE_LOG_FORMATS) - 1
assert len(LOGFILE_LOG_FORMAT) == len(CONSOLE_LOG_FORMATS), 'Log formats must be same size'
# Logger stuff
loglevel = definitions.LOGLEVEL
logger = logging.getLogger(__name__)
def setup_logging(verbosity, debug):
""" Setup root logger for iottb """
log_level = logging.ERROR
handlers = []
date_format = '%Y-%m-%d %H:%M:%S'
if verbosity > 0:
log_level = logging.WARNING
if verbosity > MAX_VERBOSITY:
verbosity = MAX_VERBOSITY
log_level = logging.INFO
assert verbosity <= MAX_VERBOSITY, f'Verbosity must be <= {MAX_VERBOSITY}'
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(logging.Formatter(CONSOLE_LOG_FORMATS[verbosity], datefmt=date_format))
console_handler.setLevel(logging.DEBUG) # can keep at debug since it depends on global level?
handlers.append(console_handler)
if debug:
log_level = logging.DEBUG
# Logfile logs INFO+, no debugs though
file_handler = RotatingFileHandler(f'{APP_NAME}.log', maxBytes=10240, backupCount=5)
file_handler.setFormatter(logging.Formatter(LOGFILE_LOG_FORMAT[verbosity], datefmt=date_format))
file_handler.setLevel(logging.INFO)
# finnish root logger setup
handlers.append(file_handler)
# Force this config to be applied to root logger
logging.basicConfig(level=log_level, handlers=handlers, force=True)
def create_default_config(cfg_file):
"""Create default iottb config file."""
logger.info(f'Creating default config file at {cfg_file}')
defaults = {
'DefaultDatabase': DB_NAME,
'DefaultDatabasePath': str(Path.home())
}
try:
cfg_file.parent.mkdir(exist_ok=True)
with cfg_file.open('w') as config_file:
defaults_dict = json.dumps(defaults, indent=4)
config_file.write(defaults_dict)
except IOError as e:
logger.error(f"Failed to create default configuration file at {cfg_file}: {e}")
raise RuntimeError(f"Failed to create configuration file: {e}") from e
return defaults
def load_config(cfg_file):
"""Loads or creates default configuration from given file path."""
logger.info("Loading iottb config file")
if not cfg_file.is_file():
return create_default_config(cfg_file)
with open(cfg_file, 'r') as config_file:
return json.load(config_file)
@click.group()
@click.option('-v', '--verbosity', count=True, type=click.IntRange(0, 3), default=0,
help='Set verbosity')
@ -95,32 +35,26 @@ def load_config(cfg_file):
envvar='IOTTB_CONF_HOME', help='Path to iottb config file')
@click.pass_context
def cli(ctx, verbosity, debug, cfg_file):
setup_logging(verbosity, debug) # Setup logging based on the loaded configuration and other options
ctx.ensure_object(dict) # Make sure context is ready for use
logger.info("Starting execution.")
ctx.obj['CONFIG'] = IottbConfig(cfg_file) # Load configuration directly
setup_logging(verbosity, debug) # Setup logging based on the loaded configuration and other options
@click.command()
@click.option('-d', '--dest', default=str(Path.home()), type=str, help='Location to put (new) iottb database')
@click.option('--name', default=DB_NAME, type=str, help='Name of new database.')
@click.option('--update-default/ --no-update-default', default=True,
help='If new db should be set as the new default')
@click.option('--file', default=DB_NAME)
@click.option('--table', type=str, default='DefaultDatabase')
@click.option('--key')
@click.option('--value')
@click.pass_context
def init_db(ctx, dest, name, update_default):
logger.debug('TOP init_db')
@click.command
@click.option('--obj', '--object', type=click.Choice(['cfg', 'db', 'dev', 'cap']),
help='Type of file to edit.')
@click.option('--file', type=click.Choice(['iottb.cfg', '']))
def edit(ctx, file, table, key, value):
def set_key_in_table_to(ctx, file, table, key, value):
"""Edit config or metadata files. TODO: Implement"""
click.echo(f'set_key_in_table_to invoked')
logger.warning("Unimplemented subcommand invoked.")
pass
@click.command
@click.command()
@click.confirmation_option(prompt="Are you certain that you want to delete the cfg file?")
def rm_cfg():
""" Removes the cfg file from the filesystem.
@ -132,12 +66,27 @@ def rm_cfg():
click.echo(f'Iottb configuration removed at {CFG_FILE_PATH}')
@click.command
@click.option('--db', default=str(Path.home() / 'iottb.db'), help='Delete database')
@click.confirmation_option(prompt="Are you certain that you want to delete the database file?")
def rm_db(db):
""" Removes database from the filesystem.
This is mostly a utility during development. Once non-standard database locations are implemented,
deleting this would lead to iottb not being able to find them anymore.
"""
Path(db).unlink()
click.echo(f'Iottb configuration removed at {CFG_FILE_PATH}')
##################################################################################
# Add all subcommands to group here
#################################################################################
cli.add_command(init_db)
cli.add_command(rm_cfg)
cli.add_command(edit)
cli.add_command(set_key_in_table_to)
cli.add_command(rm_db)
if __name__ == '__main__':
cli(auto_envvar_prefix='IOTTB')

0
iottb/utils/__init__.py Normal file
View File

44
iottb/utils/logger_config.py Normal file
View File

@ -0,0 +1,44 @@
import json
import click
from pathlib import Path
import logging
from logging.handlers import RotatingFileHandler
import sys
from iottb.contexts import IottbConfig
from iottb import definitions
from iottb.definitions import MAX_VERBOSITY, CONSOLE_LOG_FORMATS, APP_NAME, LOGFILE_LOG_FORMAT
loglevel = definitions.LOGLEVEL
def setup_logging(verbosity, debug=loglevel):
""" Setup root logger for iottb """
log_level = loglevel
handlers = []
date_format = '%Y-%m-%d %H:%M:%S'
if verbosity > 0:
log_level = logging.WARNING
if verbosity > MAX_VERBOSITY:
verbosity = MAX_VERBOSITY
log_level = logging.INFO
assert verbosity <= MAX_VERBOSITY, f'Verbosity must be <= {MAX_VERBOSITY}'
console_handler = logging.StreamHandler(sys.stdout)
print(str(sys.stdout))
console_handler.setFormatter(logging.Formatter(CONSOLE_LOG_FORMATS[verbosity], datefmt=date_format))
console_handler.setLevel(logging.DEBUG) # can keep at debug since it depends on global level?
handlers.append(console_handler)
if debug:
log_level = logging.DEBUG
# Logfile logs INFO+, no debugs though
file_handler = RotatingFileHandler(f'{str(definitions.LOGDIR / APP_NAME)}.log', maxBytes=10240, backupCount=5)
file_handler.setFormatter(logging.Formatter(LOGFILE_LOG_FORMAT[verbosity], datefmt=date_format))
file_handler.setLevel(logging.INFO)
# finnish root logger setup
handlers.append(file_handler)
# Force this config to be applied to root logger
logging.basicConfig(level=log_level, handlers=handlers, force=True)

43
tests/test_setup_logging.py Normal file
View File

@ -0,0 +1,43 @@
from unittest import mock
from unittest.mock import patch
import logging
from iottb.definitions import MAX_VERBOSITY
from iottb.utils.logger_config import setup_logging
import pytest
class TestSetupLogging:
# setup_logging sets correct log level based on verbosity
def test_sets_correct_log_level_based_on_verbosity(self, mocker):
mock_basic_config = mocker.patch('logging.basicConfig')
setup_logging(verbosity=1)
mock_basic_config.assert_called_once_with(level=logging.WARNING, handlers=[mocker.ANY], force=True)
# setup_logging handles verbosity values greater than MAX_VERBOSITY
def test_handles_verbosity_greater_than_max_verbosity(self, mocker):
mock_basic_config = mocker.patch('logging.basicConfig')
setup_logging(verbosity=MAX_VERBOSITY + 1)
mock_basic_config.assert_called_once_with(level=logging.INFO, handlers=[mocker.ANY], force=True)
# make sure the root logger has different log level before and after this is called, depending on if verbosity or
# debug is given
def test_root_logger_level_change(self):
with patch('iottb.setup_logging.logging.basicConfig') as mock_basicConfig:
setup_logging(verbosity=1, debug=False)
mock_basicConfig.assert_called_once_with(level=logging.WARNING, handlers=[mock.ANY])
mock_basicConfig.reset_mock()
setup_logging(verbosity=2, debug=False)
mock_basicConfig.assert_called_once_with(level=logging.INFO, handlers=[mock.ANY])
mock_basicConfig.reset_mock()
setup_logging(verbosity=1, debug=True)
mock_basicConfig.assert_called_once_with(level=logging.DEBUG, handlers=[mock.ANY])
mock_basicConfig.reset_mock()
setup_logging(verbosity=2, debug=True)
mock_basicConfig.assert_called_once_with(level=logging.DEBUG, handlers=[mock.ANY])