本文整理汇总了Python中seo_pysolr.Solr类的典型用法代码示例。如果您正苦于以下问题:Python Solr类的具体用法?Python Solr怎么用?Python Solr使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Solr类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: add_jobs
def add_jobs(jobs, upload_chunk_size=1024):
"""
Loads a solr-ready json list of jobs into solr.
inputs:
:jobs: A list of solr-ready, json-formatted jobs.
outputs:
The number of jobs loaded into solr.
"""
conn = Solr(settings.HAYSTACK_CONNECTIONS['default']['URL'])
num_jobs = len(jobs)
# AT&T Showed that large numbers of MOCs can cause import issues due to the size of documents.
# Therefore, when processing AT&T lower the document chunk size.
for job in jobs:
if int(job.get('buid', 0)) == 19389:
logger.warn("AT&T has large amounts of mapped_mocs, that cause problems. Reducing chunk size.")
upload_chunk_size = 64
break
# Chunk them
jobs = chunk(jobs, upload_chunk_size)
for job_group in jobs:
conn.add(list(job_group))
return num_jobs
开发者ID:AstroMatchDynamics,项目名称:MyJobs,代码行数:26,代码来源:import_jobs.py
示例2: clear_solr
def clear_solr(buid):
"""Delete all jobs for a given business unit/job source."""
conn = Solr(settings.HAYSTACK_CONNECTIONS['default']['URL'])
hits = conn.search(q="*:*", rows=1, mlt="false", facet="false").hits
logging.info("BUID:%s - SOLR - Deleting all %s jobs" % (buid, hits))
conn.delete(q="buid:%s" % buid)
logging.info("BUID:%s - SOLR - All jobs deleted." % buid)
开发者ID:DirectEmployers,项目名称:MyJobs,代码行数:7,代码来源:__init__.py
示例3: bread_box_title_heading
def bread_box_title_heading(title_slug_value, jobs=None):
if (not title_slug_value and not jobs) or not title_slug_value:
return None
if jobs:
job = jobs[0]
if title_slug_value == job.title_slug:
return job.title
else:
for job in jobs:
if title_slug_value == job.title_slug:
return job.title
# Try searching solr for a matching title.
conn = Solr(settings.HAYSTACK_CONNECTIONS['default']['URL'])
try:
search_terms = {
'q': u'title_slug:%s' % title_slug_value,
'fl': 'title, title_slug',
'rows': 1,
}
res = conn.search(**search_terms)
except SolrError:
# Poorly formated title_slug_values can sometimes cause Solr errors.
res = None
if res and res.docs[0].get('title_slug') == title_slug_value:
return res.docs[0]['title']
else:
if title_slug_value:
return title_slug_value.replace('-', ' ').title()
else:
return None
开发者ID:wejhink,项目名称:MyJobs,代码行数:33,代码来源:helpers.py
示例4: DirectSEOBase
class DirectSEOBase(TestCase):
def setUp(self):
db_backend = settings.DATABASES['default']['ENGINE'].split('.')[-1]
# Set columns that are utf8 in production to utf8
if db_backend == 'mysql':
cursor = connections['default'].cursor()
cursor.execute("alter table seo_customfacet convert to character "
"set utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_seositefacet convert to character "
"set utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_company convert to character set "
"utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_queryredirect convert to character "
"set utf8 collate utf8_unicode_ci")
cursor.execute("alter table taggit_tag convert to character set "
"utf8 collate utf8_unicode_ci")
cursor.execute("alter table taggit_taggeditem convert to "
"character set "
"utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_seositeredirect convert to "
"character set utf8 collate utf8_unicode_ci")
cursor.execute("alter table django_redirect convert to "
"character set utf8 collate utf8_unicode_ci")
setattr(settings, 'ROOT_URLCONF', 'dseo_urls')
setattr(settings, "PROJECT", 'dseo')
clear_url_caches()
self.base_middleware_classes = settings.MIDDLEWARE_CLASSES
middleware_classes = self.base_middleware_classes + (
'wildcard.middleware.WildcardMiddleware',
'middleware.RedirectOverrideMiddleware')
setattr(settings, 'MIDDLEWARE_CLASSES', middleware_classes)
self.base_context_processors = settings.TEMPLATE_CONTEXT_PROCESSORS
context_processors = self.base_context_processors + (
"social_links.context_processors.social_links_context",
"seo.context_processors.site_config_context",
)
setattr(settings, 'TEMPLATE_CONTEXT_PROCESSORS', context_processors)
context._standard_context_processors = None
self.conn = Solr('http://127.0.0.1:8983/solr/seo')
self.conn.delete(q="*:*")
cache.clear()
clear_url_caches()
setattr(settings, 'MEMOIZE', False)
def tearDown(self):
from django.conf import settings
from django.template import context
setattr(settings, 'TEMPLATE_CONTEXT_PROCESSORS',
self.base_context_processors)
context._standard_context_processors = None
setattr(settings, 'MIDDLEWARE_CLASSES',
self.base_middleware_classes)
开发者ID:wejhink,项目名称:MyJobs,代码行数:59,代码来源:setup.py
示例5: _solr_results_chunk
def _solr_results_chunk(tup, buid, step):
"""
Takes a (start_index, stop_index) tuple and gets the results in that
range from the Solr index.
"""
conn = Solr(settings.HAYSTACK_CONNECTIONS['default']['URL'])
results = conn.search("*:*", fq="buid:%s" % buid, fl="uid",
rows=step, start=tup[0], facet="false",
mlt="false").docs
return set([i['uid'] for i in results if 'uid' in i])
开发者ID:DirectEmployers,项目名称:MyJobs,代码行数:11,代码来源:__init__.py
示例6: setUp
def setUp(self):
super(ImportJobsTestCase, self).setUp()
self.businessunit = BusinessUnitFactory(id=0)
self.buid_id = self.businessunit.id
self.filepath = os.path.join(DATA_DIR, '0', 'dseo_feed_%s.xml' % self.buid_id)
self.solr_settings = {
'default': {'URL': 'http://127.0.0.1:8983/solr/seo'}
}
self.solr = Solr(settings.HAYSTACK_CONNECTIONS['default']['URL'])
开发者ID:vfulco,项目名称:MyJobs,代码行数:9,代码来源:test_import_jobs.py
示例7: remove_expired_jobs
def remove_expired_jobs(buid, active_jobs, upload_chunk_size=1024):
"""
Given a job source id and a list of active jobs for that job source,
Remove the jobs on solr that are not among the active jobs.
"""
conn = Solr(settings.HAYSTACK_CONNECTIONS['default']['URL'])
count = conn.search("*:*", fq="buid:%s" % buid, facet="false",
mlt="false").hits
old_jobs = conn.search("*:*", fq="buid:%s" % buid, facet="false",
rows=count, mlt="false").docs
active_ids = set(j['id'] for j in active_jobs)
old_ids = set(j['id'] for j in old_jobs)
expired = old_ids - active_ids
chunks = chunk(list(expired), upload_chunk_size)
for jobs in chunks:
query = "id:(%s)" % " OR ".join([str(x) for x in jobs])
conn.delete(q=query)
return expired
开发者ID:AstroMatchDynamics,项目名称:MyJobs,代码行数:18,代码来源:import_jobs.py
示例8: add_jobs
def add_jobs(jobs, upload_chunk_size=1024):
"""
Loads a solr-ready json list of jobs into solr.
inputs:
:jobs: A list of solr-ready, json-formatted jobs.
outputs:
The ids of jobs loaded into solr.
"""
conn = Solr(settings.HAYSTACK_CONNECTIONS["default"]["URL"])
# Chunk them
jobs = chunk(jobs, upload_chunk_size)
job_ids = list()
for job_group in jobs:
job_group = list(job_group)
conn.add(job_group)
job_ids.extend(j["id"] for j in job_group)
return job_ids
开发者ID:DirectEmployers,项目名称:MyJobs,代码行数:20,代码来源:solr.py
示例9: MyJobsBase
class MyJobsBase(TestCase):
def setUp(self):
from django.conf import settings
setattr(settings, 'ROOT_URLCONF', 'myjobs_urls')
cache.clear()
clear_url_caches()
self.ms_solr = Solr('http://127.0.0.1:8983/solr/seo')
self.ms_solr.delete(q='*:*')
setattr(settings, "PROJECT", 'myjobs')
self.base_context_processors = settings.TEMPLATE_CONTEXT_PROCESSORS
context_processors = self.base_context_processors + (
'mymessages.context_processors.message_lists',
)
setattr(settings, 'TEMPLATE_CONTEXT_PROCESSORS', context_processors)
setattr(settings, 'MEMOIZE', False)
def tearDown(self):
self.ms_solr.delete(q='*:*')
setattr(settings, 'TEMPLATE_CONTEXT_PROCESSORS',
self.base_context_processors)
开发者ID:AstroMatchDynamics,项目名称:MyJobs,代码行数:21,代码来源:setup.py
示例10: delete_by_guid
def delete_by_guid(guids):
"""
Removes a jobs from solr by guid.
inputs:
:guids: A list of guids
outputs:
The number of jobs that were requested to be deleted. This may
be higher than the number of actual jobs deleted if a guid
passed in did not correspond to a job in solr.
"""
conn = Solr(settings.HAYSTACK_CONNECTIONS['default']['URL'])
if not guids:
return 0
num_guids = len(guids)
guids = chunk(guids)
for guid_group in guids:
delete_str = " OR ".join(guid_group)
conn.delete(q="guid: (%s)" % delete_str)
return num_guids
开发者ID:AstroMatchDynamics,项目名称:MyJobs,代码行数:21,代码来源:import_jobs.py
示例11: setUp
def setUp(self):
db_backend = settings.DATABASES['default']['ENGINE'].split('.')[-1]
# Set columns that are utf8 in production to utf8
if db_backend == 'mysql':
cursor = connections['default'].cursor()
cursor.execute("alter table seo_customfacet convert to character "
"set utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_seositefacet convert to character "
"set utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_company convert to character set "
"utf8 collate utf8_unicode_ci")
cursor.execute("alter table taggit_tag convert to character set "
"utf8 collate utf8_unicode_ci")
cursor.execute("alter table taggit_taggeditem convert to "
"character set "
"utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_seositeredirect convert to "
"character set utf8 collate utf8_unicode_ci")
cursor.execute("alter table django_redirect convert to "
"character set utf8 collate utf8_unicode_ci")
setattr(settings, 'ROOT_URLCONF', 'dseo_urls')
setattr(settings, "PROJECT", 'dseo')
clear_url_caches()
self.base_middleware_classes = settings.MIDDLEWARE_CLASSES
middleware_classes = self.base_middleware_classes + (
'wildcard.middleware.WildcardMiddleware', )
setattr(settings, 'MIDDLEWARE_CLASSES', middleware_classes)
self.base_context_processors = settings.TEMPLATE_CONTEXT_PROCESSORS
context_processors = self.base_context_processors + (
"social_links.context_processors.social_links_context",
"seo.context_processors.site_config_context",
)
setattr(settings, 'TEMPLATE_CONTEXT_PROCESSORS', context_processors)
context._standard_context_processors = None
self.conn = Solr('http://127.0.0.1:8983/solr/seo')
self.conn.delete(q="*:*")
cache.clear()
clear_url_caches()
# Change the solr engine to one that has been extended
# for testing purposes.
self.default_engine = settings.HAYSTACK_CONNECTIONS['default']['ENGINE']
self.engine = 'seo.tests.setup.TestDESolrEngine'
settings.HAYSTACK_CONNECTIONS['default']['ENGINE'] = self.engine
haystack_connections.reload('default')
setattr(settings, 'MEMOIZE', False)
开发者ID:AstroMatchDynamics,项目名称:MyJobs,代码行数:52,代码来源:setup.py
示例12: __init__
def __init__(self, connection_alias, **connection_options):
"""
Inputs:
:HTTP_AUTH_USERNAME: Username used for http authentication
:HTTP_AUTH_PASSWORD: Password used for http authentication
"""
super(DESolrSearchBackend, self).__init__(connection_alias,
**connection_options)
user = connection_options.get("HTTP_AUTH_USERNAME")
passwd = connection_options.get("HTTP_AUTH_PASSWORD")
self.conn = Solr(connection_options['URL'], auth=(user, passwd),
timeout=self.timeout)
开发者ID:AstroMatchDynamics,项目名称:MyJobs,代码行数:13,代码来源:search_backend.py
示例13: setUp
def setUp(self):
super(SiteTestCase, self).setUp()
self.conn = Solr("http://127.0.0.1:8983/solr/seo")
self.conn.delete(q="*:*")
self.businessunit = factories.BusinessUnitFactory(id=0)
self.buid = self.businessunit.id
self.filepath = os.path.join(import_jobs.DATA_DIR, "dseo_feed_%s.xml" % self.buid)
SeoSite.objects.all().delete()
self.site = factories.SeoSiteFactory(id=1)
self.configuration = factories.ConfigurationFactory(status=2)
self.configuration.save()
self.site.configurations.clear()
self.site.configurations.add(self.configuration)
开发者ID:DirectEmployers,项目名称:MyJobs,代码行数:14,代码来源:test_integration.py
示例14: setUp
def setUp(self):
settings.ROOT_URLCONF = "myjobs_urls"
settings.PROJECT = "myjobs"
self.app_access = AppAccessFactory()
self.activities = [
ActivityFactory(name=activity, app_access=self.app_access)
for activity in [
"create communication record", "create contact",
"create partner saved search", "create partner", "create role",
"create tag", "create user", "delete tag", "delete partner",
"delete role", "delete user", "read contact",
"read communication record", "read partner saved search",
"read partner", "read role", "read user", "read tag",
"update communication record", "update contact",
"update partner", "update role", "update tag", "update user",
"read outreach email address", "create outreach email address",
"delete outreach email address",
"update outreach email address", "read outreach record",
"convert outreach record"]]
self.company = CompanyFactory(app_access=[self.app_access])
# this role will be populated by activities on a test-by-test basis
self.role = RoleFactory(company=self.company, name="Admin")
self.user = UserFactory(roles=[self.role], is_staff=True)
cache.clear()
clear_url_caches()
self.ms_solr = Solr(settings.SOLR['seo_test'])
self.ms_solr.delete(q='*:*')
self.base_context_processors = settings.TEMPLATE_CONTEXT_PROCESSORS
context_processors = self.base_context_processors + (
'mymessages.context_processors.message_lists',
)
setattr(settings, 'TEMPLATE_CONTEXT_PROCESSORS', context_processors)
setattr(settings, 'MEMOIZE', False)
self.patcher = patch('urllib2.urlopen', return_file())
self.mock_urlopen = self.patcher.start()
self.client = TestClient()
self.client.login_user(self.user)
开发者ID:DirectEmployers,项目名称:MyJobs,代码行数:43,代码来源:setup.py
示例15: setUp
def setUp(self):
super(JobFeedTestCase, self).setUp()
self.businessunit = BusinessUnitFactory(id=0)
self.buid_id = self.businessunit.id
self.numjobs = 14
self.testdir = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'data')
self.company = CompanyFactory()
self.company.job_source_ids.add(self.businessunit)
self.company.save()
self.conn = Solr("http://127.0.0.1:8983/solr/seo")
self.emptyfeed = os.path.join(self.testdir, "dseo_feed_0.no_jobs.xml")
self.malformed_feed = os.path.join(self.testdir, 'dseo_malformed_feed_0.xml')
self.invalid_feed = os.path.join(self.testdir, 'dseo_invalid_feed_0.xml')
self.unused_field_feed = os.path.join(self.testdir, 'dseo_feed_1.xml')
self.no_onet_feed = os.path.join(self.testdir, 'dseo_feed_no_onets.xml')
#Ensures DATA_DIR used by import_jobs.download_feed_file exists
data_path = DATA_DIR
if not os.path.exists(data_path):
os.mkdir(data_path)
开发者ID:AstroMatchDynamics,项目名称:MyJobs,代码行数:21,代码来源:test_xmlparse.py
示例16: setUp
def setUp(self):
db_backend = settings.DATABASES['default']['ENGINE'].split('.')[-1]
# Set columns that are utf8 in production to utf8
if db_backend == 'mysql':
cursor = connections['default'].cursor()
cursor.execute("alter table seo_customfacet convert to character "
"set utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_seositefacet convert to character "
"set utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_company convert to character set "
"utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_queryredirect convert to character "
"set utf8 collate utf8_unicode_ci")
cursor.execute("alter table taggit_tag convert to character set "
"utf8 collate utf8_unicode_ci")
cursor.execute("alter table taggit_taggeditem convert to "
"character set "
"utf8 collate utf8_unicode_ci")
cursor.execute("alter table seo_seositeredirect convert to "
"character set utf8 collate utf8_unicode_ci")
cursor.execute("alter table django_redirect convert to "
"character set utf8 collate utf8_unicode_ci")
# We have a data migration that does this, but we don't run
# migrations during tests (Django 1.6.5
cursor.execute("ALTER TABLE django_flatpage CONVERT TO "
"CHARACTER SET utf8 COLLATE utf8_general_ci")
cursor.execute("ALTER TABLE seo_custompage CONVERT TO "
"CHARACTER SET utf8 COLLATE utf8_general_ci")
setattr(settings, 'ROOT_URLCONF', 'dseo_urls')
setattr(settings, "PROJECT", 'dseo')
clear_url_caches()
self.base_middleware_classes = settings.MIDDLEWARE_CLASSES
middleware_classes = self.base_middleware_classes + (
'wildcard.middleware.WildcardMiddleware',
'middleware.RedirectOverrideMiddleware')
setattr(settings, 'MIDDLEWARE_CLASSES', middleware_classes)
self.base_context_processors = settings.TEMPLATE_CONTEXT_PROCESSORS
context_processors = self.base_context_processors + (
"social_links.context_processors.social_links_context",
"seo.context_processors.site_config_context",
)
setattr(settings, 'TEMPLATE_CONTEXT_PROCESSORS', context_processors)
context._standard_context_processors = None
self.conn = Solr('http://127.0.0.1:8983/solr/seo')
self.conn.delete(q="*:*")
cache.clear()
clear_url_caches()
setattr(settings, 'MEMOIZE', False)
# As we added tests that created more and more companies, we
# approached the hardcoded companies in import_jobs_testdata.json.
# When we hit those ids, we began to get IntegrityErrors during
# testing. Reset the sequence used by CompanyFactory to clear this
# build-up.
CompanyFactory.reset_sequence()
开发者ID:DirectEmployers,项目名称:MyJobs,代码行数:61,代码来源:setup.py
示例17: SiteTestCase
class SiteTestCase(DirectSEOBase):
"""
We're adding these tests to ensure unicode jobs descriptions and titles
make it through the import process and work with high-level features.
We should use http requests wherever possible since it's difficult to
predict which modules will have issues with unicode.
"""
def setUp(self):
super(SiteTestCase, self).setUp()
self.conn = Solr('http://127.0.0.1:8983/solr/seo')
self.conn.delete(q="*:*")
self.businessunit = factories.BusinessUnitFactory(id=0)
self.buid = self.businessunit.id
self.filepath = os.path.join(import_jobs.DATA_DIR,
'dseo_feed_%s.xml' % self.buid)
SeoSite.objects.all().delete()
self.site = factories.SeoSiteFactory(id=1)
self.configuration = factories.ConfigurationFactory(status=2)
self.configuration.save()
self.site.configurations.clear()
self.site.configurations.add(self.configuration)
def tearDown(self):
super(SiteTestCase, self).tearDown()
self.conn.delete(q="*:*")
def test_unicode_title(self):
# Test imports
group = factories.GroupFactory()
self.site.group = group
self.site.business_units.add(self.businessunit)
self.site.save()
import_jobs.update_solr(self.buid, download=False, delete_feed=False,
data_dir='seo/tests/data/')
solr_jobs = self.conn.search("*:*")
resp = self.client.get('/')
self.assertEqual(resp.context['total_jobs_count'], solr_jobs.hits)
# test standard facets against Haystack query
standard_cf = factories.CustomFacetFactory.build(
# default facet will return both jobs
name="Keyword Facet",
group=group,
show_production=True)
standard_cf.save()
standard_cf.keyword.add(u'Ключевые')
standard_cf.save()
standard_site_facet = factories.SeoSiteFacetFactory(
seosite=self.site,
customfacet=standard_cf,
facet_type=factories.SeoSiteFacet.STANDARD)
standard_site_facet.save()
# test standard facets against Haystack query
standard_cf2 = factories.CustomFacetFactory.build(
# default facet will return both jobs
name='Country Facet',
country='United States',
group=group,
show_production=True)
standard_cf2.save()
standard_site_facet2 = factories.SeoSiteFacetFactory(
seosite=self.site,
customfacet=standard_cf2,
facet_type=factories.SeoSiteFacet.STANDARD)
standard_site_facet2.save()
resp = self.client.get('/keyword-facet/new-jobs/',
HTTP_HOST=self.site.domain, follow=True)
sqs = DESearchQuerySet().filter(text=u'Ключевые')
self.assertEqual(len(resp.context['default_jobs']), sqs.count())
for facet_widget in resp.context['widgets']:
# Ensure that no standard facet has more results than current
# search results
for count_tuple in facet_widget.items:
self.assertTrue(sqs.count() >= count_tuple[1])
# Test default site facets against PySolr query
from django.core.cache import cache
cache.clear()
default_cf = factories.CustomFacetFactory.build(
name="Default Facet",
title=u"Специалист",
group=group,
show_production=True)
default_cf.save()
default_site_facet = factories.SeoSiteFacetFactory(
seosite=self.site,
facet_type=factories.SeoSiteFacet.DEFAULT,
customfacet=default_cf)
default_site_facet.save()
resp = self.client.get('/jobs/', HTTP_HOST=self.site.domain,
follow=True)
total_jobs = resp.context['total_jobs_count']
solr_jobs = self.conn.search(q=u"title:Специалист")
self.assertEqual(total_jobs, solr_jobs.hits)
self.assertEqual(len(resp.context['default_jobs']), total_jobs)
#.........这里部分代码省略.........
开发者ID:kepinae,项目名称:MyJobs,代码行数:101,代码来源:test_integration.py
示例18: MyJobsBase
class MyJobsBase(TestCase):
def setUp(self):
settings.ROOT_URLCONF = "myjobs_urls"
settings.PROJECT = "myjobs"
self.app_access = AppAccessFactory()
self.activities = [
ActivityFactory(name=activity, app_access=self.app_access)
for activity in [
"create communication record", "create contact",
"create partner saved search", "create partner", "create role",
"create tag", "create user", "delete tag", "delete partner",
"delete role", "delete user", "read contact",
"read communication record", "read partner saved search",
"read partner", "read role", "read user", "read tag",
"update communication record", "update contact",
"update partner", "update role", "update tag", "update user",
"read outreach email address", "create outreach email address",
"delete outreach email address",
"update outreach email address", "read outreach record",
"convert outreach record"]]
self.company = CompanyFactory(app_access=[self.app_access])
# this role will be populated by activities on a test-by-test basis
self.role = RoleFactory(company=self.company, name="Admin")
self.user = UserFactory(roles=[self.role], is_staff=True)
cache.clear()
clear_url_caches()
self.ms_solr = Solr(settings.SOLR['seo_test'])
self.ms_solr.delete(q='*:*')
self.base_context_processors = settings.TEMPLATE_CONTEXT_PROCESSORS
context_processors = self.base_context_processors + (
'mymessages.context_processors.message_lists',
)
setattr(settings, 'TEMPLATE_CONTEXT_PROCESSORS', context_processors)
setattr(settings, 'MEMOIZE', False)
self.patcher = patch('urllib2.urlopen', return_file())
self.mock_urlopen = self.patcher.start()
self.client = TestClient()
self.client.login_user(self.user)
def tearDown(self):
self.ms_solr.delete(q='*:*')
setattr(settings, 'TEMPLATE_CONTEXT_PROCESSORS',
self.base_context_processors)
try:
self.patcher.stop()
except RuntimeError:
# patcher was stopped in a test
pass
def assertRequires(self, view_name, *activities, **kwargs):
"""
Asserts that the given view is only accessible when a user has a role
with the given activities.
"""
url = reverse(view_name, kwargs=kwargs.get('kwargs'))
method = kwargs.get("method", "get").lower()
response = getattr(self.client, method)(path=url)
self.assertEqual(type(response), MissingActivity)
self.role.activities = [activity for activity in self.activities
if activity.name in activities]
response = getattr(self.client, method)(path=url)
self.assertNotEqual(type(response), MissingActivity)
self.role.activities.clear()
开发者ID:DirectEmployers,项目名称:MyJobs,代码行数:74,代码来源:setup.py
示例19: setUp
def setUp(self):
super(SitemapTestCase, self).setUp()
self.conn = Solr('http://127.0.0.1:8983/solr/seo')
self.conn.add(SOLR_FIXTURE)
开发者ID:AstroMatchDynamics,项目名称:MyJobs,代码行数:4,代码来源:test_sitemap.py
示例20: SitemapTestCase
class SitemapTestCase(DirectSEOBase):
def setUp(self):
super(SitemapTestCase, self).setUp()
self.conn = Solr('http://127.0.0.1:8983/solr/seo')
self.conn.add(SOLR_FIXTURE)
def test_index(self):
resp = self.client.get("/sitemap.xml")
self.assertEqual(resp.status_code, 200)
def test_no_buid_sitemap(self):
"""
Test to verify that a sitemap is generated with sites that have no
BUID.
"""
site = SeoSite.objects.get(id=1)
site.business_units = []
site.save()
today = datetime.datetime.today()
dt = datetime.date(*today.timetuple()[0:3]).isoformat()
resp = self.client.get("/sitemap-" + dt + ".xml")
self.assertTrue("<url>" in resp.content)
def test_noreverse(self):
"""
Test to ensure that jobs with bad/ugly data do not block the
creation of a sitemap page, but instead are just skipped over in
`SolrSitemap.get_urls().`
This is a regression test. It was prompted by a job in a job feed
file having "~" in the "city" field. Because our URL pattern
doesn't recognize that character in its regex, it caused a
`NoReverseMatch` exception to be thrown. Instead of adding a
tilde, we want to be able to handle any weird characters not
specified in our URL config.
"""
# Sometimes the site settings are messed up from other tests. Ensure
# that the settings are compatible with actually searching for the
# jobs we're adding.
settings.SITE_BUIDS = []
site = SeoSite.objects.get(pk=1)
site.business_units = []
site.save()
# These are kwargs from the actual error that created this error in the
# first place.
kwargs = {
'location': '~, WV',
'title': '911 Coordinator',
'uid': '25901630'
}
job = dict(SOLR_FIXTURE[0])
job.update(kwargs)
self.conn.add([job])
today = datetime.datetime.now()
dt = today.date().isoformat()
resp = self.client.get("/sitemap-" + dt + ".xml")
self.assertEqual(resp.status_code, 200)
self.assertTrue("<url>" in resp.content)
def tearDown(self):
super(SitemapTestCase, self).tearDown()
self.conn.delete("*:*")
开发者ID:AstroMatchDynamics,项目名称:MyJobs,代码行数:65,代码来源:test_sitemap.py
注:本文中的seo_pysolr.Solr类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论