<|file_name|>test_actor.py<|end_file_name|><｜fim▁begin｜># -*- coding: utf-8 -*-

# Copyright (c) 2015 Ericsson AB
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import pytest

from mock import Mock
from calvin.tests import DummyNode
from calvin.runtime.north.actormanager import ActorManager
from calvin.runtime.south.endpoint import LocalOutEndpoint, LocalInEndpoint
from calvin.actor.actor import Actor

pytestmark = pytest.mark.unittest


def create_actor(node):
    actor_manager = ActorManager(node)
    actor_id = actor_manager.new('std.Identity', {})
    actor = actor_manager.actors[actor_id]
    actor._calvinsys = Mock()
    return actor


@pytest.fixture
def actor():
    return create_actor(DummyNode())


@pytest.mark.parametrize("port_type,port_name,port_property,value,expected", [
    ("invalid", "", "", "", False),
    ("in", "missing", "", "", False),
    ("out", "missing", "", "", False),
    ("out", "token", "missing", "", False),
    ("in", "token", "missing", "", False),
    ("out", "token", "name", "new_name", True),
    ("out", "token", "name", "new_name", True),
])
def test_set_port_property(port_type, port_name, port_property, value, expected):
    assert actor().set_port_property(port_type, port_name, port_property, value) is expected


@pytest.mark.parametrize("inport_ret_val,outport_ret_val,expected", [
    (False, False, False),
    (False, True, False),
    (True, False, False),
    (True, True, True),
])
def test_did_connect(actor, inport_ret_val, outport_ret_val, expected):
    for port in actor.inports.values():
        port.is_connected = Mock(return_value=inport_ret_val)
    for port in actor.outports.values():
        port.is_connected = Mock(return_value=outport_ret_val)

    actor.fsm = Mock()
    actor.did_connect(None)
    if expected:
        actor.fsm.transition_to.assert_called_with(Actor.STATUS.ENABLED)
        assert actor._calvinsys.scheduler_wakeup.called
    else:
        assert not actor.fsm.transition_to.called
        assert not actor._calvinsys.scheduler_wakeup.called


@pytest.mark.parametrize("inport_ret_val,outport_ret_val,expected", [
    (True, True, False),
    (True, False, False),
    (False, True, False),
    (False, False, True),
])
def test_did_disconnect(actor, inport_ret_val, outport_ret_val, expected):
    for port in actor.inports.values():
        port.is_connected = Mock(return_value=inport_ret_val)
    for port in actor.outports.values():
        port.is_connected = Mock(return_value=outport_ret_val)

    actor.fsm = Mock()
    actor.did_disconnect(None)
    if expected:
        actor.fsm.transition_to.assert_called_with(Actor.STATUS.READY)
    else:
        assert not actor.fsm.transition_to.called


def test_enabled(actor):
    actor.enable()
    assert actor.enabled()
    actor.disable()
    assert not actor.enabled()


def test_connections():
    node = DummyNode()
    node.id = "node_id"
    actor = create_actor(node)
    inport = actor.inports['token']
    outport = actor.outports['token']

    port = Mock()
    port.id = "x"
    peer_port = Mock()
    peer_port.id = "y"

    inport.attach_endpoint(LocalInEndpoint(port, peer_port))
    outport.attach_endpoint(LocalOutEndpoint(port, peer_port))

    assert actor.connections(node) == {
        'actor_id': actor.id,
        'actor_name': actor.name,
        'inports': {inport.id: (node, "y")},
        'outports': {outport.id: [(node, "y")]}
    }


def test_state(actor):
    inport = actor.inports['token']
    outport = actor.outports['token']
    correct_state = {
        '_component_members': set([actor.id]),
        '_deployment_requirements': [],
        '_managed': set(['dump', '_signature', 'id', '_deployment_requirements', 'name', 'credentials']),
        '_signature': None,
        'dump': False,
        'id': actor.id,
        'inports': {'token': {'fifo': {'N': 5,
                                       'fifo': [{'data': 0, 'type': 'Token'},
                                                {'data': 0, 'type': 'Token'},
                                                {'data': 0, 'type': 'Token'},
                                                {'data': 0, 'type': 'Token'},
                                                {'data': 0, 'type': 'Token'}],
                                       'read_pos': {inport.id: 0},
                                       'readers': [inport.id],
                                       'tentative_read_pos': {inport.id: 0},
                                       'write_pos': 0},
                              'id': inport.id,
                              'name': 'token'}},
        'name': '',
        'outports': {'token': {'fanout': 1,
                               'fifo': {'N': 5,
                                        'fifo': [{'data': 0, 'type': 'Token'},
                                                 {'data': 0, 'type': 'Token'},
                                                 {'data': 0, 'type': 'Token'},
                                                 {'data': 0, 'type': 'Token'},
                                                 {'data': 0, 'type': 'Token'}],
                                        'read_pos': {},
                                        'readers': [],
                                        'tentative_read_pos': {},
                                        'write_pos': 0},
                               'id': outport.id,
                               'name': 'token'}}}
    test_state = actor.state()
    for k, v in correct_state.iteritems():
        # Read state use list to support JSON serialization
        if isinstance(v, set):
            assert set(test_state[k]) == v
        else:
            assert test_state[k] == v

@pytest.mark.parametrize("prev_signature,new_signature,expected", [
    (None, "new_val", "new_val"),
    ("old_val", "new_val", "old_val")
])
def test_set_signature(actor, prev_signature, new_signature, expected):
    <|fim_middle|>



def test_component(actor):
    actor.component_add(1)
    assert 1 in actor.component_members()

    actor.component_add([2, 3])
    assert 2 in actor.component_members()
    assert 3 in actor.component_members()

    actor.component_remove(1)
    assert 1 not in actor.component_members()
    actor.component_remove([2, 3])
    assert 2 not in actor.component_members()
    assert 3 not in actor.component_members()


def test_requirements(actor):
    assert actor.requirements_get() == []
    actor.requirements_add([1, 2, 3])
    assert actor.requirements_get() == [1, 2, 3]
    actor.requirements_add([4, 5])
    assert actor.requirements_get() == [4, 5]
    actor.requirements_add([6, 7], extend=True)
    assert actor.requirements_get() == [4, 5, 6, 7]
<｜fim▁end｜>
    actor.signature_set(prev_signature)
    actor.signature_set(new_signature)
    assert actor._signature == expected