Testing an Agent

See IoTPy/IoTPy/tests

An application may consist of a single process, or a multiple processes on a multicore shared-memory computer, or a distributed network of virtual machines where each machine represents a multicore (or single core) computer. A process consists of sources and sinks and an agent that reads sources and writes to sinks. The section single process applications tells you how to test and execute applications consisting of single processes, while multiprocess applications deals with multiple shared-memory processes and distributed applications with networks of message-passing VMs.

You may want to test an agent, buy itself, without the sources of data that drive it. This section tells you how to do that.

Test Files

Look at the tests in the package IoTPy/IoTPy/tests to see examples of a single process with fixed input. Files in this package include: element_test.py, list_test.py, window_test.py, merge_test.py, split_test.py, source_test.py, sink_test.py, multi_test.py, and shared_variable_test.py.

The examples are as follows:

  1. Test single agent. The first three examples test a single agent.

  2. Test an acyclic network. Most applications use acyclic networks. The next examples illustrate simple tests of acyclic networks.

  3. Test cyclic network. This example tests a feedback network: a network consisting of a cycle - agent A's output is agent B's input and agent B's output is agent A's input.

  4. Change agent parameters during test. This example shows how you can change parameters of an agent during a test.

test an agent in small steps

You can drive an agent in small steps and inspect the output at each step. A step in testing an agent consists of the following three operations:

  1. Input data. Put data into input streams of the network.

  2. Step. Call Stream.scheduler.step()

  3. Inspect output. Inspect the values in the output streams of the network. The most recent values in a stream y are in recent_values(y) which is a list (or NumPy array). Only the most recent values of the stream need to be inspected because the output stream will have only a few elements after a small number of steps.

TEST SINGLE AGENT

Example: Testing map_element

The following code gives an example of the steps to test a map_element agent.

# SPECIFY IMPORTS
import ...

# A TEST FUNCTION
def test_example_1():
    # -----------------------------------------
    # SPECIFY THE AGENT AND DECLARE STREAMS
    def f(v): return 2*v
    x = Stream('x')
    y = Stream('y')
    map_element(func=f, in_stream=x, out_stream=y)

    # -----------------------------------------
    # TEST THE AGENT IN STEPS.
    # (a) PUT TEST VALUES IN THE INPUT STREAMS.
    x.extend([0, 1, 2])
    # (b) EXECUTE A STEP.
    Stream.scheduler.step()
    # (c) INSPECT THE OUTPUT STREAMS.
    assert recent_values(y) == [0, 2, 4]
    assert recent_values(x) == [0, 1, 2]

    # EXECUTE ANOTHER STEP
    # (a) PUT TEST VALUES IN THE INPUT STREAMS.
    x.extend([10, 20, 30])
    # (b) EXECUTE A STEP.
    Stream.scheduler.step()
    # (c) INSPECT THE OUTPUT STREAMS.
    assert recent_values(y) == [0, 2, 4, 20, 40, 60]
    assert recent_values(x) == [0, 1, 2, 10, 20, 30]

In the first step, the input stream, x,  becomes [0, 1, 2] and so the network makes the output stream, y, become [0, 2, 4]. In the second step, the input stream, x, becomes [0, 1, 2, 10, 20, 30] and so the network makes the output stream, y, become [0, 2, 4, 20, 40, 60].we 

Example: Testing filter_element

This example illustrates that exactly the same steps are used for testing any agent.

# SPECIFY IMPORTS
import ...

# A TEST FUNCTION
def test_example_2():
    # SPECIFY THE AGENT AND DECLARE STREAMS
    def f(v): return v < 3
    x = Stream('x')
    y = Stream('y')
    filter_element(func=f, in_stream=x, out_stream=y)

    # TEST THE AGENT IN STEPS
    # (a) PUT TEST VALUES IN THE INPUT STREAMS
    x.extend([0, 1, 2, 3, 4])
    # (b) EXECUTE A STEP.
    Stream.scheduler.step()
    # (c) INSPECT THE OUTPUT STREAMS
    assert recent_values(y) == [3, 4]

    # EXECUTE ANOTHER STEP
    # (a) PUT TEST VALUES IN THE INPUT STREAMS
    x.extend([-10, 10])
    # (b) EXECUTE A STEP.
    Stream.scheduler.step()
    # (c) INSPECT THE OUTPUT STREAMS
    assert recent_values(y) == [3, 4, 10]

