test_streams.py

"""
Unit test of the streams system
"""
from collections import defaultdict
from unittest import SkipTest

import pandas as pd
import param
from panel.widgets import IntSlider

from holoviews.core.spaces import DynamicMap
from holoviews.core.util import Version
from holoviews.element import Points, Scatter, Curve, Histogram, Polygons
from holoviews.element.comparison import ComparisonTestCase
from holoviews.streams import * # noqa (Test all available streams)
from holoviews.util import Dynamic, extension
from holoviews.util.transform import dim

from .utils import LoggingComparisonTestCase


def test_all_stream_parameters_constant():
    all_stream_cls = [v for v in globals().values() if
                      isinstance(v, type) and issubclass(v, Stream)]
    for stream_cls in all_stream_cls:
        for name, p in stream_cls.param.objects().items():
            if name == 'name': continue
            if p.constant != True:
                raise TypeError(f'Parameter {name} of stream {stream_cls.__name__} not declared constant')


def test_all_linked_stream_parameters_owners():
    "Test to ensure operations can accept parameters in streams dictionary"
    stream_classes = param.concrete_descendents(LinkedStream)
    for stream_class in stream_classes.values():
        for name, p in stream_class.param.objects().items():
            if name != 'name' and (p.owner != stream_class):
                msg = ("Linked stream %r has parameter %r which is "
                       "inherited from %s. Parameter needs to be redeclared "
                       "in the class definition of this linked stream.")
                raise Exception(msg % (stream_class, name, p.owner))

class TestStreamsDefine(ComparisonTestCase):

    def setUp(self):
        self.XY = Stream.define('XY', x=0.0, y=5.0)
        self.TypesTest = Stream.define('TypesTest',
                                       t=True,
                                       u=0,
                                       v=1.2,
                                       w= (1,'a'),
                                       x='string',
                                       y= [],
                                       z = np.array([1,2,3]))

        test_param = param.Integer(default=42, doc='Test docstring')
        self.ExplicitTest = Stream.define('ExplicitTest',
                                          test=test_param)

    def test_XY_types(self):
        self.assertEqual(isinstance(self.XY.param['x'], param.Number),True)
        self.assertEqual(isinstance(self.XY.param['y'], param.Number),True)

    def test_XY_defaults(self):
        self.assertEqual(self.XY.param['x'].default,0.0)
        self.assertEqual(self.XY.param['y'].default, 5.0)

    def test_XY_instance(self):
        xy = self.XY(x=1,y=2)
        self.assertEqual(xy.x, 1)
        self.assertEqual(xy.y, 2)

    def test_XY_set_invalid_class_x(self):
        regexp = "Parameter 'x' only takes numeric values"
        with self.assertRaisesRegex(ValueError, regexp):
            self.XY.x = 'string'

    def test_XY_set_invalid_class_y(self):
        regexp = "Parameter 'y' only takes numeric values"
        with self.assertRaisesRegex(ValueError, regexp):
            self.XY.y = 'string'

    def test_XY_set_invalid_instance_x(self):
        xy = self.XY(x=1,y=2)
        regexp = "Parameter 'x' only takes numeric values"
        with self.assertRaisesRegex(ValueError, regexp):
            xy.x = 'string'

    def test_XY_set_invalid_instance_y(self):
        xy = self.XY(x=1,y=2)
        regexp = "Parameter 'y' only takes numeric values"
        with self.assertRaisesRegex(ValueError, regexp):
            xy.y = 'string'

    def test_XY_subscriber_triggered(self):

        class Inner:
            def __init__(self): self.state=None
            def __call__(self, x,y): self.state=(x,y)

        inner = Inner()
        xy = self.XY(x=1,y=2)
        xy.add_subscriber(inner)
        xy.event(x=42,y=420)
        self.assertEqual(inner.state, (42,420))

    def test_custom_types(self):
        self.assertEqual(isinstance(self.TypesTest.param['t'], param.Boolean),True)
        self.assertEqual(isinstance(self.TypesTest.param['u'], param.Integer),True)
        self.assertEqual(isinstance(self.TypesTest.param['v'], param.Number),True)
        self.assertEqual(isinstance(self.TypesTest.param['w'], param.Tuple),True)
        self.assertEqual(isinstance(self.TypesTest.param['x'], param.String),True)
        self.assertEqual(isinstance(self.TypesTest.param['y'], param.List),True)
        self.assertEqual(isinstance(self.TypesTest.param['z'], param.Array),True)

    def test_explicit_parameter(self):
        self.assertEqual(isinstance(self.ExplicitTest.param['test'], param.Integer),True)
        self.assertEqual(self.ExplicitTest.param['test'].default,42)
        self.assertEqual(self.ExplicitTest.param['test'].doc, 'Test docstring')


class _TestSubscriber:

    def __init__(self, cb=None):
        self.call_count = 0
        self.kwargs = None
        self.cb = cb

    def __call__(self, **kwargs):
        self.call_count += 1
        self.kwargs = kwargs
        if self.cb:
            self.cb()


class TestPointerStreams(ComparisonTestCase):

    def test_positionX_init(self):
        PointerX()

    def test_positionXY_init_contents(self):
        position = PointerXY(x=1, y=3)
        self.assertEqual(position.contents, dict(x=1, y=3))

    def test_positionXY_update_contents(self):
        position = PointerXY()
        position.event(x=5, y=10)
        self.assertEqual(position.contents, dict(x=5, y=10))

    def test_positionY_const_parameter(self):
        position = PointerY()
        try:
            position.y = 5
            raise Exception('No constant parameter exception')
        except TypeError as e:
            self.assertEqual(str(e), "Constant parameter 'y' cannot be modified")



