def test_data_change_shall_notify_all_observers_once(cls):
     with patch.object(cls.dec_obs, 'update') as mock_dec_obs_update, patch.object(
         cls.hex_obs, 'update'
     ) as mock_hex_obs_update:
         cls.sub.data = 10
         cls.assertEqual(mock_dec_obs_update.call_count, 1)
         cls.assertEqual(mock_hex_obs_update.call_count, 1)

async def setup_gateway(hass, mock_gateway, mock_api,
                        generate_psk=generate_psk,
                        known_hosts=None):
    """Load the Tradfri platform with a mock gateway."""
    def request_config(_, callback, description, submit_caption, fields):
        """Mock request_config."""
        hass.async_add_job(callback, {'security_code': 'mock'})

    if known_hosts is None:
        known_hosts = {}

    with patch('pytradfri.api.aiocoap_api.APIFactory.generate_psk',
               generate_psk), \
            patch('pytradfri.api.aiocoap_api.APIFactory.request', mock_api), \
            patch('pytradfri.Gateway', return_value=mock_gateway), \
            patch.object(tradfri, 'load_json', return_value=known_hosts), \
            patch.object(hass.components.configurator, 'request_config',
                         request_config):

        await async_setup_component(hass, tradfri.DOMAIN,
                                    {
                                        tradfri.DOMAIN: {
                                            'host': 'mock-host',
                                            'allow_tradfri_groups': True
                                        }
                                    })
        await hass.async_block_till_done()

 def test_delegates_simple_and_complex_objects(self):
     mock_simple_objects = [[self.direction], [self.gender], [self.item]]
     mock_game = 'game'
     with patch.object(
             GameLoader,
             '_reconstitute_simple_objects',
             return_value=mock_simple_objects,
     ) as mock_simple_reconstitute:
         with patch.object(
                 GameLoader,
                 '_reconstitute_complex_objects',
                 return_value=mock_game,
         ) as mock_complex_reconstitute:
             rebuilt_game = self.loader._reconstitute_all_game_objects(
                 self.serialized_objects
             )
     self.assertEqual(rebuilt_game, mock_game)
     mock_simple_reconstitute.assert_called_once_with(
         self.serialized_objects['directions'],
         self.serialized_objects['genders'],
         self.serialized_objects['items']
     )
     mock_complex_reconstitute.assert_called_once_with(
         self.serialized_objects['game'],
         self.serialized_objects['player'],
         self.serialized_objects['people'],
         self.serialized_objects['locations'],
         self.serialized_objects['exits'],
         *mock_simple_objects
     )

    def test_auto_indent(self):
        script = '''
a
'''
        wnd = self._getwnd(script)
        with patch.object(wnd.cursor, 'pos', new=2):
            wnd.document.mode.on_auto_indent(wnd)
        assert wnd.document.gettext(0, wnd.document.endpos()) == '\na\n\n'

        script = '''
    a
'''
        wnd = self._getwnd(script)
        with patch.object(wnd.cursor, 'pos', new=6):
            wnd.document.mode.on_auto_indent(wnd)
        assert wnd.document.gettext(0, wnd.document.endpos()
                                    ) == '\n    a\n    \n'

        script = '''
    a
'''
        wnd = self._getwnd(script)
        with patch.object(wnd.cursor, 'pos', new=5):
            wnd.document.mode.on_auto_indent(wnd)
        assert wnd.document.gettext(0, wnd.document.endpos()
                                    ) == '\n\n    a\n'

    def test_infer(self, json_loads) -> None:
        arguments = mock_arguments()
        arguments.recursive = False
        arguments.strict = False

        configuration = mock_configuration()

        with patch.object(commands.Command, '_call_client') as call_client:
            Infer(arguments, configuration, source_directory='.').run()
            call_client.assert_called_once_with(
                command=commands.Check.NAME,
                flags=['-show-error-traces', '-project-root', '.', '-infer'])

        with patch.object(commands.Command, '_call_client') as call_client:
            arguments.recursive = True
            Infer(arguments, configuration, source_directory='.').run()
            call_client.assert_called_once_with(
                command=commands.Check.NAME,
                flags=[
                    '-show-error-traces',
                    '-project-root',
                    '.',
                    '-infer',
                    '-recursive-infer',
                ])

    def test_attribute(self):
        first_items = [1, 2, 3]
        second_items = [1, 2, 4]
        third_items = [1, 2, 3, 4]

        for item in third_items:
            attribute_cache.delete('attribute:%s' % item)

        with patch.object(AttributeContainer, 'pull', side_effect=_attribute_pull) as mock_pull:
            # first execution should run pull
            self.assertEqual(AttributeContainer(first_items).get(), _attribute_pull(first_items))
            mock_pull.assert_called_once_with(first_items)

            # second execution should not run pull again
            self.assertEqual(AttributeContainer(first_items).get(), _attribute_pull(first_items))
            mock_pull.assert_called_once_with(first_items)

        with patch.object(AttributeContainer, 'pull', side_effect=_attribute_pull) as mock_pull:
            # running with second items should only call pull on 4
            self.assertEqual(AttributeContainer(second_items).get(), _attribute_pull(second_items))
            mock_pull.assert_called_once_with([4])

            # second execution should not run pull again
            self.assertEqual(AttributeContainer(second_items).get(), _attribute_pull(second_items))
            mock_pull.assert_called_once_with([4])

        with patch.object(AttributeContainer, 'pull', side_effect=_attribute_pull) as mock_pull:
            # running with third items should not call pull
            self.assertEqual(AttributeContainer(third_items).get(), _attribute_pull(third_items))
            self.assertEqual(mock_pull.called, False)