example: testing zip_map

This example is a test of a merge agent.

def simple_zip_map_test():
    # SPECIFY THE AGENT AND DECLARE STREAMS
    def f(lst):
       return 2*sum(lst)
    x = Stream('x')
    y = Stream('y')
    z = Stream('z')
    zip_map(func=f, in_streams=[x,y], out_stream=z)

    TEST THE AGENT IN STEPS
    # (a) PUT TEST VALUES IN THE INPUT STREAMS.
    x.extend([0, 1, 2, 3])
    y.extend([10, 12, 14, 16, 18])
    # (b) EXECUTE A STEP.
    Stream.scheduler.step()
    # (c) INSPECT THE OUTPUT STREAMS.
    assert recent_values(z) == [20, 26, 32, 38]

TEST A COMPOSED AGENT

Example: test an agents with component agents.

import ....

# A TEST FUNCTION
def test_example():
    # ---------------------------------------------
    # SPECIFY THE AGENT AND DECLARE STREAMS
    def f(v): return 2*v
    def g(v): return v%10 == 0
    x = Stream('x')
    y = Stream('y')
    z = Stream('z')
    map_element(func=f, in_stream=x, out_stream=y)
    filter_element(func=g, in_stream=y, out_stream=z)

    # TEST THE AGENT IN STEPS.
    # (a) PUT TEST VALUES IN THE INPUT STREAMS.
    x.extend([0, 5, 8])
    # (b) EXECUTE A STEP.
    Stream.scheduler.step()
    # (c) INSPECT THE STREAMS.
    assert recent_values(x) == [0, 5, 8]
    assert recent_values(y) == [0, 10, 16]
    assert recent_values(z) == [16]

    # EXECUTE ANOTHER STEP
    # (a) PUT TEST VALUES IN THE INPUT STREAMS.
    x.extend([10, 22, 25])
    # (b) EXECUTE A STEP.
    Stream.scheduler.step()
    # (c) INSPECT THE OUTPUT STREAMS.
    assert recent_values(x) == [0, 5, 8, 10, 22, 25]
    assert recent_values(y) == [0, 10, 16, 20, 44, 50]
    assert recent_values(z) == [16, 44]

TEST A CYCLIC NETWORK

Example: Testing a cycle of agents

This example has two agents fibonacci and controller. Each of these agents has a single input stream and a single output stream. The output from each agent is the input to the other agent. The agent, fibonacci, has an output stream x and an input stream y, while controller has an output stream y and an input stream x. Thus the network is a cycle with two nodes and two edges.

The agent, fibonacci, outputs the fibonacci sequence 1, 1, 2, 3, 5, 8, 13, 21, ... on stream x.  Each element of the sequence (after the first two 1s) is the sum of the previous two elements. The agent appends the k-th element of the sequence to x when it reads the k-th element of stream y.

The agent, controller, filters out elements of x that are divisible by a parameter, divisor, In the test divisor is set to 4. Thus y contains the elements of x that are not divisible by 4.

Slide1.jpg

Initially, we set y to have a single element, 0, while x is empty. At the first step, fibonacci reads this 0 in y and appends the first element, 1, of the sequence to x. Controller reads this element of x, and appends it to y since it is not divisible by 4. Fibonacci reads this element (1) in y and so it appends the second element, 1, of the fibonacci sequence to x. This cycle of agent actions continues until controller reads 8 on stream x, and filters it out, leaving y unchanged. At this point stream x consists of elements 1, 1, 2, 3, 5, 8 and stream y consists of 0, 1, 1, 2, 3, 5, 8 ; and the network is quiescent, and so this terminates execution of the step.

To execute another step we append another 0 to y and the cycle repeats until x contains the elements: 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, and y is [0, 1, 1, 2, 3, 5, 0, 13, 21, 34, 55, 89]

# SPECIFY IMPORTS
import sys
import os
sys.path.append(os.path.abspath("../helper_functions"))
sys.path.append(os.path.abspath("../core"))
sys.path.append(os.path.abspath("../agent_types"))