class TestParamsStream(LoggingComparisonTestCase):

    def setUp(self):
        super().setUp()
        class Inner(param.Parameterized):

            x = param.Number(default = 0)
            y = param.Number(default = 0)

        class InnerAction(Inner):

            action = param.Action(default=lambda o: o.param.trigger('action'))

        self.inner = Inner
        self.inner_action = InnerAction

    def test_param_stream_class(self):
        stream = Params(self.inner)
        self.assertEqual(set(stream.parameters), {self.inner.param.x,
                                                  self.inner.param.y})
        self.assertEqual(stream.contents, {'x': 0, 'y': 0})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)

        stream.add_subscriber(subscriber)
        self.inner.x = 1
        self.assertEqual(values, [{'x': 1, 'y': 0}])

    def test_param_stream_instance(self):
        inner = self.inner(x=2)
        stream = Params(inner)
        self.assertEqual(set(stream.parameters), {inner.param.x, inner.param.y})
        self.assertEqual(stream.contents, {'x': 2, 'y': 0})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)

        stream.add_subscriber(subscriber)
        inner.y = 2
        self.assertEqual(values, [{'x': 2, 'y': 2}])

    def test_param_stream_instance_separate_parameters(self):
        inner = self.inner()

        xparam = Params(inner, ['x'])
        yparam = Params(inner, ['y'])

        valid, invalid = Stream._process_streams([xparam, yparam])
        self.assertEqual(len(valid), 2)
        self.assertEqual(len(invalid), 0)

    def test_param_stream_instance_overlapping_parameters(self):
        inner = self.inner()

        params1 = Params(inner)
        params2 = Params(inner)

        Stream._process_streams([params1, params2])
        self.log_handler.assertContains('WARNING', "['x', 'y']")

    def test_param_parameter_instance_separate_parameters(self):
        inner = self.inner()

        valid, invalid = Stream._process_streams([inner.param.x, inner.param.y])
        xparam, yparam = valid

        self.assertIs(xparam.parameterized, inner)
        self.assertEqual(xparam.parameters, [inner.param.x])
        self.assertIs(yparam.parameterized, inner)
        self.assertEqual(yparam.parameters, [inner.param.y])

    def test_param_parameter_instance_overlapping_parameters(self):
        inner = self.inner()
        Stream._process_streams([inner.param.x, inner.param.x])
        self.log_handler.assertContains('WARNING', "['x']")

    def test_param_stream_parameter_override(self):
        inner = self.inner(x=2)
        stream = Params(inner, parameters=['x'])
        self.assertEqual(stream.parameters, [inner.param.x])
        self.assertEqual(stream.contents, {'x': 2})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)

        stream.add_subscriber(subscriber)
        inner.x = 3
        self.assertEqual(values, [{'x': 3}])

    def test_param_stream_rename(self):
        inner = self.inner(x=2)
        stream = Params(inner, rename={'x': 'X', 'y': 'Y'})
        self.assertEqual(set(stream.parameters), {inner.param.x, inner.param.y})
        self.assertEqual(stream.contents, {'X': 2, 'Y': 0})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)

        stream.add_subscriber(subscriber)
        inner.y = 2
        self.assertEqual(values, [{'X': 2, 'Y': 2}])

    def test_param_stream_action(self):
        inner = self.inner_action()
        stream = Params(inner, ['action'])
        self.assertEqual(set(stream.parameters), {inner.param.action})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)
            self.assertEqual(set(stream.hashkey),
                             {f'{id(inner)} action', '_memoize_key'})

        stream.add_subscriber(subscriber)
        inner.action(inner)
        self.assertEqual(values, [{'action': inner.action}])

    def test_param_stream_memoization(self):
        inner = self.inner_action()
        stream = Params(inner, ['action', 'x'])
        self.assertEqual(set(stream.parameters), {inner.param.action, inner.param.x})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)
            self.assertEqual(
                set(stream.hashkey),
                {f'{id(inner)} action', f'{id(inner)} x', '_memoize_key'})

        stream.add_subscriber(subscriber)
        inner.action(inner)
        inner.x = 0
        self.assertEqual(values, [{'action': inner.action, 'x': 0}])

    def test_params_stream_batch_watch(self):
        tap = Tap(x=0, y=1)
        params = Params(parameters=[tap.param.x, tap.param.y])

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)
        params.add_subscriber(subscriber)

        tap.param.trigger('x', 'y')

        assert values == [{'x': 0, 'y': 1}]

        tap.event(x=1, y=2)

        assert values == [{'x': 0, 'y': 1}, {'x': 1, 'y': 2}]

    def test_params_no_names(self):
        a = IntSlider()
        b = IntSlider()
        p = Params(parameters=[a.param.value, b.param.value])
        assert len(p.hashkey) == 3  # the two widgets + _memoize_key

    def test_params_identical_names(self):
        a = IntSlider(name="Name")
        b = IntSlider(name="Name")
        p = Params(parameters=[a.param.value, b.param.value])
        assert len(p.hashkey) == 3  # the two widgets + _memoize_key


