<commit_before>from __future__ import print_function

import unittest

from flowpipe.node import INode
from flowpipe.plug import InputPlug, OutputPlug


class SquareNode(INode):
    """Square the given value."""

    def __init__(self, name=None):
        """Init the node."""
        super(SquareNode, self).__init__(name)

        # Inputs
        InputPlug('in', self)

        # Outputs
        OutputPlug('out', self)
    # end def __init__

    def compute(self):
        """Square the given input and send to the output."""
        squared = self.inputs['in'].value**2
        self.outputs['out'].value = squared
    # end def compute
# end class SquareNode


class TestPlugs(unittest.TestCase):
    """Test the INode."""

    def test_downstream_upstream_nodes(self):
        """Verify downstream and upstream Nodes."""
        node_a = SquareNode('NodeA')
        node_b = SquareNode('NodeB')
        node_c = SquareNode('NodeC')
        node_a.outputs['out'] >> node_b.inputs['in']
        node_a.outputs['out'] >> node_c.inputs['in']

        self.assertEqual(2, len(node_a.downstream_nodes))
        self.assertIn(node_b, node_a.downstream_nodes)
        self.assertIn(node_c, node_a.downstream_nodes)

        self.assertEqual(1, len(node_b.upstream_nodes))
        self.assertIn(node_a, node_b.upstream_nodes)
    # end def test_downstream_upstream_nodes

    def test_evaluate(self):
        """Evaluate the Node will push the new data to it's output."""
        node = SquareNode()
        test_input = 2
        self.assertIsNone(node.outputs['out'].value)
        node.inputs['in'].value = test_input
        node.evaluate()
        self.assertEqual(test_input**2, node.outputs['out'].value)
    # end def test_evaluate

    def test_dirty_depends_on_inputs(self):
        """Dirty status of a Node depends on it's Plugs."""
        node = SquareNode()
        self.assertTrue(node.is_dirty)

        node.inputs['in'].is_dirty = False
        self.assertFalse(node.is_dirty)

        node.inputs['in'].value = 2
        self.assertTrue(node.is_dirty)
    # end def test_dirty_depends_on_inputs

    def test_evaluate_sets_all_inputs_clean(self):
        """After the evaluation, the inputs are considered clean."""
        node = SquareNode()
        node.inputs['in'].value = 2
        self.assertTrue(node.is_dirty)
        node.evaluate()
        self.assertFalse(node.is_dirty)
    # end def test_evaluate_sets_all_inputs_clean
    
    def test_cannot_connect_node_to_itself(self):
        """A node can not create a cycle by connecting to itself."""
        node = SquareNode()
        self.assertRaises(Exception):
            node.outputs['out'] >> node.inputs['in']
    # end def test_cannot_connect_node_to_itself
# end class TestPlugs<commit_msg>Update test_node.py<commit_after>from __future__ import print_function

import unittest

from flowpipe.node import INode
from flowpipe.plug import InputPlug, OutputPlug


class SquareNode(INode):
    """Square the given value."""

    def __init__(self, name=None):
        """Init the node."""
        super(SquareNode, self).__init__(name)

        # Inputs
        InputPlug('in', self)

        # Outputs
        OutputPlug('out', self)
    # end def __init__

    def compute(self):
        """Square the given input and send to the output."""
        squared = self.inputs['in'].value**2
        self.outputs['out'].value = squared
    # end def compute
# end class SquareNode


class TestPlugs(unittest.TestCase):
    """Test the INode."""

    def test_downstream_upstream_nodes(self):
        """Verify downstream and upstream Nodes."""
        node_a = SquareNode('NodeA')
        node_b = SquareNode('NodeB')
        node_c = SquareNode('NodeC')
        node_a.outputs['out'] >> node_b.inputs['in']
        node_a.outputs['out'] >> node_c.inputs['in']

        self.assertEqual(2, len(node_a.downstream_nodes))
        self.assertIn(node_b, node_a.downstream_nodes)
        self.assertIn(node_c, node_a.downstream_nodes)

        self.assertEqual(1, len(node_b.upstream_nodes))
        self.assertIn(node_a, node_b.upstream_nodes)
    # end def test_downstream_upstream_nodes

    def test_evaluate(self):
        """Evaluate the Node will push the new data to it's output."""
        node = SquareNode()
        test_input = 2
        self.assertIsNone(node.outputs['out'].value)
        node.inputs['in'].value = test_input
        node.evaluate()
        self.assertEqual(test_input**2, node.outputs['out'].value)
    # end def test_evaluate

    def test_dirty_depends_on_inputs(self):
        """Dirty status of a Node depends on it's Plugs."""
        node = SquareNode()
        self.assertTrue(node.is_dirty)

        node.inputs['in'].is_dirty = False
        self.assertFalse(node.is_dirty)

        node.inputs['in'].value = 2
        self.assertTrue(node.is_dirty)
    # end def test_dirty_depends_on_inputs

    def test_evaluate_sets_all_inputs_clean(self):
        """After the evaluation, the inputs are considered clean."""
        node = SquareNode()
        node.inputs['in'].value = 2
        self.assertTrue(node.is_dirty)
        node.evaluate()
        self.assertFalse(node.is_dirty)
    # end def test_evaluate_sets_all_inputs_clean
    
    def test_cannot_connect_node_to_itself(self):
        """A node can not create a cycle by connecting to itself."""
        node = SquareNode()
        with self.assertRaises(Exception):
            node.outputs['out'] >> node.inputs['in']
    # end def test_cannot_connect_node_to_itself
# end class TestPlugs
{'fraction_non_alphanumeric': 0.11414704296243035, 'fraction_numerical': 0.002516627718856732, 'mean_word_length': 2.8692628650904033, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4637357', 'n_tokens_mistral': 1723, 'n_tokens_neox': 1658, 'n_words': 375}