kodjo/MyManagingTools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactoring DbEngine

Browse Source
This commit is contained in:
Kodjo Sossouvi
2025-05-10 20:40:03 +02:00
parent 2daff83e67
commit e1c10183eb
12 changed files with 515 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/components/addstuff/AddStuffApp.py
View File

@@ -13,8 +13,9 @@ add_stuff_app, rt = fast_app()
@rt(Routes.AddRepository)
def get(session):
repositories_instance = InstanceManager.get(session, Repositories.create_component_id(session))
return repositories_instance.request_new_repository()
_id = Repositories.create_component_id(session) # there is only one instance of Repositories
instance = InstanceManager.get(session, _id)
return instance.request_new_repository()
@rt(Routes.AddRepository)
@@ -25,6 +26,18 @@ def post(session, _id: str, tab_id: str, form_id: str, repository: str, table: s
return instance.add_new_repository(tab_id, form_id, repository, table)
@rt(Routes.AddTable)
def get(session, _id: str, repository_name: str):
instance = InstanceManager.get(session, _id)
return instance.request_new_table(repository_name)
@rt(Routes.AddTable)
def post(session, _id: str, tab_id: str, form_id: str, repository_name: str, table_name: str):
instance = InstanceManager.get(session, _id)
return instance.add_new_table(tab_id, form_id, repository_name, table_name)
@rt(Routes.SelectRepository)
def put(session, _id: str, repository: str):
logger.debug(f"Entering {Routes.SelectRepository} with args {debug_session(session)}, {_id=}, {repository=}")

23
src/components/addstuff/commands.py Normal file
View File

@@ -0,0 +1,23 @@
from components.addstuff.constants import ROUTE_ROOT, Routes
class Commands:
def __init__(self, owner):
self._owner = owner
self._id = owner.get_id()
def request_add_table(self, repository_name):
return {
"hx-get": f"{ROUTE_ROOT}{Routes.AddTable}",
"hx-target": f"#{self._owner.tabs_manager.get_id()}",
"hx-swap": "outerHTML",
"hx-vals": f'{{"_id": "{self._id}", "repository_name": "{repository_name}"}}',
}
def add_table(self):
return {
"hx-post": f"{ROUTE_ROOT}{Routes.AddTable}",
"hx-target": f"#{self._owner.tabs_manager.get_id()}",
"hx-swap": "outerHTML",
# The repository_name and the table_name will be given by the form
}

4
src/components/addstuff/components/AddStuffMenu.py
View File

@@ -2,7 +2,7 @@ from fasthtml.components import *
from components.BaseComponent import BaseComponent
from components.addstuff.constants import ADD_STUFF_INSTANCE_ID, ROUTE_ROOT, Routes
from components.addstuff.settings import AddStuffSettingsManager
from components.addstuff.settings import RepositoriesDbManager
class AddStuffMenu(BaseComponent):
@@ -10,7 +10,7 @@ class AddStuffMenu(BaseComponent):
super().__init__(session, _id)
self.tabs_manager = tabs_manager # MyTabs component id
self.mappings = {} # to keep track of when element is displayed on which tab
self.settings = AddStuffSettingsManager(session, settings_manager)
self.settings = RepositoriesDbManager(session, settings_manager)
def __ft__(self):
return Div(

65
src/components/addstuff/components/Repositories.py
View File

@@ -5,8 +5,9 @@ from fasthtml.xtend import Script
from components.BaseComponent import BaseComponent
from components.addstuff.assets.icons import icon_database, icon_table
from components.addstuff.commands import Commands
from components.addstuff.constants import REPOSITORIES_INSTANCE_ID, ROUTE_ROOT, Routes
from components.addstuff.settings import AddStuffSettingsManager, MyTable, Repository
from components.addstuff.settings import RepositoriesDbManager, Repository
from components.datagrid_new.components.DataGrid import DataGrid
from components.form.components.MyForm import MyForm, FormField
from components_helpers import mk_icon, mk_ellipsis, mk_tooltip_container
@@ -19,9 +20,10 @@ class Repositories(BaseComponent):
def __init__(self, session: dict, _id: str, settings_manager=None, tabs_manager=None):
super().__init__(session, _id)
self._settings_manager = settings_manager
self.repo_settings_manager = AddStuffSettingsManager(session, settings_manager)
self.db = RepositoriesDbManager(session, settings_manager)
self.tabs_manager = tabs_manager
self._contents = {} # ket tracks of already displayed contents
self._commands = Commands(self)
def request_new_repository(self):
# request for a new tab_id
@@ -34,6 +36,17 @@ class Repositories(BaseComponent):
self.tabs_manager.add_tab("Add Database", add_repository_form, tab_id=new_tab_id)
return self.tabs_manager
def request_new_table(self, repository_name: str):
# request for a new tab_id
new_tab_id = self.tabs_manager.request_new_tab_id()
# create a new form to ask for the details of the new database
add_table_form = self._mk_add_table_form(new_tab_id, repository_name)
# create and display the form in a new tab
self.tabs_manager.add_tab("Add New Table", add_table_form, tab_id=new_tab_id)
return self.tabs_manager
def add_new_repository(self, tab_id: str, form_id: str, repository_name: str, table_name: str):
"""
@@ -46,7 +59,7 @@ class Repositories(BaseComponent):
try:
# Add the new repository and its default table to the list of repositories
tables = [MyTable(table_name, {})] if table_name else []
repository = self.repo_settings_manager.add_repository(repository_name, tables)
repository = self.db.add_repository(repository_name, tables)
# update the tab content with table content
key = (repository_name, table_name)
@@ -65,8 +78,37 @@ class Repositories(BaseComponent):
return self.tabs_manager.refresh()
def add_new_table(self, tab_id: str, form_id: str, repository_name: str, table_name: str):
"""
:param tab_id: tab id where the table content will be displayed (and where the form was displayed)
:param form_id: form used to give the repository name (to be used in case of error)
:param repository_name: new repository name
:param table_name: default table name
:return:
"""
try:
self.db.add_table(repository_name, table_name, {})
repository = self.db.get_repository(repository_name)
# update the tab content with table content
key = (repository_name, table_name)
self.tabs_manager.set_tab_content(tab_id,
self._get_table_content(key),
title=table_name,
key=key,
active=True)
return self._mk_repository(repository, True), self.tabs_manager.refresh()
except ValueError as ex:
logger.debug(f" Repository '{repository_name}' already exists.")
add_repository_form = InstanceManager.get(self._session, form_id)
add_repository_form.set_error(ex)
return self.tabs_manager.refresh()
def select_repository(self, repository_name: str):
self.repo_settings_manager.select_repository(repository_name)
self.db.select_repository(repository_name)
def show_table(self, repository_name: str, table_name: str):
key = (repository_name, table_name)
@@ -86,11 +128,10 @@ class Repositories(BaseComponent):
)
def _mk_repositories(self, oob=False):
settings = self.repo_settings_manager.get_settings()
settings = self.db._get_settings()
return Div(
*[self._mk_repository(repo, repo.name == settings.selected_repository_name)
for repo in settings.repositories],
id=self._id,
hx_swap_oob="true" if oob else None,
)
@@ -120,6 +161,7 @@ class Repositories(BaseComponent):
cls="flex")
for table in repo.tables
],
Div("+ Add Table", **self._commands.request_add_table(repo.name)),
cls="collapse-content pr-0! truncate",
),
tabindex="0", cls="collapse mb-2")
@@ -137,6 +179,17 @@ class Repositories(BaseComponent):
htmx_params=htmx_params,
extra_values={"_id": self._id, "tab_id": tab_id})
def _mk_add_table_form(self, tab_id: str, repository_name: str = None):
htmx_request = self._commands.add_table()
return InstanceManager.get(self._session, MyForm.create_component_id(self._session), MyForm,
title="Add Table",
fields=[FormField("repository_name", 'Repository Name', 'input',
value=repository_name,
disabled=True),
FormField("table_name", 'Table Name', 'input')],
htmx_request=htmx_request,
extra_values={"_id": self._id, "tab_id": tab_id, "repository_name": repository_name})
def _get_table_content(self, key):
if key in self._contents:

1
src/components/addstuff/constants.py
View File

@@ -6,4 +6,5 @@ ROUTE_ROOT = "/add"
class Routes:
AddRepository = "/add-repository"
SelectRepository = "/select-repository"
AddTable = "/add-table"
ShowTable = "/show-table"

45
src/components/addstuff/settings.py
View File

@@ -5,38 +5,33 @@ from core.settings_management import SettingsManager
from core.settings_objects import BaseSettingObj
ADD_STUFF_SETTINGS_ENTRY = "AddStuffSettings"
REPOSITORIES_SETTINGS_ENTRY = "Repositories"
logger = logging.getLogger("AddStuffSettings")
@dataclasses.dataclass
class MyTable:
name: str
settings: dict | None = None
@dataclasses.dataclass
class Repository:
name: str
tables: list[MyTable]
tables: list[str]
@dataclasses.dataclass
class AddStuffSettings:
class RepositoriesSettings:
repositories: list[Repository] = dataclasses.field(default_factory=list)
selected_repository_name: str = None
class AddStuffSettingsManager(BaseSettingObj):
__ENTRY_NAME__ = ADD_STUFF_SETTINGS_ENTRY
class RepositoriesDbManager:
def __init__(self, session: dict, settings_manager: SettingsManager):
self.session = session
self.settings_manager = settings_manager
def get_settings(self):
return self.settings_manager.get(self.session, ADD_STUFF_SETTINGS_ENTRY, default=AddStuffSettings())
def _get_settings(self):
return self.settings_manager.get(self.session, REPOSITORIES_SETTINGS_ENTRY, default=RepositoriesSettings())
def add_repository(self, repository_name: str, tables: list[MyTable] = None):
def add_repository(self, repository_name: str, tables: list[str] = None):
"""
Adds a new repository to the list of repositories. The repository is identified
by its name and can optionally include a list of associated tables.
@@ -48,7 +43,7 @@ class AddStuffSettingsManager(BaseSettingObj):
:return: None
"""
settings = self.get_settings()
settings = self._get_settings()
if repository_name is None or repository_name == "":
raise ValueError("Repository name cannot be empty.")
@@ -61,9 +56,26 @@ class AddStuffSettingsManager(BaseSettingObj):
repository = Repository(repository_name, tables or [])
settings.repositories.append(repository)
self.settings_manager.put(self.session, ADD_STUFF_SETTINGS_ENTRY, settings)
self.settings_manager.put(self.session, REPOSITORIES_SETTINGS_ENTRY, settings)
return repository
def get_repository(self, repository_name: str):
if repository_name is None or repository_name == "":
raise ValueError("Repository name cannot be empty.")
settings = self._get_settings()
if repository_name not in [repo.name for repo in settings.repositories]:
raise ValueError(f"Repository '{repository_name}' does not exists.")
return next(filter(lambda r: r.name == repository_name, settings.repositories))
def modify_repository(self, repository_name: str, tables: list[str]):
repository = self.get_repository(repository_name)
def get_repositories(self):
return self._get_settings().repositories
def add_table(self, repository_name: str, table_name: str, table_settings: dict):
"""
Adds a table to the specified repository
@@ -74,7 +86,7 @@ class AddStuffSettingsManager(BaseSettingObj):
of the table.
:return: None
"""
settings = self.get_settings()
settings = self._get_settings()
repository = next(filter(lambda r: r.name == repository_name, settings.repositories), None)
if repository is None:
@@ -94,6 +106,7 @@ class AddStuffSettingsManager(BaseSettingObj):
:type repository_name: str
:return: None
"""
settings = self.get_settings()
settings = self._get_settings()
settings.selected_repository_name = repository_name
self.settings_manager.put(self.session, ADD_STUFF_SETTINGS_ENTRY, settings)

19
src/components/form/components/MyForm.py
View File

@@ -15,6 +15,8 @@ class FormField:
name: str
label: str
type: str
value: str = None
disabled: bool = False
class MyForm(BaseComponent):
@@ -23,7 +25,7 @@ class MyForm(BaseComponent):
fields: list[FormField] = None,
state: dict = None, # to remember the values of the fields
submit: str = "Submit", # submit button
htmx_params: dict = None, # htmx parameters
htmx_request: dict = None, # htmx parameters
extra_values: dict = None, # hx_vals parameters, but using python dict rather than javascript
success: str = None,
error: str = None
@@ -33,7 +35,7 @@ class MyForm(BaseComponent):
self.fields = fields
self.state: dict = {} if state is None else state
self.submit = submit
self.htmx_params = htmx_params
self.htmx_request = htmx_request
self.extra_values = extra_values
self.success = success
self.error = error
@@ -80,10 +82,10 @@ class MyForm(BaseComponent):
Button(
self.submit,
hx_post=self.htmx_params.get("hx-post", None),
hx_target=self.htmx_params.get("hx-target", None),
hx_swap=self.htmx_params.get("hx-swap", None),
hx_vals=self.htmx_params.get("hx-vals", f"js:{{...{self.extra_values} }}" if self.extra_values else None),
hx_post=self.htmx_request.get("hx-post", None),
hx_target=self.htmx_request.get("hx-target", None),
hx_swap=self.htmx_request.get("hx-swap", None),
hx_vals=self.htmx_request.get("hx-vals", f"js:{{...{self.extra_values} }}" if self.extra_values else None),
cls="btn w-full font-bold py-2 px-4 rounded button-xs"
),
@@ -106,7 +108,8 @@ class MyForm(BaseComponent):
name=field.name,
placeholder=field.label,
required=True,
value=self.state.get(field.name, None),
value=self.state.get(field.name, field.value),
disabled=field.disabled,
hx_put=f"{ROUTE_ROOT}{Routes.OnUpdate}",
hx_trigger="keyup changed delay:300ms",
@@ -120,4 +123,4 @@ class MyForm(BaseComponent):
@staticmethod
def create_component_id(session):
prefix = f"{MY_FORM_INSTANCE_ID}{session['user_id']}"
return get_unique_id(prefix)
return get_unique_id(prefix)

114
src/core/dbengine.py
View File

@@ -25,8 +25,8 @@ class DbException(Exception):
class RefHelper:
def __init__(self, get_obj_path):
self.get_obj_path = get_obj_path
def __init__(self, get_ref_path):
self.get_ref_path = get_ref_path
def save_ref(self, obj):
"""
@@ -40,12 +40,12 @@ class RefHelper:
digest = get_stream_digest(buffer)
target_path = self.get_obj_path(digest)
target_path = self.get_ref_path(digest)
if not os.path.exists(os.path.dirname(target_path)):
os.makedirs(os.path.dirname(target_path))
buffer.seek(0)
with open(self.get_obj_path(digest), "wb") as file:
with open(self.get_ref_path(digest), "wb") as file:
while chunk := buffer.read(BUFFER_SIZE):
file.write(chunk)
@@ -58,7 +58,7 @@ class RefHelper:
:param digest:
:return:
"""
with open(self.get_obj_path(digest), 'rb') as file:
with open(self.get_ref_path(digest), 'rb') as file:
return pickle.load(file)
@@ -73,40 +73,50 @@ class DbEngine:
def __init__(self, root: str = None):
self.root = root or ".mytools_db"
self.serializer = Serializer(RefHelper(self._get_obj_path))
self.debug_serializer = DebugSerializer(RefHelper(self.debug_load))
self.serializer = Serializer(RefHelper(self._get_ref_path))
self.debug_serializer = DebugSerializer(RefHelper(self._get_ref_path))
self.lock = RLock()
def is_initialized(self):
def is_initialized(self, user_id: str):
"""
:return:
"""
return os.path.exists(self.root)
return os.path.exists(self._get_user_root(user_id))
def init(self):
def init(self, user_id: str):
"""
Make sure that the DbEngine is properly initialized
:return:
"""
if not os.path.exists(self.root):
logger.debug(f"Creating root folder in {os.path.abspath(self.root)}.")
os.mkdir(self.root)
if not os.path.exists(self._get_user_root(user_id)):
logger.debug(f"Creating root folder in {os.path.abspath(self._get_user_root(user_id))}.")
os.makedirs(self._get_user_root(user_id))
def save(self, user_id: str, entry: str, obj: object) -> str:
def save(self, user_id: str, user_email: str, entry: str, obj: object) -> str:
"""
Save a snapshot of an entry
:param user_id:
:param user_email:
:param entry:
:param obj: snapshot to save
:return:
"""
with self.lock:
logger.info(f"Saving {user_id=}, {entry=}, {obj=}")
if not user_id:
raise DbException("user_id is None")
if not user_email:
raise DbException("user_email is None")
if not entry:
raise DbException("entry is None")
# prepare the data
as_dict = self._serialize(obj)
as_dict[TAG_PARENT] = [self._get_entry_digest(entry)]
as_dict[TAG_USER] = user_id
as_dict[TAG_PARENT] = [self._get_entry_digest(user_id, entry)]
as_dict[TAG_USER] = user_email
as_dict[TAG_DATE] = datetime.datetime.now().strftime('%Y%m%d %H:%M:%S %z')
# transform into a stream
@@ -117,7 +127,7 @@ class DbEngine:
# compute the digest to know where to store it
digest = hashlib.sha256(byte_stream).hexdigest()
target_path = self._get_obj_path(digest)
target_path = self._get_obj_path(user_id, digest)
if os.path.exists(target_path):
# the same object is already saved. Noting to do
return digest
@@ -128,8 +138,8 @@ class DbEngine:
with open(target_path, "wb") as file:
file.write(byte_stream)
# update the head to remember where is the latest entry
self._update_head(entry, digest)
# update the head to remember where the latest entry is
self._update_head(user_id, entry, digest)
logger.debug(f"New head for entry '{entry}' is {digest}")
return digest
@@ -144,24 +154,25 @@ class DbEngine:
with self.lock:
logger.info(f"Loading {user_id=}, {entry=}, {digest=}")
digest_to_use = digest or self._get_entry_digest(entry)
digest_to_use = digest or self._get_entry_digest(user_id, entry)
logger.debug(f"Using digest {digest_to_use}.")
if digest_to_use is None:
raise DbException(entry)
target_file = self._get_obj_path(digest_to_use)
target_file = self._get_obj_path(user_id, digest_to_use)
with open(target_file, 'r', encoding='utf-8') as file:
as_dict = json.load(file)
return self._deserialize(as_dict)
def put(self, user_id: str, entry, key: str, value: object):
def put(self, user_id: str, user_email, entry, key: str, value: object):
"""
Save a specific record.
This will create a new snapshot is the record is new or different
You should not mix the usage of put_many() and save() as it's two different way to manage the db
:param user_email:
:param user_id:
:param entry:
:param key:
@@ -182,16 +193,17 @@ class DbEngine:
return False
entry_content[key] = value
self.save(user_id, entry, entry_content)
self.save(user_id, user_email, entry, entry_content)
return True
def put_many(self, user_id: str, entry, items: list):
def put_many(self, user_id: str, user_email, entry, items: list):
"""
Save a list of item as one single snapshot
A new snapshot will not be created if all the items already exist
You should not mix the usage of put_many() and save() as it's two different way to manage the db
:param user_id:
:param user_email:
:param entry:
:param items:
:return:
@@ -213,12 +225,12 @@ class DbEngine:
is_dirty = True
if is_dirty:
self.save(user_id, entry, entry_content)
self.save(user_id, user_email, entry, entry_content)
return True
return False
def exists(self, entry: str):
def exists(self, user_id, entry: str):
"""
Tells if an entry exist
:param user_id:
@@ -226,7 +238,7 @@ class DbEngine:
:return:
"""
with self.lock:
return self._get_entry_digest(entry) is not None
return self._get_entry_digest(user_id, entry) is not None
def get(self, user_id: str, entry: str, key: str | None = None, digest=None):
"""
@@ -247,9 +259,19 @@ class DbEngine:
return entry_content[key]
def debug_head(self):
def debug_root(self):
"""
Lists all folders in the root directory
:return: List of folder names
"""
with self.lock:
head_path = os.path.join(self.root, self.HeadFile)
if not os.path.exists(self.root):
return []
return [f for f in os.listdir(self.root) if os.path.isdir(os.path.join(self.root, f))]
def debug_head(self, user_id):
with self.lock:
head_path = os.path.join(self.root, user_id, self.HeadFile)
# load
try:
with open(head_path, 'r') as file:
@@ -259,14 +281,17 @@ class DbEngine:
return head
def debug_load(self, digest):
def debug_load(self, user_id, digest):
with self.lock:
target_file = self._get_obj_path(digest)
target_file = self._get_obj_path(user_id, digest)
with open(target_file, 'r', encoding='utf-8') as file:
as_dict = json.load(file)
return self.debug_serializer.deserialize(as_dict)
def debug_get_digest(self, user_id, entry):
return self._get_entry_digest(user_id, entry)
def _serialize(self, obj):
"""
Just call the serializer
@@ -280,14 +305,14 @@ class DbEngine:
def _deserialize(self, as_dict):
return self.serializer.deserialize(as_dict)
def _update_head(self, entry, digest):
def _update_head(self, user_id, entry, digest):
"""
Actually dumps the snapshot in file system
:param entry:
:param digest:
:return:
"""
head_path = os.path.join(self.root, self.HeadFile)
head_path = os.path.join(self.root, user_id, self.HeadFile)
# load
try:
with open(head_path, 'r') as file:
@@ -302,13 +327,16 @@ class DbEngine:
with open(head_path, 'w') as file:
json.dump(head, file)
def _get_entry_digest(self, entry):
def _get_user_root(self, user_id):
return os.path.join(self.root, user_id)
def _get_entry_digest(self, user_id, entry):
"""
Search for the latest digest, for a given entry
:param entry:
:return:
"""
head_path = os.path.join(self.root, self.HeadFile)
head_path = os.path.join(self._get_user_root(user_id), self.HeadFile)
try:
with open(head_path, 'r') as file:
head = json.load(file)
@@ -319,17 +347,25 @@ class DbEngine:
except KeyError:
return None
def _get_head_path(self):
def _get_head_path(self, user_id: str):
"""
Location of the Head file
:return:
"""
return os.path.join(self.root, self.HeadFile)
return os.path.join(self._get_user_root(user_id), self.HeadFile)
def _get_obj_path(self, digest):
def _get_obj_path(self, user_id, digest):
"""
Location of objects
:param digest:
:return:
"""
return os.path.join(self.root, "objects", digest[:24], digest)
return os.path.join(self._get_user_root(user_id), "objects", digest[:24], digest)
def _get_ref_path(self, digest):
"""
Location of reference. They are not linked to the user folder
:param digest:
:return:
"""
return os.path.join(self.root, "refs", digest[:24], digest)

18
src/core/settings_management.py
View File

@@ -169,6 +169,24 @@ class SettingsManager:
else:
return default
def remove(self, session: dict, key: str):
user_id = session["user_id"] if session else NO_SESSION
user_email = session["user_email"] if session else NOT_LOGGED
return self._db_engine.remove(user_email, user_id, key)
def update(self, session: dict, old_key: str, key: str, value: object):
user_id = session["user_id"] if session else NO_SESSION
user_email = session["user_email"] if session else NOT_LOGGED
def _update_helper(_old_key, _key, _value):
pass
if hasattr(self._db_engine, "lock"):
with self._db_engine.lock:
_update_helper(old_key, key, value)
else:
_update_helper(old_key, key, value)
def init_user(self, user_id: str, user_email: str):
"""
Init the settings block space for a user

322
tests/test_dbengine.py
View File

@@ -4,178 +4,234 @@ import shutil
import pandas as pd
import pytest
from core.dbengine import DbEngine, TAG_PARENT
from core.settings_objects import BudgetTrackerSettings, BudgetTrackerFile, BudgetTrackerFiles
from core.dbengine import DbEngine, DbException, TAG_PARENT, TAG_USER, TAG_DATE
DB_ENGINE_ROOT = "TestDBEngineRoot"
FAKE_USER_ID = "FakeUserId"
FAKE_USER_EMAIL = "fake_user@me.com"
class DummyObj:
def __init__(self, a, b, c):
self.a = a
self.b = b
self.c = c
def __eq__(self, other):
if id(self) == id(other):
return True
if not isinstance(other, DummyObj):
return False
return self.a == other.a and self.b == other.b and self.c == other.c
def __hash__(self):
return hash((self.a, self.b, self.c))
class DummyObjWithRef(DummyObj):
@staticmethod
def use_refs() -> set:
return {"c"}
class DummyObjWithKey(DummyObj):
def get_key(self) -> set:
return self.a
@pytest.fixture()
def engine():
if os.path.exists(DB_ENGINE_ROOT):
shutil.rmtree(DB_ENGINE_ROOT)
engine = DbEngine(DB_ENGINE_ROOT)
engine.init()
return engine
if os.path.exists(DB_ENGINE_ROOT):
shutil.rmtree(DB_ENGINE_ROOT)
engine = DbEngine(DB_ENGINE_ROOT)
engine.init(FAKE_USER_ID)
yield engine
shutil.rmtree(DB_ENGINE_ROOT)
@pytest.fixture()
def dummy_obj():
return BudgetTrackerSettings(
spread_sheet="spread_sheet",
col_row_num="row_number",
col_project="project",
col_owner="owner",
col_capex="capex",
col_details="details",
col_supplier="supplier",
col_budget_amt="budget",
col_actual_amt="actual",
col_forecast5_7_amt="forecast5_7",
)
return DummyObj(1, "a", False)
@pytest.fixture()
def dummy_obj2():
return BudgetTrackerSettings(
spread_sheet="spread_sheet2",
col_row_num="row_number2",
col_project="project2",
col_owner="owner2",
col_capex="capex2",
col_details="details2",
col_supplier="supplier2",
col_budget_amt="budget2",
col_actual_amt="actual2",
col_forecast5_7_amt="forecast5_72",
)
return DummyObj(2, "b", True)
@pytest.fixture()
def dummy_obj_with_ref():
data = {
'Key1': ['A', 'B', 'C'],
'Key2': ['X', 'Y', 'Z'],
'Percentage': [0.1, 0.2, 0.15],
}
df = pd.DataFrame(data)
return DummyObjWithRef(1, "a", df)
def test_i_can_test_init():
if os.path.exists(DB_ENGINE_ROOT):
shutil.rmtree(DB_ENGINE_ROOT)
engine = DbEngine(DB_ENGINE_ROOT)
assert not engine.is_initialized()
engine.init()
assert engine.is_initialized()
if os.path.exists(DB_ENGINE_ROOT):
shutil.rmtree(DB_ENGINE_ROOT)
engine = DbEngine(DB_ENGINE_ROOT)
assert not engine.is_initialized(FAKE_USER_ID)
engine.init(FAKE_USER_ID)
assert engine.is_initialized(FAKE_USER_ID)
def test_i_can_save_and_load(engine, dummy_obj):
engine.save(FAKE_USER_ID, "MyEntry", dummy_obj)
res = engine.load(FAKE_USER_ID, "MyEntry")
assert isinstance(res, BudgetTrackerSettings)
assert res.spread_sheet == dummy_obj.spread_sheet
assert res.col_row_num == dummy_obj.col_row_num
assert res.col_project == dummy_obj.col_project
assert res.col_owner == dummy_obj.col_owner
assert res.col_capex == dummy_obj.col_capex
assert res.col_details == dummy_obj.col_details
assert res.col_supplier == dummy_obj.col_supplier
assert res.col_budget_amt == dummy_obj.col_budget_amt
assert res.col_actual_amt == dummy_obj.col_actual_amt
assert res.col_forecast5_7_amt == dummy_obj.col_forecast5_7_amt
digest = engine.save(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", dummy_obj)
res = engine.load(FAKE_USER_ID, "MyEntry")
assert digest is not None
assert isinstance(res, DummyObj)
assert res.a == dummy_obj.a
assert res.b == dummy_obj.b
assert res.c == dummy_obj.c
# check that the files are created
assert os.path.exists(os.path.join(DB_ENGINE_ROOT, FAKE_USER_ID, "objects"))
assert os.path.exists(os.path.join(DB_ENGINE_ROOT, FAKE_USER_ID, "head"))
def test_i_can_save_using_ref(engine):
data = {
'Key1': ['A', 'B', 'C'],
'Key2': ['X', 'Y', 'Z'],
'Percentage': [0.1, 0.2, 0.15],
}
df = pd.DataFrame(data)
obj = BudgetTrackerFile(2024, 8, data=df)
engine.save(FAKE_USER_ID, "MyEntry", obj)
res = engine.load(FAKE_USER_ID, "MyEntry")
assert isinstance(res, BudgetTrackerFile)
assert res.year == obj.year
assert res.month == obj.month
assert res.data.to_dict() == obj.data.to_dict()
def test_save_invalid_inputs(engine):
"""
Test save with invalid inputs.
"""
with pytest.raises(DbException):
engine.save(None, FAKE_USER_EMAIL, "InvalidEntry", DummyObj(1, 2, 3))
with pytest.raises(DbException):
engine.save(FAKE_USER_ID, None, "InvalidEntry", DummyObj(1, 2, 3))
with pytest.raises(DbException):
engine.save(FAKE_USER_ID, FAKE_USER_EMAIL, "", DummyObj(1, 2, 3))
with pytest.raises(DbException):
engine.save(FAKE_USER_ID, FAKE_USER_EMAIL, None, DummyObj(1, 2, 3))
def test_i_can_use_ref_when_subclass(engine):
data1 = {'Key': ['A'], 'Value': [0.1]}
data2 = {'Key': ['B'], 'Value': [0.2]}
file1 = BudgetTrackerFile(2024, 8, data=pd.DataFrame(data1))
file2 = BudgetTrackerFile(2024, 9, data=pd.DataFrame(data2))
files = BudgetTrackerFiles([file1, file2])
def test_i_can_save_using_ref(engine, dummy_obj_with_ref):
engine.save(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", dummy_obj_with_ref)
res = engine.load(FAKE_USER_ID, "MyEntry")
assert isinstance(res, DummyObjWithRef)
assert res.a == dummy_obj_with_ref.a
assert res.b == dummy_obj_with_ref.b
assert res.c.to_dict() == dummy_obj_with_ref.c.to_dict()
# check that the files are created
assert os.path.exists(os.path.join(DB_ENGINE_ROOT, FAKE_USER_ID, "objects"))
assert os.path.exists(os.path.join(DB_ENGINE_ROOT, FAKE_USER_ID, "head"))
assert os.path.exists(os.path.join(DB_ENGINE_ROOT, "refs"))
engine.save(FAKE_USER_ID, "MyEntry", files)
res = engine.load(FAKE_USER_ID, "MyEntry")
assert isinstance(res, BudgetTrackerFiles)
assert len(res.files) == 2
def test_refs_are_share_across_users(engine, dummy_obj_with_ref):
engine.save(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", dummy_obj_with_ref)
engine.save("AnotherUserId", "AnotherUser", "AnotherMyEntry", dummy_obj_with_ref)
refs_path = os.path.join(DB_ENGINE_ROOT, "refs")
assert len(os.listdir(refs_path)) == 1
def test_metadata_are_correctly_set(engine, dummy_obj):
digest = engine.save(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", dummy_obj)
as_dict = engine.debug_load(FAKE_USER_ID, digest)
assert as_dict[TAG_PARENT] == [None]
assert as_dict[TAG_USER] == FAKE_USER_EMAIL
assert as_dict[TAG_DATE] is not None
def test_i_can_track_parents(engine):
digest = engine.save(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", DummyObj(1, "a", False))
second_digest = engine.save(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", DummyObj(1, "a", True))
as_dict = engine.debug_load(FAKE_USER_ID, second_digest)
assert as_dict[TAG_PARENT] == [digest]
def test_i_can_put_and_get_one_object(engine, dummy_obj):
engine.put(FAKE_USER_ID, "MyEntry", "key1", dummy_obj)
from_db = engine.get(FAKE_USER_ID, "MyEntry", "key1")
assert from_db == dummy_obj
engine.put(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", "key1", dummy_obj)
from_db = engine.get(FAKE_USER_ID, "MyEntry", "key1")
assert from_db == dummy_obj
def test_i_can_put_and_get_multiple_objects(engine, dummy_obj, dummy_obj2):
engine.put(FAKE_USER_ID, "MyEntry", "key1", dummy_obj)
engine.put(FAKE_USER_ID, "MyEntry", "key2", dummy_obj2)
from_db1 = engine.get(FAKE_USER_ID, "MyEntry", "key1")
from_db2 = engine.get(FAKE_USER_ID, "MyEntry", "key2")
assert from_db1 == dummy_obj
assert from_db2 == dummy_obj2
all_items = engine.get(FAKE_USER_ID, "MyEntry")
assert all_items == [dummy_obj, dummy_obj2]
engine.put(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", "key1", dummy_obj)
engine.put(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", "key2", dummy_obj2)
from_db1 = engine.get(FAKE_USER_ID, "MyEntry", "key1")
from_db2 = engine.get(FAKE_USER_ID, "MyEntry", "key2")
assert from_db1 == dummy_obj
assert from_db2 == dummy_obj2
as_dict = engine.load(FAKE_USER_ID, "MyEntry")
assert "key1" in as_dict
assert "key2" in as_dict
assert as_dict["key1"] == dummy_obj
assert as_dict["key2"] == dummy_obj2
def test_i_automatically_replace_keys(engine, dummy_obj, dummy_obj2):
engine.put(FAKE_USER_ID, "MyEntry", "key1", dummy_obj)
engine.put(FAKE_USER_ID, "MyEntry", "key1", dummy_obj2)
from_db1 = engine.get(FAKE_USER_ID, "MyEntry", "key1")
assert from_db1 == dummy_obj2
all_items = engine.get(FAKE_USER_ID, "MyEntry")
assert all_items == [dummy_obj2]
engine.put(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", "key1", dummy_obj)
engine.put(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", "key1", dummy_obj2)
from_db1 = engine.get(FAKE_USER_ID, "MyEntry", "key1")
assert from_db1 == dummy_obj2
all_items = engine.get(FAKE_USER_ID, "MyEntry")
assert all_items == [dummy_obj2]
def test_i_do_not_save_twice_when_the_entries_are_the_same(engine, dummy_obj):
engine.put(FAKE_USER_ID, "MyEntry", "key1", dummy_obj)
entry_content = engine.load(FAKE_USER_ID, "MyEntry")
assert entry_content[TAG_PARENT] == [None]
# Save the same entry again
engine.put(FAKE_USER_ID, "MyEntry", "key1", dummy_obj)
entry_content = engine.load(FAKE_USER_ID, "MyEntry")
assert entry_content[TAG_PARENT] == [None] # still no other parent
engine.put(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", "key1", dummy_obj)
entry_content = engine.load(FAKE_USER_ID, "MyEntry")
assert entry_content[TAG_PARENT] == [None]
# Save the same entry again
engine.put(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", "key1", dummy_obj)
entry_content = engine.load(FAKE_USER_ID, "MyEntry")
assert entry_content[TAG_PARENT] == [None] # still no other parent
def test_i_can_put_many(engine, dummy_obj, dummy_obj2):
engine.put_many(FAKE_USER_ID, "MyEntry", [dummy_obj, dummy_obj2])
from_db1 = engine.get(FAKE_USER_ID, "MyEntry", "spread_sheet")
from_db2 = engine.get(FAKE_USER_ID, "MyEntry", "spread_sheet2")
assert from_db1 == dummy_obj
assert from_db2 == dummy_obj2
entry_content = engine.load(FAKE_USER_ID, "MyEntry")
assert entry_content[TAG_PARENT] == [None] # only one save was made
def test_i_can_put_many(engine):
dummy_obj = DummyObjWithKey("1", "a", True)
dummy_obj2 = DummyObjWithKey("2", "b", False)
engine.put_many(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", [dummy_obj, dummy_obj2])
from_db1 = engine.get(FAKE_USER_ID, "MyEntry", "1")
from_db2 = engine.get(FAKE_USER_ID, "MyEntry", "2")
assert from_db1 == dummy_obj
assert from_db2 == dummy_obj2
entry_content = engine.load(FAKE_USER_ID, "MyEntry")
assert entry_content[TAG_PARENT] == [None] # only one save was made
def test_i_can_do_not_save_in_not_necessary(engine, dummy_obj, dummy_obj2):
engine.put_many(FAKE_USER_ID, "MyEntry", [dummy_obj, dummy_obj2])
engine.put_many(FAKE_USER_ID, "MyEntry", [dummy_obj, dummy_obj2])
entry_content = engine.load(FAKE_USER_ID, "MyEntry")
assert entry_content[TAG_PARENT] == [None] # Still None, nothing was saved
def test_put_many_save_only_if_necessary(engine):
dummy_obj = DummyObjWithKey("1", "a", True)
dummy_obj2 = DummyObjWithKey("2", "b", False)
engine.put_many(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", [dummy_obj, dummy_obj2])
engine.put_many(FAKE_USER_ID, FAKE_USER_EMAIL, "MyEntry", [dummy_obj, dummy_obj2])
entry_content = engine.load(FAKE_USER_ID, "MyEntry")
assert entry_content[TAG_PARENT] == [None] # Still None, nothing was save

8
tests/test_repositories.py
View File

@@ -2,7 +2,7 @@ import pytest
from fasthtml.components import *
from components.addstuff.constants import ROUTE_ROOT, Routes
from components.addstuff.settings import Repository, MyTable, AddStuffSettings
from components.addstuff.settings import Repository, RepositoriesSettings
from core.settings_management import SettingsManager, MemoryDbEngine
from helpers import matches, StartsWith, div_icon, find_first_match, search_elements_by_path
from src.components.addstuff.components.Repositories import Repositories
@@ -75,7 +75,7 @@ def test_render_no_repository(repositories):
def test_render_when_repo_and_tables(db_engine, repositories):
db_engine.init_db(USER_ID, 'AddStuffSettings', AddStuffSettings([
db_engine.init_db(USER_ID, 'AddStuffSettings', RepositoriesSettings([
Repository("repo 1", [MyTable("table 1"), MyTable("table 2")]),
Repository("repo 2", [MyTable("table 3")]),
]))
@@ -126,7 +126,7 @@ def test_i_can_add_new_repository(repositories):
def test_i_can_click_on_repo(db_engine, repositories):
db_engine.init_db(USER_ID, 'AddStuffSettings', AddStuffSettings([
db_engine.init_db(USER_ID, 'AddStuffSettings', RepositoriesSettings([
Repository("repo 1", [])
]))
@@ -141,7 +141,7 @@ def test_i_can_click_on_repo(db_engine, repositories):
def test_render_i_can_click_on_table(db_engine, repositories, tabs_manager):
db_engine.init_db(USER_ID, 'AddStuffSettings', AddStuffSettings([
db_engine.init_db(USER_ID, 'AddStuffSettings', RepositoriesSettings([
Repository("repo 1", [MyTable("table 1")])
]))

89
tests/test_repositories_db_manager.py Normal file
View File

@@ -0,0 +1,89 @@
import pytest
from components.addstuff.settings import RepositoriesDbManager, RepositoriesSettings, Repository, \
REPOSITORIES_SETTINGS_ENTRY
from core.settings_management import SettingsManager, MemoryDbEngine
@pytest.fixture
def settings_manager():
return SettingsManager(MemoryDbEngine())
@pytest.fixture
def db(session, settings_manager):
return RepositoriesDbManager(session, settings_manager)
def test_add_new_repository(db, settings_manager):
"""Test adding a new repository with valid data."""
db.add_repository("NewRepo", ["Table1", "Table2"])
settings = settings_manager.get(db.session, REPOSITORIES_SETTINGS_ENTRY)
assert len(settings.repositories) == 1
assert settings.repositories[0].name == "NewRepo"
assert settings.repositories[0].tables == ["Table1", "Table2"]
def test_add_repository_duplicate_name(db, settings_manager):
"""Test adding a repository with an existing name."""
settings = RepositoriesSettings()
settings.repositories.append(Repository(name="ExistingRepo", tables=[]))
settings_manager.put(db.session, REPOSITORIES_SETTINGS_ENTRY, settings)
with pytest.raises(ValueError, match="Repository 'ExistingRepo' already exists."):
db.add_repository("ExistingRepo")
def test_add_repository_empty_name(db):
"""Test adding a repository with an empty name."""
with pytest.raises(ValueError, match="Repository name cannot be empty."):
db.add_repository("")
def test_add_repository_none_name(db):
"""Test adding a repository with a None name."""
with pytest.raises(ValueError, match="Repository name cannot be empty."):
db.add_repository(None)
def test_add_repository_no_tables(db, settings_manager):
"""Test adding a repository without specifying tables."""
db.add_repository("RepoWithoutTables")
settings = settings_manager.get(db.session, "Repositories")
assert len(settings.repositories) == 1
assert settings.repositories[0].name == "RepoWithoutTables"
assert settings.repositories[0].tables == []
def test_get_existing_repository(db, settings_manager):
"""Test retrieving an existing repository."""
# Pre-populate settings with a repository
settings = RepositoriesSettings()
repo = Repository(name="ExistingRepo", tables=["Table1"])
settings.repositories.append(repo)
settings_manager.put(db.session, "Repositories", settings)
# Retrieve the repository
retrieved_repo = db.get_repository("ExistingRepo")
# Verify the repository is correctly returned
assert retrieved_repo.name == "ExistingRepo"
assert retrieved_repo.tables == ["Table1"]
def test_get_repository_not_found(db):
"""Test retrieving a repository that does not exist."""
with pytest.raises(ValueError, match="Repository 'NonExistentRepo' does not exists."):
db.get_repository("NonExistentRepo")
def test_get_repository_empty_name(db):
"""Test retrieving a repository with an empty name."""
with pytest.raises(ValueError, match="Repository name cannot be empty."):
db.get_repository("")
def test_get_repository_none_name(db):
"""Test retrieving a repository with a None name."""
with pytest.raises(ValueError, match="Repository name cannot be empty."):
db.get_repository(None)