class TestParamMethodStream(ComparisonTestCase):

    def setUp(self):
        if Version(param.__version__) < Version('1.8.0'):
            raise SkipTest('Params stream requires param >= 1.8.0')

        class Inner(param.Parameterized):

            action = param.Action(default=lambda o: o.param.trigger('action'))
            x = param.Number(default = 0)
            y = param.Number(default = 0)
            count = param.Integer(default=0)

            @param.depends('x')
            def method(self):
                self.count += 1
                return Points([])

            @param.depends('action')
            def action_method(self):
                pass

            @param.depends('action', 'x')
            def action_number_method(self):
                self.count += 1
                return Points([])

            @param.depends('y')
            def op_method(self, obj):
                pass

            def method_no_deps(self):
                pass

        class InnerSubObj(Inner):

            sub = param.Parameter()

            @param.depends('sub.x')
            def subobj_method(self):
                pass

        self.inner = Inner
        self.innersubobj = InnerSubObj

    def test_param_method_depends(self):
        inner = self.inner()
        stream = ParamMethod(inner.method)
        self.assertEqual(set(stream.parameters), {inner.param.x})
        self.assertEqual(stream.contents, {})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)

        stream.add_subscriber(subscriber)
        inner.x = 2
        inner.y = 2
        self.assertEqual(values, [{}])

    def test_param_function_depends(self):
        inner = self.inner()

        @param.depends(inner.param.x)
        def test(x):
            return Points([x])

        dmap = DynamicMap(test)

        inner.x = 10
        self.assertEqual(dmap[()], Points([10]))


    def test_param_instance_steams_dict(self):
        inner = self.inner()

        def test(x):
            return Points([x])

        dmap = DynamicMap(test, streams=dict(x=inner.param.x))

        inner.x = 10
        self.assertEqual(dmap[()], Points([10]))

    def test_param_class_steams_dict(self):
        class ClassParamExample(param.Parameterized):
            x = param.Number(default=1)

        def test(x):
            return Points([x])

        dmap = DynamicMap(test, streams=dict(x=ClassParamExample.param.x))

        ClassParamExample.x = 10
        self.assertEqual(dmap[()], Points([10]))

    def test_panel_param_steams_dict(self):
        import panel as pn
        widget = pn.widgets.FloatSlider(value=1)

        def test(x):
            return Points([x])

        dmap = DynamicMap(test, streams=dict(x=widget))

        widget.value = 10
        self.assertEqual(dmap[()], Points([10]))


    def test_param_method_depends_no_deps(self):
        inner = self.inner()
        stream = ParamMethod(inner.method_no_deps)
        self.assertEqual(set(stream.parameters), {
            inner.param.x, inner.param.y, inner.param.action,
            inner.param.name, inner.param.count
        })
        self.assertEqual(stream.contents, {})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)

        stream.add_subscriber(subscriber)
        inner.x = 2
        inner.y = 2
        self.assertEqual(values, [{}, {}])

    def test_param_method_depends_on_subobj(self):
        inner = self.innersubobj(sub=self.inner())
        stream = ParamMethod(inner.subobj_method)
        self.assertEqual(set(stream.parameters), {inner.sub.param.x})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)

        stream.add_subscriber(subscriber)
        inner.sub.x = 2
        self.assertEqual(values, [{}])

    def test_dynamicmap_param_method_deps(self):
        inner = self.inner()
        dmap = DynamicMap(inner.method)
        self.assertEqual(len(dmap.streams), 1)
        stream = dmap.streams[0]
        self.assertIsInstance(stream, ParamMethod)
        self.assertEqual(stream.contents, {})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)

        stream.add_subscriber(subscriber)
        inner.x = 2
        inner.y = 2
        self.assertEqual(values, [{}])

    def test_param_method_depends_trigger_no_memoization(self):
        inner = self.inner()
        stream = ParamMethod(inner.method)
        self.assertEqual(set(stream.parameters), {inner.param.x})
        self.assertEqual(stream.contents, {})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)

        stream.add_subscriber(subscriber)
        inner.x = 2
        inner.param.trigger('x')
        self.assertEqual(values, [{}, {}])

    def test_dynamicmap_param_method_deps_memoization(self):
        inner = self.inner()
        dmap = DynamicMap(inner.method)
        stream = dmap.streams[0]
        self.assertEqual(set(stream.parameters), {inner.param.x})
        self.assertEqual(stream.contents, {})

        dmap[()]
        dmap[()]
        self.assertEqual(inner.count, 1)

    def test_dynamicmap_param_method_no_deps(self):
        inner = self.inner()
        dmap = DynamicMap(inner.method_no_deps)
        self.assertEqual(dmap.streams, [])

    def test_dynamicmap_param_method_action_param(self):
        inner = self.inner()
        dmap = DynamicMap(inner.action_method)
        self.assertEqual(len(dmap.streams), 1)
        stream = dmap.streams[0]
        self.assertEqual(set(stream.parameters), {inner.param.action})
        self.assertIsInstance(stream, ParamMethod)
        self.assertEqual(stream.contents, {})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)
            self.assertEqual(set(stream.hashkey),
                             {f'{id(inner)} action', '_memoize_key'})

        stream.add_subscriber(subscriber)
        inner.action(inner)
        self.assertEqual(values, [{}])

    def test_dynamicmap_param_action_number_method_memoizes(self):
        inner = self.inner()
        dmap = DynamicMap(inner.action_number_method)
        self.assertEqual(len(dmap.streams), 1)
        stream = dmap.streams[0]
        self.assertEqual(set(stream.parameters), {inner.param.action, inner.param.x})
        self.assertIsInstance(stream, ParamMethod)
        self.assertEqual(stream.contents, {})

        values = []
        def subscriber(**kwargs):
            values.append(kwargs)
            self.assertEqual(
                set(stream.hashkey),
                {f'{id(inner)} action', f'{id(inner)} x', '_memoize_key'})

        stream.add_subscriber(subscriber)
        stream.add_subscriber(lambda **kwargs: dmap[()])
        inner.action(inner)
        self.assertEqual(values, [{}])
        self.assertEqual(inner.count, 1)
        inner.x = 0
        self.assertEqual(values, [{}])
        self.assertEqual(inner.count, 1)

    def test_dynamicmap_param_method_dynamic_operation(self):
        inner = self.inner()
        dmap = DynamicMap(inner.method)
        inner_stream = dmap.streams[0]
        op_dmap = Dynamic(dmap, operation=inner.op_method)
        self.assertEqual(len(op_dmap.streams), 1)
        stream = op_dmap.streams[0]
        self.assertEqual(set(stream.parameters), {inner.param.y})
        self.assertIsInstance(stream, ParamMethod)
        self.assertEqual(stream.contents, {})

        values_x, values_y = [], []
        def subscriber_x(**kwargs):
            values_x.append(kwargs)

        def subscriber_y(**kwargs):
            values_y.append(kwargs)

        inner_stream.add_subscriber(subscriber_x)
        stream.add_subscriber(subscriber_y)
        inner.y = 3
        self.assertEqual(values_x, [])
        self.assertEqual(values_y, [{}])



