fix

Author: evanwave

sdks/fs-python/function_stream/function.py

@@ -429,7 +429,7 @@ def get_metadata(key: str) -> Any:
                         if key == "message_id":
                             return message.message_id()
                         raise KeyError(key)
-                    
+
                     context.get_metadata = get_metadata
 
                     # Call the function with context as first argument and handle both sync and async results
@@ -449,7 +449,7 @@ def get_metadata(key: str) -> Any:
                         logger.error(f"Error invoking process function: {str(e)}")
                         raise Exception(f"Error invoking process function: {str(e)}") from e
                     if response_data:
-                        resp_msgs.append(MsgWrapper(data=response_data, event_time=datetime.utcnow()))
+                        resp_msgs.append(MsgWrapper(data=response_data, event_time=datetime.now(timezone.utc)))
 
                     if not response_topic:
                         logger.warning("No response_topic provided and no sink topic available. Skip messages")
@@ -481,7 +481,7 @@ def get_metadata(key: str) -> Any:
                             await self._send_response(
                                 response_topic,
                                 request_id,
-                                [MsgWrapper(data={'error': str(e)}, event_time=datetime.utcnow())]
+                                [MsgWrapper(data={'error': str(e)}, event_time=datetime.now(timezone.utc))]
                             )
                     self.metrics.record_request_end(False, time.time() - start_time)
                     self.metrics.record_event(False)
@@ -642,7 +642,7 @@ async def close(self):
     def __del__(self):
         """
         Ensure resources are cleaned up when the object is destroyed.
-        
+
         This finalizer ensures that all resources are properly closed when the
         object is garbage collected. It provides a safety net for resource cleanup
         in case the explicit close() method is not called.

sdks/fs-python/function_stream/module.py

@@ -1,3 +1,11 @@
+"""
+FSModule module provides the base class for all FunctionStream modules.
+
+This module defines the abstract base class FSModule that all FunctionStream modules
+must inherit from. It provides a common interface for module initialization and
+data processing, ensuring consistency across different module implementations.
+"""
+
 from abc import ABC, abstractmethod
 from typing import Dict, Any
 
@@ -8,32 +16,51 @@ class FSModule(ABC):
     """
     Base class for all FunctionStream modules.
     
-    This class provides a common interface for all modules in the FunctionStream SDK.
-    Each module must implement the process method to handle incoming data.
+    This abstract base class provides a common interface for all modules in the 
+    FunctionStream SDK. Each module must implement the init and process methods
+    to handle module initialization and incoming data processing.
     
     Attributes:
-        name (str): The name of the module
+        name (str): The name of the module (to be set during initialization).
     """
 
     @abstractmethod
     def init(self, context: FSContext):
         """
-        Initialize the module with a name.
+        Initialize the module with the provided context.
+        
+        This method is called during module initialization to set up the module
+        with the necessary context and configuration. Subclasses must implement
+        this method to handle any required setup.
         
         Args:
-            name (str): The name of the module
+            context (FSContext): The context object containing configuration and 
+                               runtime information for the module.
         """
+        pass
 
     @abstractmethod
     async def process(self, context: FSContext, data: Dict[str, Any]) -> Dict[str, Any]:
         """
-        Process incoming data.
+        Process incoming data asynchronously.
+        
+        This method is the core processing function that handles incoming data.
+        Subclasses must implement this method to define the specific data processing
+        logic for their module. The method should be asynchronous to support
+        non-blocking operations.
         
         Args:
-            context (FSContext): The context object containing configuration and runtime information
-            data (Dict[str, Any]): The input data to process
+            context (FSContext): The context object containing configuration and 
+                               runtime information.
+            data (Dict[str, Any]): The input data to process. This is typically
+                                  a dictionary containing the message payload
+                                  and any associated metadata.
             
         Returns:
-            Union[Dict[str, Any], Awaitable[Dict[str, Any]]]: The processed data or an awaitable that will resolve to the processed data
+            Dict[str, Any]: The processed data that should be returned as the
+                           result of the processing operation.
+            
+        Raises:
+            NotImplementedError: This method must be implemented by subclasses.
         """
         raise NotImplementedError("Subclasses must implement process method")

sdks/fs-python/tests/test_context.py

@@ -3,7 +3,7 @@
 """
 
 from unittest.mock import Mock
-from datetime import datetime
+from datetime import datetime, timezone
 
 import pytest
 
@@ -73,16 +73,16 @@ def test_get_metadata_default_implementation(self, context):
     def test_produce_default_implementation(self, context):
         """Test that produce does nothing by default."""
         test_data = {"key": "value"}
-        test_time = datetime.utcnow()
-        
+        test_time = datetime.now(timezone.utc)
+
         # Should not raise any exception
         result = context.produce(test_data, test_time)
         assert result is None
 
     def test_produce_without_event_time(self, context):
         """Test produce method without event_time parameter."""
         test_data = {"key": "value"}
-        
+
         # Should not raise any exception
         result = context.produce(test_data)
         assert result is None

sdks/fs-python/tests/test_function.py

@@ -115,9 +115,7 @@ async def process_func(context: FSContext, data: Dict[str, Any]) -> Dict[str, An
     async def test_init(self):
         """Test FSFunction initialization."""
         with patch('function_stream.function.Config.from_yaml') as mock_from_yaml, \
-                patch('function_stream.function.Client') as mock_client, \
-                patch('function_stream.function.Metrics') as mock_metrics, \
-                patch('function_stream.function.MetricsServer') as mock_metrics_server:
+                patch('function_stream.function.Client') as mock_client:
             mock_config = Mock(spec=Config)
             mock_config.module = "test_module"
             mock_config.subscriptionName = "test_subscription"