Scenarios

1.3 Scenarios

The following scenarios will now be described.

Environment variables

Environment variables are initialised in a separate file. It is not good practice to do this in a function block.

import os
from dotenv import load_dotenv

load_dotenv()

LOCALHOST = os.getenv("LOCALHOST")
PORT = os.getenv("PORT")
USERNAME = os.getenv("USERNAME")
PASSWORD = os.getenv("PASSWORD")

@Patch and Mypy

Because @patch does not retrieve the type of the class or method it is mocking, mypy will always give an Any type error. Therefore, it temporarily ignores that error using: # type: ignore

Function block

Given the class and their correct types.

class FunctionBlock(Generic[ParamsT]):
    @final
    def __init__(self) -> None:
        self.access_token: str = ""
        self.client: Optional[Client] = self.configure_client()
        self.params: Optional[ParamsT] = None

    def configure_client(self) -> Client:
        headers: Dict[str, str] = {"Content-Type": "application/json", "Accept": "application/json"}

        if self.access_token:
            headers["Authorization"] = f"Bearer {self.access_token}"

        return Client(url=f"{Config.url_graphql}", headers=headers)

    def save_token(self, token: str) -> None:
        self.access_token = token
        self.client = self.configure_client()

    @abstractmethod
    async def run(self, context: WorkflowContext, client: Client) -> FunctionBlockResult:
        pass

    @final
    async def execute_function_block(self, context: WorkflowContext) -> FunctionBlockResult:
        try:
            if self.client is not None:
                return await self.run(context, self.client)
            return FunctionBlockResult("Function block could not be executed, check configuration")
        except Exception as e:
            return FunctionBlockResult(f"Error executing function block: {str(e)}")

To test FunctionBlock, It is important to create a subclass that inherits from it because it is an abstract class.

class FunctionBlockSubclass(FunctionBlock):
    async def run(self, context: WorkflowContext) -> FunctionBlockResult:
        return FunctionBlockResult(message="Test run successful.")

Then the test cases have to be prepared:

• Client initialisation: In this case MagicMock is used to simulate the Client.

@patch("neops_worker_sdk.function_block.function_block.FunctionBlock.configure_client")  # type: ignore
def test_client_initialization(self, mock_configure_client: MagicMock) -> None:
    # Mock the configure_client to return a mock client
    mock_client = MagicMock()
    mock_configure_client.return_value = mock_client

    function_block = FunctionBlockSubclass()

    # Assert the client was set correctly
    self.assertEqual(function_block.client, mock_client)

• The run abstract method:

@patch("neops_worker_sdk.function_block.function_block.FunctionBlock.configure_client")  # type: ignore
async def test_run_method(self, mock_configure_client: MagicMock) -> None:
    # Mock the configure_client to return a mock client
    mock_client: MagicMock = MagicMock()
    mock_configure_client.return_value = mock_client

    # Instantiate WorkflowContext and FunctionBlock subclass
    context = WorkflowContext()
    function_block = FunctionBlockSubclass()

    # Call the run method
    result = await function_block.run(context)

    # Assert the result message
    self.assertEqual(result.message, "Test run successful.")

• The save token method:
  • By using side_effect, the scenario simulated is where the function block first initilizes with a Client then reconfigure the client after save the token.
  • It is important to cast the type for the side_effect.

@patch("neops_worker_sdk.function_block.function_block.FunctionBlock.configure_client")  # type: ignore
def test_save_token(self, mock_configure_client: MagicMock) -> None:
    # Mock the configure_client to return a new client after token is set
    mock_client_initial = MagicMock()
    mock_client_after_token = MagicMock()

    # First return the initial client, then the reconfigured client after token save
    mock_configure_client.side_effect = cast(List[MagicMock], (mock_client_initial, mock_client_after_token))

    function_block = FunctionBlockSubclass()

    # Ensure the initial client is set
    self.assertEqual(function_block.client, mock_client_initial)

    # Call save_token and check if client is reconfigured
    function_block.save_token("mock_access_token")

    # Assert the access token is updated
    self.assertEqual(function_block.access_token, "mock_access_token")

    # Assert the client is reconfigured after saving the token
    self.assertEqual(function_block.client, mock_client_after_token)

Login FB

Given the class:

@register_function_block(
    Registration(
        name="ping",
        description="this fetches authenticated user data",
        package="fb.core.neops.io",
        run_on="",
        fb_type="configure",
        json_schema="",
    )
)
class Login(FunctionBlock):
async def run(self, context: WorkflowContext) -> FunctionBlockResult:
    try:
        if self.client is None:
            raise ValueError("Client is not configured correctly.")

        # Perform login
        response = await self.client.login(
            username=USERNAME,
            password=PASSWORD,
        )

        # Ensure response and login attributes are present
        if response is None or response.login is None:
            raise ValueError("Login failed, no response or login data received.")

        access_token: Optional[str] = response.login.access_token
        if access_token is None:
            raise ValueError("Login failed, no access token received.")

        self.save_token(access_token)
        return FunctionBlockResult(message="Login successful.")

    except Exception as e:
        return FunctionBlockResult(message=f"Error fetching task data: {str(e)}")

It is important to add the conditional: if self.client is None. Because mypy requires to evaluate the case of when it is None since the self.client.login method has no explicit types. In this occasion there are two cases to be tested: when login is successful, and when it has failed.

Note: Always look at the structure of the code and check whether the piece of code you want to test uses internal or external calls.

• In this case, the login function use the external call to the Client (response.login.access_token). Then this should be mocked, but you need to understand what the real method is supposed to return: object? dictrionary? It contains nested fields?
• The mock should reflect the same structure as the real return value. If the real method return an object with attributes, the mock sohuld return and object with similar attributes. Because this ensure that test are realistic and that they won't break when runing.
• Instead of just testing weather a certain value is returned, the focus should be on testing how the code interacts with the result.
• If the real method might throw an exception or behave differently in some cases, then side_effect shold be used.

Login succes: In this case the mocking is done to the acces_token

@patch("neops_worker_sdk.function_block.function_block.FunctionBlock.configure_client")  # type: ignore
async def test_login_success(self, mock_configure_client: AsyncMock) -> None:
    # Mock the client and its login method to return a valid access_token
    mock_client: AsyncMock = AsyncMock()
    mock_client.login = AsyncMock(return_value=AsyncMock(login=AsyncMock(access_token="mock_access_token")))
    mock_configure_client.return_value = mock_client

    # Instantiate WorkflowContext and Login block
    context: WorkflowContext = WorkflowContext()
    login_block: FunctionBlock = Login()

    # Run the login block
    result: FunctionBlockResult = await login_block.run(context)

    # Assert that login was successful
    self.assertEqual(result.message, "Login successful.")
    # Assert that the token is saved correctly
    self.assertEqual(login_block.access_token, "mock_access_token")

Login failure: In this case the side_effect is used to simulted the exception.

@patch("neops_worker_sdk.function_block.function_block.FunctionBlock.configure_client")  # type: ignore
async def test_login_failure(self, mock_configure_client: AsyncMock) -> None:
    # Mock the client and its login method to raise an exception
    mock_client: AsyncMock = AsyncMock()
    mock_client.login = AsyncMock(side_effect=Exception("Login failed"))
    mock_configure_client.return_value = mock_client

    # Instantiate WorkflowContext and Login block
    context: WorkflowContext = WorkflowContext()
    login_block: FunctionBlock = Login()

    # Run the login block
    result: FunctionBlockResult = await login_block.run(context)

    # Assert that the error message is returned
    self.assertEqual(result.message, "Error fetching task data: Login failed")

Get Device FB

Given the class:

@register_function_block(
    Registration(
        name="get_random_device",
        description="This fetches a random device from the list of available devices",
        package="fb.core.neops.io",
        run_on="",
        fb_type="query",
        json_schema="",
    )
)
class GetRandomDevice(FunctionBlock):
async def run(self, context: WorkflowContext) -> FunctionBlockResult:
    try:
        if self.client is None:
            raise ValueError("Client is not configured correctly.")

        # Fetch devices using the client method
        response: Devices = await self.client.devices(
            query="",  # Optional query string can be empty to get all devices
            page=1,  # Fetch the first page
            page_size=1,
        )

        # Check if devices_elastic or results are None
        devices_elastic = response.devices_elastic
        if devices_elastic is None or not devices_elastic.results:
            return FunctionBlockResult(message="No devices found.")

        # Device details
        device: Optional[DevicesDevicesElasticResults] = devices_elastic.results[0]

        if device is not None:
            # Ensure connection_state and other attributes are not None
            connection_state: str = device.connection_state.name if device.connection_state else "Unknown"

            message = (
                f"Random device fetched:\n"
                f"  Hostname: {device.hostname}\n"
                f"  IP: {device.ip}\n"
                f"  Connection State: {connection_state}\n"
            )
            return FunctionBlockResult(message=message)
        else:
            return FunctionBlockResult(message="No devices found.")

    except Exception as e:
        return FunctionBlockResult(message=f"Error fetching random device: {str(e)}")