class TestSubscribers(ComparisonTestCase):

    def test_exception_subscriber(self):
        subscriber = _TestSubscriber()
        position = PointerXY(subscribers=[subscriber])
        kwargs = dict(x=3, y=4)
        position.event(**kwargs)
        self.assertEqual(subscriber.kwargs, kwargs)

    def test_subscriber_disabled(self):
        subscriber = _TestSubscriber()
        position = PointerXY(subscribers=[subscriber])
        kwargs = dict(x=3, y=4)
        position.update(**kwargs)
        self.assertEqual(subscriber.kwargs, None)

    def test_subscribers(self):
        subscriber1 = _TestSubscriber()
        subscriber2 = _TestSubscriber()
        position = PointerXY(subscribers=[subscriber1, subscriber2])
        kwargs = dict(x=3, y=4)
        position.event(**kwargs)
        self.assertEqual(subscriber1.kwargs, kwargs)
        self.assertEqual(subscriber2.kwargs, kwargs)

    def test_batch_subscriber(self):
        subscriber = _TestSubscriber()

        positionX = PointerX(subscribers=[subscriber])
        positionY = PointerY(subscribers=[subscriber])

        positionX.update(x=5)
        positionY.update(y=10)

        Stream.trigger([positionX, positionY])
        self.assertEqual(subscriber.kwargs, dict(x=5, y=10))
        self.assertEqual(subscriber.call_count, 1)

    def test_batch_subscribers(self):
        subscriber1 = _TestSubscriber()
        subscriber2 = _TestSubscriber()

        positionX = PointerX(subscribers=[subscriber1, subscriber2])
        positionY = PointerY(subscribers=[subscriber1, subscriber2])

        positionX.update(x=50)
        positionY.update(y=100)

        Stream.trigger([positionX, positionY])

        self.assertEqual(subscriber1.kwargs, dict(x=50, y=100))
        self.assertEqual(subscriber1.call_count, 1)

        self.assertEqual(subscriber2.kwargs, dict(x=50, y=100))
        self.assertEqual(subscriber2.call_count, 1)

    def test_pipe_memoization(self):
        def points(data):
            subscriber.call_count += 1
            return Points([(0, data)])

        stream = Pipe(data=0)
        dmap = DynamicMap(points, streams=[stream])
        def cb():
            dmap[()]
        subscriber = _TestSubscriber(cb)
        stream.add_subscriber(subscriber)
        dmap[()]
        stream.send(1)

        # Ensure call count was incremented on init, the subscriber
        # and the callback
        self.assertEqual(subscriber.call_count, 3)



class TestStreamSource(ComparisonTestCase):

    def tearDown(self):
        with param.logging_level('ERROR'):
            Stream.registry = defaultdict(list)

    def test_source_empty_element(self):
        points = Points([])
        stream = PointerX(source=points)
        self.assertIs(stream.source, points)

    def test_source_empty_element_remap(self):
        points = Points([])
        stream = PointerX(source=points)
        self.assertIs(stream.source, points)
        curve = Curve([])
        stream.source = curve
        self.assertNotIn(points, Stream.registry)
        self.assertIn(curve, Stream.registry)

    def test_source_empty_dmap(self):
        points_dmap = DynamicMap(lambda x: Points([]), kdims=['X'])
        stream = PointerX(source=points_dmap)
        self.assertIs(stream.source, points_dmap)

    def test_source_registry(self):
        points = Points([(0, 0)])
        PointerX(source=points)
        self.assertIn(points, Stream.registry)

    def test_source_registry_empty_element(self):
        points = Points([])
        PointerX(source=points)
        self.assertIn(points, Stream.registry)


class TestParameterRenaming(ComparisonTestCase):

    def test_simple_rename_constructor(self):
        xy = PointerXY(rename={'x':'xtest', 'y':'ytest'}, x=0, y=4)
        self.assertEqual(xy.contents, {'xtest':0, 'ytest':4})

    def test_invalid_rename_constructor(self):
        regexp = '(.+?)is not a stream parameter'
        with self.assertRaisesRegex(KeyError, regexp):
            PointerXY(rename={'x':'xtest', 'z':'ytest'}, x=0, y=4)
            self.assertEqual(str(cm).endswith(), True)

    def test_clashing_rename_constructor(self):
        regexp = '(.+?)parameter of the same name'
        with self.assertRaisesRegex(KeyError, regexp):
            PointerXY(rename={'x':'xtest', 'y':'x'}, x=0, y=4)

    def test_simple_rename_method(self):
        xy = PointerXY(x=0, y=4)
        renamed = xy.rename(x='xtest', y='ytest')
        self.assertEqual(renamed.contents, {'xtest':0, 'ytest':4})

    def test_invalid_rename_method(self):
        xy = PointerXY(x=0, y=4)
        regexp = '(.+?)is not a stream parameter'
        with self.assertRaisesRegex(KeyError, regexp):
            xy.rename(x='xtest', z='ytest')


    def test_clashing_rename_method(self):
        xy = PointerXY(x=0, y=4)
        regexp = '(.+?)parameter of the same name'
        with self.assertRaisesRegex(KeyError, regexp):
            xy.rename(x='xtest', y='x')

    def test_update_rename_valid(self):
        xy = PointerXY(x=0, y=4)
        renamed = xy.rename(x='xtest', y='ytest')
        renamed.event(x=4, y=8)
        self.assertEqual(renamed.contents, {'xtest':4, 'ytest':8})

    def test_update_rename_invalid(self):
        xy = PointerXY(x=0, y=4)
        renamed = xy.rename(y='ytest')
        regexp = "ytest' is not a parameter of(.+?)"
        with self.assertRaisesRegex(ValueError, regexp):
            renamed.event(ytest=8)

    def test_rename_suppression(self):
        renamed = PointerXY(x=0,y=0).rename(x=None)
        self.assertEqual(renamed.contents, {'y':0})

    def test_rename_suppression_reenable(self):
        renamed = PointerXY(x=0,y=0).rename(x=None)
        self.assertEqual(renamed.contents, {'y':0})
        reenabled = renamed.rename(x='foo')
        self.assertEqual(reenabled.contents, {'foo':0, 'y':0})



class TestPlotSizeTransform(ComparisonTestCase):

    def test_plotsize_initial_contents_1(self):
        plotsize = PlotSize(width=300, height=400, scale=0.5)
        self.assertEqual(plotsize.contents, {'width':300, 'height':400, 'scale':0.5})

    def test_plotsize_update_1(self):
        plotsize = PlotSize(scale=0.5)
        plotsize.event(width=300, height=400)
        self.assertEqual(plotsize.contents, {'width':150, 'height':200, 'scale':0.5})

    def test_plotsize_initial_contents_2(self):
        plotsize = PlotSize(width=600, height=100, scale=2)
        self.assertEqual(plotsize.contents, {'width':600, 'height':100, 'scale':2})

    def test_plotsize_update_2(self):
        plotsize = PlotSize(scale=2)
        plotsize.event(width=600, height=100)
        self.assertEqual(plotsize.contents, {'width':1200, 'height':200, 'scale':2})


class TestPipeStream(ComparisonTestCase):

    def test_pipe_send(self):
        def subscriber(data):
            subscriber.test = data
        subscriber.test = None

        pipe = Pipe()
        pipe.add_subscriber(subscriber)
        pipe.send('Test')
        self.assertEqual(pipe.data, 'Test')
        self.assertEqual(subscriber.test, 'Test')

    def test_pipe_event(self):
        def subscriber(data):
            subscriber.test = data
        subscriber.test = None

        pipe = Pipe()
        pipe.add_subscriber(subscriber)
        pipe.event(data='Test')
        self.assertEqual(pipe.data, 'Test')
        self.assertEqual(subscriber.test, 'Test')

    def test_pipe_update(self):
        pipe = Pipe()
        pipe.event(data='Test')
        self.assertEqual(pipe.data, 'Test')



