Integrate 'suggest-terms' view from ftw.solr.

This commit is contained in:
Timo Stollenwerk 2013-07-10 11:50:16 +02:00
parent b8eec23612
commit 36076d04f3
4 changed files with 195 additions and 0 deletions

View File

@ -4,6 +4,9 @@ Changelog
4.0 - unreleased
------------------
- Integrate 'suggest-terms' view from ftw.solr. No UI yet!
[timo, 4teamworks]
- Add plone.app.testing setup.
[timo]

View File

@ -43,4 +43,10 @@
class=".facets.SearchFacetsView"
permission="zope2.View" />
<browser:page
for="*"
name="suggest-terms"
class=".suggest.SuggestView"
permission="zope2.View"/>
</configure>

View File

@ -0,0 +1,41 @@
import json
import urllib
from collective.solr.interfaces import ISolrConnectionManager
from Products.Five.browser import BrowserView
from zope.component import getUtility
class SuggestView(BrowserView):
"""Get autocomplete suggestions using solr's suggester component."""
def __call__(self):
suggestions = []
term = self.request.get('term', '')
if not term:
return json.dumps(suggestions)
manager = getUtility(ISolrConnectionManager)
connection = manager.getConnection()
if connection is None:
return json.dumps(suggestions)
params = {}
params['q'] = term
params['wt'] = 'json'
params = urllib.urlencode(params, doseq=True)
response = connection.doPost(
connection.solrBase + '/suggest?' + params, '', {})
results = json.loads(response.read())
spellcheck = results.get('spellcheck', None)
if not spellcheck:
return json.dumps(suggestions)
spellcheck_suggestions = spellcheck.get('suggestions', None)
if not spellcheck_suggestions:
return json.dumps(suggestions)
for suggestion in spellcheck_suggestions[1]['suggestion']:
suggestions.append(dict(label=suggestion, value=suggestion))
return json.dumps(suggestions)

View File

@ -0,0 +1,145 @@
# -*- coding: UTF-8 -*-
from zope.component import getUtility
from zope.component import queryUtility
from collective.solr.interfaces import ISolrConnectionManager
from collective.solr.testing import (
COLLECTIVE_SOLR_INTEGRATION_TESTING
)
from plone.app.testing import setRoles
from plone.app.testing import TEST_USER_ID
from zope.component import getMultiAdapter
import json
import unittest2 as unittest
class MockResponse():
def read(self):
result = {
"responseHeader": {"status": 0, "QTime": 4},
"response": {
"numFound": 0,
"start": 0,
"docs": []
},
"spellcheck": {
"suggestions": [
"Plane",
{
"numFound": 1,
"startOffset": 0,
"endOffset": 5,
"origFreq": 0,
"suggestion": [
{"word":"Plone", "freq":13}
]
},
"correctlySpelled",
False,
"collation",
[
"collationQuery",
"Plone",
"hits",
0,
"misspellingsAndCorrections",
["Plane", "Plone"]
]
]
}
}
return json.dumps(result)
class MockConnection():
solrBase = '/solr'
def doPost(self, url, foo, bar):
return MockResponse()
class MockSolrConnectionManager():
def getConnection(self):
return MockConnection()
class SuggestTermsViewIntegrationTest(unittest.TestCase):
layer = COLLECTIVE_SOLR_INTEGRATION_TESTING
def setUp(self):
self.portal = self.layer['portal']
self.request = self.layer['request']
self.request['ACTUAL_URL'] = self.portal.absolute_url()
setRoles(self.portal, TEST_USER_ID, ['Manager'])
# Unregister standard SolrConnectionManager
self.gsm = self.portal.getSiteManager()
config = getUtility(ISolrConnectionManager)
self.gsm.unregisterUtility(
component=config, provided=ISolrConnectionManager)
def test_suggest_terms_view_is_registered(self):
try:
getMultiAdapter(
(self.portal, self.request),
name="suggest-terms"
)
except:
self.fail("suggest-terms view is not registered properly.")
def test_suggest_terms_view_without_param(self):
view = getMultiAdapter(
(self.portal, self.portal.REQUEST),
name="suggest-terms")
view = view.__of__(self.portal)
self.failUnless(view())
self.assertEqual(view(), '[]')
def test_suggest_terms_view_with_empty_param(self):
self.request.set('term', '')
view = getMultiAdapter(
(self.portal, self.portal.REQUEST),
name="suggest-terms")
view = view.__of__(self.portal)
self.failUnless(view())
self.assertEqual(view(), '[]')
def test_suggest_terms_view_with_param_not_in_solf(self):
self.request.set('term', 'abcdef')
view = getMultiAdapter(
(self.portal, self.portal.REQUEST),
name="suggest-terms")
view = view.__of__(self.portal)
self.failUnless(view())
self.assertEqual(view(), '[]')
def test_suggest_terms_view_with_correctly_spelled_param(self):
self.request.set('term', 'Plone')
view = getMultiAdapter(
(self.portal, self.portal.REQUEST),
name="suggest-terms")
view = view.__of__(self.portal)
self.failUnless(view())
self.assertEqual(view(), '[]')
def test_suggest_terms_view_with_incorrectly_spelled_param(self):
# Replace SolrConnectionManager with Mock
self.gsm.registerUtility(
MockSolrConnectionManager(), ISolrConnectionManager)
self.assertTrue(getUtility(ISolrConnectionManager))
self.request.set('term', 'Plane')
view = getMultiAdapter(
(self.portal, self.portal.REQUEST),
name="suggest-terms")
view = view.__of__(self.portal)
self.failUnless(view())
self.assertEqual(
view(),
json.dumps([{
"value": {"freq": 13, "word": "Plone"},
"label": {"freq": 13, "word": "Plone"}
}])
)