Begin extracting contexts into separate module
This commit is contained in:
parent
a29ffe92a6
commit
18f80fc6fe
77
iottb/contexts.py
Normal file
77
iottb/contexts.py
Normal file
@ -0,0 +1,77 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DB_NAME = 'iottb.db'
|
||||
|
||||
class IottbConfig:
|
||||
def __init__(self, cfg_file):
|
||||
self.cfg_file = cfg_file
|
||||
self.default_database = None
|
||||
self.default_path = None
|
||||
self.database_locations = {}
|
||||
self.load_config()
|
||||
|
||||
def create_default_config(self):
|
||||
"""Create default iottb config file."""
|
||||
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 = {
|
||||
DB_NAME: self.default_path
|
||||
}
|
||||
|
||||
defaults = {
|
||||
'DefaultDatabase': self.default_database,
|
||||
'DefaultDatabasePath': self.default_path,
|
||||
'DatabaseLocations': self.database_locations
|
||||
}
|
||||
|
||||
try:
|
||||
self.cfg_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
with self.cfg_file.open('w') as config_file:
|
||||
json.dump(defaults, config_file, indent=4)
|
||||
except IOError as e:
|
||||
logger.error(f"Failed to create default configuration file at {self.cfg_file}: {e}")
|
||||
raise RuntimeError(f"Failed to create configuration file: {e}") from e
|
||||
|
||||
def load_config(self):
|
||||
"""Loads or creates default configuration from given file path."""
|
||||
if not self.cfg_file.is_file():
|
||||
self.create_default_config()
|
||||
else:
|
||||
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', {})
|
||||
|
||||
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
|
||||
}
|
||||
try:
|
||||
with self.cfg_file.open('w') as config_file:
|
||||
json.dump(data, config_file, indent=4)
|
||||
except IOError as e:
|
||||
logger.error(f"Failed to save configuration file at {self.cfg_file}: {e}")
|
||||
raise RuntimeError(f"Failed to save configuration file: {e}") from e
|
||||
|
||||
def set_default_database(self, name, path):
|
||||
"""Set the default database and its path."""
|
||||
self.default_database = name
|
||||
self.default_path = path
|
||||
self.database_locations[name] = path
|
||||
|
||||
def get_database_location(self, name):
|
||||
"""Get the location of a specific database."""
|
||||
return self.database_locations.get(name)
|
||||
|
||||
def set_database_location(self, name, path):
|
||||
"""Set the location for a database."""
|
||||
self.database_locations[name] = path
|
||||
@ -4,10 +4,11 @@ from pathlib import Path
|
||||
import logging
|
||||
from logging.handlers import RotatingFileHandler
|
||||
import sys
|
||||
from iottb.contexts import IottbConfig
|
||||
|
||||
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',
|
||||
@ -60,13 +61,14 @@ def create_default_config(cfg_file):
|
||||
logger.info(f'Creating default config file at {cfg_file}')
|
||||
defaults = {
|
||||
'DefaultDatabase': DB_NAME,
|
||||
'DatabaseLocations': {
|
||||
DB_NAME: str(Path.home() / DB_NAME)
|
||||
}
|
||||
'DefaultDatabasePath': str(Path.home())
|
||||
}
|
||||
|
||||
try:
|
||||
with open(cfg_file, 'w') as config_file:
|
||||
json.dump(defaults, config_file)
|
||||
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
|
||||
@ -75,29 +77,32 @@ def create_default_config(cfg_file):
|
||||
|
||||
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')
|
||||
@click.option('-d', '--debug', is_flag=True, default=False,
|
||||
help='Enable debug mode')
|
||||
@click.option('--cfg-file', type=click.Path(exists=True),
|
||||
@click.option('--cfg-file', type=click.Path(),
|
||||
default=Path(click.get_app_dir(APP_NAME)).joinpath('iottb.cfg'),
|
||||
envvar='IOTTB_CONF_HOME', help='Path to iottb config file')
|
||||
@click.pass_context
|
||||
def cli(ctx, verbosity, debug, cfg_file):
|
||||
ctx.ensure_object(dict) # Make sure context is ready for use
|
||||
ctx.obj['CONFIG'] = load_config(cfg_file) # Load configuration directly
|
||||
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=Path.home(), type=str, help='Location to put (new) iottb database')
|
||||
@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')
|
||||
@ -106,8 +111,33 @@ def init_db(ctx, dest, name, update_default):
|
||||
logger.debug('TOP init_db')
|
||||
|
||||
|
||||
cli.add_command(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):
|
||||
"""Edit config or metadata files. TODO: Implement"""
|
||||
pass
|
||||
|
||||
|
||||
@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.
|
||||
|
||||
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(CFG_FILE_PATH).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)
|
||||
|
||||
if __name__ == '__main__':
|
||||
cli(auto_envvar_prefix='IOTTB')
|
||||
|
||||
@ -1,18 +1,33 @@
|
||||
# test_main.py
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
from iottb.main import cli # Adjust this import according to your project's structure
|
||||
from pathlib import Path
|
||||
from iottb.main import load_config
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def runner():
|
||||
"""Fixture to return a Click CliRunner instance."""
|
||||
return CliRunner()
|
||||
class TestLoadConfig:
|
||||
|
||||
# Loads configuration from an existing file
|
||||
def test_loads_config_from_existing_file(self, mocker):
|
||||
cfg_file = mocker.Mock()
|
||||
cfg_file.is_file.return_value = True
|
||||
mock_open = mocker.mock_open(
|
||||
read_data=f'{{"DefaultDatabase": "test_db", "DefaultDatabasePath": "{Path.home()}/user"}}')
|
||||
mocker.patch('builtins.open', mock_open)
|
||||
|
||||
def test_debug_option(runner):
|
||||
"""Test if the debug mode sets the appropriate flag in the context."""
|
||||
result = runner.invoke(cli, ['--debug'])
|
||||
assert result.exit_code == 0
|
||||
# If debug mode affects logging or other behavior, validate those:
|
||||
assert 'DEBUG mode on' in result.output
|
||||
result = load_config(cfg_file)
|
||||
|
||||
assert result == {"DefaultDatabase": "test_db", "DefaultDatabasePath": f"{Path.home()}/user"}
|
||||
cfg_file.is_file.assert_called_once()
|
||||
mock_open.assert_called_once_with(cfg_file, 'r')
|
||||
|
||||
# File path is invalid or inaccessible
|
||||
def test_file_path_invalid_or_inaccessible(self, mocker):
|
||||
cfg_file = mocker.Mock()
|
||||
cfg_file.is_file.return_value = False
|
||||
mock_create_default_config = mocker.patch('iottb.main.create_default_config',
|
||||
return_value={"DefaultDatabase": "default_db",
|
||||
"DefaultDatabasePath": f"{Path.home()}/default"})
|
||||
|
||||
result = load_config(cfg_file)
|
||||
|
||||
assert result == {"DefaultDatabase": "default_db", "DefaultDatabasePath": f"{Path.home()}/default"}
|
||||
cfg_file.is_file.assert_called_once()
|
||||
mock_create_default_config.assert_called_once_with(cfg_file)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user