class TestBufferArrayStream(ComparisonTestCase):

    def test_init_buffer_array(self):
        arr = np.array([[0, 1]])
        buff = Buffer(arr)
        self.assertEqual(buff.data, arr)

    def test_buffer_array_ndim_exception(self):
        error = "Only 2D array data may be streamed by Buffer."
        with self.assertRaisesRegex(ValueError, error):
            Buffer(np.array([0, 1]))

    def test_buffer_array_send(self):
        buff = Buffer(np.array([[0, 1]]))
        buff.send(np.array([[1, 2]]))
        self.assertEqual(buff.data, np.array([[0, 1], [1, 2]]))

    def test_buffer_array_larger_than_length(self):
        buff = Buffer(np.array([[0, 1]]), length=1)
        buff.send(np.array([[1, 2]]))
        self.assertEqual(buff.data, np.array([[1, 2]]))

    def test_buffer_array_patch_larger_than_length(self):
        buff = Buffer(np.array([[0, 1]]), length=1)
        buff.send(np.array([[1, 2], [2, 3]]))
        self.assertEqual(buff.data, np.array([[2, 3]]))

    def test_buffer_array_send_verify_ndim_fail(self):
        buff = Buffer(np.array([[0, 1]]))
        error = 'Streamed array data must be two-dimensional'
        with self.assertRaisesRegex(ValueError, error):
            buff.send(np.array([1]))

    def test_buffer_array_send_verify_shape_fail(self):
        buff = Buffer(np.array([[0, 1]]))
        error = "Streamed array data expected to have 2 columns, got 3."
        with self.assertRaisesRegex(ValueError, error):
            buff.send(np.array([[1, 2, 3]]))

    def test_buffer_array_send_verify_type_fail(self):
        buff = Buffer(np.array([[0, 1]]))
        error = "Input expected to be of type ndarray, got list."
        with self.assertRaisesRegex(TypeError, error):
            buff.send([1])


class TestBufferDictionaryStream(ComparisonTestCase):

    def test_init_buffer_dict(self):
        data = {'x': np.array([1]), 'y': np.array([2])}
        buff = Buffer(data)
        self.assertEqual(buff.data, data)

    def test_buffer_dict_send(self):
        data = {'x': np.array([0]), 'y': np.array([1])}
        buff = Buffer(data)
        buff.send({'x': np.array([1]), 'y': np.array([2])})
        self.assertEqual(buff.data, {'x': np.array([0, 1]), 'y': np.array([1, 2])})

    def test_buffer_dict_larger_than_length(self):
        data = {'x': np.array([0]), 'y': np.array([1])}
        buff = Buffer(data, length=1)
        chunk = {'x': np.array([1]), 'y': np.array([2])}
        buff.send(chunk)
        self.assertEqual(buff.data, chunk)

    def test_buffer_dict_patch_larger_than_length(self):
        data = {'x': np.array([0]), 'y': np.array([1])}
        buff = Buffer(data, length=1)
        chunk = {'x': np.array([1, 2]), 'y': np.array([2, 3])}
        buff.send(chunk)
        self.assertEqual(buff.data, {'x': np.array([2]), 'y': np.array([3])})

    def test_buffer_dict_send_verify_column_fail(self):
        data = {'x': np.array([0]), 'y': np.array([1])}
        buff = Buffer(data)
        error = r"Input expected to have columns \['x', 'y'\], got \['x'\]"
        with self.assertRaisesRegex(IndexError, error):
            buff.send({'x': np.array([2])})

    def test_buffer_dict_send_verify_shape_fail(self):
        data = {'x': np.array([0]), 'y': np.array([1])}
        buff = Buffer(data)
        error = "Input columns expected to have the same number of rows."
        with self.assertRaisesRegex(ValueError, error):
            buff.send({'x': np.array([2]), 'y': np.array([3, 4])})


class TestBufferDataFrameStream(ComparisonTestCase):

    def test_init_buffer_dframe(self):
        data = pd.DataFrame({'x': np.array([1]), 'y': np.array([2])})
        buff = Buffer(data, index=False)
        self.assertEqual(buff.data, data)

    def test_init_buffer_dframe_with_index(self):
        data = pd.DataFrame({'x': np.array([1]), 'y': np.array([2])})
        buff = Buffer(data)
        self.assertEqual(buff.data, data.reset_index())

    def test_buffer_dframe_send(self):
        data = pd.DataFrame({'x': np.array([0]), 'y': np.array([1])})
        buff = Buffer(data, index=False)
        buff.send(pd.DataFrame({'x': np.array([1]), 'y': np.array([2])}))
        dframe = pd.DataFrame({'x': np.array([0, 1]), 'y': np.array([1, 2])})
        self.assertEqual(buff.data.values, dframe.values)

    def test_buffer_dframe_send_with_index(self):
        data = pd.DataFrame({'x': np.array([0]), 'y': np.array([1])})
        buff = Buffer(data)
        buff.send(pd.DataFrame({'x': np.array([1]), 'y': np.array([2])}))
        dframe = pd.DataFrame({'x': np.array([0, 1]), 'y': np.array([1, 2])}, index=[0, 0])
        self.assertEqual(buff.data.values, dframe.reset_index().values)

    def test_buffer_dframe_larger_than_length(self):
        data = pd.DataFrame({'x': np.array([0]), 'y': np.array([1])})
        buff = Buffer(data, length=1, index=False)
        chunk = pd.DataFrame({'x': np.array([1]), 'y': np.array([2])})
        buff.send(chunk)
        self.assertEqual(buff.data.values, chunk.values)

    def test_buffer_dframe_patch_larger_than_length(self):
        data = pd.DataFrame({'x': np.array([0]), 'y': np.array([1])})
        buff = Buffer(data, length=1, index=False)
        chunk = pd.DataFrame({'x': np.array([1, 2]), 'y': np.array([2, 3])})
        buff.send(chunk)
        dframe = pd.DataFrame({'x': np.array([2]), 'y': np.array([3])})
        self.assertEqual(buff.data.values, dframe.values)

    def test_buffer_dframe_send_verify_column_fail(self):
        data = pd.DataFrame({'x': np.array([0]), 'y': np.array([1])})
        buff = Buffer(data, index=False)
        error = r"Input expected to have columns \['x', 'y'\], got \['x'\]"
        with self.assertRaisesRegex(IndexError, error):
            buff.send(pd.DataFrame({'x': np.array([2])}))

    def test_clear_buffer_dframe_with_index(self):
        data = pd.DataFrame({'a': [1, 2, 3]})
        buff = Buffer(data)
        buff.clear()
        self.assertEqual(buff.data, data.iloc[:0, :].reset_index())


