Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
normalize_identifier_fields: bool
Whether to normalize the identifier fields in responses. This
parameter is used only in the 'harvesting' and 'manager' suites
use_xml_canonicalisation: bool
Whether to compare results with their expected values by using xml
canonicalisation or simply by doing a diff.
save_results_directory: str
Path to a directory where to test results should be saved to. A value
of ``None`` (the default) means that results will not get saved to
disk.
"""
request_environment = _prepare_wsgi_test_environment(request_method,
request_data)
pycsw_server = server.Csw(rtconfig=configuration, env=request_environment)
encoding = "utf-8"
status, raw_contents = pycsw_server.dispatch_wsgi()
contents = raw_contents.decode(encoding)
with codecs.open(expected_result, encoding=encoding) as fh:
expected = fh.read()
normalized_result = _normalize(
contents,
normalize_identifiers=normalize_identifier_fields
)
if use_xml_canonicalisation:
print("Comparing results using XML canonicalization...")
matches_expected = _compare_with_xml_canonicalisation(
normalized_result, expected)
else:
print("Comparing results using diffs...")
matches_expected = _compare_without_xml_canonicalisation(
config = env['PYCSW_CONFIG']
if env['QUERY_STRING'].lower().find('config') != -1:
for kvp in env['QUERY_STRING'].split('&'):
if kvp.lower().find('config') != -1:
config = kvp.split('=')[1]
if not os.path.isabs(config):
config = os.path.join(app_path, config)
if 'HTTP_HOST' in env and ':' in env['HTTP_HOST']:
env['HTTP_HOST'] = env['HTTP_HOST'].split(':')[0]
env['local.app_root'] = app_path
csw = server.Csw(config, env)
gzip = False
if ('HTTP_ACCEPT_ENCODING' in env and
env['HTTP_ACCEPT_ENCODING'].find('gzip') != -1):
# set for gzip compressed response
gzip = True
# set compression level
if csw.config.has_option('server', 'gzip_compresslevel'):
gzip_compresslevel = \
int(csw.config.get('server', 'gzip_compresslevel'))
else:
gzip_compresslevel = 0
status, contents = csw.dispatch_wsgi()
return HttpResponse(message, status=status)
env = request.META.copy()
env.update({'local.app_root': os.path.dirname(__file__),
'REQUEST_URI': request.build_absolute_uri()})
# pycsw prefers absolute urls, let's get them from the request.
url = request.build_absolute_uri().split('?')[0]
PYCSW['server']['url'] = url
PYCSW['metadata:main']['provider_url'] = url
# Enable CSW-T when a catalog is defined in the
if catalog:
PYCSW['manager']['transactions'] = 'true'
csw = server.Csw(PYCSW, env)
csw.orm = 'sqlalchemy'
status, content = csw.dispatch_wsgi()
status_code = int(status[0:3])
response = HttpResponse(content,
content_type=csw.contenttype,
status=status_code,
)
return response
def application(env, start_response):
"""WSGI wrapper"""
pycsw_root = get_pycsw_root_path(os.environ, env)
configuration_path = get_configuration_path(os.environ, env, pycsw_root)
env['local.app_root'] = pycsw_root
if 'HTTP_HOST' in env and ':' in env['HTTP_HOST']:
env['HTTP_HOST'] = env['HTTP_HOST'].split(':')[0]
csw = server.Csw(configuration_path, env)
status, contents = csw.dispatch_wsgi()
headers = {
'Content-Length': str(len(contents)),
'Content-Type': str(csw.contenttype)
}
if "gzip" in env.get("HTTP_ACCEPT_ENCODING", ""):
try:
compression_level = int(
csw.config.get("server", "gzip_compresslevel"))
contents, compress_headers = compress_response(
contents, compression_level)
headers.update(compress_headers)
except configparser.NoOptionError:
print(
"The client requested a gzip compressed response. However, "
"the server does not specify the 'gzip_compresslevel' option. "
# update server.url
server_url = '%s://%s%s' % \
(request.META['wsgi.url_scheme'],
request.META['HTTP_HOST'],
request.META['PATH_INFO'])
mdict['server']['url'] = server_url
env = request.META.copy()
env.update({
'local.app_root': os.path.dirname(__file__),
'REQUEST_URI': request.build_absolute_uri(),
})
csw = server.Csw(mdict, env)
content = csw.dispatch_wsgi()
return HttpResponse(content, content_type=csw.contenttype)
# HH should be able to pass env['wsgi.input'] without hanging
# details at https://github.com/cga-harvard/HHypermap/issues/94
if request.method == 'POST':
from StringIO import StringIO
env['wsgi.input'] = StringIO(request.body)
env.update({'local.app_root': os.path.dirname(__file__),
'REQUEST_URI': request.build_absolute_uri()})
# if this is a catalog based CSW, then update settings
if url is not None:
settings.REGISTRY_PYCSW['server']['url'] = url
if catalog_id is not None:
settings.REGISTRY_PYCSW['repository']['filter'] = 'catalog_id = %d' % catalog_id
csw = server.Csw(settings.REGISTRY_PYCSW, env)
content = csw.dispatch_wsgi()
# pycsw 2.0 has an API break:
# pycsw < 2.0: content = xml_response
# pycsw >= 2.0: content = [http_status_code, content]
# deal with the API break
if isinstance(content, list): # pycsw 2.0+
content = content[1]
response = HttpResponse(content, content_type=csw.contenttype)
# TODO: Fix before 1.0 release. CORS should not be enabled blindly like this.
response['Access-Control-Allow-Origin'] = '*'
return response
mdict = dict(settings.PYCSW['CONFIGURATION'], **CONFIGURATION)
if 'server' in settings.PYCSW['CONFIGURATION']:
# override server system defaults with user specified directives
mdict['server'].update(settings.PYCSW['CONFIGURATION']['server'])
config = SafeConfigParser()
for section, options in mdict.iteritems():
config.add_section(section)
for option, value in options.iteritems():
config.set(section, option, value)
# fake HTTP environment variable
os.environ['QUERY_STRING'] = ''
# init pycsw
csw = server.Csw(config, version='2.0.2')
# fake HTTP method
csw.requesttype = 'GET'
# fake HTTP request parameters
if identifier is None: # it's a GetRecords request
formats = []
for f in self.catalogue.formats:
formats.append(METADATA_FORMATS[f][0])
csw.kvp = {
'service': 'CSW',
'version': '2.0.2',
'elementsetname': 'full',
'typenames': formats,
'resulttype': 'results',
for group in public_groups:
if isinstance(group, dict):
if 'group' in group:
groups_ids.append(group['group'])
else:
groups_ids.append(group.id)
if len(groups_ids) > 0:
groups = "(" + (", ".join(str(e) for e in groups_ids)) + ")"
groups_filter = "(group_id IS NULL OR group_id IN " + groups + ")"
mdict['repository']['filter'] += " AND " + groups_filter
else:
groups_filter = "group_id IS NULL"
mdict['repository']['filter'] += " AND " + groups_filter
csw = server.Csw(mdict, env, version='2.0.2')
content = csw.dispatch_wsgi()
# pycsw 2.0 has an API break:
# pycsw < 2.0: content = xml_response
# pycsw >= 2.0: content = [http_status_code, content]
# deal with the API break
if isinstance(content, list): # pycsw 2.0+
content = content[1]
spaces = {'csw': 'http://www.opengis.net/cat/csw/2.0.2',
'dc': 'http://purl.org/dc/elements/1.1/',
'dct': 'http://purl.org/dc/terms/',
'gmd': 'http://www.isotc211.org/2005/gmd',
'gml': 'http://www.opengis.net/gml',