Compare commits
5 Commits
AddingWorf
...
AddingNRTW
| Author | SHA1 | Date | |
|---|---|---|---|
| a406440126 | |||
| ef88d2925e | |||
| 94214125fd | |||
| 17b08be077 | |||
| db56363b1f |
7
.gitignore
vendored
7
.gitignore
vendored
@@ -8,11 +8,18 @@ htmlcov
|
||||
.venv
|
||||
tests/settings_from_unit_testing.json
|
||||
tests/TestDBEngineRoot
|
||||
tests/*.png
|
||||
src/*.png
|
||||
tests/*.html
|
||||
tests/*.txt
|
||||
test-results
|
||||
.sesskey
|
||||
tools.db
|
||||
.mytools_db
|
||||
.idea/MyManagingTools.iml
|
||||
.idea/misc.xml
|
||||
.idea/dataSources.xml
|
||||
.idea/sqldialects.xml
|
||||
.idea_bak
|
||||
**/*.prof
|
||||
|
||||
|
||||
6
.idea/inspectionProfiles/Project_Default.xml
generated
6
.idea/inspectionProfiles/Project_Default.xml
generated
@@ -1,6 +0,0 @@
|
||||
<component name="InspectionProjectProfileManager">
|
||||
<profile version="1.0">
|
||||
<option name="myName" value="Project Default" />
|
||||
<inspection_tool class="PyInitNewSignatureInspection" enabled="false" level="WARNING" enabled_by_default="false" />
|
||||
</profile>
|
||||
</component>
|
||||
34
Makefile
34
Makefile
@@ -1,4 +1,5 @@
|
||||
.PHONY: test
|
||||
.PHONY: test install-playwright install-playwright-deps test-e2e test-e2e-headed test-regression
|
||||
|
||||
|
||||
test:
|
||||
pytest
|
||||
@@ -7,6 +8,34 @@ coverage:
|
||||
coverage run --source=src -m pytest
|
||||
coverage html
|
||||
|
||||
install-playwright:
|
||||
@echo "Installing Playwright and pytest-playwright via pip..."
|
||||
pip install playwright pytest-playwright
|
||||
@echo "Installing Playwright browsers..."
|
||||
playwright install
|
||||
@echo "Installing system dependencies for Ubuntu/WSL2..."
|
||||
playwright install-deps
|
||||
@echo "Playwright installation complete!"
|
||||
|
||||
install-playwright-deps:
|
||||
@echo "Installing system dependencies for Playwright on Ubuntu/WSL2..."
|
||||
playwright install-deps
|
||||
|
||||
test-e2e:
|
||||
pytest tests/test_e2e_regression.py -m "e2e" --browser chromium
|
||||
|
||||
test-e2e-headed:
|
||||
pytest -m "e2e" --browser chromium --headed
|
||||
|
||||
test-e2e-headed-slowmo:
|
||||
pytest -m "priority1" --browser chromium --headed --slowmo=2000
|
||||
|
||||
test-regression:
|
||||
pytest tests/test_e2e_regression.py -m "regression" --browser chromium
|
||||
|
||||
test-all-browsers:
|
||||
pytest tests/test_e2e_regression.py -m "e2e" --browser chromium --browser firefox --browser webkit
|
||||
|
||||
clean:
|
||||
rm -rf build
|
||||
rm -rf htmlcov
|
||||
@@ -20,6 +49,9 @@ clean:
|
||||
rm -rf src/tools.db
|
||||
rm -rf src/*.out
|
||||
rm -rf src/*.prof
|
||||
rm -rf tests/debug*.txt
|
||||
rm -rf tests/debug*.html
|
||||
rm -rf tests/debug*.png
|
||||
find . -name '.sesskey' -exec rm -rf {} +
|
||||
find . -name '.pytest_cache' -exec rm -rf {} +
|
||||
find . -name '__pycache__' -exec rm -rf {} +
|
||||
|
||||
14
README.md
14
README.md
@@ -42,3 +42,17 @@ cd src
|
||||
python -m cProfile -o profile.out main.py
|
||||
snakeviz profile.out # 'pip install snakeviz' if snakeviz is not installed
|
||||
```
|
||||
|
||||
|
||||
# End to end testing
|
||||
```shell
|
||||
make install-playwright
|
||||
```
|
||||
|
||||
Alternatively, you can install Playwright and pytest-playwright via pip:
|
||||
```shell
|
||||
pip install playwright pytest-playwright
|
||||
playwright install
|
||||
playwright install-deps # may be required on Linux
|
||||
playwright --version
|
||||
```
|
||||
|
||||
11
pytest.ini
Normal file
11
pytest.ini
Normal file
@@ -0,0 +1,11 @@
|
||||
[pytest]
|
||||
python_files = test_*.py
|
||||
python_classes = Test*
|
||||
python_functions = test_*
|
||||
testpaths = tests
|
||||
pythonpath = src tests
|
||||
addopts = --tb=short -v
|
||||
markers =
|
||||
e2e: marks tests as end-to-end tests
|
||||
smoke: marks tests as smoke tests
|
||||
regression: marks tests as regression tests
|
||||
@@ -10,6 +10,8 @@ click==8.1.7
|
||||
et-xmlfile==1.1.0
|
||||
fastcore==1.8.5
|
||||
fastlite==0.2.1
|
||||
gprof2dot==2025.4.14
|
||||
greenlet==3.2.4
|
||||
h11==0.14.0
|
||||
httpcore==1.0.5
|
||||
httptools==0.6.1
|
||||
@@ -26,28 +28,37 @@ oauthlib==3.2.2
|
||||
openpyxl==3.1.5
|
||||
packaging==24.1
|
||||
pandas==2.2.3
|
||||
pandas-stubs==2.3.2.250827
|
||||
playwright==1.55.0
|
||||
pluggy==1.5.0
|
||||
pydantic==2.11.5
|
||||
pydantic-settings==2.9.1
|
||||
pydantic_core==2.33.2
|
||||
pyee==13.0.0
|
||||
Pygments==2.19.1
|
||||
pytest==8.3.3
|
||||
pytest-base-url==2.1.0
|
||||
pytest-mock==3.14.1
|
||||
pytest-playwright==0.7.0
|
||||
python-dateutil==2.9.0.post0
|
||||
python-dotenv==1.0.1
|
||||
python-fasthtml==0.12.21
|
||||
python-multipart==0.0.10
|
||||
python-slugify==8.0.4
|
||||
pytz==2024.2
|
||||
PyYAML==6.0.2
|
||||
requests==2.32.3
|
||||
rich==14.0.0
|
||||
shellingham==1.5.4
|
||||
six==1.16.0
|
||||
snakeviz==2.2.2
|
||||
sniffio==1.3.1
|
||||
soupsieve==2.6
|
||||
sqlite-minutils==3.37.0.post3
|
||||
sse-starlette==2.3.6
|
||||
starlette==0.38.5
|
||||
text-unidecode==1.3
|
||||
tornado==6.5.2
|
||||
typer==0.16.0
|
||||
typing-inspection==0.4.1
|
||||
typing_extensions==4.13.2
|
||||
|
||||
@@ -22,6 +22,7 @@ class AuthManager:
|
||||
session["username"] = user_data["username"]
|
||||
session["user_email"] = user_data["email"]
|
||||
session["is_admin"] = bool(user_data["is_admin"])
|
||||
session["roles"] = UserDAO.get_user_roles_by_id(user_data["id"])
|
||||
|
||||
@staticmethod
|
||||
def logout_user(session) -> None:
|
||||
@@ -130,4 +131,3 @@ class AuthManager:
|
||||
"user_email": "admin@mmt.com",
|
||||
"is_admin": True
|
||||
}
|
||||
|
||||
@@ -6,9 +6,10 @@ class BaseComponent:
|
||||
Base class for all components that need to have a session and an id
|
||||
"""
|
||||
|
||||
def __init__(self, session, _id=None, **kwargs):
|
||||
def __init__(self, session, _id=None, role_name=None, **kwargs):
|
||||
self._session = session
|
||||
self._id = _id or self.create_component_id(session)
|
||||
self._role_name = role_name
|
||||
|
||||
def get_id(self):
|
||||
return self._id
|
||||
@@ -35,6 +36,17 @@ class BaseComponent:
|
||||
def create_component_id(session):
|
||||
pass
|
||||
|
||||
def render(self):
|
||||
pass
|
||||
|
||||
def __ft__(self):
|
||||
if (not self._session["is_admin"] and
|
||||
self._role_name is not None and
|
||||
self._role_name not in self._session["roles"]):
|
||||
return None
|
||||
|
||||
return self.render()
|
||||
|
||||
|
||||
class BaseComponentSingleton(BaseComponent):
|
||||
"""
|
||||
@@ -43,8 +55,8 @@ class BaseComponentSingleton(BaseComponent):
|
||||
|
||||
COMPONENT_INSTANCE_ID = None
|
||||
|
||||
def __init__(self, session, _id=None, settings_manager=None, tabs_manager=None, **kwargs):
|
||||
super().__init__(session, _id, **kwargs)
|
||||
def __init__(self, session, _id=None, role_name=None, settings_manager=None, tabs_manager=None, **kwargs):
|
||||
super().__init__(session, _id, role_name,**kwargs)
|
||||
self._settings_manager = settings_manager
|
||||
self.tabs_manager = tabs_manager
|
||||
|
||||
|
||||
@@ -1,18 +1,18 @@
|
||||
from fasthtml.components import *
|
||||
|
||||
from components.BaseComponent import BaseComponent
|
||||
from components.addstuff.constants import ADD_STUFF_INSTANCE_ID
|
||||
from components.addstuff.constants import ADD_STUFF_INSTANCE_ID, ADD_STUFF_ROLE
|
||||
from components.repositories.components.Repositories import Repositories
|
||||
from core.instance_manager import InstanceManager
|
||||
|
||||
|
||||
class AddStuffMenu(BaseComponent):
|
||||
def __init__(self, session: dict, _id: str, settings_manager=None, tabs_manager=None):
|
||||
super().__init__(session, _id)
|
||||
super().__init__(session, _id, ADD_STUFF_ROLE)
|
||||
self.tabs_manager = tabs_manager # MyTabs component id
|
||||
self.repositories = InstanceManager.get(session, Repositories.create_component_id(session), Repositories)
|
||||
|
||||
def __ft__(self):
|
||||
def render(self):
|
||||
return Div(
|
||||
Div("Add stuff...", tabindex="0"),
|
||||
Ul(
|
||||
|
||||
@@ -1 +1,2 @@
|
||||
ADD_STUFF_INSTANCE_ID = "__AddStuff__"
|
||||
ADD_STUFF_ROLE = "add_stuff"
|
||||
@@ -8,7 +8,7 @@ from components.admin.assets.icons import icon_jira
|
||||
from components.admin.commands import AdminCommandManager
|
||||
from components.admin.components.AdminForm import AdminFormItem, AdminFormType, AdminForm, AdminButton, AdminMessageType
|
||||
from components.admin.components.ImportHolidays import ImportHolidays
|
||||
from components.admin.constants import ADMIN_INSTANCE_ID, ADMIN_AI_BUDDY_INSTANCE_ID, ADMIN_JIRA_INSTANCE_ID
|
||||
from components.admin.constants import ADMIN_INSTANCE_ID, ADMIN_AI_BUDDY_INSTANCE_ID, ADMIN_JIRA_INSTANCE_ID, ADMIN_ROLE
|
||||
from components.aibuddy.assets.icons import icon_brain_ok
|
||||
from components.hoildays.assets.icons import icon_holidays
|
||||
from components.tabs.components.MyTabs import MyTabs
|
||||
@@ -19,7 +19,7 @@ from core.jira import Jira
|
||||
|
||||
class Admin(BaseComponent):
|
||||
def __init__(self, session, _id: str = None, settings_manager=None, tabs_manager: MyTabs = None):
|
||||
super().__init__(session, _id)
|
||||
super().__init__(session, _id, ADMIN_ROLE)
|
||||
self.settings_manager = settings_manager
|
||||
self.tabs_manager = tabs_manager
|
||||
self.commands = AdminCommandManager(self)
|
||||
@@ -117,7 +117,7 @@ class Admin(BaseComponent):
|
||||
form.set_message(f"Error {res.status_code} - {res.text}", AdminMessageType.ERROR)
|
||||
return self.tabs_manager.render()
|
||||
|
||||
def __ft__(self):
|
||||
def render(self):
|
||||
return Div(
|
||||
Div(cls="divider"),
|
||||
Div(mk_ellipsis("Admin", cls="text-sm font-medium mb-1 mr-3")),
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
ADMIN_INSTANCE_ID = "__Admin__"
|
||||
ADMIN_ROLE = "admin"
|
||||
ADMIN_AI_BUDDY_INSTANCE_ID = "__AdminAIBuddy__"
|
||||
ADMIN_IMPORT_HOLIDAYS_INSTANCE_ID = "__AdminImportHolidays__"
|
||||
ADMIN_JIRA_INSTANCE_ID = "__AdminJira__"
|
||||
|
||||
@@ -21,7 +21,7 @@ from core.settings_management import GenericDbManager
|
||||
|
||||
class AIBuddy(BaseComponent):
|
||||
def __init__(self, session, _id: str = None, settings_manager=None, tabs_manager=None):
|
||||
super().__init__(session, _id)
|
||||
super().__init__(session, _id, AI_BUDDY_ROLE)
|
||||
self.settings_manager = settings_manager
|
||||
self.db = GenericDbManager(session, settings_manager, AI_BUDDY_SETTINGS_ENTRY, AIBuddySettings)
|
||||
self.tabs_manager = tabs_manager
|
||||
@@ -153,7 +153,7 @@ class AIBuddy(BaseComponent):
|
||||
for name, tool in available_tools.items()
|
||||
]
|
||||
|
||||
def __ft__(self):
|
||||
def render(self):
|
||||
return Div(
|
||||
Div(cls="divider"),
|
||||
Div(
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
AI_BUDDY_INSTANCE_ID = "__AIBuddy__"
|
||||
AI_BUDDY_ROLE = "ai_buddy"
|
||||
ROUTE_ROOT = "/ai"
|
||||
|
||||
|
||||
|
||||
@@ -2,7 +2,7 @@ from fasthtml.components import *
|
||||
|
||||
from components.BaseComponent import BaseComponent
|
||||
from components.applications.commands import Commands
|
||||
from components.applications.constants import APPLICATION_INSTANCE_ID
|
||||
from components.applications.constants import APPLICATION_INSTANCE_ID, APPLICATION_ROLE
|
||||
from components.hoildays.assets.icons import icon_holidays
|
||||
from components.hoildays.components.HolidaysViewer import HolidaysViewer
|
||||
from components.hoildays.constants import HOLIDAYS_VIEWER_INSTANCE_ID
|
||||
@@ -12,7 +12,7 @@ from core.instance_manager import InstanceManager
|
||||
|
||||
class Applications(BaseComponent):
|
||||
def __init__(self, session: dict, _id: str, settings_manager=None, tabs_manager=None):
|
||||
super().__init__(session, _id)
|
||||
super().__init__(session, _id, APPLICATION_ROLE)
|
||||
self.tabs_manager = tabs_manager
|
||||
self.settings_manager = settings_manager
|
||||
self.commands = Commands(self)
|
||||
@@ -30,7 +30,7 @@ class Applications(BaseComponent):
|
||||
|
||||
raise NotImplementedError(app_name)
|
||||
|
||||
def __ft__(self):
|
||||
def render(self):
|
||||
return Div(
|
||||
Div(cls="divider"),
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
APPLICATION_INSTANCE_ID = "__Applications__"
|
||||
APPLICATION_ROLE = "applications"
|
||||
ROUTE_ROOT = "/apps"
|
||||
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ from components.aibuddy.components.AIBuddy import AIBuddy
|
||||
from components.debugger.assets.icons import icon_dbengine
|
||||
from components.debugger.commands import Commands
|
||||
from components.debugger.components.JsonViewer import JsonViewer
|
||||
from components.debugger.constants import DBENGINE_DEBUGGER_INSTANCE_ID
|
||||
from components.debugger.constants import DBENGINE_DEBUGGER_INSTANCE_ID, DEBUGGER_ROLE
|
||||
from components_helpers import mk_ellipsis, mk_accordion_section
|
||||
from core.instance_manager import InstanceManager
|
||||
from core.utils import get_unique_id
|
||||
@@ -20,7 +20,7 @@ logger = logging.getLogger("Debugger")
|
||||
|
||||
class Debugger(BaseComponent):
|
||||
def __init__(self, session, _id, settings_manager, tabs_manager):
|
||||
super().__init__(session, _id)
|
||||
super().__init__(session, _id, DEBUGGER_ROLE)
|
||||
self.settings_manager = settings_manager
|
||||
self.db_engine = settings_manager.get_db_engine()
|
||||
self.tabs_manager = tabs_manager
|
||||
@@ -104,7 +104,7 @@ class Debugger(BaseComponent):
|
||||
self.tabs_manager.add_tab(title, content, key=tab_key)
|
||||
return self.tabs_manager.render()
|
||||
|
||||
def __ft__(self):
|
||||
def render(self):
|
||||
return Div(
|
||||
Div(cls="divider"),
|
||||
mk_ellipsis("Debugger", cls="text-sm font-medium mb-1"),
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
DBENGINE_DEBUGGER_INSTANCE_ID = "debugger"
|
||||
DEBUGGER_ROLE = "debugger"
|
||||
ROUTE_ROOT = "/debugger"
|
||||
|
||||
INDENT_SIZE = 20
|
||||
|
||||
@@ -8,7 +8,7 @@ from components.datagrid_new.components.DataGrid import DataGrid
|
||||
from components.form.components.MyForm import MyForm, FormField
|
||||
from components.repositories.assets.icons import icon_database, icon_table
|
||||
from components.repositories.commands import Commands
|
||||
from components.repositories.constants import REPOSITORIES_INSTANCE_ID, ROUTE_ROOT, Routes
|
||||
from components.repositories.constants import REPOSITORIES_INSTANCE_ID, ROUTE_ROOT, Routes, REPOSITORIES_ROLE
|
||||
from components.repositories.db_management import RepositoriesDbManager, Repository
|
||||
from components_helpers import mk_icon, mk_ellipsis
|
||||
from core.instance_manager import InstanceManager
|
||||
@@ -19,7 +19,7 @@ logger = logging.getLogger("Repositories")
|
||||
|
||||
class Repositories(BaseComponent):
|
||||
def __init__(self, session: dict, _id: str, settings_manager=None, tabs_manager=None):
|
||||
super().__init__(session, _id)
|
||||
super().__init__(session, _id, REPOSITORIES_ROLE)
|
||||
self.commands = Commands(self)
|
||||
self.tabs_manager = tabs_manager
|
||||
self.db = RepositoriesDbManager(session, settings_manager)
|
||||
@@ -121,7 +121,7 @@ class Repositories(BaseComponent):
|
||||
def refresh(self):
|
||||
return self._mk_repositories(oob=True)
|
||||
|
||||
def __ft__(self):
|
||||
def render(self):
|
||||
return Div(
|
||||
Div(cls="divider"),
|
||||
mk_ellipsis("Repositories", cls="text-sm font-medium mb-1"),
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
REPOSITORIES_INSTANCE_ID = "__Repositories__"
|
||||
REPOSITORIES_ROLE = "repositories"
|
||||
ROUTE_ROOT = "/repositories"
|
||||
USERS_REPOSITORY_NAME = "__USERS___"
|
||||
HOLIDAYS_TABLE_NAME = "__HOLIDAYS__"
|
||||
|
||||
@@ -28,7 +28,7 @@ class UndoRedo(BaseComponentSingleton):
|
||||
COMPONENT_INSTANCE_ID = UNDO_REDO_INSTANCE_ID
|
||||
|
||||
def __init__(self, session, _id, settings_manager=None, tabs_manager=None):
|
||||
super().__init__(session, _id, settings_manager, tabs_manager)
|
||||
super().__init__(session, _id, None, settings_manager, tabs_manager)
|
||||
self.index = -1
|
||||
self.history = []
|
||||
self._commands = UndoRedoCommandManager(self)
|
||||
|
||||
@@ -7,7 +7,7 @@ from components.BaseComponent import BaseComponentSingleton
|
||||
from components.form.components.MyForm import MyForm, FormField
|
||||
from components.workflows.commands import WorkflowsCommandManager
|
||||
from components.workflows.components.WorkflowDesigner import WorkflowDesigner
|
||||
from components.workflows.constants import WORKFLOWS_INSTANCE_ID
|
||||
from components.workflows.constants import WORKFLOWS_INSTANCE_ID, WORKFLOWS_ROLE
|
||||
from components.workflows.db_management import WorkflowsDbManager, WorkflowsDesignerSettings
|
||||
from components_helpers import mk_ellipsis, mk_icon
|
||||
from core.instance_manager import InstanceManager
|
||||
@@ -19,7 +19,7 @@ class Workflows(BaseComponentSingleton):
|
||||
COMPONENT_INSTANCE_ID = WORKFLOWS_INSTANCE_ID
|
||||
|
||||
def __init__(self, session, _id, settings_manager=None, tabs_manager=None):
|
||||
super().__init__(session, _id, settings_manager, tabs_manager)
|
||||
super().__init__(session, _id, WORKFLOWS_ROLE, settings_manager, tabs_manager)
|
||||
self.commands = WorkflowsCommandManager(self)
|
||||
self.db = WorkflowsDbManager(session, settings_manager)
|
||||
|
||||
@@ -80,7 +80,7 @@ class Workflows(BaseComponentSingleton):
|
||||
def refresh(self):
|
||||
return self._mk_workflows(True)
|
||||
|
||||
def __ft__(self):
|
||||
def render(self):
|
||||
return Div(
|
||||
Div(cls="divider"),
|
||||
Div(
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
WORKFLOWS_INSTANCE_ID = "__Workflows__"
|
||||
WORKFLOWS_ROLE = "workflows"
|
||||
WORKFLOW_DESIGNER_INSTANCE_ID = "__WorkflowDesigner__"
|
||||
WORKFLOW_PLAYER_INSTANCE_ID = "__WorkflowPlayer__"
|
||||
WORKFLOWS_DB_ENTRY = "Workflows"
|
||||
|
||||
@@ -13,6 +13,10 @@ load_dotenv()
|
||||
DB_PATH = os.getenv("DB_PATH", "tools.db")
|
||||
logger.info(f"{DB_PATH=}")
|
||||
|
||||
# Custom database engine settings
|
||||
DBENGINE_PATH = os.getenv("DBENGINE_PATH", ".mytools_db")
|
||||
logger.info(f"{DBENGINE_PATH=}")
|
||||
|
||||
# Authentication settings
|
||||
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key-change-in-production")
|
||||
|
||||
|
||||
18
src/core/database_manager.py
Normal file
18
src/core/database_manager.py
Normal file
@@ -0,0 +1,18 @@
|
||||
"""Database manager for application initialization."""
|
||||
|
||||
import logging
|
||||
from .user_database import Database, set_user_db
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def initialize_database():
|
||||
"""Initialize the application database."""
|
||||
try:
|
||||
# Create default database instance
|
||||
db_instance = Database()
|
||||
set_user_db(db_instance)
|
||||
logger.info("Database initialized successfully")
|
||||
return db_instance
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to initialize database: {e}")
|
||||
raise
|
||||
@@ -1,6 +1,7 @@
|
||||
import logging
|
||||
from datetime import datetime
|
||||
|
||||
import config
|
||||
from constants import NOT_LOGGED, NO_SESSION
|
||||
from core.dbengine import DbEngine, TAG_PARENT, TAG_USER, TAG_DATE, DbException
|
||||
from core.settings_objects import *
|
||||
@@ -92,7 +93,7 @@ class MemoryDbEngine:
|
||||
|
||||
class SettingsManager:
|
||||
def __init__(self, engine=None):
|
||||
self._db_engine = engine or DbEngine()
|
||||
self._db_engine = engine or DbEngine(config.DBENGINE_PATH)
|
||||
|
||||
def save(self, session: dict, entry: str, obj: object):
|
||||
user_id, user_email = self._get_user(session)
|
||||
@@ -263,3 +264,15 @@ class NestedSettingsManager:
|
||||
raise AttributeError(f"Settings '{self._obj_entry}' has no attribute '{self._obj_attribute}'.")
|
||||
|
||||
return settings, getattr(settings, self._obj_attribute)
|
||||
|
||||
|
||||
_settings_manager = SettingsManager()
|
||||
|
||||
|
||||
def set_settings_manager(_setting_manager):
|
||||
global _settings_manager
|
||||
_settings_manager = _setting_manager
|
||||
|
||||
|
||||
def get_settings_manager():
|
||||
return _settings_manager
|
||||
|
||||
@@ -3,15 +3,17 @@ import secrets
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from .user_database import user_db
|
||||
from .user_database import get_user_db
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class UserDAO:
|
||||
"""Data Access Object for user management."""
|
||||
|
||||
@staticmethod
|
||||
def create_user(username: str, email: str, password: str | None = None, github_id: str | None = None) -> int:
|
||||
def create_user(username: str, email: str, password: str | None = None, github_id: str | None = None,
|
||||
db_instance=None) -> int:
|
||||
"""
|
||||
Create a new user with email/password or GitHub authentication.
|
||||
|
||||
@@ -20,10 +22,12 @@ class UserDAO:
|
||||
email: The user's email
|
||||
password: The user's password (optional)
|
||||
github_id: GitHub user ID (optional)
|
||||
db_instance: Database instance (optional, uses default if None)
|
||||
|
||||
Returns:
|
||||
int: ID of the new user or 0 if creation failed
|
||||
"""
|
||||
user_db = db_instance or get_user_db()
|
||||
try:
|
||||
with user_db.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
@@ -52,24 +56,39 @@ class UserDAO:
|
||||
VALUES (?, ?, ?, ?, ?, 0)
|
||||
''', (username, email, password_hash, salt, github_id))
|
||||
|
||||
# Get the ID of the newly created user
|
||||
user_id = cursor.lastrowid
|
||||
|
||||
# Add default roles to the new user
|
||||
default_roles = ['repositories', 'datagrid', 'ai_buddy']
|
||||
for role in default_roles:
|
||||
cursor.execute('''
|
||||
INSERT INTO roles (user_id, role_name)
|
||||
VALUES (?, ?)
|
||||
''', (user_id, role))
|
||||
|
||||
conn.commit()
|
||||
return cursor.lastrowid
|
||||
return user_id
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating user: {e}")
|
||||
return 0
|
||||
|
||||
|
||||
@staticmethod
|
||||
def authenticate_email(email: str, password: str) -> dict[str, Any] | None:
|
||||
def authenticate_email(email: str, password: str, db_instance=None) -> dict[str, Any] | None:
|
||||
"""
|
||||
Authenticate a user with email and password.
|
||||
|
||||
Args:
|
||||
email: The user's email
|
||||
password: The user's password
|
||||
db_instance: Database instance (optional, uses default if None)
|
||||
|
||||
Returns:
|
||||
Dict or None: User record if authentication succeeds, None otherwise
|
||||
"""
|
||||
user_db = db_instance or get_user_db()
|
||||
with user_db.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
@@ -100,8 +119,10 @@ class UserDAO:
|
||||
# Return user info
|
||||
return dict(user)
|
||||
|
||||
|
||||
@staticmethod
|
||||
def find_or_create_github_user(github_id: str, username: str, email: str | None) -> dict[str, Any] | None:
|
||||
def find_or_create_github_user(github_id: str, username: str, email: str | None, db_instance=None) -> dict[
|
||||
str, Any] | None:
|
||||
"""
|
||||
Find existing GitHub user or create a new one.
|
||||
|
||||
@@ -109,10 +130,12 @@ class UserDAO:
|
||||
github_id: GitHub user ID
|
||||
username: The username from GitHub
|
||||
email: The email from GitHub (may be None)
|
||||
db_instance: Database instance (optional, uses default if None)
|
||||
|
||||
Returns:
|
||||
Dict or None: User record if found or created, None on error
|
||||
"""
|
||||
user_db = db_instance or get_user_db()
|
||||
with user_db.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
@@ -156,17 +179,20 @@ class UserDAO:
|
||||
logger.error(f"Error creating GitHub user: {e}")
|
||||
return None
|
||||
|
||||
|
||||
@staticmethod
|
||||
def get_user_by_id(user_id: int) -> dict[str, Any] | None:
|
||||
def get_user_by_id(user_id: int, db_instance=None) -> dict[str, Any] | None:
|
||||
"""
|
||||
Get a user by ID.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
db_instance: Database instance (optional, uses default if None)
|
||||
|
||||
Returns:
|
||||
Dict or None: User record if found, None otherwise
|
||||
"""
|
||||
user_db = db_instance or get_user_db()
|
||||
with user_db.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
@@ -179,22 +205,57 @@ class UserDAO:
|
||||
return dict(user) if user else None
|
||||
|
||||
@staticmethod
|
||||
def get_all_users(limit: int = 100, offset: int = 0) -> list[dict[str, Any]]:
|
||||
def get_user_roles_by_id(user_id: int, db_instance=None) -> list[str]:
|
||||
"""
|
||||
Retrieve the roles associated with a user.
|
||||
|
||||
Args:
|
||||
user_id (int): User ID.
|
||||
db_instance: Database instance (optional, uses default if None)
|
||||
|
||||
Returns:
|
||||
list[str]: List of role names associated with the user.
|
||||
"""
|
||||
db_instance = db_instance or get_user_db()
|
||||
|
||||
with db_instance.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('''
|
||||
SELECT role_name
|
||||
FROM roles
|
||||
WHERE user_id = ?
|
||||
''', (user_id,))
|
||||
|
||||
# get the roles
|
||||
roles = [row["role_name"] for row in cursor.fetchall()]
|
||||
|
||||
return roles
|
||||
|
||||
|
||||
@staticmethod
|
||||
def get_all_users(limit: int = 100, offset: int = 0, db_instance=None) -> list[dict[str, Any]]:
|
||||
"""
|
||||
Get all users with pagination.
|
||||
|
||||
Args:
|
||||
limit: Maximum number of users to return
|
||||
offset: Number of users to skip
|
||||
db_instance: Database instance (optional, uses default if None)
|
||||
|
||||
Returns:
|
||||
List of user records
|
||||
"""
|
||||
user_db = db_instance or get_user_db()
|
||||
with user_db.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute('''
|
||||
SELECT id, username, email, is_admin, created_at, last_login,
|
||||
SELECT id,
|
||||
username,
|
||||
email,
|
||||
is_admin,
|
||||
created_at,
|
||||
last_login,
|
||||
(github_id IS NOT NULL) as is_github_user
|
||||
FROM users
|
||||
ORDER BY created_at DESC
|
||||
@@ -203,18 +264,21 @@ class UserDAO:
|
||||
|
||||
return [dict(user) for user in cursor.fetchall()]
|
||||
|
||||
|
||||
@staticmethod
|
||||
def set_admin_status(user_id: int, is_admin: bool) -> bool:
|
||||
def set_admin_status(user_id: int, is_admin: bool, db_instance=None) -> bool:
|
||||
"""
|
||||
Change a user's admin status.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
is_admin: True to make admin, False to remove admin status
|
||||
db_instance: Database instance (optional, uses default if None)
|
||||
|
||||
Returns:
|
||||
bool: True if successful, False otherwise
|
||||
"""
|
||||
user_db = db_instance or get_user_db()
|
||||
try:
|
||||
with user_db.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
@@ -230,17 +294,20 @@ class UserDAO:
|
||||
logger.error(f"Error setting admin status: {e}")
|
||||
return False
|
||||
|
||||
|
||||
@staticmethod
|
||||
def delete_user(user_id: int) -> bool:
|
||||
def delete_user(user_id: int, db_instance=None) -> bool:
|
||||
"""
|
||||
Delete a user and all their data.
|
||||
|
||||
Args:
|
||||
user_id: The user ID
|
||||
db_instance: Database instance (optional, uses default if None)
|
||||
|
||||
Returns:
|
||||
bool: True if successful, False otherwise
|
||||
"""
|
||||
user_db = db_instance or get_user_db()
|
||||
try:
|
||||
with user_db.get_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
@@ -46,6 +46,17 @@ class Database:
|
||||
''')
|
||||
logger.info("Created users table")
|
||||
|
||||
# Create the roles table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS roles (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
user_id INTEGER,
|
||||
role_name TEXT NOT NULL,
|
||||
assigned_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE
|
||||
)
|
||||
''')
|
||||
logger.info("Created roles table")
|
||||
|
||||
# Check if we need to create an admin user
|
||||
cursor.execute("SELECT COUNT(*) FROM users WHERE is_admin = 1")
|
||||
@@ -111,5 +122,18 @@ class Database:
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
# Create a singleton instance
|
||||
user_db = Database()
|
||||
|
||||
# Global instance for backward compatibility (will be modified in tests)
|
||||
_default_instance = None
|
||||
|
||||
def get_user_db():
|
||||
"""Get the default database instance."""
|
||||
global _default_instance
|
||||
if _default_instance is None:
|
||||
_default_instance = Database()
|
||||
return _default_instance
|
||||
|
||||
def set_user_db(instance):
|
||||
"""Set a custom database instance (for testing)."""
|
||||
global _default_instance
|
||||
_default_instance = instance
|
||||
29
src/main.py
29
src/main.py
@@ -1,6 +1,7 @@
|
||||
# global layout
|
||||
import logging.config
|
||||
|
||||
import click
|
||||
import yaml
|
||||
from fasthtml.common import *
|
||||
|
||||
@@ -19,9 +20,10 @@ from components.register.constants import ROUTE_ROOT as REGISTER_ROUTE_ROOT
|
||||
from components.register.constants import Routes as RegisterRoutes
|
||||
from config import APP_PORT
|
||||
from constants import Routes
|
||||
from core.database_manager import initialize_database
|
||||
from core.dbengine import DbException
|
||||
from core.instance_manager import InstanceManager
|
||||
from core.settings_management import SettingsManager
|
||||
from core.settings_management import get_settings_manager
|
||||
from pages.admin_import_settings import AdminImportSettings, IMPORT_SETTINGS_PATH, import_settings_app
|
||||
from pages.another_grid import get_datagrid2
|
||||
from pages.basic_test import BASIC_TEST_PATH, basic_test_app, get_basic_test
|
||||
@@ -38,6 +40,9 @@ logging.config.dictConfig(config)
|
||||
|
||||
logger = logging.getLogger("MainApp")
|
||||
|
||||
# Initialize database
|
||||
initialize_database()
|
||||
|
||||
# daisy_ui_links_v4 = (
|
||||
# Link(href="https://cdn.jsdelivr.net/npm/daisyui@latest/dist/full.min.css", rel="stylesheet", type="text/css"),
|
||||
# Script(src="https://cdn.tailwindcss.com"),
|
||||
@@ -231,9 +236,7 @@ async def timing_middleware(request, call_next):
|
||||
return response
|
||||
|
||||
|
||||
settings_manager = SettingsManager()
|
||||
|
||||
import_settings = AdminImportSettings(settings_manager, None)
|
||||
import_settings = AdminImportSettings(get_settings_manager(), None)
|
||||
pages = [
|
||||
Page("My table", get_datagrid, id="my_table"),
|
||||
Page("new settings", import_settings, id="import_settings"),
|
||||
@@ -242,8 +245,8 @@ pages = [
|
||||
Page("Another Table", get_datagrid2, id="another_table"),
|
||||
]
|
||||
|
||||
login = Login(settings_manager)
|
||||
register = Register(settings_manager)
|
||||
login = Login(get_settings_manager())
|
||||
register = Register(get_settings_manager())
|
||||
InstanceManager.register_many(login, register)
|
||||
|
||||
|
||||
@@ -253,8 +256,8 @@ def get(session):
|
||||
main = InstanceManager.get(session,
|
||||
DrawerLayout.create_component_id(session),
|
||||
DrawerLayout,
|
||||
settings_manager=settings_manager)
|
||||
return page_layout_lite(session, settings_manager, main)
|
||||
settings_manager=get_settings_manager())
|
||||
return page_layout_lite(session, get_settings_manager(), main)
|
||||
except DbException:
|
||||
return RedirectResponse(LOGIN_ROUTE_ROOT + LoginRoutes.Logout, status_code=303)
|
||||
|
||||
@@ -298,7 +301,7 @@ def not_found(path: str, session=None):
|
||||
|
||||
return page_layout_new(
|
||||
session=session,
|
||||
settings_manager=settings_manager,
|
||||
settings_manager=get_settings_manager(),
|
||||
content=error_content
|
||||
)
|
||||
|
||||
@@ -306,9 +309,11 @@ def not_found(path: str, session=None):
|
||||
setup_toasts(app)
|
||||
|
||||
|
||||
def main():
|
||||
logger.info(f" Starting FastHTML server on http://localhost:{APP_PORT}")
|
||||
serve(port=APP_PORT)
|
||||
@click.command()
|
||||
@click.option('--port', default=APP_PORT, help='Port to run the server on')
|
||||
def main(port):
|
||||
logger.info(f" Starting FastHTML server on http://localhost:{port}")
|
||||
serve(port=port)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -1,12 +1,161 @@
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from components.datagrid.DataGrid import reset_instances
|
||||
from playwright_config import BROWSER_CONFIG, BASE_URL
|
||||
from tests.fixtures.test_database import TestDatabaseManager
|
||||
from tests.fixtures.test_users import TestUsers
|
||||
|
||||
USER_EMAIL = "test@mail.com"
|
||||
USER_ID = "test_user"
|
||||
APP_PORT = 5002
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_database():
|
||||
"""Configure temporary database for tests"""
|
||||
# Create temporary DB
|
||||
temp_db_path = TestDatabaseManager.create_temp_db_path()
|
||||
|
||||
# Save original environment
|
||||
original_env = {
|
||||
"DB_PATH": os.environ.get("DB_PATH"),
|
||||
"ADMIN_EMAIL": os.environ.get("ADMIN_EMAIL"),
|
||||
"ADMIN_PASSWORD": os.environ.get("ADMIN_PASSWORD"),
|
||||
"SECRET_KEY": os.environ.get("SECRET_KEY")
|
||||
}
|
||||
|
||||
# Configure test environment
|
||||
TestDatabaseManager.setup_test_environment(temp_db_path)
|
||||
|
||||
print(f"Test database created at: {temp_db_path}")
|
||||
|
||||
yield temp_db_path
|
||||
|
||||
# Cleanup: restore environment and clean up DB
|
||||
TestDatabaseManager.restore_original_environment(original_env)
|
||||
TestDatabaseManager.cleanup_test_db(temp_db_path)
|
||||
print(f"Test database cleaned up: {temp_db_path}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_users():
|
||||
"""Define available test users"""
|
||||
return {
|
||||
"admin": TestUsers.ADMIN,
|
||||
"regular_user": TestUsers.REGULAR_USER,
|
||||
"invalid_cases": TestUsers.INVALID_CREDENTIALS,
|
||||
"protected_urls": TestUsers.PROTECTED_URLS
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_setting_manager():
|
||||
# create a dedicated folder for the db
|
||||
db_engine_folder = tempfile.mkdtemp(prefix="test_db_engine")
|
||||
|
||||
yield db_engine_folder
|
||||
|
||||
# clean up folder & restore settings manager
|
||||
if os.path.exists(db_engine_folder):
|
||||
shutil.rmtree(db_engine_folder, ignore_errors=True)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def app_server(test_database, test_users, test_setting_manager):
|
||||
"""Start application server with test database"""
|
||||
# Use the same Python executable that's running pytest
|
||||
python_executable = sys.executable
|
||||
|
||||
# Get the absolute path to the src directory
|
||||
project_root = Path(__file__).parent.parent
|
||||
src_path = project_root / "src"
|
||||
|
||||
# Create test environment
|
||||
test_env = os.environ.copy()
|
||||
test_env["DB_PATH"] = test_database
|
||||
test_env["DBENGINE_PATH"] = test_setting_manager
|
||||
test_env["ADMIN_EMAIL"] = test_users["admin"]["email"]
|
||||
test_env["ADMIN_PASSWORD"] = test_users["admin"]["password"]
|
||||
test_env["SECRET_KEY"] = "test-secret-key-for-e2e-tests"
|
||||
|
||||
# Start the application server
|
||||
print(f"Starting server on url {BASE_URL} with test database...")
|
||||
port = BASE_URL.split(':')[-1].split('/')[0] if ':' in BASE_URL else APP_PORT
|
||||
print(f"Using port {port}")
|
||||
print(f"Test DB path: {test_database}")
|
||||
|
||||
server_process = subprocess.Popen(
|
||||
[python_executable, "main.py", "--port", "5002"],
|
||||
cwd=str(src_path), # Change to src directory where main.py is located
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
env=test_env # Use test environment
|
||||
)
|
||||
|
||||
# Wait for the server to start
|
||||
max_retries = 10 # Wait up to 30 seconds
|
||||
for i in range(max_retries):
|
||||
try:
|
||||
print(f"Waiting retry {i}/{max_retries}")
|
||||
response = requests.get(BASE_URL, timeout=1)
|
||||
if response.status_code in [200, 302, 404]: # Server is responding
|
||||
print(f"Server started successfully after {i + 1} attempts")
|
||||
break
|
||||
except requests.exceptions.RequestException:
|
||||
time.sleep(1)
|
||||
else:
|
||||
# If we get here, the server didn't start in time
|
||||
print(f"Failed to start after {max_retries} attempts.")
|
||||
server_process.kill()
|
||||
stdout, stderr = server_process.communicate()
|
||||
raise RuntimeError(
|
||||
f"Server failed to start within {max_retries} seconds.\n"
|
||||
f"STDOUT: {stdout.decode()}\n"
|
||||
f"STDERR: {stderr.decode()}"
|
||||
)
|
||||
|
||||
# Yield control to the tests
|
||||
print('Test server started with isolated database!')
|
||||
yield server_process
|
||||
|
||||
# Cleanup: terminate the server after tests
|
||||
server_process.terminate()
|
||||
try:
|
||||
server_process.wait(timeout=5)
|
||||
print('Test server stopped.')
|
||||
except subprocess.TimeoutExpired:
|
||||
server_process.kill()
|
||||
server_process.wait()
|
||||
print('Test server killed!')
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def browser_context_args(browser_context_args):
|
||||
"""Configure browser context arguments"""
|
||||
return {
|
||||
**browser_context_args,
|
||||
**BROWSER_CONFIG,
|
||||
"record_video_dir": "test-results/videos/",
|
||||
"record_har_path": "test-results/har/trace.har",
|
||||
"record_video_size": {"width": 1280, "height": 720},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def app_url():
|
||||
"""Base URL for the application"""
|
||||
return BASE_URL
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
177
tests/e2e/Readme.md
Normal file
177
tests/e2e/Readme.md
Normal file
@@ -0,0 +1,177 @@
|
||||
# README - End-to-End Authentication Testing Guide
|
||||
|
||||
This README provides details on how to set up, configure, and run the end-to-end (e2e) authentication tests for the **My
|
||||
Managing Tools** application. The tests are implemented using `pytest` with `playwright` for browser automation.
|
||||
|
||||
## Purpose
|
||||
|
||||
The e2e tests verify that the authentication system works as expected, including login functionality, session
|
||||
validation, and database isolation for testing purposes. These tests ensure the login page behaves properly under
|
||||
different scenarios such as unauthenticated access, form validation, and successful user login.
|
||||
|
||||
---
|
||||
|
||||
## Table of Contents
|
||||
|
||||
1. [Setup Instructions](#setup-instructions)
|
||||
- [Application Setup](#application-setup)
|
||||
- [Database Setup](#database-setup)
|
||||
2. [Installing Dependencies](#installing-dependencies)
|
||||
3. [Running Tests](#running-tests)
|
||||
4. [Debugging and Logs](#debugging-and-logs)
|
||||
5. [Test Scenarios](#test-scenarios)
|
||||
6. [Troubleshooting](#troubleshooting)
|
||||
|
||||
---
|
||||
|
||||
## Installing Dependencies
|
||||
|
||||
Install the required packages for the tests:
|
||||
|
||||
1. **Playwright and Pytest**:
|
||||
Install `pytest-playwright` and playwright dependencies:
|
||||
```bash
|
||||
pip install pytest-playwright
|
||||
playwright install
|
||||
playwright install-deps # For Linux systems
|
||||
```
|
||||
|
||||
2. **Project Dependencies**:
|
||||
Use the provided `requirements.txt` to install other necessary packages:
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
3. **Verify Installation**:
|
||||
Confirm Playwright is installed and functional:
|
||||
```bash
|
||||
playwright --version
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
|
||||
### Database Setup
|
||||
|
||||
The e2e tests are designed to interact with an **isolated test database**. During setup, a temporary SQLite database will be created for each test session.
|
||||
|
||||
1. **Test Database Initialization**:
|
||||
The database is initialized dynamically during the test execution using a custom temporary path. The `test_database` fixture creates a database with the following characteristics:
|
||||
- **Temporary Directory**: Ensures no conflicts with the production database.
|
||||
- **Admin User Auto-Creation**: Adds an admin account if the environment variables `ADMIN_EMAIL` and `ADMIN_PASSWORD` are set.
|
||||
|
||||
2. **Environment Variables**:
|
||||
Ensure the following environment variables are defined for the test database:
|
||||
```env
|
||||
DB_PATH="test_mmt_tools.db"
|
||||
ADMIN_EMAIL="test.admin@test.com"
|
||||
ADMIN_PASSWORD="TestAdmin123"
|
||||
```
|
||||
|
||||
3. **Configuration**:
|
||||
The `test_database.py` script handles the database isolation and cleanup after each test session. It ensures that no state is leaked between tests.
|
||||
|
||||
---
|
||||
|
||||
## Running Tests
|
||||
|
||||
To execute the end-to-end authentication tests, run the following commands from the project root.
|
||||
|
||||
1. **Run All Tests**:
|
||||
Execute all e2e tests:
|
||||
```bash
|
||||
pytest -m "e2e"
|
||||
```
|
||||
|
||||
2. **Run Specific Test**:
|
||||
Use the test path and `-k` argument to target specific tests:
|
||||
```bash
|
||||
pytest tests/test_e2e_authentication.py -k "test_unauthenticated_user_redirected_to_login"
|
||||
```
|
||||
|
||||
3. **Record Test Results**:
|
||||
Save test results (HTML report):
|
||||
```bash
|
||||
pytest --html=test-results/report.html --self-contained-html
|
||||
```
|
||||
|
||||
4. **Parallel Execution**:
|
||||
If desired, run tests in parallel:
|
||||
```bash
|
||||
pytest -n auto
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Debugging and Logs
|
||||
|
||||
1. **Console Logs**:
|
||||
Console errors during page navigation are captured using the `test_page_loads_without_errors` test. Check the pytest
|
||||
output for details.
|
||||
|
||||
2. **Debugging Tools**:
|
||||
- Videos and HAR traces are recorded in the `test-results` directory.
|
||||
- Use the `page.pause()` method inside a test for interactive debugging with Playwright Inspector.
|
||||
|
||||
3. **Screenshots and Debug HTML**:
|
||||
If a test fails, screenshots and full HTML are saved for post-mortem debugging:
|
||||
- `debug_after_login.png`: Screenshot post-login.
|
||||
- `debug_page.html`: Captured HTML structure of the page.
|
||||
|
||||
4. **Manual Debugging**:
|
||||
```python
|
||||
# Screenshot
|
||||
page.screenshot(path="debug_after_login.png")
|
||||
|
||||
# HTML content
|
||||
with open("debug_page.html", "w", encoding="utf-8") as f:
|
||||
f.write(page.content())
|
||||
```
|
||||
|
||||
5. **Server Logs**:
|
||||
When using the `app_server` fixture, server `stdout` and `stderr` are recorded during the test execution.
|
||||
|
||||
---
|
||||
|
||||
|
||||
|
||||
## Test Scenarios
|
||||
|
||||
The main authentication tests cover the following scenarios:
|
||||
|
||||
1. **Unauthenticated User Redirect**:
|
||||
Verify that unauthenticated users accessing protected resources are redirected to the login page.
|
||||
|
||||
2. **Login Page Elements**:
|
||||
Validate that key login page elements are visible, such as email/password fields and the submit button.
|
||||
|
||||
3. **Successful Login**:
|
||||
Ensure users can log in successfully with valid credentials and are redirected to the home page.
|
||||
|
||||
4. **Test Database Isolation**:
|
||||
Confirm that the test environment uses an isolated temporary database and does not affect production data.
|
||||
|
||||
5. **Page Load Without Errors**:
|
||||
Ensure that the login page loads without console or HTTP errors.
|
||||
|
||||
For a full list of test cases, see `tests/test_e2e_authentication.py`.
|
||||
|
||||
---
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
- **Server Not Starting**:
|
||||
If the server fails to start during testing, confirm the `DB_PATH`, `BASE_URL`, and all dependencies are correctly
|
||||
configured in the `.env` file.
|
||||
|
||||
- **Playwright Issues**:
|
||||
Ensure Playwright dependencies (`playwright install`) are installed and functional. Missing browsers or outdated
|
||||
Playwright versions can cause tests to fail.
|
||||
|
||||
- **Database Error**:
|
||||
Check that the `tools.db` file or the temporary test database exists and is accessible. Ensure the test database
|
||||
environment variables are set correctly during tests.
|
||||
|
||||
---
|
||||
|
||||
igurations are hardcoded in the test scripts.
|
||||
0
tests/e2e/__init__.py
Normal file
0
tests/e2e/__init__.py
Normal file
4
tests/e2e/e2e_constants.py
Normal file
4
tests/e2e/e2e_constants.py
Normal file
@@ -0,0 +1,4 @@
|
||||
class Login:
|
||||
email = ["input[type='email']", "input[name='email']"]
|
||||
password = ["input[type='password']", "input[name='password']"]
|
||||
submit = "button[type='submit']"
|
||||
94
tests/e2e/e2e_helper.py
Normal file
94
tests/e2e/e2e_helper.py
Normal file
@@ -0,0 +1,94 @@
|
||||
from playwright.sync_api import Page, expect
|
||||
|
||||
|
||||
def locate(page: Page, name, check_visible=True):
|
||||
"""
|
||||
Locate an element or elements on the provided page by name.
|
||||
|
||||
This function identifies elements on a web page using the locator API of the
|
||||
provided page object. The `name` parameter can either be a single locator
|
||||
string or a list of locator strings. If a list is provided, its elements are
|
||||
joined into a single comma-separated string before locating the elements.
|
||||
|
||||
Args:
|
||||
page (Page): The web page object in which to locate the element(s).
|
||||
name (Union[str, List[str]]): Locator or list of locators to identify elements.
|
||||
check_visible(bool): Whether to check if the element is visible before returning it.
|
||||
|
||||
Returns:
|
||||
Locator: The locator object for the given name.
|
||||
"""
|
||||
if isinstance(name, list):
|
||||
name = ",".join(name)
|
||||
|
||||
element = page.locator(name)
|
||||
|
||||
if check_visible:
|
||||
expect(element).to_be_visible()
|
||||
|
||||
return element
|
||||
|
||||
|
||||
def fill(element, value):
|
||||
"""
|
||||
Fills an element with a specified value and verifies that the element's value matches the provided value.
|
||||
|
||||
Parameters:
|
||||
element : The UI element to be filled with the value.
|
||||
value : The value to fill into the element.
|
||||
|
||||
Raises:
|
||||
AssertionError: If the element's value does not match the provided value after the fill operation.
|
||||
"""
|
||||
element.fill(value)
|
||||
expect(element).to_have_value(value)
|
||||
|
||||
|
||||
def debug(page: Page, checkpoint_name: str):
|
||||
"""
|
||||
Takes a screenshot of the current state of a page and saves its full HTML content
|
||||
to support debugging at a specified checkpoint. The screenshot and HTML file are saved
|
||||
with checkpoint-specific filenames in the current working directory.
|
||||
|
||||
This function is useful for capturing the visual state and content of the page at
|
||||
any moment during the script execution, helping in resolving issues or verifying
|
||||
state transitions.
|
||||
|
||||
Arguments:
|
||||
page (Page): The page object whose state and content are to be captured.
|
||||
checkpoint_name (str): A descriptive name for the checkpoint. This is used to
|
||||
generate unique filenames for the screenshot and HTML file.
|
||||
"""
|
||||
# Take screenshot
|
||||
page.screenshot(path=f"debug_{checkpoint_name}.png")
|
||||
|
||||
# Save full HTML for inspection
|
||||
with open(f"debug_{checkpoint_name}.html", "w", encoding="utf-8") as f:
|
||||
f.write(page.content())
|
||||
|
||||
with open(f"debug_{checkpoint_name}.txt", "w", encoding="utf-8") as f:
|
||||
f.writelines([
|
||||
f"Checkpoint: {checkpoint_name}\n",
|
||||
f"URL: {page.url}\n",
|
||||
f"Title: {page.title()}\n",
|
||||
])
|
||||
|
||||
|
||||
def check_not_in_content(page, keywords):
|
||||
"""
|
||||
Checks that none of the specified keywords are present in the page content.
|
||||
|
||||
This function iterates through a list of keywords and verifies that none of
|
||||
them are present in the given page's content. If any keyword is found, an
|
||||
AssertionError is raised with details about the number of times it was found.
|
||||
|
||||
Args:
|
||||
page: A page object that provides access to retrieve text content.
|
||||
keywords: List of strings representing keywords to search in the page content.
|
||||
|
||||
Raises:
|
||||
AssertionError: If any of the keywords are found in the page content.
|
||||
"""
|
||||
for keyword in keywords:
|
||||
occurrences = page.get_by_text(keyword, exact=False)
|
||||
assert occurrences.count() == 0, f"Found {occurrences.count()} of '{keyword}' in page content which should not be visible"
|
||||
212
tests/e2e/test_e2e_authentication.py
Normal file
212
tests/e2e/test_e2e_authentication.py
Normal file
@@ -0,0 +1,212 @@
|
||||
# tests/test_e2e_authentication.py
|
||||
|
||||
import pytest
|
||||
from playwright.sync_api import Page, expect
|
||||
|
||||
from e2e.e2e_constants import Login
|
||||
from e2e.e2e_helper import locate, fill, debug, check_not_in_content
|
||||
from playwright_config import BASE_URL
|
||||
|
||||
|
||||
class TestAuthentication:
|
||||
"""Tests for authentication and login functionality"""
|
||||
|
||||
@pytest.mark.e2e
|
||||
@pytest.mark.smoke
|
||||
def test_unauthenticated_user_redirected_to_login(self, app_server, page: Page):
|
||||
"""Test that when not logged in, the default page is the login page"""
|
||||
# Navigate to the root URL
|
||||
page.goto(BASE_URL)
|
||||
|
||||
# Wait for the page to fully load
|
||||
page.wait_for_load_state("networkidle")
|
||||
|
||||
# Check that we're on the login page
|
||||
# Option 1: Check URL contains login-related path
|
||||
expect(page).to_have_url(f"{BASE_URL}/authlogin/login")
|
||||
|
||||
# Option 2: Check for login form elements
|
||||
# Look for typical login form elements
|
||||
login_indicators = [
|
||||
"input[type='email']",
|
||||
"input[type='password']",
|
||||
"input[name*='username']",
|
||||
"input[name*='email']",
|
||||
"input[name*='password']",
|
||||
"button[type='submit']",
|
||||
"form"
|
||||
]
|
||||
|
||||
# At least one login indicator should be present
|
||||
login_form_found = False
|
||||
for selector in login_indicators:
|
||||
if page.locator(selector).count() > 0:
|
||||
login_form_found = True
|
||||
break
|
||||
|
||||
assert login_form_found, "No login form elements found on the page"
|
||||
|
||||
# Option 3: Check for login-related text content
|
||||
page_content = page.content().lower()
|
||||
login_keywords = ["login", "sign in", "authenticate", "username", "password", "email"]
|
||||
|
||||
has_login_content = any(keyword in page_content for keyword in login_keywords)
|
||||
assert has_login_content, "Page does not contain login-related content"
|
||||
|
||||
# Option 4: Ensure we're not on a protected page
|
||||
# Check that we don't see protected content like "dashboard", "logout", "profile", "settings"]
|
||||
protected_content = ["dashboard", "logout", "profile", "settings"]
|
||||
page_text = page.locator("body").inner_text().lower()
|
||||
|
||||
for protected_word in protected_content:
|
||||
assert protected_word not in page_text, f"Found protected content '{protected_word}' when not logged in"
|
||||
|
||||
@pytest.mark.e2e
|
||||
@pytest.mark.regression
|
||||
def test_login_page_has_required_elements(self, app_server, page: Page):
|
||||
"""Test that the login page contains all required elements"""
|
||||
page.goto(BASE_URL)
|
||||
page.wait_for_load_state("networkidle")
|
||||
|
||||
# Check for essential login form elements
|
||||
expect(page.locator("form")).to_be_visible()
|
||||
|
||||
# Check for input fields (at least email/username and password)
|
||||
username_input = page.locator("input[type='email'], input[name*='username'], input[name*='email']")
|
||||
expect(username_input.first).to_be_visible()
|
||||
|
||||
password_input = page.locator("input[type='password'], input[name*='password']")
|
||||
expect(password_input.first).to_be_visible()
|
||||
|
||||
# Check for submit button
|
||||
submit_button = page.locator("button[type='submit'], input[type='submit']")
|
||||
expect(submit_button.first).to_be_visible()
|
||||
|
||||
def test_page_loads_without_errors(self, app_server, page: Page):
|
||||
"""Test that the login page loads without console errors"""
|
||||
console_errors = []
|
||||
|
||||
def handle_console(msg):
|
||||
if msg.type == "error":
|
||||
console_errors.append(msg.text)
|
||||
|
||||
page.on("console", handle_console)
|
||||
page.goto(BASE_URL)
|
||||
page.wait_for_load_state("networkidle")
|
||||
|
||||
# Check for console errors
|
||||
assert len(console_errors) == 0, f"Console errors found: {console_errors}"
|
||||
|
||||
# Check that the page actually loaded (not a 404 or 500 error)
|
||||
expect(page.locator("body")).to_be_visible()
|
||||
|
||||
# Check that the page has a title
|
||||
expect(page).to_have_title("My Managing Tools")
|
||||
|
||||
# New test to validate database isolation
|
||||
@pytest.mark.e2e
|
||||
@pytest.mark.smoke
|
||||
def test_test_database_isolation(self, app_server, test_database, test_users):
|
||||
"""Test that application uses isolated test database"""
|
||||
import os
|
||||
|
||||
# Verify test environment variables are configured
|
||||
assert os.environ.get("DB_PATH") == test_database
|
||||
assert os.environ.get("ADMIN_EMAIL") == test_users["admin"]["email"]
|
||||
assert os.environ.get("ADMIN_PASSWORD") == test_users["admin"]["password"]
|
||||
|
||||
# Verify temporary database file exists
|
||||
assert os.path.exists(test_database), f"Test database file not found: {test_database}"
|
||||
|
||||
# Verify path contains 'test_mmt_' to confirm isolation
|
||||
assert "test_mmt_" in test_database, "Database path should contain test prefix"
|
||||
|
||||
print(f"✅ Test database isolation confirmed: {test_database}")
|
||||
|
||||
# =============================================================================
|
||||
# PRIORITY 1 TESTS - CRITICAL AUTHENTICATION FLOWS
|
||||
# =============================================================================
|
||||
|
||||
@pytest.mark.e2e
|
||||
@pytest.mark.priority1
|
||||
@pytest.mark.smoke
|
||||
def test_successful_login_with_valid_credentials(self, app_server, test_users, page: Page):
|
||||
"""
|
||||
Priority 1 Test: Validate complete successful login flow with valid credentials
|
||||
|
||||
This test ensures that:
|
||||
1. User can access the login page
|
||||
2. User can enter valid credentials
|
||||
3. Form submission works correctly
|
||||
4. User is redirected to home page after successful authentication
|
||||
5. User session is properly established
|
||||
"""
|
||||
admin_user = test_users["admin"]
|
||||
|
||||
# Step 1: Navigate to login page
|
||||
page.goto(BASE_URL)
|
||||
page.wait_for_load_state("networkidle")
|
||||
|
||||
# Verify we're on the login page
|
||||
expect(page).to_have_url(f"{BASE_URL}/authlogin/login")
|
||||
|
||||
# Step 2: Locate form elements
|
||||
email_input = locate(page, Login.email)
|
||||
password_input = locate(page, Login.password)
|
||||
submit_button = locate(page, Login.submit)
|
||||
|
||||
fill(email_input, admin_user["email"])
|
||||
fill(password_input, admin_user["password"])
|
||||
|
||||
with page.expect_navigation(wait_until="networkidle"):
|
||||
submit_button.click()
|
||||
|
||||
debug(page, "after_login")
|
||||
check_not_in_content(page, ["sign in", "login", "authenticate"])
|
||||
|
||||
page.pause()
|
||||
|
||||
# Step 5: Verify successful authentication and redirect
|
||||
# Should be redirected to the home page
|
||||
expect(page).to_have_url(BASE_URL + "/")
|
||||
|
||||
# Step 6: Verify we're actually authenticated (not redirected back to login)
|
||||
# The page should load without being redirected back to login
|
||||
page.wait_for_load_state("networkidle")
|
||||
current_url = page.url
|
||||
assert not current_url.endswith("/authlogin/login"), f"User was redirected back to login page: {current_url}"
|
||||
|
||||
# Step 7: Verify authenticated content is present
|
||||
# Check that we don't see login-related content anymore
|
||||
page_content = page.content().lower()
|
||||
login_keywords = ["sign in", "login", "authenticate"]
|
||||
|
||||
# Should not see login form elements on authenticated page
|
||||
has_login_content = any(keyword in page_content for keyword in login_keywords)
|
||||
assert not has_login_content, "Login content still visible after successful authentication"
|
||||
|
||||
# Step 8: Verify no error messages are displayed
|
||||
# Look for common error message containers
|
||||
error_selectors = [
|
||||
".error",
|
||||
".alert-error",
|
||||
".bg-error",
|
||||
"[class*='error']",
|
||||
".text-red",
|
||||
"[class*='text-red']"
|
||||
]
|
||||
|
||||
for error_selector in error_selectors:
|
||||
error_elements = page.locator(error_selector)
|
||||
if error_elements.count() > 0:
|
||||
# If error elements exist, they should not be visible or should be empty
|
||||
for i in range(error_elements.count()):
|
||||
element = error_elements.nth(i)
|
||||
if element.is_visible():
|
||||
element_text = element.inner_text().strip()
|
||||
assert not element_text, f"Error message found after successful login: {element_text}"
|
||||
|
||||
# Step 9: Verify the page has expected title
|
||||
expect(page).to_have_title("My Managing Tools")
|
||||
|
||||
print(f"✅ Successful login test completed - User {admin_user['email']} authenticated successfully")
|
||||
0
tests/fixtures/__init__.py
vendored
Normal file
0
tests/fixtures/__init__.py
vendored
Normal file
0
tests/fixtures/app_factory.py
vendored
Normal file
0
tests/fixtures/app_factory.py
vendored
Normal file
56
tests/fixtures/test_database.py
vendored
Normal file
56
tests/fixtures/test_database.py
vendored
Normal file
@@ -0,0 +1,56 @@
|
||||
import tempfile
|
||||
import os
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class TestDatabaseManager:
|
||||
"""Manager for temporary test databases"""
|
||||
|
||||
@staticmethod
|
||||
def create_temp_db_path():
|
||||
"""Create a unique temporary path for test database"""
|
||||
temp_dir = tempfile.mkdtemp(prefix="test_mmt_")
|
||||
return os.path.join(temp_dir, "test_tools.db")
|
||||
|
||||
@staticmethod
|
||||
def setup_test_environment(db_path):
|
||||
"""Configure environment to use test database"""
|
||||
os.environ["DB_PATH"] = db_path
|
||||
os.environ["ADMIN_EMAIL"] = "test.admin@test.com"
|
||||
os.environ["ADMIN_PASSWORD"] = "TestAdmin123"
|
||||
os.environ["SECRET_KEY"] = "test-secret-key-for-e2e-tests"
|
||||
|
||||
@staticmethod
|
||||
def inject_test_database(db_path):
|
||||
"""Inject test database instance into the application"""
|
||||
# Import here to avoid circular imports
|
||||
from src.core.user_database import Database, set_user_db
|
||||
|
||||
# Create test database instance
|
||||
test_db = Database(db_path)
|
||||
|
||||
# Set it as the active instance
|
||||
set_user_db(test_db)
|
||||
|
||||
return test_db
|
||||
|
||||
@staticmethod
|
||||
def cleanup_test_db(db_path):
|
||||
"""Clean up temporary database"""
|
||||
if os.path.exists(db_path):
|
||||
# Clean up the complete temporary directory
|
||||
temp_dir = os.path.dirname(db_path)
|
||||
if os.path.exists(temp_dir):
|
||||
shutil.rmtree(temp_dir, ignore_errors=True)
|
||||
|
||||
@staticmethod
|
||||
def restore_original_environment(original_env):
|
||||
"""Restore original environment"""
|
||||
for key, value in original_env.items():
|
||||
if value is None:
|
||||
# Variable didn't exist before
|
||||
if key in os.environ:
|
||||
del os.environ[key]
|
||||
else:
|
||||
os.environ[key] = value
|
||||
61
tests/fixtures/test_users.py
vendored
Normal file
61
tests/fixtures/test_users.py
vendored
Normal file
@@ -0,0 +1,61 @@
|
||||
"""Test user definitions for E2E authentication testing"""
|
||||
|
||||
|
||||
class TestUsers:
|
||||
"""Class containing test user data"""
|
||||
|
||||
# Admin user (automatically created on startup)
|
||||
ADMIN = {
|
||||
"email": "test.admin@test.com",
|
||||
"password": "TestAdmin123"
|
||||
}
|
||||
|
||||
# Regular user to create for certain tests
|
||||
REGULAR_USER = {
|
||||
"username": "testuser",
|
||||
"email": "user@test.com",
|
||||
"password": "TestUser123"
|
||||
}
|
||||
|
||||
# Failed authentication test cases
|
||||
INVALID_CREDENTIALS = [
|
||||
{
|
||||
"name": "wrong_password",
|
||||
"email": "test.admin@test.com",
|
||||
"password": "wrongpass",
|
||||
"expected_error": "Invalid email or password"
|
||||
},
|
||||
{
|
||||
"name": "nonexistent_user",
|
||||
"email": "nonexistent@test.com",
|
||||
"password": "TestAdmin123",
|
||||
"expected_error": "Invalid email or password"
|
||||
},
|
||||
{
|
||||
"name": "invalid_email_format",
|
||||
"email": "invalid.email",
|
||||
"password": "TestAdmin123",
|
||||
"expected_error": "Invalid email or password"
|
||||
},
|
||||
{
|
||||
"name": "empty_email",
|
||||
"email": "",
|
||||
"password": "TestAdmin123",
|
||||
"expected_error": "Email and password are required"
|
||||
},
|
||||
{
|
||||
"name": "empty_password",
|
||||
"email": "test.admin@test.com",
|
||||
"password": "",
|
||||
"expected_error": "Email and password are required"
|
||||
}
|
||||
]
|
||||
|
||||
# Protected URLs to test for unauthorized access
|
||||
PROTECTED_URLS = [
|
||||
"/",
|
||||
"/admin",
|
||||
"/repositories",
|
||||
"/workflows",
|
||||
"/applications"
|
||||
]
|
||||
28
tests/playwright_config.py
Normal file
28
tests/playwright_config.py
Normal file
@@ -0,0 +1,28 @@
|
||||
import os
|
||||
from playwright.sync_api import Playwright
|
||||
|
||||
# Playwright configuration
|
||||
BASE_URL = os.getenv("APP_URL", "http://localhost:5002")
|
||||
TIMEOUT = 30000
|
||||
EXPECT_TIMEOUT = 5000
|
||||
|
||||
def pytest_configure():
|
||||
"""Configure pytest for Playwright"""
|
||||
pass
|
||||
|
||||
# Browser configuration
|
||||
BROWSER_CONFIG = {
|
||||
#"headless": os.getenv("HEADLESS", "true").lower() == "true",
|
||||
"viewport": {"width": 1280, "height": 720},
|
||||
"ignore_https_errors": True,
|
||||
}
|
||||
|
||||
# Test configuration
|
||||
TEST_CONFIG = {
|
||||
"base_url": BASE_URL,
|
||||
"timeout": TIMEOUT,
|
||||
"expect_timeout": EXPECT_TIMEOUT,
|
||||
"screenshot": "only-on-failure",
|
||||
"video": "retain-on-failure",
|
||||
"trace": "retain-on-failure",
|
||||
}
|
||||
Reference in New Issue
Block a user