153 lines
4.9 KiB
Python
153 lines
4.9 KiB
Python
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from components.workflows.constants import WORKFLOWS_DB_ENTRY
|
|
from components.workflows.db_management import WorkflowsDbManager, WorkflowsSettings
|
|
from core.settings_management import SettingsManager, MemoryDbEngine
|
|
|
|
USER_EMAIL = "test@mail.com"
|
|
USER_ID = "test_user"
|
|
|
|
|
|
@pytest.fixture
|
|
def session():
|
|
return {"user_id": USER_ID, "user_email": USER_EMAIL}
|
|
|
|
|
|
@pytest.fixture
|
|
def settings_manager():
|
|
return SettingsManager(engine=MemoryDbEngine())
|
|
|
|
|
|
@pytest.fixture
|
|
def workflows_db_manager(session, settings_manager):
|
|
return WorkflowsDbManager(session=session, settings_manager=settings_manager)
|
|
|
|
|
|
def test_add_workflow(workflows_db_manager):
|
|
# Test adding a new workflow
|
|
assert workflows_db_manager.add_workflow("workflow1") is True
|
|
|
|
# Verify workflow was added
|
|
workflows = workflows_db_manager.get_workflows()
|
|
assert "workflow1" in workflows
|
|
assert len(workflows) == 1
|
|
|
|
|
|
def test_add_workflow_empty_name(workflows_db_manager):
|
|
# Test adding a workflow with empty name raises ValueError
|
|
with pytest.raises(ValueError, match="Workflow name cannot be empty."):
|
|
workflows_db_manager.add_workflow("")
|
|
|
|
|
|
def test_add_workflow_duplicate(workflows_db_manager):
|
|
# Add a workflow
|
|
workflows_db_manager.add_workflow("workflow1")
|
|
|
|
# Test adding duplicate workflow raises ValueError
|
|
with pytest.raises(ValueError, match="Workflow 'workflow1' already exists."):
|
|
workflows_db_manager.add_workflow("workflow1")
|
|
|
|
|
|
def test_get_workflow(workflows_db_manager):
|
|
# Add a workflow
|
|
workflows_db_manager.add_workflow("workflow1")
|
|
|
|
# Test getting the workflow
|
|
workflow = workflows_db_manager.get_workflow("workflow1")
|
|
assert workflow == "workflow1"
|
|
|
|
|
|
def test_get_workflow_empty_name(workflows_db_manager):
|
|
# Test getting a workflow with empty name raises ValueError
|
|
with pytest.raises(ValueError, match="Workflow name cannot be empty."):
|
|
workflows_db_manager.get_workflow("")
|
|
|
|
|
|
def test_get_workflow_nonexistent(workflows_db_manager):
|
|
# Test getting a non-existent workflow raises ValueError
|
|
with pytest.raises(ValueError, match="Workflow 'nonexistent' does not exist."):
|
|
workflows_db_manager.get_workflow("nonexistent")
|
|
|
|
|
|
def test_remove_workflow(workflows_db_manager):
|
|
# Add a workflow
|
|
workflows_db_manager.add_workflow("workflow1")
|
|
|
|
# Test removing the workflow
|
|
assert workflows_db_manager.remove_workflow("workflow1") is True
|
|
|
|
# Verify workflow was removed
|
|
assert len(workflows_db_manager.get_workflows()) == 0
|
|
|
|
|
|
def test_remove_workflow_empty_name(workflows_db_manager):
|
|
# Test removing a workflow with empty name raises ValueError
|
|
with pytest.raises(ValueError, match="Workflow name cannot be empty."):
|
|
workflows_db_manager.remove_workflow("")
|
|
|
|
|
|
def test_remove_workflow_nonexistent(workflows_db_manager):
|
|
# Test removing a non-existent workflow raises ValueError
|
|
with pytest.raises(ValueError, match="workflow 'nonexistent' does not exist."):
|
|
workflows_db_manager.remove_workflow("nonexistent")
|
|
|
|
|
|
def test_exists_workflow(workflows_db_manager):
|
|
# Add a workflow
|
|
workflows_db_manager.add_workflow("workflow1")
|
|
|
|
# Test workflow exists
|
|
assert workflows_db_manager.exists_workflow("workflow1") is True
|
|
|
|
# Test non-existent workflow
|
|
assert workflows_db_manager.exists_workflow("nonexistent") is False
|
|
|
|
|
|
def test_exists_workflow_empty_name(workflows_db_manager):
|
|
# Test checking existence of workflow with empty name raises ValueError
|
|
with pytest.raises(ValueError, match="workflow name cannot be empty."):
|
|
workflows_db_manager.exists_workflow("")
|
|
|
|
|
|
def test_get_workflows(workflows_db_manager):
|
|
# Initially, no workflows
|
|
assert len(workflows_db_manager.get_workflows()) == 0
|
|
|
|
# Add workflows
|
|
workflows_db_manager.add_workflow("workflow1")
|
|
workflows_db_manager.add_workflow("workflow2")
|
|
|
|
# Test getting all workflows
|
|
workflows = workflows_db_manager.get_workflows()
|
|
assert "workflow1" in workflows
|
|
assert "workflow2" in workflows
|
|
assert len(workflows) == 2
|
|
|
|
|
|
def test_select_workflow(workflows_db_manager):
|
|
# Add a workflow
|
|
workflows_db_manager.add_workflow("workflow1")
|
|
|
|
# Select the workflow
|
|
workflows_db_manager.select_workflow("workflow1")
|
|
|
|
# Verify workflow was selected
|
|
assert workflows_db_manager.get_selected_workflow() == "workflow1"
|
|
|
|
|
|
def test_get_selected_workflow_none(workflows_db_manager):
|
|
# Initially, no selected workflow
|
|
assert workflows_db_manager.get_selected_workflow() is None
|
|
|
|
|
|
def test_get_settings_default(workflows_db_manager, session, settings_manager):
|
|
# Test _get_settings returns default settings when none exist
|
|
with patch.object(settings_manager, 'load', return_value=WorkflowsSettings()) as mock_load:
|
|
settings = workflows_db_manager._get_settings()
|
|
mock_load.assert_called_once_with(session, WORKFLOWS_DB_ENTRY, default=WorkflowsSettings())
|
|
assert isinstance(settings, WorkflowsSettings)
|
|
assert settings.workflows == []
|
|
assert settings.selected_workflow is None
|