class Sum(Derived):
    v = param.Number(constant=True)

    def __init__(self, val_streams, exclusive=False, base=0):
        self.base = base
        super().__init__(input_streams=val_streams, exclusive=exclusive)

    @property
    def constants(self):
        return dict(base=self.base)

    @classmethod
    def transform_function(cls, stream_values, constants):
        v = sum([val["v"] for val in stream_values if val["v"]])
        return dict(v=v + constants['base'])


Val = Stream.define("Val", v=0.0)


class TestDerivedStream(ComparisonTestCase):

    def test_simple_derived_stream(self):
        # Define input streams
        v0 = Val(v=1.0)
        v1 = Val(v=2.0)

        # Build Sum stream
        s0 = Sum([v0, v1])

        # Check outputs
        self.assertEqual(s0.v, 3.0)

        # Update v0
        v0.event(v=7.0)
        self.assertEqual(s0.v, 9.0)

        # Update v1
        v1.event(v=-8.0)
        self.assertEqual(s0.v, -1.0)

    def test_nested_derived_stream(self):
        v0 = Val(v=1.0)
        v1 = Val(v=4.0)
        v2 = Val(v=7.0)

        # Build nested sum stream
        s1 = Sum([v0, v1])
        s0 = Sum([s1, v2])

        # Check initial value
        self.assertEqual(s0.v, 12.0)

        # Update top-level value
        v2.event(v=8.0)
        self.assertEqual(s0.v, 13.0)

        # Update nested value
        v1.event(v=5.0,)
        self.assertEqual(s0.v, 14.0)

    def test_derived_stream_constants(self):
        v0 = Val(v=1.0)
        v1 = Val(v=4.0)
        v2 = Val(v=7.0)

        # Build Sum stream with base value
        s0 = Sum([v0, v1, v2], base=100)

        # Check initial value
        self.assertEqual(s0.v, 112.0)

        # Update value
        v2.event(v=8.0)
        self.assertEqual(s0.v, 113.0)

    def test_exclusive_derived_stream(self):
        # Define input streams
        v0 = Val()
        v1 = Val(v=2.0)

        # Build exclusive Sum stream
        # In this case, all streams except the most recently updated will be reset on
        # update
        s0 = Sum([v0, v1], exclusive=True)

        # Check outputs
        self.assertEqual(s0.v, 2.0)

        # Update v0
        v0.event(v=7.0)
        self.assertEqual(s0.v, 7.0)

        # Update v1
        v1.event(v=-8.0)
        self.assertEqual(s0.v, -8.0)


class TestHistoryStream(ComparisonTestCase):
    def test_initial_history_stream_values(self):
        # Check values list is initialized with initial contents of input stream
        val = Val(v=1.0)
        history = History(val)
        self.assertEqual(history.contents, {"values": [val.contents]})

    def test_history_stream_values_appended(self):
        val = Val(v=1.0)
        history = History(val)
        # Perform a few updates on val stream
        val.event(v=2.0)
        val.event(v=3.0)
        self.assertEqual(
            history.contents,
            {"values": [{"v": 1.0}, {"v": 2.0}, {"v": 3.0}]}
        )

        # clear
        history.clear_history()
        self.assertEqual(history.contents, {"values": []})

    def test_history_stream_trigger_callbacks(self):
        # Construct history stream
        val = Val(v=1.0)
        history = History(val)

        # Register callback
        callback_input = []
        def cb(**kwargs):
            callback_input.append(kwargs)
        history.add_subscriber(cb)
        self.assertEqual(callback_input, [])

        # Perform updates on val stream and make sure history callback is triggered
        del callback_input[:]
        val.event(v=2.0)
        self.assertEqual(
            callback_input[0],
            {"values": [{"v": 1.0}, {"v": 2.0}]}
        )

        del callback_input[:]
        val.event(v=3.0)
        self.assertEqual(
            callback_input[0],
            {"values": [{"v": 1.0}, {"v": 2.0}, {"v": 3.0}]}
        )

        # clearing history should trigger callback
        del callback_input[:]
        history.clear_history()
        history.event()
        self.assertEqual(
            callback_input[0],
            {"values": []}
        )

        # Update after clearing
        del callback_input[:]
        val.event(v=4.0)
        self.assertEqual(
            callback_input[0],
            {"values": [{"v": 4.0}]}
        )


