diff --git a/.vscode/settings.json b/.vscode/settings.json
index 9a329fa..084a4a9 100644
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@@ -17,4 +17,5 @@
"editor.defaultFormatter": "charliermarsh.ruff",
"ruff.enable": true,
"ruff.organizeImports": true,
+ "editor.renderWhitespace": "all",
}
diff --git a/.vscode/tasks.json b/.vscode/tasks.json
index d4db1b4..2d57213 100644
--- a/.vscode/tasks.json
+++ b/.vscode/tasks.json
@@ -61,8 +61,7 @@
"clear": true,
"focus": true
}, "problemMatcher": []
- },
- {
+ }, {
"label": "UV Sync ",
"type": "shell",
"command": "uv sync",
@@ -77,6 +76,22 @@
"focus": true
},
"problemMatcher": []
+ },
+ {
+ "label": "Fix Code with Ruff",
+ "type": "shell",
+ "command": "ruff check --fix .",
+ "options": {
+ "cwd": "${workspaceFolder}/src"
+ },
+ "group": "none",
+ "presentation": {
+ "reveal": "always",
+ "panel": "dedicated",
+ "clear": true,
+ "focus": true
+ },
+ "problemMatcher": []
}
]
}
diff --git a/pyproject.toml b/pyproject.toml
index 4804f36..d178565 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -31,6 +31,7 @@ plugins.jedi_definition.enabled = true
[tool.ruff]
line-length = 100
target-version = "py313"
+fix = true
[tool.ruff.lint]
select = ["E", "F", "I", "W"] # Error, F: PyFlakes, I: isort, W: pycodestyle warnings
diff --git a/src/config.py b/src/config.py
index 975c89e..851119d 100644
--- a/src/config.py
+++ b/src/config.py
@@ -10,6 +10,7 @@ def __init__(self):
self.CONFIG_FILE_DIRECTORY = "config_files"
self.LOGGING_CONFIG = self.load_config_file("logging_config.yaml")
self.DATABASE_SCHEMA = self.load_config_file("verity_schema.yaml")
+ self.DEFAULT_DATA = self.load_config_file("default_data.yaml")
def load_config_file(self, file):
config = ""
diff --git a/src/config_files/default_data.yaml b/src/config_files/default_data.yaml
new file mode 100644
index 0000000..fb9cac2
--- /dev/null
+++ b/src/config_files/default_data.yaml
@@ -0,0 +1,4 @@
+category:
+ - name: 'internal_master_category'
+ - name: 'tesing more categories'
+ budget_value: '100'
diff --git a/src/currency_handler.py b/src/currency_handler.py
new file mode 100644
index 0000000..cb92036
--- /dev/null
+++ b/src/currency_handler.py
@@ -0,0 +1,23 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+# TODO: Make this into a proper class so we can scale currrency handling
+
+
+def convert_to_universal_currency(input_value: float) -> int:
+ """
+ Converts the the input value to remove all decimal places and return an int.
+ This will be the starting point for our universal currency,
+ (see docs/data_dictionary).
+ for now, we will just focus on making this an int.
+ it will need change later once we have the basics done
+ """
+ logger.info(f"received {input_value} to convert to universal currency")
+ input_value = float(input_value)
+ while input_value % 1 != 0:
+ logger.debug(f"input value is not a whole number {input_value}")
+ input_value = input_value * 10
+ logger.info(f"returning {int(input_value)}")
+ return int(input_value)
+
diff --git a/src/data_handler.py b/src/data_handler.py
index 5fd0b0f..c1d970d 100644
--- a/src/data_handler.py
+++ b/src/data_handler.py
@@ -14,10 +14,13 @@ def __init__(self, config) -> None:
self.verity_config = config
self.schema = self.verity_config.DATABASE_SCHEMA
self.database = self.verity_config.DATABASE
+ self.default_data = self.verity_config.DEFAULT_DATA
- def execute_sql(self, sql_statement: str, return_id: bool = False) -> (bool, int):
+ def execute_sql(
+ self, sql_statement: str, params: tuple = (), return_id: bool = False, seed: bool = False
+ ) -> (bool, int):
"send the query here, returns true if successful, false if fail"
- logging.debug(f"received request to execute {sql_statement}")
+ logger.debug(f"received request to execute {sql_statement} with params {params}")
new_id: int = 0
is_success: bool = False
try:
@@ -25,45 +28,50 @@ def execute_sql(self, sql_statement: str, return_id: bool = False) -> (bool, int
database=self.database,
timeout=10, # seconds i hope
)
- logging.debug("opened connection to database")
+ logger.info("opened connection to database")
cursor = connection.cursor()
- cursor.execute(sql_statement, {})
+ cursor.execute("PRAGMA foreign_keys = ON;")
+ if seed:
+ cursor.execute("PRAGMA foreign_keys = OFF;")
+ cursor.execute(sql_statement, params)
connection.commit()
+ logger.debug(f"executed {sql_statement} with params: {params}")
+ logger.info("executed sql command")
if return_id:
new_id = cursor.lastrowid
+ logger.debug(f"New id created {new_id}")
is_success = True
except Exception as e: # TODO: better Exception handling
- logging.error(e)
+ logger.error(e)
is_success = False
finally:
+ cursor.execute("PRAGMA foreign_keys = ON;")
connection.close()
+ logger.info("closed connection to database")
if return_id:
return (is_success, new_id)
else:
return is_success
- def read_database(self, sql_statement: str):
+ def read_database(self, sql_statement: str, params: tuple = ()) -> list:
"reads the database query and returns the results"
- logging.debug(f"received request to read {sql_statement}")
- results = 0
+ logger.debug(f"received request to read {sql_statement} with params {params}")
+ results = []
try:
connection = sqlite3.connect(self.database)
- logging.debug("opened connection to database")
cursor = connection.cursor()
- results = cursor.execute(sql_statement, {})
+ results = cursor.execute(sql_statement, params)
results = results.fetchall()
- logging.debug(f"query returned: {results}")
+ logger.debug(f"query returned: {results}")
except Exception as e:
- logging.error(f"Read error occured: {e}")
+ logger.error(f"Read error occurred: {e}")
finally:
connection.close()
return results
@staticmethod
def _build_column(column: dict) -> str:
- logging.debug(
- f"building column {column}"
- )
+ logging.debug(f"building column {column}")
name = column["column_name"]
is_pk = column["is_pk"]
is_fk = column.get("is_fk")
@@ -78,7 +86,6 @@ def _build_column(column: dict) -> str:
column_string += " NOT NULL"
return column_string
-
def _add_table_to_db(self, table: dict) -> bool:
"Creates the table in the verity database, based on the schema yaml"
sql = f"""CREATE TABLE IF NOT EXISTS {table["table_name"]} (
@@ -133,9 +140,7 @@ def print_table_schema(self, table_name):
# Print the schema
logger.debug(f"Schema for table: {table_name}")
for row in cursor.fetchall():
- logger.debug(
- f"Column Name: {row[1]}, Data Type: {row[2]}, Not Null: {row[3]}"
- )
+ logger.debug(f"Column Name: {row[1]}, Data Type: {row[2]}, Not Null: {row[3]}")
except Exception as e:
logger.error(f"An error occurred: {e}")
@@ -148,42 +153,51 @@ def print_table_schema(self, table_name):
def add_user_name(self, user_name: str) -> int:
"takes user name string, returns user id"
- logger.debug(
- f"attempting to insert values into user table {user_name}"
- )
- try:
- connection = sqlite3.connect(self.database)
- logger.debug("connection to db open")
- cursor = connection.cursor()
- logger.debug("cursor activated")
- cursor.execute(
- """INSERT INTO user (
- name
- )
- VALUES (?)
- """,
- (user_name,),
- )
- logger.debug("cursor executed")
- connection.commit()
- user_id = cursor.lastrowid
- logger.debug(f"insert attempt seems successful, user id is {user_id}")
-
- if user_id is None:
- user_id = 0
- except Exception as e:
- logger.error(f"Failed to insert user name, error: {e}")
- user_id = 0
- finally:
- try:
- connection.close()
- logger.debug("connection to db closed")
- return user_id
- except Exception as e:
- logger.error(f"failed to close connection message: {e}")
- return 0
+ logger.debug(f"attempting to insert values into user table {user_name}")
+ sql_statement = """
+ INSERT INTO user (name)
+ VALUES (?)
+ """
+ params = (user_name,)
+ success, user_id = self.execute_sql(sql_statement, params, True)
+ if not success:
+ logger.error("Failed to execute sql, check the logs")
+ self.add_category(user_id, "internal_master_category", seed=True)
+ return user_id
def get_users(self) -> list:
- get_user_sql = "SELECT name FROM user"
- users = self.read_database(get_user_sql)
- return users
+ """Returns all users in the database."""
+ get_user_sql = "SELECT id, name FROM user"
+ return self.read_database(get_user_sql)
+
+ def add_category(
+ self, user_id: int, category_name: str, budget_value: int = 0, parent_id=None, seed=False
+ ) -> int:
+ """Inserts a new category. Returns the category id."""
+ logger.debug(f"attempting to insert category '{category_name}' for user {user_id}")
+ sql_statement = """
+ INSERT INTO category (user_id, name, budget_value, parent_id)
+ VALUES (?, ?, ?, ?)"""
+ if not seed:
+ if not parent_id:
+ parent_id = self.read_database(
+ "SELECT id FROM category WHERE user_id = ? AND name = ?",
+ (user_id, "internal_master_category"),
+ )
+ try:
+ parent_id = parent_id[0][0]
+ except IndexError:
+ parent_id = None
+ params = (user_id, category_name, budget_value, parent_id)
+
+ success, category_id = self.execute_sql(sql_statement, params, True, seed)
+ if not success:
+ logger.error("Failed to execute sql, check the logs")
+ return category_id
+
+ def get_categories(self, user_id: int) -> list:
+ """Returns all categories for a given user."""
+ sql = (
+ "SELECT id, name, budget_value, parent_id FROM category WHERE user_id = ? and name != ?"
+ )
+ return self.read_database(sql_statement=sql, params=(user_id, "internal_master_category"))
diff --git a/src/front/home.py b/src/front/home.py
index 160989a..0bf3eb0 100644
--- a/src/front/home.py
+++ b/src/front/home.py
@@ -2,6 +2,7 @@
from flask import Blueprint, flash, redirect, render_template, request, session, url_for
+import currency_handler
import data_handler
from config import VerityConfig
@@ -16,25 +17,119 @@ def home_page():
logger.info("home page hit")
db_call = data_handler.database(verity_config)
users = db_call.get_users()
- return render_template("home.html", users=users)
+ # Get user info directly from session
+ user_id = session.get("user_id")
+ selected_user_name = session.get("user_name")
+ # Debug log to see what's being passed to the template
+ logger.debug(f"User ID from session: {user_id}, User Name from session: {selected_user_name}")
+ logger.info(f"{selected_user_name} is logged in")
-@home_bp.route("/submit", methods=["POST"])
+ categories = db_call.get_categories(user_id) if user_id else []
+ return render_template(
+ "home.html",
+ users=users,
+ categories=categories,
+ selected_user_id=user_id,
+ selected_user_name=selected_user_name,
+ )
+
+
+@home_bp.route("/submit_user", methods=["POST"])
def submit_user_name():
user_name = request.form.get("userName")
logger.debug(f"Request Received: {request.form}")
- if user_name:
- logger.info(f"User submitted new user name: {user_name}")
- session["user_name"] = user_name
- db_call = data_handler.database(verity_config)
- user_id = db_call.add_user_name(user_name)
- if user_id == 0:
- # db entry failed, throw error message
- flash("User Name not saved, please check the logs", "error")
- return redirect(url_for("home.home_page"))
- session["user_id"] = user_id
- flash("user name saved!", "success")
+ if not user_name:
+ flash("Please enter a user name.", "danger")
+ return redirect(url_for("home.home_page"))
+
+ db_call = data_handler.database(verity_config)
+ # More efficient check for existing user
+ exists = db_call.read_database("SELECT 1 FROM user WHERE name = ?", (user_name,))
+ if exists:
+ logger.warning(f"username already exists in the database: {user_name}")
+ flash("A user with that name already exists.", "danger")
+ return redirect(url_for("home.home_page"))
+ logger.info(f"User submitted new user name: {user_name}")
+ user_id = db_call.add_user_name(user_name)
+ if user_id == 0:
+ flash("User Name not saved, please check the logs", "danger")
return redirect(url_for("home.home_page"))
+ session["user_id"] = int(user_id)
+ session["user_name"] = user_name
+ flash("user name saved! Start adding categories.", "success")
+ return redirect(url_for("home.home_page"))
+
+
+@home_bp.route("/select_user", methods=["POST"])
+def select_user():
+ selected_user_id = request.form.get("selectedUserId")
+ if not selected_user_id:
+ flash("No user selected", "error")
+ session.pop("user_id", None)
+ session.pop("user_name", None)
+ return redirect(url_for("home.home_page"))
+
+ # Convert selected_user_id to int and handle invalid input
+ try:
+ selected_user_id = int(selected_user_id)
+ except ValueError:
+ logger.error(f"Invalid user ID: {selected_user_id}")
+ flash("Invalid user ID", "error")
+ return redirect(url_for("home.home_page"))
+
+ # Get the user details directly from the database using the ID
+ db_call = data_handler.database(verity_config)
+ result = db_call.read_database("SELECT name FROM user WHERE id = ?", (selected_user_id,))
+
+ if result and result[0]:
+ # Store both ID and name in the session
+ session["user_id"] = selected_user_id
+ session["user_name"] = result[0][0]
+ flash("User selected!", "success")
else:
- flash("Please enter a user name.", "error")
+ session.pop("user_id", None)
+ session.pop("user_name", None)
+ flash("User not found!", "danger")
+
+ return redirect(url_for("home.home_page"))
+
+
+@home_bp.route("/submit_category", methods=["POST"])
+def submit_category():
+ user_id = session.get("user_id")
+ if not user_id:
+ logger.warning("No user selected and trying to add a category... how?")
+ flash("No user selected!", "danger")
return redirect(url_for("home.home_page"))
+
+ # Get and validate category name (required)
+ category_name = request.form.get("categoryName")
+
+ # Convert budget_value to our universal currency
+ budget_value: int = 0
+ budget_value_input = request.form.get("budgetValue", "0")
+ logger.debug(f"budget value input: {budget_value_input}")
+ if not budget_value_input:
+ logger.info("no budget value assigned to category")
+ else:
+ logger.info(f"Category:{category_name}, Assigned amount: {budget_value_input}")
+ # User might either add a whole currency or a decimal of. we need to handle both
+ if float(budget_value_input) < 0:
+ logger.info("user is stupid and tried to assign a negative amount to the category")
+ flash("Negative amounts don`t really make sense here, removed budget amount", "danger")
+ budget_value_input = 0
+ budget_value = currency_handler.convert_to_universal_currency(budget_value_input)
+ # Convert parent_id to int if provided
+ parent_id = request.form.get("parentId", "0").strip()
+
+ db_call = data_handler.database(verity_config)
+ if parent_id == 0:
+ category_id = db_call.add_category(user_id, category_name, budget_value)
+ else:
+ category_id = db_call.add_category(user_id, category_name, budget_value, parent_id)
+ if category_id == 0:
+ flash("Category not saved, please check the logs", "danger")
+ else:
+ flash("Category saved!", "success")
+ return redirect(url_for("home.home_page"))
diff --git a/src/front/templates/home.html b/src/front/templates/home.html
index 5f5d5ed..667cec8 100644
--- a/src/front/templates/home.html
+++ b/src/front/templates/home.html
@@ -1,25 +1,65 @@
-{% extends "base.html" %} {% block title %} Home {% endblock %} {% block content
-%}
+{% extends "base.html" %}
+{% block title %} Home {% endblock %}
+{% block content %}
Home of Verity
+
+
+
+{% if selected_user_id %}
+
+
- {% for user in users %}
-- {{ user[0] }}
+ Your Categories:
+ {% for category in categories %}
+ - {{ category[1] }}{% if category[3] %} (Sub of ID {{ category[3] }}){% endif %}{% if category[2] %} [Budget: {{ category[2] }}]{% endif %}
{% endfor %}
-{% endblock %}
+{% else %}
+
+ Please select a user before adding a category.
+
+{% endif %}
+{% endblock %}
\ No newline at end of file
diff --git a/src/tests/test_currency_handler.py b/src/tests/test_currency_handler.py
new file mode 100644
index 0000000..8111018
--- /dev/null
+++ b/src/tests/test_currency_handler.py
@@ -0,0 +1,21 @@
+
+from src import currency_handler
+
+
+def test_convert_to_universal_currency():
+ input = 87.82
+ result = currency_handler.convert_to_universal_currency(input)
+ assert result == 8782
+
+
+def test_whole_number():
+ input = 123
+ result = currency_handler.convert_to_universal_currency(input)
+ assert result == input
+
+
+def test_negative_number():
+ input = -123.45
+ result = currency_handler.convert_to_universal_currency(input)
+ print(result)
+ assert result == -12345
diff --git a/src/tests/test_data_handler.py b/src/tests/test_data_handler.py
index 85f8f01..283ae80 100644
--- a/src/tests/test_data_handler.py
+++ b/src/tests/test_data_handler.py
@@ -1,3 +1,6 @@
+import os
+import tempfile
+
import pytest
from src import data_handler
@@ -6,54 +9,178 @@
@pytest.fixture
def test_db_call():
+ """Create a test database instance with a unique test database file"""
config = VerityConfig()
+ # Use a test-specific database file
+ with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as temp_db:
+ test_db_file = temp_db.name
+
+ # Create a config with the test DB
+ config.DATABASE = test_db_file
db_call = data_handler.database(config)
db_call.build_database()
yield db_call
+ # Cleanup after tests
+ if os.path.exists(test_db_file):
+ os.remove(test_db_file)
def test_execute_sql_success(test_db_call):
+ """Test executing valid SQL statements"""
sql_statement = """INSERT INTO user (
name
)
VALUES ('a_user_name')
"""
result = test_db_call.execute_sql(sql_statement)
- assert result
+ assert result is True
def test_execute_sql_error(test_db_call):
+ """Test executing invalid SQL statements"""
sql_statement = "SELECT * FROM non_existent_table"
result = test_db_call.execute_sql(sql_statement)
- assert not result
- # Check if an error message is logged
- # You might need to add a logging assertion here
+ assert result is False
+
+
+def test_execute_sql_with_return_id(test_db_call):
+ """Test executing SQL with return_id flag"""
+ sql_statement = """INSERT INTO user (
+ name
+ )
+ VALUES ('return_id_test')
+ """
+ result, new_id = test_db_call.execute_sql(sql_statement, return_id=True)
+ assert result is True
+ assert new_id > 0
def test_read_database(test_db_call):
- # Test reading database data
- sql_statement = "SELECT name FROM user"
+ """Test reading database data"""
+ # First insert a user
+ test_db_call.add_user_name("read_test_user")
+ # Then read it back
+ sql_statement = "SELECT id, name FROM user WHERE name = 'read_test_user'"
results = test_db_call.read_database(sql_statement)
assert isinstance(results, list)
- # Check the contents of the list
- # self.assertIsNotNone(results)
- # self.assertIsInstance(results[0], str)
+ assert len(results) > 0
+ assert results[0][1] == "read_test_user"
+
+
+def test_read_database_with_params(test_db_call):
+ """Test reading database with parameterized query"""
+ # First insert a user
+ test_db_call.add_user_name("param_test_user")
+ # Then read it back with params
+ sql_statement = "SELECT id, name FROM user WHERE name = ?"
+ params = ("param_test_user",)
+ results = test_db_call.read_database(sql_statement, params)
+ assert isinstance(results, list)
+ assert len(results) > 0
+ assert results[0][1] == "param_test_user"
+
+
+def test_build_database(test_db_call):
+ """Test that the database builds correctly with all required tables"""
+ # Check if all tables have been created
+ config = VerityConfig()
+ for table in config.DATABASE_SCHEMA["tables"]:
+ table_name = table["table_name"]
+ sql = f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table_name}';"
+ results = test_db_call.read_database(sql)
+ assert len(results) == 1, f"Table '{table}' should exist"
def test_add_user_name(test_db_call):
- # Test adding a new user name
+ """Test adding a new user name"""
user_name = "Test user"
user_id = test_db_call.add_user_name(user_name)
- assert user_id is not None
+ assert user_id > 0
+
# Check if the user name was actually added
- # You might need to query the database to verify
- # self.assertEqual(user_id, 1)
+ sql = "SELECT id, name FROM user WHERE name = ?"
+ results = test_db_call.read_database(sql, (user_name,))
+ assert len(results) == 1
+ assert results[0][1] == user_name
+
+
+def test_add_duplicate_user_name(test_db_call):
+ """Test adding a duplicate user name (should succeed at DB level without constraints)"""
+ user_name = "Duplicate User"
+ # Add first user
+ user_id1 = test_db_call.add_user_name(user_name)
+ assert user_id1 > 0
+
+ # Add second user with same name
+ user_id2 = test_db_call.add_user_name(user_name)
+ assert user_id2 > 0
+ assert user_id2 != user_id1 # They should have different IDs
def test_get_users(test_db_call):
- # Test getting users
+ """Test getting all users"""
+ # Add some test users first
+ test_db_call.add_user_name("User One")
+ test_db_call.add_user_name("User Two")
+
+ # Get users
users = test_db_call.get_users()
assert isinstance(users, list)
- # Check the contents of the list
- # self.assertIsNotNone(users)
- # self.assertIsInstance(users[0], str)
+ assert len(users) >= 2
+
+ # Check that users are returned as (id, name) tuples
+ for user in users:
+ assert len(user) == 2
+ assert isinstance(user[0], int) # ID
+ assert isinstance(user[1], str) # Name
+
+
+def test_add_category_success(test_db_call):
+ # Add a user first, since category requires a user_id
+ user_id = test_db_call.add_user_name("CategoryTestUser")
+ assert user_id != 0
+ # Add a category with all fields
+ category_id = test_db_call.add_category(user_id, "Groceries", 200.0, None)
+ assert category_id != 0
+ # Add a category with only required fields
+ category_id2 = test_db_call.add_category(user_id, "Utilities")
+ assert category_id2 != 0
+
+
+def test_add_category_null_budget_and_parent(test_db_call):
+ user_id = test_db_call.add_user_name("NullBudgetParentUser")
+ assert user_id != 0
+ # Add a category with None for budget_value and parent_id
+ category_id = test_db_call.add_category(user_id, "NoBudgetOrParent", None, None)
+ assert category_id != 0
+
+
+def test_add_category_and_get_categories(test_db_call):
+ # Add a user
+ user_name = "CategoryTestUser"
+ user_id = test_db_call.add_user_name(user_name)
+ assert user_id != 0
+ # Add a category for this user
+ category_name = "Groceries"
+ budget_value = 100.0
+ parent_id = None
+ category_id = test_db_call.add_category(user_id, category_name, budget_value, parent_id)
+ assert category_id != 0
+ # Add a subcategory
+ subcategory_name = "Supermarket"
+ subcategory_id = test_db_call.add_category(user_id, subcategory_name, 50.0, category_id)
+ assert subcategory_id != 0
+ # Retrieve categories for this user
+ categories = test_db_call.get_categories(user_id)
+ assert isinstance(categories, list)
+ names = [cat[1] for cat in categories]
+ assert category_name in names
+ assert subcategory_name in names
+
+
+def test_add_category_invalid_user(test_db_call):
+ # Try to add a category with a non-existent user_id
+ # (should still succeed in SQLite unless foreign keys are enforced)
+ category_id = test_db_call.add_category(99999, "InvalidUserCategory")
+ assert category_id == 0
+ assert isinstance(category_id, int)
diff --git a/src/verity.py b/src/verity.py
index 69df55b..0bf866c 100644
--- a/src/verity.py
+++ b/src/verity.py
@@ -28,6 +28,7 @@ def set_up_logging(config):
# database initialise
verity = database(verity_config)
+ logger.debug(verity_config.DEFAULT_DATA)
verity.build_database()
# app initialise