To avoid problems with mypy. Note the types used in the attributes returned by the self.client.devices() method.

In this scenario, do the unit tests for these cases: When a device is found, when it is not found and when there is a client error.

Note: When writing unit tests that involve objects with attributes, such as a device, it is crucial to create the expected object using the relevant attributes rather than relying solely on mocks. This ensures that the test is more accurate and reflects the actual structure and behavior of the object being tested.

Test random device found

• DevicesDevicesElasticResults is a clear example of the importance of ensuring that the test behaves in a predictable way, even if the device changes in the future.
• mock_client.devices = AsyncMock(...) The divice method of mock_client is being mocked using AsyncMock. This means that whenever devices() is called in the code, it will return a mocked asynchronous result instead of executing the actual method.
• return_value=MagicMock(...) This indicates that when devices() is called, it will return a MagicMock object. The MagicMock represents the full result object returned by the devices() method.
• devices_elastic=MagicMock(...) Inside the mocked result of devices(), the device_elastic is the mocked attribute.
• results=cast(List[DevicesDevicesElasticResults], [device])} The cast function is used here to ensure that results is explicitly treated as a list of DevicesDevicesElasticResults objects, which helps with type checking.

@patch("neops_worker_sdk.function_block.function_block.FunctionBlock.configure_client")  # type: ignore
async def test_random_device_found(self, mock_configure_client: AsyncMock) -> None:
    # Mock the client
    mock_client = AsyncMock()

    # Create an DevicesDevicesElasticResults object
    device: DevicesDevicesElasticResults = DevicesDevicesElasticResults(
        id="device1",
        createdAt=None,
        updatedAt=None,
        hostname="device1",
        ip="192.168.1.1",
        connectionState=ConnectionStateEnum.NEW,
        connectionOptions=None,
        facts=None,
        softwareRelease=None,
        platform=None,
        vendor=None,
        model=None,
        serial=None,
        username=None,
        currentConfiguration=None,
        groups=[],
        interfaces=[],
        permission=None,
        checks=None,
        checkElements=None,
    )

    # Mocking the device object in the DevicesDevicesElastic
    mock_client.devices = AsyncMock(
        return_value=MagicMock(
            devices_elastic=MagicMock(
                results=cast(List[DevicesDevicesElasticResults], [device])  # Explicitly cast the list
            )
        )
    )

    mock_configure_client.return_value = mock_client

    # Instantiate WorkflowContext and GetRandomDevice block
    context = WorkflowContext()
    function_block = GetRandomDevice()

    # Call the run method
    result: FunctionBlockResult = await function_block.run(context)

    # Assert the result message
    expected_message = (
        "Random device fetched:\n" "  Hostname: device1\n" "  IP: 192.168.1.1\n" "  Connection State: NEW\n"
    )
    self.assertEqual(result.message, expected_message)

Test no device found

Here, an empty list of the type is returned: List[DevicesDevicesElasticResults]

@patch("neops_worker_sdk.function_block.function_block.FunctionBlock.configure_client")  # type: ignore
async def test_no_device_found(self, mock_configure_client: AsyncMock) -> None:
    # Mock the client and the devices method to return an empty list
    mock_client = AsyncMock()
    mock_client.devices = AsyncMock(
        return_value=MagicMock(devices_elastic=MagicMock(results=cast(List[DevicesDevicesElasticResults], [])))
    )
    mock_configure_client.return_value = mock_client

    # Instantiate WorkflowContext and GetRandomDevice block
    context = WorkflowContext()
    function_block = GetRandomDevice()

    # Call the run method
    result: FunctionBlockResult = await function_block.run(context)

    # Assert the result message for no device found
    self.assertEqual(result.message, "No devices found.")

Test client error

@patch("neops_worker_sdk.function_block.function_block.FunctionBlock.configure_client")  # type: ignore
async def test_client_error(self, mock_configure_client: AsyncMock) -> None:
    # Mock the client and the devices method to raise an exception
    mock_client = AsyncMock()
    mock_client.devices = AsyncMock(side_effect=Exception("Client error"))
    mock_configure_client.return_value = mock_client

    # Instantiate WorkflowContext and GetRandomDevice block
    context = WorkflowContext()
    function_block = GetRandomDevice()

    # Call the run method and expect it to handle the exception
    result: FunctionBlockResult = await function_block.run(context)

    # Assert the error message is returned
    self.assertEqual(result.message, "Error fetching random device: Client error")