class TestExprSelectionStream(ComparisonTestCase):

    def setUp(self):
        extension("bokeh")

    def test_selection_expr_stream_2D_elements(self):
        element_type_2D = [Points]
        for element_type in element_type_2D:
            # Create SelectionExpr on element
            element = element_type(([1, 2, 3], [1, 5, 10]))
            expr_stream = SelectionExpr(element)

            # Check stream properties
            self.assertEqual(len(expr_stream.input_streams), 3)
            self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
            self.assertIsInstance(expr_stream.input_streams[1], Lasso)
            self.assertIsInstance(expr_stream.input_streams[2], Selection1D)
            self.assertIsNone(expr_stream.bbox)
            self.assertIsNone(expr_stream.selection_expr)

            # Simulate interactive update by triggering source stream
            expr_stream.input_streams[0].event(bounds=(1, 1, 3, 4))

            # Check SelectionExpr values
            self.assertEqual(
                repr(expr_stream.selection_expr),
                repr(((dim('x')>=1)&(dim('x')<=3))&((dim('y')>=1)&(dim('y')<=4)))
            )
            self.assertEqual(
                expr_stream.bbox,
                {'x': (1, 3), 'y': (1, 4)}
            )

    def test_selection_expr_stream_1D_elements(self):
        element_type_1D = [Scatter]
        for element_type in element_type_1D:
            # Create SelectionExpr on element
            element = element_type(([1, 2, 3], [1, 5, 10]))
            expr_stream = SelectionExpr(element)

            # Check stream properties
            self.assertEqual(len(expr_stream.input_streams), 1)
            self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
            self.assertIsNone(expr_stream.bbox)
            self.assertIsNone(expr_stream.selection_expr)

            # Simulate interactive update by triggering source stream
            expr_stream.input_streams[0].event(bounds=(1, 1, 3, 4))

            # Check SelectionExpr values
            self.assertEqual(
                repr(expr_stream.selection_expr),
                repr((dim('x')>=1)&(dim('x')<=3))
            )
            self.assertEqual(
                expr_stream.bbox,
                {'x': (1, 3)}
            )


    def test_selection_expr_stream_invert_axes_2D_elements(self):
        element_type_2D = [Points]
        for element_type in element_type_2D:
            # Create SelectionExpr on element
            element = element_type(([1, 2, 3], [1, 5, 10])).opts(invert_axes=True)
            expr_stream = SelectionExpr(element)

            # Check stream properties
            self.assertEqual(len(expr_stream.input_streams), 3)
            self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
            self.assertIsInstance(expr_stream.input_streams[1], Lasso)
            self.assertIsInstance(expr_stream.input_streams[2], Selection1D)
            self.assertIsNone(expr_stream.bbox)
            self.assertIsNone(expr_stream.selection_expr)

            # Simulate interactive update by triggering source stream
            expr_stream.input_streams[0].event(bounds=(1, 1, 3, 4))

            # Check SelectionExpr values
            self.assertEqual(
                repr(expr_stream.selection_expr),
                repr(((dim('y')>=1)&(dim('y')<=3))&((dim('x')>=1)&(dim('x')<=4)))
            )
            self.assertEqual(
                expr_stream.bbox,
                {'y': (1, 3), 'x': (1, 4)}
            )

    def test_selection_expr_stream_invert_axes_1D_elements(self):
        element_type_1D = [Scatter]
        for element_type in element_type_1D:
            # Create SelectionExpr on element
            element = element_type(([1, 2, 3], [1, 5, 10])).opts(invert_axes=True)
            expr_stream = SelectionExpr(element)

            # Check stream properties
            self.assertEqual(len(expr_stream.input_streams), 1)
            self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
            self.assertIsNone(expr_stream.bbox)
            self.assertIsNone(expr_stream.selection_expr)

            # Simulate interactive update by triggering source stream
            expr_stream.input_streams[0].event(bounds=(1, 1, 3, 4))

            # Check SelectionExpr values
            self.assertEqual(
                repr(expr_stream.selection_expr),
                repr((dim('x')>=1)&(dim('x')<=4))
            )
            self.assertEqual(
                expr_stream.bbox,
                {'x': (1, 4)}
            )

    def test_selection_expr_stream_invert_xaxis_yaxis_2D_elements(self):
        element_type_2D = [Points]
        for element_type in element_type_2D:

            # Create SelectionExpr on element
            element = element_type(([1, 2, 3], [1, 5, 10])).opts(
                invert_xaxis=True,
                invert_yaxis=True,
            )
            expr_stream = SelectionExpr(element)

            # Check stream properties
            self.assertEqual(len(expr_stream.input_streams), 3)
            self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
            self.assertIsInstance(expr_stream.input_streams[1], Lasso)
            self.assertIsInstance(expr_stream.input_streams[2], Selection1D)
            self.assertIsNone(expr_stream.bbox)
            self.assertIsNone(expr_stream.selection_expr)

            # Simulate interactive update by triggering source stream
            expr_stream.input_streams[0].event(bounds=(3, 4, 1, 1))

            # Check SelectionExpr values
            self.assertEqual(
                repr(expr_stream.selection_expr),
                repr(((dim('x')>=1)&(dim('x')<=3))&((dim('y')>=1)&(dim('y')<=4)))
            )
            self.assertEqual(
                expr_stream.bbox,
                {'x': (1, 3), 'y': (1, 4)}
            )

    def test_selection_expr_stream_invert_xaxis_yaxis_1D_elements(self):
        element_type_1D = [Scatter]
        for element_type in element_type_1D:

            # Create SelectionExpr on element
            element = element_type(([1, 2, 3], [1, 5, 10])).opts(
                invert_xaxis=True,
                invert_yaxis=True,
            )
            expr_stream = SelectionExpr(element)

            # Check stream properties
            self.assertEqual(len(expr_stream.input_streams), 1)
            self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
            self.assertIsNone(expr_stream.bbox)
            self.assertIsNone(expr_stream.selection_expr)

            # Simulate interactive update by triggering source stream
            expr_stream.input_streams[0].event(bounds=(3, 4, 1, 1))

            # Check SelectionExpr values
            self.assertEqual(
                repr(expr_stream.selection_expr),
                repr((dim('x')>=1)&(dim('x')<=3))
            )
            self.assertEqual(
                expr_stream.bbox,
                {'x': (1, 3)}
            )

    def test_selection_expr_stream_hist(self):
        # Create SelectionExpr on element
        hist = Histogram(([1, 2, 3, 4, 5], [1, 5, 2, 3, 7]))
        expr_stream = SelectionExpr(hist)

        # Check stream properties
        self.assertEqual(len(expr_stream.input_streams), 1)
        self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
        self.assertIsNone(expr_stream.bbox)
        self.assertIsNone(expr_stream.selection_expr)

        # Simulate interactive update by triggering source stream.
        # Select second and forth bar.
        expr_stream.input_streams[0].event(bounds=(1.5, 2.5, 4.6, 6))
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr((dim('x')>=1.5)&(dim('x')<=4.6))
        )
        self.assertEqual(expr_stream.bbox, {'x': (1.5, 4.6)})

        # Select third, forth, and fifth bar.  Make sure there is special
        # handling when last bar is selected to include values exactly on the
        # upper edge in the selection
        expr_stream.input_streams[0].event(bounds=(2.5, -10, 8, 10))
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr((dim('x')>=2.5)&(dim('x')<=8))
        )
        self.assertEqual(expr_stream.bbox, {'x': (2.5, 8)})

    def test_selection_expr_stream_hist_invert_axes(self):
        # Create SelectionExpr on element
        hist = Histogram(([1, 2, 3, 4, 5], [1, 5, 2, 3, 7])).opts(
            invert_axes=True
        )
        expr_stream = SelectionExpr(hist)

        # Check stream properties
        self.assertEqual(len(expr_stream.input_streams), 1)
        self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
        self.assertIsNone(expr_stream.bbox)
        self.assertIsNone(expr_stream.selection_expr)

        # Simulate interactive update by triggering source stream.
        # Select second and forth bar.
        expr_stream.input_streams[0].event(bounds=(2.5, 1.5, 6, 4.6))
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr((dim('x')>=1.5)&(dim('x')<=4.6))
        )
        self.assertEqual(expr_stream.bbox, {'x': (1.5, 4.6)})

        # Select third, forth, and fifth bar.  Make sure there is special
        # handling when last bar is selected to include values exactly on the
        # upper edge in the selection
        expr_stream.input_streams[0].event(bounds=(-10, 2.5, 10, 8))
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr((dim('x')>=2.5)&(dim('x')<=8))
        )
        self.assertEqual(expr_stream.bbox, {'x': (2.5, 8)})

    def test_selection_expr_stream_hist_invert_xaxis_yaxis(self):
        # Create SelectionExpr on element
        hist = Histogram(([1, 2, 3, 4, 5], [1, 5, 2, 3, 7])).opts(
                invert_xaxis=True,
                invert_yaxis=True,
        )
        expr_stream = SelectionExpr(hist)

        # Check stream properties
        self.assertEqual(len(expr_stream.input_streams), 1)
        self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
        self.assertIsNone(expr_stream.bbox)
        self.assertIsNone(expr_stream.selection_expr)

        # Simulate interactive update by triggering source stream.
        # Select second and forth bar.
        expr_stream.input_streams[0].event(bounds=(4.6, 6, 1.5, 2.5))
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr((dim('x')>=1.5)&(dim('x')<=4.6))
        )
        self.assertEqual(expr_stream.bbox, {'x': (1.5, 4.6)})

        # Select third, forth, and fifth bar.  Make sure there is special
        # handling when last bar is selected to include values exactly on the
        # upper edge in the selection
        expr_stream.input_streams[0].event(bounds=(8, 10, 2.5, -10))
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr((dim('x')>=2.5)&(dim('x')<=8))
        )
        self.assertEqual(expr_stream.bbox, {'x': (2.5, 8)})


    def test_selection_expr_stream_polygon_index_cols(self):
        # Create SelectionExpr on element
        try: import shapely # noqa
        except ImportError:
            try: import spatialpandas # noqa
            except ImportError: raise SkipTest('Shapely required for polygon selection')
        poly = Polygons([
            [(0, 0, 'a'), (2, 0, 'a'), (1, 1, 'a')],
            [(2, 0, 'b'), (4, 0, 'b'), (3, 1, 'b')],
            [(1, 1, 'c'), (3, 1, 'c'), (2, 2, 'c')]
        ], vdims=['cat'])

        events = []
        expr_stream = SelectionExpr(poly, index_cols=['cat'])
        expr_stream.add_subscriber(lambda **kwargs: events.append(kwargs))

        # Check stream properties
        self.assertEqual(len(expr_stream.input_streams), 3)
        self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
        self.assertIsInstance(expr_stream.input_streams[1], Lasso)
        self.assertIsInstance(expr_stream.input_streams[2], Selection1D)
        self.assertIsNone(expr_stream.bbox)
        self.assertIsNone(expr_stream.selection_expr)

        expr_stream.input_streams[2].event(index=[0, 1])
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr(dim('cat').isin(['a', 'b']))
        )
        self.assertEqual(expr_stream.bbox, None)
        self.assertEqual(len(events), 1)

        # Ensure bounds event does not trigger another update
        expr_stream.input_streams[0].event(bounds=(0, 0, 4, 1))
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr(dim('cat').isin(['a', 'b']))
        )
        self.assertEqual(len(events), 1)

        # Ensure geometry event does trigger another update
        expr_stream.input_streams[1].event(geometry=np.array([(0, 0), (4, 0), (4, 2), (0, 2)]))
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr(dim('cat').isin(['a', 'b', 'c']))
        )
        self.assertEqual(len(events), 2)

        # Ensure index event does trigger another update
        expr_stream.input_streams[2].event(index=[1, 2])
        self.assertEqual(
            repr(expr_stream.selection_expr),
            repr(dim('cat').isin(['b', 'c']))
        )
        self.assertEqual(expr_stream.bbox, None)
        self.assertEqual(len(events), 3)

    def test_selection_expr_stream_dynamic_map_2D_elements(self):
        element_type_2D = [Points]
        for element_type in element_type_2D: # Scatter,
            # Create SelectionExpr on element
            dmap = Dynamic(element_type(([1, 2, 3], [1, 5, 10])))
            expr_stream = SelectionExpr(dmap)

            # Check stream properties
            self.assertEqual(len(expr_stream.input_streams), 3)
            self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
            self.assertIsNone(expr_stream.bbox)
            self.assertIsNone(expr_stream.selection_expr)

            # Simulate interactive update by triggering source stream
            expr_stream.input_streams[0].event(bounds=(1, 1, 3, 4))

            # Check SelectionExpr values
            self.assertEqual(
                repr(expr_stream.selection_expr),
                repr(((dim('x')>=1)&(dim('x')<=3))&((dim('y')>=1)&(dim('y')<=4)))
            )
            self.assertEqual(
                expr_stream.bbox,
                {'x': (1, 3), 'y': (1, 4)}
            )

    def test_selection_expr_stream_dynamic_map_1D_elements(self):
        element_type_1D = [Scatter]
        for element_type in element_type_1D:
            # Create SelectionExpr on element
            dmap = Dynamic(element_type(([1, 2, 3], [1, 5, 10])))
            expr_stream = SelectionExpr(dmap)

            # Check stream properties
            self.assertEqual(len(expr_stream.input_streams), 1)
            self.assertIsInstance(expr_stream.input_streams[0], SelectionXY)
            self.assertIsNone(expr_stream.bbox)
            self.assertIsNone(expr_stream.selection_expr)

            # Simulate interactive update by triggering source stream
            expr_stream.input_streams[0].event(bounds=(1, 1, 3, 4))

            # Check SelectionExpr values
            self.assertEqual(
                repr(expr_stream.selection_expr),
                repr((dim('x')>=1)&(dim('x')<=3))
            )
            self.assertEqual(
                expr_stream.bbox,
                {'x': (1, 3)}
            )