def test_call_decorated_kwargs_on_trait_change():
    """test calling @interact(foo=bar) decorated functions"""
    d = {}
    with patch.object(interaction, 'display', record_display):
        @interact(a='kwarg')
        def foo(a='default'):
            d['a'] = a
            return a
    nt.assert_equal(len(displayed), 1)
    w = displayed[0].children[0]
    check_widget(w,
        cls=widgets.Text,
        value='kwarg',
    )
    # test calling the function directly
    a = foo('hello')
    nt.assert_equal(a, 'hello')
    nt.assert_equal(d['a'], 'hello')

    # test that setting trait values calls the function
    with patch.object(interaction, 'display', record_display):
        w.value = 'called'
    nt.assert_equal(d['a'], 'called')
    nt.assert_equal(len(displayed), 2)
    nt.assert_equal(w.value, displayed[-1])

    def test_delete_blocks(self):
        blocks = self.add_blocks(self.add_operations())
        bc = get_blockchain()

        with patch.object(BlockChain, '_get_new_blocks', return_value=blocks):
            bc.update_blocks()

        with self.assertRaisesRegex(Block.ChainOperationBlockedError, "can't remove: blocked by another block"):
            blocks[4].remove()

        self.assertEqual(bc.max_depth, 5)
        self.assertEqual(bc.head, blocks[5].id)

        for block_to_remove, max_depth, heads in (
                (blocks[5], 5, [blocks[6].id]),
                (blocks[6], 4, [blocks[3].id, blocks[4].id]),
                (blocks[4], 4, [blocks[3].id]),
                (blocks[3], 3, [blocks[2].id]),
                (blocks[2], 2, [blocks[1].id]),
                (blocks[1], 1, [blocks[0].id]),
                (blocks[0], 0, [BlockRev().id])
        ):
            block_to_remove.remove()
            self.assertEqual(bc.max_depth, max_depth)
            self.assertIn(bc.head, heads)

        with patch.object(BlockChain, '_get_new_blocks', return_value=blocks):
            bc.update_blocks()

        self.assertEqual(bc.max_depth, 5)
        self.assertEqual(bc.head, blocks[5].id)

        self.assertCountEqual([op.uuid for op in blocks[0].operations + blocks[2].operations[1:2]],
                              Identifier.get_uuid_list())

 def test_request(self, client):
     response = Mock(spec=Response)
     with patch.object(requests,
                       "request",
                       return_value=response):
         mapper = ObjectMapper(GitterClient.Room)
         with patch.object(response.content,
                           "decode",
                           return_value=json.dumps(room_info)):
             ret = client.request("GET",
                                  "http://localhost/rooms",
                                  ObjectMapper(GitterClient.Room),
                                  {'param': "spam"},
                                  {'body': "ham"})
             method, endpoint = requests.request.call_args[0]
             kwargs = requests.request.call_args[1]
             assert_that(ret).is_equal_to([mapper.map(obj)
                                           for obj in room_info])
             assert_that(method).is_equal_to("GET")
             assert_that(endpoint).is_equal_to("http://localhost/rooms")
             assert_that(kwargs['params']).is_equal_to({'param': "spam"})
             assert_that(kwargs['json']).is_equal_to({'body': "ham"})
             assert_that(kwargs['headers']['Authorization']).is_not_empty()
             assert_that(kwargs['headers']['Accept']).is_not_empty()
             assert_that(kwargs['headers']['Content-Type']).is_not_empty()

