def test_basic_creation(self):

        message = self.Message()

        payload = StoragePayload()
        payload.about = 'urn:rho:instance:owner'
        payload.add_type(FOAF.Person, RHO.Owner)

        rdf = RDFStanza()
        rdf['type'] = RDFStanzaType.CREATE.value
        rdf.append(payload.populate_payload())

        message.append(rdf)

        self.check(message, """
            <message>
                <rdf xmlns="urn:rho:rdf" type="create">
                    <x xmlns="jabber:x:data" type="form">
                      <field var="%s" type="text-single">
                        <value>%s</value>
                      </field>
                      <field var="%s" type="list-multi">
                        <value>%s</value>
                        <value>%s</value>
                      </field>
                    </x>
                </rdf>
            </message>
        """ % (str(RDF.about), payload.about,
               str(RDF.type), str(FOAF.Person), str(RHO.Owner), ), use_values=False)

    def _find_work_node(self, session):
        """
        Find a node to work over.
        :param session:
        :return:
        """

        def _handle_cypher_result(_result):
            """
            Handle the results from the cypher result.  If there as no result found then reject the promise, otherwise
            update the session with the uri of the node, and
            :param _result:
            :return:
            """
            if not _result.results:
                raise Exception("No results to work")

            node_uri = _result.results[0].about

            session["node_uri"] = node_uri

            return session

        payload = StoragePayload()
        payload.add_property(key=NEO4J.cypher, value=self.query)
        promise = self._storage_client.execute_cypher(payload).then(_handle_cypher_result)

        return promise

    def test_publish_create(self):

        publish_urn = 'rho:instances.owner'

        payload = StoragePayload()
        payload.add_type(FOAF.Person, RHO.Owner)
        payload.about = publish_urn

        self.rdf_publisher.publish_create(payload)

        self.assertEqual(1, self.roster_plugin.send_message.call_count)

        args, kwargs = self.roster_plugin.send_message.call_args

        self.assertIn('thread_id', kwargs)
        self.assertIn('payload', kwargs)

        self.assertIsNone(kwargs['thread_id'])
        self.assertIsNotNone(kwargs['payload'])

        payload = kwargs['payload']

        self.assertEqual(payload['type'], RDFStanzaType.CREATE.value)

        form = payload['form']

        self.assertIsNotNone(form.get_fields()[str(RDF.type)])
        values = form.get_fields()[str(RDF.type)].get_value()
        self.assertIn(str(FOAF.Person), values)
        self.assertIn(str(RHO.Owner), values)

        self.assertIsNotNone(form.get_fields()[str(RDF.about)])
        values = form.get_fields()[str(RDF.about)].get_value()
        self.assertEqual(publish_urn, values)

    def test_send_request(self):

        payload = StoragePayload()
        payload.add_type(FOAF.Person, RHO.Owner)

        request_promise = self.rdf_publisher.send_out_request(payload)
        self.assertIsNotNone(request_promise)
        self.assertTrue(hasattr(request_promise, 'then'))

        # Verify the payload request.
        self.assertEqual(1, self.roster_plugin.send_message.call_count)

        args, kwargs = self.roster_plugin.send_message.call_args

        self.assertIn('thread_id', kwargs)
        self.assertIn('payload', kwargs)

        # Must have a payload, and a thread_id
        self.assertIsNotNone(kwargs['thread_id'])
        self.assertIsNotNone(kwargs['payload'])

        payload = kwargs['payload']
        self.assertEqual(payload['type'], RDFStanzaType.REQUEST.value)
        self.assertIn(str(FOAF.Person), payload['form'].get_fields()[str(RDF.type)].get_value())
        self.assertIn(str(RHO.Owner), payload['form'].get_fields()[str(RDF.type)].get_value())

    def command_start(self, request, initial_session):
        """
        Provide the configuration details back to the requester and end the command.
        :param request:
        :param initial_session:
        :return:
        """

        storage = StoragePayload()
        storage.add_type(FOAF.Person, RHO.Owner)

        promise = self._storage_client.find_nodes(storage)

        def find_nodes_processor(results):
            """
            Process the results and place the payload into the initial session value.
            :param results:
            :return: the initial session value.
            """
            initial_session['payload'] = results.populate_payload()

            return initial_session

        # Finish populating the rest of initial_session values.
        initial_session['next'] = None
        initial_session['has_next'] = False

        return promise.then(find_nodes_processor)

    def test_response_received(self):

        publish_urn = 'rho:instances.owner'

        payload = StoragePayload()
        payload.add_type(FOAF.Person, RHO.Owner)

        promise = self.rdf_publisher.send_out_request(payload)

        args, kwargs = self.roster_plugin.send_message.call_args
        payload = kwargs['payload']
        thread_id = kwargs['thread_id']

        response_payload = ResultCollectionPayload()
        response_payload.append(ResultPayload(about=publish_urn, types=[FOAF.Person, RHO.Owner]))

        self.rdf_publisher._send_message(RDFStanzaType.RESPONSE, response_payload, thread_id)
        response_args, response_kwargs = self.roster_plugin.send_message.call_args

        payload = response_kwargs['payload']
        response_message = Message()
        response_message.append(payload)
        response_message['thread'] = response_kwargs['thread_id']

        with mock.patch.object(promise, attribute='resolved') as mock_promise_resolve:
            self.rdf_publisher._receive_message(response_message)

            self.assertEqual(1, mock_promise_resolve.call_count)
            args, kwargs = mock_promise_resolve.call_args

            result = [rdf.about for rdf in args[0].results]

            self.assertEqual(result, [publish_urn])

    def store_results(self, payload, session):
        """
        Store the results into the database.
        :param payload:
        :param session:
        :return:
        """

        logger.info('Payload: %s' % payload)
        logger.info('Session: %s' % session)

        storage_session = dict()

        if 'locations' in payload.get_values():
            storage_payload = StoragePayload()
            storage_payload.about = payload.get_values()['locations']
            storage_payload.add_type(WGS_84.SpatialThing)

            promise = self._get_or_lookup(storage_payload).then(
                self._scheduler.generate_promise_handler(self._update_session, storage_session, 'location'))
        else:
            promise = self._scheduler.defer(lambda: storage_session)

        promise = promise.then(self._get_owner)
        promise = promise.then(self._scheduler.generate_promise_handler(self._create_interval, payload))
        promise = promise.then(self._scheduler.generate_promise_handler(self._create_event, payload))

        promise = promise.then(lambda s: session)

        session['payload'] = None
        session['next'] = None
        session['has_next'] = False

        return promise

    def test_execute_cypher(self):

        self.storage_client._store_found('storag[email protected]/storage')

        payload = StoragePayload()
        payload.add_property(NEO4J.cypher, 'Match (n) RETURN n LIMIT 25')

        promise = self.storage_client.execute_cypher(payload)

        def handle_result(result):
            self.session['result'] = result

        promise.then(handle_result)

        self.assertTrue(hasattr(promise, 'then'))

        self.send("""
                <iq type="set" to="[email protected]/storage" id="1">
                    <command xmlns="http://jabber.org/protocol/commands"
                        node="cypher"
                        action="execute">
                        <x xmlns="jabber:x:data" type="form">
                            <field var="http://www.neo4j.com/terms/#cypher" type="list-multi">
                                <value>Match (n) RETURN n LIMIT 25</value>
                                <validate xmlns="http://jabber.org/protocol/xdata-validate" datatype="xs:string" />
                            </field>
                        </x>
                    </command>
                </iq>
            """)

        self.assertNotIn('result', self.session)

        result_payload = ResultCollectionPayload()
        result_payload.append(ResultPayload(about='http://www.example.org/instance/01',
                                            types=[FOAF.Person]))

        self.recv("""
                <iq type='result' from='[email protected]/storage' to='[email protected]/full' id='1'>
                    <command xmlns='http://jabber.org/protocol/commands'
                           sessionid='list:20020923T213616Z-700'
                           node='cypher'
                           status='completed'>
                            %s
                    </command>
                </iq>
            """ % result_payload.populate_payload())

        time.sleep(0.2)

        self.assertIn('result', self.session)

        result = self.session['result']

        self.assertEqual(1, len(result.results))

        self.assertEqual(result.results[0].about, 'http://www.example.org/instance/01')
        self.assertEquals(result.results[0].types[0], str(FOAF.Person))

    def _publish_update(self, storage_result):

        publish_payload = StoragePayload()
        publish_payload.about = storage_result.results[0].about
        publish_payload.add_type(*storage_result.results[0].types)

        self._node_id = publish_payload.about

        self._rdf_publish.publish_update(publish_payload)

    def test_handling(self):

        payload = StoragePayload()
        payload.add_type(WGS_84.SpatialThing)

        self.assertTrue(self.search_handler._process_payload(payload))

        payload.add_type(FOAF.Agent)
        self.assertTrue(self.search_handler._process_payload(payload))

    def _start(self, event):
        payload = StoragePayload()
        payload.add_type(FOAF.Agent, RDFS.Resource)
        payload.add_property(RDFS.seeAlso, self.xmpp.get_uri())

        promise = self._storage_client.find_nodes(payload)

        node_found_promise = promise.then(self._node_found)

        node_found_promise.then(self._update_node, self._create_node)

    def _get_node(self, uri):
        """
        Retrieve the actual node.
        :param uri:
        :return:
        """
        get_payload = StoragePayload()
        get_payload.about = uri

        return self._storage_client.get_node(get_payload)

    def _create_interval(self, session, form_payload):
        """
        Convert the form payload into a storage payload for creating a new interval.
        :param session:
        :param form_payload:
        :return:
        """
        def _translate_search_to_about(result):
            if result.results:
                return result.results[0].about

            raise RuntimeError('Interval was not created.')

        # Handle the creation of the interval.
        create_interval_payload = StoragePayload()
        create_interval_payload.add_type(TIMELINE.Interval)

        if 'event_start' in form_payload.get_values():
            create_interval_payload.add_property(TIMELINE.start, form_payload.get_values()['event_start'])

        if 'event_stop' in form_payload.get_values():
            create_interval_payload.add_property(TIMELINE.end, form_payload.get_values()['event_stop'])

        creator = self._representation_manager.representation_uri
        if creator:
            create_interval_payload.add_property(DCTERMS.creator, creator)

        promise = self._storage_client.create_node(create_interval_payload)
        promise = promise.then(_translate_search_to_about)
        promise = promise.then(self._scheduler.generate_promise_handler(self._update_session, session, 'interval'))

        return promise

    def test_sources_retrieved(self):

        publish_urn = 'rho:instances.owner'

        payload = StoragePayload()
        payload.add_type(FOAF.Person, RHO.Owner)

        promise = self.rdf_publisher.send_out_request(payload, allow_multiple=True)

        args, kwargs = self.roster_plugin.send_message.call_args
        payload = kwargs['payload']
        thread_id = kwargs['thread_id']

        args, kwargs = self.scheduler_plugin.schedule_task.call_args
        callback = kwargs['callback']

        response_payload = ResultCollectionPayload()
        response_payload.append(ResultPayload(about=publish_urn, types=[FOAF.Person, RHO.Owner]))

        search_command_node = 'xmpp:[email protected]/bot?command;node=search_command'

        rdf_payload = self.rdf_publisher.create_rdf(mtype=RDFStanzaType.SEARCH_RESPONSE,
                                                    payload=response_payload, source_name='Search Command',
                                                    source_command=search_command_node)

        response_message = Message()
        response_message.append(rdf_payload)
        response_message['thread'] = thread_id

        with mock.patch.object(promise, attribute='resolved') as mock_promise_resolve:
            self.rdf_publisher._receive_message(response_message)

            mock_promise_resolve.assert_not_called()

        with mock.patch.object(promise, attribute='resolved') as mock_promise_resolve:

            callback()

            self.assertEqual(1, mock_promise_resolve.call_count)
            args, kwargs = mock_promise_resolve.call_args

            result_payload = args[0]

            result = [rdf.about for rdf in result_payload.results]

            self.assertEqual(result, [publish_urn, ])

            self.assertTrue(hasattr(result_payload, 'sources'))

            sources = list(result_payload.sources)

            self.assertEqual(1, len(sources))

            self.assertEqual('Search Command', sources[0][0])
            self.assertEqual('xmpp:[email protected]/bot?command;node=search_command', sources[0][1])

    def test_request_data_with_payload(self):

        foursquare_uri = 'foursquare://venues/4be0b4f0652b0f475f607311'

        storage_payload = StoragePayload()

        storage_payload.add_type(WGS_84.SpatialThing)
        storage_payload.add_property(RDFS.seeAlso, foursquare_uri)

        result = get_foursquare_venue(storage_payload)

        self.assertEqual(result, foursquare_uri.split('/')[-1])

    def test_retrieve_all(self):

        publish_urn = 'rho:instances.owner'

        payload = StoragePayload()
        payload.add_type(FOAF.Person, RHO.Owner)

        promise = self.rdf_publisher.send_out_request(payload, allow_multiple=True)

        args, kwargs = self.roster_plugin.send_message.call_args
        payload = kwargs['payload']
        thread_id = kwargs['thread_id']

        args, kwargs = self.scheduler_plugin.schedule_task.call_args
        callback = kwargs['callback']

        response_payload = ResultCollectionPayload()
        response_payload.append(ResultPayload(about=publish_urn, types=[FOAF.Person, RHO.Owner]))

        self.rdf_publisher._send_message(RDFStanzaType.RESPONSE, response_payload, thread_id)
        response_args, response_kwargs = self.roster_plugin.send_message.call_args

        payload = response_kwargs['payload']
        response_message = Message()
        response_message.append(payload)
        response_message['thread'] = response_kwargs['thread_id']

        with mock.patch.object(promise, attribute='resolved') as mock_promise_resolve:
            self.rdf_publisher._receive_message(response_message)

            mock_promise_resolve.assert_not_called()

        with mock.patch.object(promise, attribute='resolved') as mock_promise_resolve:
            self.rdf_publisher._receive_message(response_message)

            mock_promise_resolve.assert_not_called()

        with mock.patch.object(promise, attribute='resolved') as mock_promise_resolve:

            callback()

            self.assertEqual(1, mock_promise_resolve.call_count)
            args, kwargs = mock_promise_resolve.call_args

            result = [rdf.about for rdf in args[0].results]

            self.assertEqual(result, [publish_urn, publish_urn])

            # Should not have any sources defined.
            callback_results = args[0]

            self.assertFalse(hasattr(callback_results, 'sources'))

    def test_timeout_request(self):
        payload = StoragePayload()
        payload.add_type(FOAF.Person, RHO.Owner)

        promise = self.rdf_publisher.send_out_request(payload)

        args, kwargs = self.scheduler_plugin.schedule_task.call_args
        callback = kwargs['callback']

        with mock.patch.object(target=promise, attribute='resolved') as mocked_method:
            callback()
            self.assertEqual(1, mocked_method.call_count)
            self.assertEqual([], mocked_method.call_args[0][0].results)

    def _find_work_node(self, session):
        """
        Find node to do work over.
        :return:
        """
        logger.debug('Executing query: %s' % self.query)

        payload = StoragePayload()
        payload.add_property(key=NEO4J.cypher, value=self.query)
        promise = self._storage_client.execute_cypher(payload).then(
            self._scheduler.generate_promise_handler(self._handle_results, session))

        return promise

 def publish_all_results(self, result_collection, created=True):
     """
     Publish all of the results in the collection to the correct type.
     :param result_collection: collection to publish the results of.
     :return:
     """
     for res in result_collection.results:
         publish_payload = StoragePayload()
         publish_payload.about = res.about
         publish_payload.add_type(*res.types)
         if created:
             self.publish_create(publish_payload)
         else:
             self.publish_update(publish_payload)

    def test_multiple_responses(self):

        publish_urn = "rho:instances.owner"

        payload = StoragePayload()
        payload.add_type(FOAF.Person, RHO.Owner)

        promise = self.rdf_publisher.send_out_request(payload, allow_multiple=True)

        args, kwargs = self.roster_plugin.send_message.call_args
        payload = kwargs["payload"]
        thread_id = kwargs["thread_id"]

        args, kwargs = self.scheduler_plugin.schedule_task.call_args
        callback = kwargs["callback"]

        response_payload = ResultCollectionPayload()
        response_payload.append(ResultPayload(about=publish_urn, types=[FOAF.Person, RHO.Owner]))

        self.rdf_publisher._send_message(RDFStanzaType.RESPONSE, response_payload, thread_id)
        response_args, response_kwargs = self.roster_plugin.send_message.call_args

        payload = response_kwargs["payload"]
        response_message = Message()
        response_message.append(payload)
        response_message["thread"] = response_kwargs["thread_id"]

        with mock.patch.object(promise, attribute="resolved") as mock_promise_resolve:
            self.rdf_publisher._receive_message(response_message)

            mock_promise_resolve.assert_not_called()

        with mock.patch.object(promise, attribute="resolved") as mock_promise_resolve:
            self.rdf_publisher._receive_message(response_message)

            mock_promise_resolve.assert_not_called()

        with mock.patch.object(promise, attribute="resolved") as mock_promise_resolve:

            callback()

            self.assertEqual(1, mock_promise_resolve.call_count)
            args, kwargs = mock_promise_resolve.call_args

            result = [rdf.about for rdf in args[0].results]

            self.assertEqual(result, [publish_urn, publish_urn])