# agent and stream are in ../core
from agent import Agent
from stream import Stream, _no_value
# recent_values is in ../helper_functions
from recent_values import recent_values
# op is in ../agent_types
from op import map_element, filter_element

def test_example_3():
    #-------------------------------------------
    # SPECIFY THE AGENT NETWORK
    # (a) SPECIFY ENCAPSULATED FUNCTIONS
    def f(v, state):
        # Generate the fibonacci sequence
        # prefinal, final are the last two values
        # of the fibonacci sequence generated so
        # far.
        final, prefinal = state
        next_output = final + prefinal
        # In the next state:
        # prefinal becomes final
        # final becomes next_output
        next_state = next_output, final
        return next_output, next_state
    
    def g(v, divisor):
        return v % divisor == 0
            
    # (b) SPECIFY STREAMS
    x = Stream('x')
    y = Stream('y')

    # (c) SPECIFY AGENTS.
    map_element(func=f, in_stream=y, out_stream=x, state=(0, 1))
    filter_element(func=g, in_stream=x, out_stream=y, divisor=4)
    #-------------------------------------------


    # DRIVE THE NETWORK IN STEPS
    # (a) PUT TEST VALUES IN THE INPUT STREAMS.
    y.append(0)
    # (b) EXECUTE STEP
    Stream.scheduler.step()
    # (c) INSPECT OUTPUT STREAMS.
    assert recent_values(x) == [1, 1, 2, 3, 5, 8]

    # EXECUTE ANOTHER STEP
    # (a) PUT TEST VALUES IN THE INPUT STREAMS.
    y.append(0)
    # (b) EXECUTE STEP
    Stream.scheduler.step()
    # (c) INSPECT OUTPUT STREAMS.
    assert recent_values(x) == \
      [1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144]

CHANGE AGENT'S PARAMETERS IN TEST

This example illustrates how to change an agent's parameters during its execution. In this example, the value of divisor (see previous example) is changed during execution of the agent. The example is the same as the previous one with three changes: (1) The encapsulated function g of the previous example is replaced by obj.g where obj is an instance of a class G. (2) The parameter, divisor, in the previous example is replaced by obj.divisor. (3) Before executing the second step, the value of obj.divisor is changed from 4 to 2.

def test_example_4():
    #-------------------------------------------
    # SPECIFY THE AGENT NETWORK
    # (a) SPECIFY ENCAPSULATED FUNCTIONS
    def f(v, state):
        # Generate the fibonacci sequence
        # prefinal, final are the last two values
        # of the fibonacci sequence generated so
        # far.
        final, prefinal = state
        next_output = final + prefinal
        # In the next state:
        # prefinal becomes final
        # final becomes next_output
        next_state = next_output, final
        return next_output, next_state

    class G(object):
        # This class has parameters such as divisor
        # and the encapsulated function g.
        def __init__(self):
            self.divisor = 4
        def g(self, v):
            return v % self.divisor == 0
            
    # (b) SPECIFY STREAMS
    x = Stream('x')
    y = Stream('y')

    # (c) SPECIFY AGENTS.
    map_element(func=f, in_stream=y, out_stream=x, state=(0, 1))
    # Create obj, an instance of class G, and call the
    # encapsulator filter_element to encapsulate
    # function obj.g
    obj = G()
    filter_element(func=obj.g, in_stream=x, out_stream=y)
    #-------------------------------------------


    # DRIVE THE NETWORK IN STEPS
    # (a) PUT TEST VALUES IN THE INPUT STREAMS.
    y.append(0)
    # (b) EXECUTE A STEP
    Stream.scheduler.step()
    # (c) INSPECT OUTPUT STREAMS.
    assert recent_values(x) == [1, 1, 2, 3, 5, 8]

    # EXECUTE A STEP AFTER CHANGING AGENT PARAMETERS
    obj.divisor = 2
    # (a) PUT TEST VALUES IN THE INPUT STREAMS.
    y.append(0)
    # (b) EXECUTE STEP
    Stream.scheduler.step()
    # (c) INSPECT OUTPUT STREAMS.
    assert recent_values(x) == \
      [1, 1, 2, 3, 5, 8, 13, 21, 34]