async def test_setup_with_discovery_no_known_auth(hass, aioclient_mock):
    """Test discovering a bridge and not having known auth."""
    aioclient_mock.get(hue.API_NUPNP, json=[
        {
            'internalipaddress': '0.0.0.0',
            'id': 'abcd1234'
        }
    ])

    with patch.object(hass, 'config_entries') as mock_config_entries, \
            patch.object(hue, 'configured_hosts', return_value=[]):
        mock_config_entries.flow.async_init.return_value = mock_coro()
        assert await async_setup_component(hass, hue.DOMAIN, {
            hue.DOMAIN: {}
        }) is True

    # Flow started for discovered bridge
    assert len(mock_config_entries.flow.mock_calls) == 1
    assert mock_config_entries.flow.mock_calls[0][2]['data'] == {
        'host': '0.0.0.0',
        'path': '.hue_abcd1234.conf',
    }

    # Config stored for domain.
    assert hass.data[hue.DOMAIN] == {
        '0.0.0.0': {
            hue.CONF_HOST: '0.0.0.0',
            hue.CONF_FILENAME: '.hue_abcd1234.conf',
            hue.CONF_ALLOW_HUE_GROUPS: hue.DEFAULT_ALLOW_HUE_GROUPS,
            hue.CONF_ALLOW_UNREACHABLE: hue.DEFAULT_ALLOW_UNREACHABLE,
        }
    }

        def wrapper(*args, **kargs):
            future = Future()
            future.set_result(self._response)

            with patch.object(AsyncHTTPClient, "fetch", return_value=future):
                with patch.object(Client, "fetch", return_value=future):
                    yield coroutine(*args, **kargs)

def test_migrate_one(td):
    src = pjoin(td, 'src')
    srcdir = pjoin(td, 'srcdir')
    dst = pjoin(td, 'dst')
    dstdir = pjoin(td, 'dstdir')
    
    touch(src, 'test file')
    touch(pjoin(srcdir, 'f'), 'test dir file')
    
    called = {}
    def notice_m_file(src, dst):
        called['migrate_file'] = True
        return migrate_file(src, dst)
    
    def notice_m_dir(src, dst):
        called['migrate_dir'] = True
        return migrate_dir(src, dst)
    
    with patch.object(migrate_mod, 'migrate_file', notice_m_file), \
            patch.object(migrate_mod, 'migrate_dir', notice_m_dir):
        assert migrate_one(src, dst)
        assert called == {'migrate_file': True}
        called.clear()
        assert migrate_one(srcdir, dstdir)
        assert called == {'migrate_dir': True}
        called.clear()
        assert not migrate_one(pjoin(td, 'dne'), dst)
        assert called == {}

