Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_wms_getmap():
# Lastly, test the getcapabilities and getmap methods
wms = WebMapService(SERVICE_URL, version='1.1.1')
def test_wms_getmap_111_service_exception():
"""GetMap 1.1.1 ServiceException for an invalid CRS"""
wms = WebMapService(SERVICE_URL, version='1.1.1')
try:
wms.getmap(
layers=['nexrad_base_reflect'],
styles=['default'],
srs='EPSG:4328',
bbox=(-126, 24, -66, 50),
size=(250, 250),
format='image/jpeg',
transparent=True)
except ServiceException as e:
assert "msWMSLoadGetMapParams(): WMS server error. Invalid SRS given : SRS must be valid for all requested layers." in str(e) # noqa
else:
assert False
def setUp(self):
super(WmsServiceHandlerTestCase, self).setUp()
self.phony_url = ("http://a-really-long-and-fake-name-here-so-that-"
"we-use-it-in-tests")
self.phony_title = "a generic title"
self.phony_version = "some.version"
self.phony_layer_name = "phony_name"
self.phony_keywords = ["first", "second"]
mock_parsed_wms = mock.MagicMock(OwsWebMapService).return_value
(url, mock_parsed_wms) = mock.MagicMock(WebMapService,
return_value=(self.phony_url,
mock_parsed_wms)).return_value
mock_parsed_wms.url = self.phony_url
mock_parsed_wms.identification.title = self.phony_title
mock_parsed_wms.identification.version = self.phony_version
mock_parsed_wms.identification.keywords = self.phony_keywords
mock_layer_meta = mock.MagicMock(ContentMetadata)
mock_layer_meta.name = self.phony_layer_name
mock_layer_meta.children = []
mock_layer_meta.crsOptions = ["EPSG:3857"]
mock_parsed_wms.contents = {
mock_layer_meta.name: mock_layer_meta,
}
self.parsed_wms = mock_parsed_wms
def test_getmap_130_national_map():
"""National Map"""
# TODO: use flaky tests or fix it: https://pypi.python.org/pypi/pytest-ignore-flaky
url = SERVICE_URL_NATIONAL_MAP
wms = WebMapService(url, version='1.3.0')
rsp = wms.getmap(
layers=['3'],
styles=['default'],
srs='CRS:84',
bbox=(-176.646, 17.7016, -64.8017, 71.2854),
size=(500, 300),
format='image/png',
transparent=True)
assert type(rsp) is ResponseWrapper
assert "service=WMS" in wms.request
assert "version=1.3.0" in wms.request
assert "request=GetMap" in wms.request
assert "layers=3" in wms.request
assert "styles=default" in wms.request
assert "crs=CRS%3A84" in wms.request
assert "box=-176.646%2C17.7016%2C-64.8017%2C71.2854" in wms.request
def capcache_query(self):
self.capcache_tstamp = datetime.utcnow()
reader = WMSCapabilitiesReader(self.version, url=self.url,
un=self.username,
pw=self.password,
headers=env.wmsclient.headers)
self.capcache_xml = etree.tostring(reader.read(self.url))
service = WebMapService(
url=self.url, version=self.version,
username=self.username,
password=self.password,
xml=str(self.capcache_xml))
layers = []
for lid, layer in service.contents.iteritems():
layers.append(OrderedDict((
('id', lid), ('title', layer.title),
('index', map(int, layer.index.split('.'))),
)))
layers.sort(key=lambda i: i['index'])
for l in layers:
del l['index']
def get_metadata(self, resource, version='1.1.1'):
"""
Get metadata, specific per Resource type.
:param resource:
:param version:
:return: Metadata object
"""
return WebMapService(resource.url, version=version,
headers=self.get_request_headers())
def _process_wms_service(url, type, username, password, wms=None, owner=None, parent=None):
"""
Create a new WMS/OWS service, cascade it if necessary (i.e. if Web Mercator not available)
"""
if wms is None:
wms = WebMapService(url)
try:
base_url = _clean_url(
wms.getOperationByName('GetMap').methods['Get']['url'])
if base_url and base_url != url:
url = base_url
wms = WebMapService(base_url)
except:
logger.info(
"Could not retrieve GetMap url, using originally supplied URL %s" % url)
pass
try:
service = Service.objects.get(base_url=url)
return_dict = [{'status': 'ok',
'msg': _("This is an existing service"),
'service_id': service.pk,
'service_name': service.name,
'service_title': service.title
}]
return HttpResponse(json.dumps(return_dict),
mimetype='application/json',
status=200)
@staticmethod
def retreive_WMS_metadata(typename):
workspace, layername = decodeTypeName(typename)
# workspace is hard-coded in the importer
url = settings.OGC_SERVER['default']['LOCATION'] + workspace+"/"
url += layername + "/wms?request=GetCapabilities&version=1.1.1"
get_cap_data = CreateStoryLayerThumbnailTask.request_geoserver_with_credentials(
url)
wms = WebMapService(url, xml=get_cap_data)
# I found that some dataset advertise illegal bounds - fix them up
xmin = wms[layername].boundingBoxWGS84[0]
if math.isnan(xmin) or math.isinf(xmin) or xmin < -180:
xmin = -180
ymin = wms[layername].boundingBoxWGS84[1]
if math.isnan(ymin) or math.isinf(ymin) or ymin < -90:
ymin = -90
xmax = wms[layername].boundingBoxWGS84[2]
if math.isnan(xmax) or math.isinf(xmax) or xmax > 180:
xmax = 180
ymax = wms[layername].boundingBoxWGS84[3]
if math.isnan(ymax) or math.isinf(ymax) or ymax > 90:
def _process_wms_service(url, type, username, password, wms=None, owner=None, parent=None):
"""
Create a new WMS/OWS service, cascade it if necessary (i.e. if Web Mercator not available)
"""
if wms is None:
wms = WebMapService(url)
try:
base_url = _clean_url(
wms.getOperationByName('GetMap').methods['Get']['url'])
if base_url and base_url != url:
url = base_url
wms = WebMapService(base_url)
except:
logger.info(
"Could not retrieve GetMap url, using originally supplied URL %s" % url)
pass
try:
service = Service.objects.get(base_url=url)
return_dict = [{'status': 'ok',
'msg': _("This is an existing service"),
def check_service(service):
"""
Tests a service and provide run metrics.
"""
success = True
start_time = datetime.datetime.utcnow()
message = ''
try:
if service.type == 'OGC:WMS':
ows = WebMapService(service.url)
if service.type == 'OGC:WMTS':
ows = WebMapTileService(service.url)
# TODO add more service types here
if service.type.startswith('OGC:'):
title = ows.identification.title
if title is None:
title = '%s %s' % (resource_type, service.url)
# update title
service.title = title
service.save()
except Exception, err:
message = str(err)
success = False