From f60b5cc2abc1a33a77e7276fa45edb8bdcb816da Mon Sep 17 00:00:00 2001 From: Lane Date: Wed, 11 Sep 2024 14:36:17 -0700 Subject: [PATCH] operations utils updated w/ tests --- .gitignore | 2 + .../operations/core/utils.py | 45 ++++- .../operations/core/test_utils.py | 159 ++++++++++++++++++ 3 files changed, 201 insertions(+), 5 deletions(-) create mode 100644 tests/operation_manager/operations/core/test_utils.py diff --git a/.gitignore b/.gitignore index f65926b..137e298 100644 --- a/.gitignore +++ b/.gitignore @@ -219,3 +219,5 @@ identifier.sqlite /tests/MagicMock/* /.idea *.db +*/None +None diff --git a/research_analytics_suite/operation_manager/operations/core/utils.py b/research_analytics_suite/operation_manager/operations/core/utils.py index bb42997..eed76ef 100644 --- a/research_analytics_suite/operation_manager/operations/core/utils.py +++ b/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 diff --git a/tests/operation_manager/operations/core/test_utils.py b/tests/operation_manager/operations/core/test_utils.py new file mode 100644 index 0000000..fe53e25 --- /dev/null +++ b/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]'