test_commands.py 文件源码

def test_update_old_threads(self):
        """ Test updating the index with old threads.

        If there was a thread that was previously in the index and has
        since been deleted, then it should be removed from the index.
        """
        thread = create_thread()
        thread_pk = thread.pk

        backend = ElasticSearch()
        backend.add(thread)

        thread.delete()

        call_command('updateindex', stdout=self.out)

        es = Elasticsearch([{'host': 'localhost', 'port': 9200}])

        with self.assertRaises(NotFoundError):
            es.get_source(
                index='test',
                doc_type='thread',
                id=thread_pk)