operations utils updated w/ tests

.gitignore

@@ -219,3 +219,5 @@ identifier.sqlite
 /tests/MagicMock/*
 /.idea
 *.db
+*/None
+None

research_analytics_suite/operation_manager/operations/core/utils.py

@@ -1,5 +1,22 @@
+"""
+utils.py
+
+This module contains utility functions for extracting operation attributes from different sources such as disk,
+module, operation, and dictionary. The attributes are used to initialize the OperationAttributes class.
+
+Author: Lane
+Copyright: Lane
+Credits: Lane
+License: BSD 3-Clause License
+Version: 0.0.0.1
+Maintainer: Lane
+Email: justlane@uw.edu
+Status: Prototype
+"""
+
 import ast
 import inspect
+import textwrap
 from typing import Optional
 
 from research_analytics_suite.operation_manager.operations.core.OperationAttributes import OperationAttributes
@@ -8,6 +25,15 @@
 
 
 def translate_item(item):
+    """
+    Translates an AST item into a Python object.
+
+    Args:
+        item: The AST item to translate.
+
+    Returns:
+        object: The Python object.
+    """
     if isinstance(item, ast.Str):
         return item.s
     elif isinstance(item, ast.Num):
@@ -39,8 +65,13 @@ async def get_attributes_from_disk(file_path: str) -> Optional[OperationAttribut
     Returns:
         OperationAttributes: The operation attributes.
     """
-    attributes = await load_from_disk(file_path=file_path, operation_group=None)
-    return attributes
+    base_operation = await load_from_disk(file_path=file_path, operation_group=None)
+
+    # Extract OperationAttributes from the BaseOperation
+    if hasattr(base_operation, '_attributes') and isinstance(base_operation.attributes, OperationAttributes):
+        return base_operation.attributes
+    else:
+        raise TypeError("Loaded object is not a valid BaseOperation with OperationAttributes")
 
 
 async def get_attributes_from_module(module) -> OperationAttributes:
@@ -55,6 +86,7 @@ async def get_attributes_from_module(module) -> OperationAttributes:
     """
     # Get the source code of the class
     source = inspect.getsource(module)
+    source = textwrap.dedent(source)
 
     # Parse the source code into an AST
     tree = ast.parse(source)
@@ -102,10 +134,13 @@ async def get_attributes_from_module(module) -> OperationAttributes:
                             CustomLogger().error(AttributeError(f"Invalid attribute: {prop_name}"),
                                                  OperationAttributes.__name__)
     if _op_props.action is None:
-        if callable(module.execute):
+        # Check if the module has an execute method and if it is callable
+        if hasattr(module, 'execute') and callable(getattr(module, 'execute', None)):
             _op_props.action = module.execute
         else:
-            CustomLogger().error(TypeError("operation.execute is not callable"), _op_props.__class__.__name__)
+            # If no execute method, set action to None or handle accordingly
+            _op_props.action = None
+            CustomLogger().warning(f"{module.__name__} has no 'execute' method")
 
     return _op_props
 
@@ -137,7 +172,7 @@ async def get_attributes_from_dict(attributes: dict) -> Optional[OperationAttrib
     Returns:
         OperationAttributes: The operation attributes.
     """
-    del attributes['_initialized']
+    attributes.pop('_initialized', None)  # Safely remove if it exists
     _op = OperationAttributes(**attributes)
     await _op.initialize()
     return _op

tests/operation_manager/operations/core/test_utils.py

@@ -0,0 +1,159 @@
+import json
+
+import pytest
+import ast
+import tempfile
+import os
+from unittest.mock import patch, MagicMock
+
+from research_analytics_suite.operation_manager import translate_item, get_attributes_from_disk, \
+    get_attributes_from_operation, get_attributes_from_dict, get_attributes_from_module
+from research_analytics_suite.operation_manager.operations.core.OperationAttributes import OperationAttributes
+
+
+# Helper fixtures
+@pytest.fixture
+def sample_dict():
+    return {
+        'name': 'test_name',
+        'version': '0.0.2',
+        'description': 'test_description',
+        'category_id': 1,
+        'author': 'test_author'
+    }
+
+
+@pytest.fixture
+def sample_module():
+    class SampleModule:
+        name = "sample"
+        version = "1.0"
+        description = "A sample module"
+
+        def execute(self):
+            return "Executing"
+
+    return SampleModule
+
+
+@pytest.fixture
+async def sample_operation():
+    operation = MagicMock()
+    operation.export_attributes.return_value = {
+        'name': 'operation_name',
+        'version': '0.1.0',
+        'description': 'An operation'
+    }
+    return operation
+
+
+@pytest.fixture
+def temp_file():
+    # Create a temporary file and write valid JSON content
+    temp = tempfile.NamedTemporaryFile(delete=False)
+    with open(temp.name, 'w') as f:
+        json.dump({"key": "value"}, f)  # Example JSON content
+    temp.close()
+    yield temp.name
+    # Ensure cleanup after test
+    if os.path.exists(temp.name):
+        os.remove(temp.name)
+
+
+@pytest.fixture
+def temp_dir():
+    # Create a temporary directory
+    temp_directory = tempfile.TemporaryDirectory()
+    yield temp_directory.name
+    # Cleanup the directory after the test
+    temp_directory.cleanup()
+
+
+class TestUtils:
+
+    # Test translate_item function
+    def test_translate_item_str(self):
+        item = ast.Str(s='test_string')
+        assert translate_item(item) == 'test_string'
+
+    def test_translate_item_num(self):
+        item = ast.Num(n=42)
+        assert translate_item(item) == 42
+
+    def test_translate_item_nameconstant(self):
+        item = ast.NameConstant(value=True)
+        assert translate_item(item) is True
+
+    def test_translate_item_list(self):
+        item = ast.List(elts=[ast.Num(n=1), ast.Num(n=2)])
+        assert translate_item(item) == [1, 2]
+
+    def test_translate_item_dict(self):
+        item = ast.Dict(keys=[ast.Str(s='key')], values=[ast.Num(n=42)])
+        assert translate_item(item) == {'key': 42}
+
+    def test_translate_item_tuple(self):
+        item = ast.Tuple(elts=[ast.Str(s='a'), ast.Str(s='b')])
+        assert translate_item(item) == ('a', 'b')
+
+    def test_translate_item_name(self):
+        item = ast.Name(id='variable_name')
+        assert translate_item(item) == 'variable_name'
+
+    def test_translate_item_attribute(self):
+        item = ast.Attribute(attr='attribute_name')
+        assert translate_item(item) == 'attribute_name'
+
+    # Test get_attributes_from_module function
+    @pytest.mark.asyncio
+    async def test_get_attributes_from_module(self, sample_module):
+        class SampleModule:
+            name = "sample"
+            version = "1.0"
+            description = "A sample module"
+
+            def execute(self):
+                return "Executing"
+
+        # Make sure the source code is stripped of any leading whitespace
+        attributes = await get_attributes_from_module(SampleModule)
+        assert isinstance(attributes, OperationAttributes)
+        assert attributes.name == "sample"
+
+    # Test get_attributes_from_operation function
+    @pytest.mark.asyncio
+    async def test_get_attributes_from_operation(self, sample_operation):
+        attributes = await get_attributes_from_operation(sample_operation)
+        assert isinstance(attributes, OperationAttributes)
+        sample_operation.export_attributes.assert_called_once()
+
+    # Test get_attributes_from_dict function
+    @pytest.mark.asyncio
+    async def test_get_attributes_from_dict(self, sample_dict):
+        attributes = await get_attributes_from_dict(sample_dict)
+        assert isinstance(attributes, OperationAttributes)
+        assert attributes.name == sample_dict['name']
+
+    # Additional tests for edge cases
+    @pytest.mark.asyncio
+    async def test_get_attributes_from_disk_invalid_path(self):
+        with patch('research_analytics_suite.operation_manager.operations.core.workspace.load_from_disk',
+                   side_effect=FileNotFoundError):
+            with pytest.raises(FileNotFoundError):
+                await get_attributes_from_disk('invalid/path')
+
+    @pytest.mark.asyncio
+    async def test_get_attributes_from_module_no_execute(self):
+        class NoExecuteModule:
+            name = "sample"
+            version = "1.0"
+            description = "A sample module"
+
+        attributes = await get_attributes_from_module(NoExecuteModule)
+        assert attributes.action is None
+
+    @pytest.mark.asyncio
+    async def test_get_attributes_from_dict_missing_values(self, sample_dict):
+        sample_dict.pop('name')
+        attributes = await get_attributes_from_dict(sample_dict)
+        assert attributes.name == '[no-name]'