def test_lock_get_usercode_service(hass, mock_openzwave):
    """Test the zwave lock get_usercode service."""
    node = MockNode(node_id=12)
    value0 = MockValue(data=None, node=node, index=0)
    value1 = MockValue(data='1234', node=node, index=1)
    yield from zwave.async_setup_platform(
        hass, {}, MagicMock())

    node.get_values.return_value = {
        value0.value_id: value0,
        value1.value_id: value1,
    }

    with patch.object(zwave.zwave, 'NETWORK') as mock_network:
        with patch.object(zwave, '_LOGGER') as mock_logger:
            mock_network.nodes = {
                node.node_id: node
            }
            yield from hass.services.async_call(
                zwave.DOMAIN, zwave.SERVICE_GET_USERCODE, {
                    const.ATTR_NODE_ID: node.node_id,
                    zwave.ATTR_CODE_SLOT: 1,
                    })
            yield from hass.async_block_till_done()

            # This service only seems to write to the log
            assert mock_logger.info.called
            assert len(mock_logger.info.mock_calls) == 1
            assert mock_logger.info.mock_calls[0][1][2] == '1234'

 def test_install_without_pip(self):
     mgr = PipManager('/usr/bin', pip_installed=False)
     with patch.object(helpers, 'logged_exec') as mocked_exec:
         with patch.object(mgr, '_brute_force_install_pip') as mocked_install_pip:
             mgr.install('foo')
             self.assertEqual(mocked_install_pip.call_count, 1)
         mocked_exec.assert_called_with(['/usr/bin/pip', 'install', 'foo'])

    def test_tooltip(self):
        data = self.iris
        self.send_signal(self.widget.Inputs.data, data)

        event = MagicMock()
        with patch.object(self.widget.imageplot.plot.vb, "mapSceneToView"), \
                patch.object(QToolTip, "showText") as show_text:

            sel = np.zeros(len(data), dtype="bool")

            sel[3] = 1  # a single instance
            with patch.object(self.widget.imageplot, "_points_at_pos",
                              return_value=(sel, 2)):
                self.assertTrue(self.widget.imageplot.help_event(event))
                (_, text), _ = show_text.call_args
                self.assertIn("iris = {}".format(data[3, "iris"]), text)
                self.assertIn("value = {}".format(data[3, 2]), text)
                self.assertEqual(1, text.count("iris ="))

            sel[51] = 1  # add a data point
            with patch.object(self.widget.imageplot, "_points_at_pos",
                              return_value=(sel, 2)):
                self.assertTrue(self.widget.imageplot.help_event(event))
                (_, text), _ = show_text.call_args
                self.assertIn("iris = {}".format(data[3, "iris"]), text)
                self.assertIn("iris = {}".format(data[51, "iris"]), text)
                self.assertIn("value = {}".format(data[3, 2]), text)
                self.assertIn("value = {}".format(data[51, 2]), text)
                self.assertEqual(2, text.count("iris ="))

