Files
pygeoapi/tests/test_geojson_provider.py
T
Richard Law 60202129ec implement processing manager (#365)
* add process manager

* process manager updating

* control connectivity from inside manager

* missing method on execute_process function signature

* adds tinydb depenedency for requirements-provider

* adds some processes tests

* process manager under server config, not a process

* additional html rendering for jobs and processes

* fixes typo in docstring

* complete html templating for jobs and processes

* considers timezone when formatting processing start/end datetimes

* expanded locales; locale and tz environment variables

* makes environment TZ a global from pygeoapi/__init__.py

* don't redefine built-in format

* adds placeholder values in process form if given as examples

* placeholder in form handles null example

* changes to process progress update

* updates and changes to processing web UI

* adds python3-distutils dep, and env changes for TZs and locales

* adds tinydb dependency for processes

* Adds libpq-dev and related deps for processing - should eventually be reconciled

* minor margin change

* fix bug in postJob.js that prevented results from displaying

* move process elements around to economise on space

* adds second, complex process

* post data may be bytes

* make external process example

* adds redis process manager (may be moved to third-party plugin eventually)

* reintroduce default async processing; refactoring managers; improving sync/async processing ui

* throw pygeoapi ManagerExecuteError, not redis exception

* adds test for async-execute

* restores original test host

* separates async execute handler function

* adds support for file uploads as processing inputs

* adds endpoint for file download as process output

* naming consistency

* Fix job submission form in case server url is relative (#549)

`new URL()` fails if the first argument isn't a complete url

* Implement job deletion in UI, api and tinydb

* Also allow deletion from job list

The UI could be fancier by reloading the page or just removing the
deleted job right away, but I don't want to start building that now.

* Delete output file when deleting a job

* typo in test case

* move delete button in job view

* remove Dockerfile changes (cf #470)

* removes some docs/source*.rst absent in master

* fixes whitespace

* fixes pip installs

* removes more locale-specific content, cf 70

* fixes duplicate app following merge

* removes redis as default plugin

* removes ENV_TZ, cf #407

* fix typo

* fixes syntax error

* process execution test cleanup

* Fix format detection in execute_process (#589)

Before, html was always selected.

* remove db files in code directories, fix flake8, specific functionality

* remove interactive capabilities in lieu of Swagger

* Add basic openapi documentation for job detail endpoint (#590)

It's somewhat superficial still, e.g. jobId is missing as
proper parameter in the UI.

* Show elapsed time for currently running jobs (#591)

* flake8, jobs workflow

* update docs, DB path in test config

* more code cleanup

* manager workflow cleanup

* add default dummy manager

* update docs

* fix routes

* fixed per comments

* fix per comments

* fix per comments

* fix travis

* simply describe process logic

* simplify describe process logic

Co-authored-by: Tom Kralidis <tomkralidis@gmail.com>
Co-authored-by: Jorge de Jesus <jorge.jesus@gmail.com>
Co-authored-by: totycro <bernhard.mallinger@eox.at>
Co-authored-by: Richard Law <richard.law@cartoncloud.com.au>
2020-12-22 06:53:31 -05:00

152 lines
4.1 KiB
Python

# =================================================================
#
# Authors: Matthew Perry <perrygeo@gmail.com>
#
# Copyright (c) 2018 Matthew Perry
# Copyright (c) 2019 Tom Kralidis
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use,
# copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following
# conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
#
# =================================================================
import json
import pytest
from pygeoapi.provider.base import ProviderItemNotFoundError
from pygeoapi.provider.geojson import GeoJSONProvider
path = '/tmp/test.geojson'
@pytest.fixture()
def fixture():
data = {
'type': 'FeatureCollection',
'features': [{
'type': 'Feature',
'id': '123-456',
'geometry': {
'type': 'Point',
'coordinates': [125.6, 10.1]},
'properties': {
'name': 'Dinagat Islands',
'foo': 'bar'
}}
]
}
with open(path, 'w') as fh:
fh.write(json.dumps(data))
return path
@pytest.fixture()
def config():
return {
'name': 'GeoJSON',
'type': 'feature',
'data': path,
'id_field': 'id'
}
def test_query(fixture, config):
p = GeoJSONProvider(config)
fields = p.get_fields()
assert len(fields) == 2
assert fields['name'] == 'string'
results = p.query()
assert len(results['features']) == 1
assert results['numberMatched'] == 1
assert results['numberReturned'] == 1
assert results['features'][0]['id'] == '123-456'
results = p.query(select_properties=['foo'])
assert len(results['features'][0]['properties']) == 1
results = p.query(skip_geometry=True)
assert results['features'][0]['geometry'] is None
def test_get(fixture, config):
p = GeoJSONProvider(config)
results = p.get('123-456')
assert isinstance(results, dict)
assert 'Dinagat' in results['properties']['name']
def test_get_not_existing_item_raise_exception(
fixture, config
):
"""Testing query for a not existing object"""
p = GeoJSONProvider(config)
with pytest.raises(ProviderItemNotFoundError):
p.get(-1)
def test_delete(fixture, config):
p = GeoJSONProvider(config)
p.delete('123-456')
results = p.query()
assert len(results['features']) == 0
def test_create(fixture, config):
p = GeoJSONProvider(config)
new_feature = {
'type': 'Feature',
'id': '123-456',
'geometry': {
'type': 'Point',
'coordinates': [0.0, 0.0]},
'properties': {
'name': 'Null Island'}}
p.create(new_feature)
results = p._load()
assert len(results['features']) == 2
assert 'Dinagat' in results['features'][0]['properties']['name']
assert 'Null' in results['features'][1]['properties']['name']
def test_update(fixture, config):
p = GeoJSONProvider(config)
new_feature = {
'type': 'Feature',
'id': '123-456',
'geometry': {
'type': 'Point',
'coordinates': [0.0, 0.0]},
'properties': {
'name': 'Null Island'}}
p.update('123-456', new_feature)
# Should be changed
results = p.get('123-456')
assert 'Null' in results['properties']['name']