async def test_cloudhook_app_created_then_show_wait_form(
        hass, app, app_oauth_client, smartthings_mock):
    """Test SmartApp is created with a cloudhoko and shows wait form."""
    # Unload the endpoint so we can reload it under the cloud.
    await smartapp.unload_smartapp_endpoint(hass)

    mock_async_active_subscription = Mock(return_value=True)
    mock_create_cloudhook = Mock(return_value=mock_coro(
        return_value="http://cloud.test"))
    with patch.object(cloud, 'async_active_subscription',
                      new=mock_async_active_subscription), \
        patch.object(cloud, 'async_create_cloudhook',
                     new=mock_create_cloudhook):

        await smartapp.setup_smartapp_endpoint(hass)

        flow = SmartThingsFlowHandler()
        flow.hass = hass
        smartthings = smartthings_mock.return_value
        smartthings.apps.return_value = mock_coro(return_value=[])
        smartthings.create_app.return_value = \
            mock_coro(return_value=(app, app_oauth_client))
        smartthings.update_app_settings.return_value = mock_coro()
        smartthings.update_app_oauth.return_value = mock_coro()

        result = await flow.async_step_user({'access_token': str(uuid4())})

        assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
        assert result['step_id'] == 'wait_install'
        assert mock_create_cloudhook.call_count == 1

 def test_provider_shall_update_affected_subscribers_with_published_subscription(cls):
     pro = Provider()
     pub = Publisher(pro)
     sub1 = Subscriber('sub 1 name', pro)
     sub1.subscribe('sub 1 msg 1')
     sub1.subscribe('sub 1 msg 2')
     sub2 = Subscriber('sub 2 name', pro)
     sub2.subscribe('sub 2 msg 1')
     sub2.subscribe('sub 2 msg 2')
     with patch.object(sub1, 'run') as mock_subscriber1_run,\
          patch.object(sub2, 'run') as mock_subscriber2_run:
         pro.update()
         cls.assertEqual(mock_subscriber1_run.call_count, 0)
         cls.assertEqual(mock_subscriber2_run.call_count, 0)
     pub.publish('sub 1 msg 1')
     pub.publish('sub 1 msg 2')
     pub.publish('sub 2 msg 1')
     pub.publish('sub 2 msg 2')
     with patch.object(sub1, 'run') as mock_subscriber1_run,\
          patch.object(sub2, 'run') as mock_subscriber2_run:
         pro.update()
         expected_sub1_calls = [call('sub 1 msg 1'), call('sub 1 msg 2')]
         mock_subscriber1_run.assert_has_calls(expected_sub1_calls)
         expected_sub2_calls = [call('sub 2 msg 1'), call('sub 2 msg 2')]
         mock_subscriber2_run.assert_has_calls(expected_sub2_calls)

    def test_use_cache(self):
        widget = self.widget
        widget.k = 3
        widget.optimize_k = False

        self.send_signal(self.widget.Inputs.data, self.iris[:50])
        widget.unconditional_apply()

        widget.k_from = 2
        widget.k_to = 3
        widget.optimize_k = True
        with patch.object(widget, "_compute_clustering",
                          wraps=widget._compute_clustering) as compute, \
            patch.object(widget, "progressBar",
                         wraps=widget.progressBar) as progressBar:
            widget.unconditional_apply()
            self.assertEqual(compute.call_count, 1)
            compute.assert_called_with(2)
            self.assertEqual(progressBar.call_count, 1)
            progressBar.assert_called_with(1)

            compute.reset_mock()
            progressBar.reset_mock()

            widget.unconditional_apply()
            # compute.assert_not_called unfortunately didn't exist before 3.5
            self.assertFalse(compute.called)
            self.assertFalse(progressBar.called)

    def test_attribute_dirty(self):
        first_items = [1, 2, 3]
        second_items = [1, 2, 4]
        third_items = [1, 2, 3, 4]

        for item in third_items:
            attribute_cache.delete('attribute:%s' % item)

        with patch.object(AttributeContainer, 'pull', side_effect=_attribute_pull) as mock_pull:
            # first execution should run pull
            self.assertEqual(AttributeContainer(third_items).get(), _attribute_pull(third_items))
            mock_pull.assert_called_once_with(third_items)

            # second execution should not run pull again
            self.assertEqual(AttributeContainer(third_items).get(), _attribute_pull(third_items))
            mock_pull.assert_called_once_with(third_items)

        with patch.object(AttributeContainer, 'pull', side_effect=_attribute_pull) as mock_pull:
            # dirtying the cache means we should call pull again
            AttributeContainer(third_items).dirty()
            self.assertEqual(AttributeContainer(third_items).get(), _attribute_pull(third_items))
            mock_pull.assert_called_once_with(third_items)

            # second execution should not run pull again
            self.assertEqual(AttributeContainer(third_items).get(), _attribute_pull(third_items))
            mock_pull.assert_called_once_with(third_items)

        with patch.object(AttributeContainer, 'pull', side_effect=_attribute_pull) as mock_pull:
            # dirtying subset of items should call pull for dirtied items
            AttributeContainer(first_items).dirty()
            self.assertEqual(AttributeContainer(third_items).get(), _attribute_pull(third_items))
            mock_pull.assert_called_once_with(first_items)

    def test_non_existing_dep(self):
        requested = {
            REPO_PYPI: [get_req('dep1 == 1000')]
        }
        interpreter = 'python3'
        is_current = True
        options = {'virtualenv_options': [],
                   'pyvenv_options': [],
                   }
        pip_options = []

        with patch.object(envbuilder._FadesEnvBuilder, 'create_env') as mock_create:
            with patch.object(envbuilder, 'PipManager') as mock_mgr_c:
                mock_create.return_value = ('env_path', 'env_bin_path', 'pip_installed')
                mock_mgr_c.return_value = self.FailInstallManager()
                with patch.object(envbuilder, 'destroy_venv', spec=True) as mock_destroy:
                    with self.assertRaises(FadesError) as cm:
                        envbuilder.create_venv(
                            requested,
                            interpreter,
                            is_current,
                            options,
                            pip_options)
                    self.assertEqual(str(cm.exception), 'Dependency installation failed')
                    mock_destroy.assert_called_once_with('env_path')

        self.assertLoggedDebug("Installation Step failed, removing virtualenv")