[Git][debian-gis-team/python-osmapi][master] 5 commits: New upstream version 3.0.0+ds
Bas Couwenberg (@sebastic)
gitlab at salsa.debian.org
Sun Feb 13 07:03:10 GMT 2022
Bas Couwenberg pushed to branch master at Debian GIS Project / python-osmapi
Commits:
5d341129 by Bas Couwenberg at 2022-02-13T07:51:13+01:00
New upstream version 3.0.0+ds
- - - - -
8b20174b by Bas Couwenberg at 2022-02-13T07:51:14+01:00
Update upstream source from tag 'upstream/3.0.0+ds'
Update to upstream version '3.0.0+ds'
with Debian dir f4a58f9a2d3f582836f063c32642880c255518b1
- - - - -
1f1b2805 by Bas Couwenberg at 2022-02-13T07:51:29+01:00
New upstream release.
- - - - -
d4904c8d by Bas Couwenberg at 2022-02-13T07:54:30+01:00
Add python3-{dotenv,responses} to build dependencies.
- - - - -
1aedd422 by Bas Couwenberg at 2022-02-13T07:54:42+01:00
Set distribution to unstable.
- - - - -
28 changed files:
- .github/workflows/build.yml
- .gitignore
- CHANGELOG.md
- README.md
- debian/changelog
- debian/control
- + examples/write_to_osm.py
- osmapi/OsmApi.py
- osmapi/__init__.py
- + osmapi/dom.py
- + osmapi/http.py
- + osmapi/parser.py
- + osmapi/xmlbuilder.py
- requirements.txt
- setup.py
- test-requirements.txt
- tests/changeset_test.py
- + tests/conftest.py
- + tests/dom_test.py
- + tests/fixtures/test_Changeset_create.xml
- + tests/fixtures/test_Changeset_create_node.xml
- + tests/fixtures/test_Changeset_upload.xml
- tests/helper_test.py
- tests/node_test.py
- tests/notes_test.py
- tests/osmapi_test.py
- tests/relation_test.py
- tests/way_test.py
Changes:
=====================================
.github/workflows/build.yml
=====================================
@@ -10,7 +10,7 @@ jobs:
strategy:
fail-fast: false
matrix:
- python-version: [3.7, 3.8, 3.9]
+ python-version: ["3.7", "3.8", "3.9", "3.10"]
steps:
- uses: actions/checkout at v2
=====================================
.gitignore
=====================================
@@ -6,3 +6,4 @@ MANIFEST
.tox
.pycache/*
.pytest_cache/*
+.env
=====================================
CHANGELOG.md
=====================================
@@ -4,6 +4,21 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/) and this p
## [Unreleased]
+## [3.0.0] - 2022-02-12
+### Added
+- Add context manager `Changeset()` to open/close changesets
+- Add `session` parameter to provide a custom http session object
+
+### Changed
+- Refactor code into several modules/files to improve maintainability
+- Use `logging` module to log debug information
+
+### Removed
+- **BC-Break**: Remove `debug` parameter of OsmApi, replaced debug messages with `logging` module
+
+### Fixed
+- Added `python_requires` to setup.py to define Python 3.7 as minimum version
+
## [2.0.2] - 2021-11-24
### Changed
- Set `long_description` format to markdown
@@ -299,7 +314,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/) and this p
- `Fixed` for any bug fixes.
- `Security` to invite users to upgrade in case of vulnerabilities.
-[Unreleased]: https://github.com/metaodi/osmapi/compare/v2.0.2...HEAD
+[Unreleased]: https://github.com/metaodi/osmapi/compare/v3.0.0...HEAD
+[3.0.0]: https://github.com/metaodi/osmapi/compare/v2.0.2...v3.0.0
[2.0.2]: https://github.com/metaodi/osmapi/compare/v2.0.1...v2.0.2
[2.0.1]: https://github.com/metaodi/osmapi/compare/v2.0.0...v2.0.1
[2.0.0]: https://github.com/metaodi/osmapi/compare/v1.3.0...v2.0.0
=====================================
README.md
=====================================
@@ -23,7 +23,7 @@ The build the documentation locally, you can use
make docs
This project uses GitHub Pages to publish its documentation.
-To update the online documentation, you need to re-generate the documentation with the above command and update the `gh-pages` branch of this repository.
+To update the online documentation, you need to re-generate the documentation with the above command and update the `master` branch of this repository.
## Examples
@@ -32,14 +32,14 @@ To test this library, please create an account on the [development server of Ope
### Read from OpenStreetMap
```python
-import osmapi
-api = osmapi.OsmApi()
-print(api.NodeGet(123))
-# {u'changeset': 532907, u'uid': 14298,
-# u'timestamp': u'2007-09-29T09:19:17Z',
-# u'lon': 10.790009299999999, u'visible': True,
-# u'version': 1, u'user': u'Mede',
-# u'lat': 59.9503044, u'tag': {}, u'id': 123}
+>>> import osmapi
+>>> api = osmapi.OsmApi()
+>>> print(api.NodeGet(123))
+{u'changeset': 532907, u'uid': 14298,
+u'timestamp': u'2007-09-29T09:19:17Z',
+u'lon': 10.790009299999999, u'visible': True,
+u'version': 1, u'user': u'Mede',
+u'lat': 59.9503044, u'tag': {}, u'id': 123}
```
### Constructor
@@ -56,12 +56,12 @@ Note: Each line in the password file should have the format _user:password_
### Write to OpenStreetMap
```python
-import osmapi
-api = osmapi.OsmApi(api="https://api06.dev.openstreetmap.org", username = u"metaodi", password = u"*******")
-api.ChangesetCreate({u"comment": u"My first test"})
-print(api.NodeCreate({u"lon":1, u"lat":1, u"tag": {}}))
-# {u'changeset': 532907, u'lon': 1, u'version': 1, u'lat': 1, u'tag': {}, u'id': 164684}
-api.ChangesetClose()
+>>> import osmapi
+>>> api = osmapi.OsmApi(api="https://api06.dev.openstreetmap.org", username = u"metaodi", password = u"*******")
+>>> api.ChangesetCreate({u"comment": u"My first test"})
+>>> print(api.NodeCreate({u"lon":1, u"lat":1, u"tag": {}}))
+{u'changeset': 532907, u'lon': 1, u'version': 1, u'lat': 1, u'tag': {}, u'id': 164684}
+>>> api.ChangesetClose()
```
## Note
@@ -93,10 +93,10 @@ To create a new release, follow these steps (please respect [Semantic Versioning
1. Adapt the version number in `osmapi/__init__.py`
1. Update the CHANGELOG with the version
+1. Re-build the documentation (`make docs`)
1. Create a pull request to merge develop into master (make sure the tests pass!)
1. Create a [new release/tag on GitHub](https://github.com/metaodi/osmapi/releases) (on the master branch)
1. The [publication on PyPI](https://pypi.python.org/pypi/osmapi) happens via [GitHub Actions](https://github.com/metaodi/osmapi/actions/workflows/publish_python.yml) on every tagged commit
-1. Re-build the documentation (see above) and copy the generated files to the `gh-pages` branch
## Attribution
=====================================
debian/changelog
=====================================
@@ -1,3 +1,10 @@
+python-osmapi (3.0.0+ds-1) unstable; urgency=medium
+
+ * New upstream release.
+ * Add python3-{dotenv,responses} to build dependencies.
+
+ -- Bas Couwenberg <sebastic at debian.org> Sun, 13 Feb 2022 07:54:32 +0100
+
python-osmapi (2.0.2+ds-1) unstable; urgency=medium
* New upstream release.
=====================================
debian/control
=====================================
@@ -7,12 +7,14 @@ Priority: optional
Build-Depends: debhelper-compat (= 12),
dh-python,
python3-all,
- python3-setuptools,
+ python3-dotenv,
python3-flake8,
python3-mock,
python3-pytest,
python3-pytest-cov,
python3-requests,
+ python3-responses,
+ python3-setuptools,
python3-xmltodict,
tox
Standards-Version: 4.6.0
=====================================
examples/write_to_osm.py
=====================================
@@ -0,0 +1,26 @@
+import osmapi
+from dotenv import load_dotenv, find_dotenv
+import os
+
+load_dotenv(find_dotenv())
+user = os.getenv('OSM_USER')
+pw = os.getenv('OSM_PASS')
+
+api = osmapi.OsmApi(api="https://api06.dev.openstreetmap.org", username=user, password=pw)
+with api.Changeset({u"comment": u"My first test"}) as changeset_id:
+ print(f"Part of Changeset {changeset_id}")
+ node1 = api.NodeCreate({u"lon": 1, u"lat": 1, u"tag": {}})
+ print(node1)
+ node2 = api.NodeCreate({u"lon": 2, u"lat": 2, u"tag": {}})
+ print(node2)
+ way = api.WayCreate({
+ 'nd': [
+ node1['id'],
+ node2['id'],
+ ],
+ 'tag': {
+ 'highway': 'unclassified',
+ 'name': 'Osmapi Street',
+ }
+ })
+ print(way)
=====================================
osmapi/OsmApi.py
=====================================
@@ -29,40 +29,26 @@ Find all information about changes of the different versions of this module
import xml.dom.minidom
import xml.parsers.expat
-import time
-import sys
import urllib.parse
import re
-import requests
-from datetime import datetime
+import logging
+from contextlib import contextmanager
from osmapi import __version__
-from .errors import AlreadySubscribedApiError
-from .errors import ApiError
-from .errors import ChangesetAlreadyOpenError
-from .errors import ChangesetClosedApiError
-from .errors import ElementDeletedApiError
-from .errors import MaximumRetryLimitReachedError
-from .errors import NoChangesetOpenError
-from .errors import NotSubscribedApiError
-from .errors import NoteClosedApiError
-from .errors import OsmApiError
-from .errors import OsmTypeAlreadyExistsError
-from .errors import PreconditionFailedApiError
-from .errors import ResponseEmptyApiError
-from .errors import UsernamePasswordMissingError
-from .errors import VersionMismatchApiError
-from .errors import XmlResponseInvalidError
+from . import dom
+from . import errors
+from . import http
+from . import parser
+from . import xmlbuilder
+
+
+logger = logging.getLogger(__name__)
class OsmApi:
"""
Main class of osmapi, instanciate this class to use osmapi
"""
-
- MAX_RETRY_LIMIT = 5
- """Maximum retries if a call to the remote API fails (default: 5)"""
-
def __init__(
self,
username=None,
@@ -75,7 +61,7 @@ class OsmApi:
changesetautotags={},
changesetautosize=500,
changesetautomulti=1,
- debug=False):
+ session=None):
"""
Initialized the OsmApi object.
@@ -107,14 +93,10 @@ class OsmApi:
upload (default: 500) and `changesetautomulti` defines how many
uploads should be made before closing a changeset and opening a new
one (default: 1).
-
- The `debug` parameter can be used to generate a more verbose output.
"""
- # debug
- self._debug = debug
-
# Get username
+ self._username = None
if username:
self._username = username
elif passwordfile:
@@ -123,6 +105,7 @@ class OsmApi:
self._username = pass_line.split(":")[0].strip()
# Get password
+ self._password = None
if password:
self._password = password
elif passwordfile:
@@ -160,15 +143,23 @@ class OsmApi:
self._CurrentChangesetId = 0
# Http connection
- self._session = self._get_http_session()
-
- def __del__(self):
- self.close()
-
- return None
+ self.http_session = session
+ auth = None
+ if self._username and self._password:
+ auth = (self._username, self._password)
+ self._session = http.OsmApiSession(
+ self._api,
+ self._created_by,
+ auth=auth,
+ session=self.http_session
+ )
def __enter__(self):
- self._session = self._get_http_session()
+ self._session = http.OsmApiSession(
+ self._api,
+ self._created_by,
+ session=self.http_session
+ )
return self
def __exit__(self, *args):
@@ -178,7 +169,7 @@ class OsmApi:
try:
if self._changesetauto:
self._changesetautoflush(True)
- except ResponseEmptyApiError:
+ except errors.ResponseEmptyApiError:
pass
if self._session:
@@ -224,9 +215,9 @@ class OsmApi:
gain insights of the server in use.
"""
uri = "/api/capabilities"
- data = self._get(uri)
+ data = self._session._get(uri)
- data = self._OsmResponseToDom(data, tag="api", single=True)
+ data = dom.OsmResponseToDom(data, tag="api", single=True)
result = {}
for elem in data.childNodes:
if elem.nodeType != elem.ELEMENT_NODE:
@@ -270,9 +261,9 @@ class OsmApi:
uri = "/api/0.6/node/%s" % (NodeId)
if NodeVersion != -1:
uri += "/%s" % (NodeVersion)
- data = self._get(uri)
- data = self._OsmResponseToDom(data, tag="node", single=True)
- return self._DomParseNode(data)
+ data = self._session._get(uri)
+ data = dom.OsmResponseToDom(data, tag="node", single=True)
+ return dom.DomParseNode(data)
def NodeCreate(self, NodeData):
"""
@@ -415,11 +406,11 @@ class OsmApi:
`NodeId` is the unique identifier of a node.
"""
uri = "/api/0.6/node/%s/history" % NodeId
- data = self._get(uri)
- nodes = self._OsmResponseToDom(data, tag="node")
+ data = self._session._get(uri)
+ nodes = dom.OsmResponseToDom(data, tag="node")
result = {}
for node in nodes:
- data = self._DomParseNode(node)
+ data = dom.DomParseNode(node)
result[data["version"]] = data
return result
@@ -447,11 +438,11 @@ class OsmApi:
The `NodeId` is a unique identifier for a node.
"""
uri = "/api/0.6/node/%d/ways" % NodeId
- data = self._get(uri)
- ways = self._OsmResponseToDom(data, tag="way", allow_empty=True)
+ data = self._session._get(uri)
+ ways = dom.OsmResponseToDom(data, tag="way", allow_empty=True)
result = []
for way in ways:
- data = self._DomParseWay(way)
+ data = dom.DomParseWay(way)
result.append(data)
return result
@@ -488,11 +479,11 @@ class OsmApi:
The `NodeId` is a unique identifier for a node.
"""
uri = "/api/0.6/node/%d/relations" % NodeId
- data = self._get(uri)
- relations = self._OsmResponseToDom(data, tag="relation", allow_empty=True)
+ data = self._session._get(uri)
+ relations = dom.OsmResponseToDom(data, tag="relation", allow_empty=True)
result = []
for relation in relations:
- data = self._DomParseRelation(relation)
+ data = dom.DomParseRelation(relation)
result.append(data)
return result
@@ -513,11 +504,11 @@ class OsmApi:
"""
node_list = ",".join([str(x) for x in NodeIdList])
uri = "/api/0.6/nodes?nodes=%s" % node_list
- data = self._get(uri)
- nodes = self._OsmResponseToDom(data, tag="node")
+ data = self._session._get(uri)
+ nodes = dom.OsmResponseToDom(data, tag="node")
result = {}
for node in nodes:
- data = self._DomParseNode(node)
+ data = dom.DomParseNode(node)
result[data["id"]] = data
return result
@@ -551,9 +542,9 @@ class OsmApi:
uri = "/api/0.6/way/%s" % (WayId)
if WayVersion != -1:
uri += "/%s" % (WayVersion)
- data = self._get(uri)
- way = self._OsmResponseToDom(data, tag="way", single=True)
- return self._DomParseWay(way)
+ data = self._session._get(uri)
+ way = dom.OsmResponseToDom(data, tag="way", single=True)
+ return dom.DomParseWay(way)
def WayCreate(self, WayData):
"""
@@ -693,11 +684,11 @@ class OsmApi:
`WayId` is the unique identifier of a way.
"""
uri = "/api/0.6/way/%s/history" % (WayId)
- data = self._get(uri)
- ways = self._OsmResponseToDom(data, tag="way")
+ data = self._session._get(uri)
+ ways = dom.OsmResponseToDom(data, tag="way")
result = {}
for way in ways:
- data = self._DomParseWay(way)
+ data = dom.DomParseWay(way)
result[data["version"]] = data
return result
@@ -734,11 +725,11 @@ class OsmApi:
The `WayId` is a unique identifier for a way.
"""
uri = "/api/0.6/way/%d/relations" % WayId
- data = self._get(uri)
- relations = self._OsmResponseToDom(data, tag="relation", allow_empty=True)
+ data = self._session._get(uri)
+ relations = dom.OsmResponseToDom(data, tag="relation", allow_empty=True)
result = []
for relation in relations:
- data = self._DomParseRelation(relation)
+ data = dom.DomParseRelation(relation)
result.append(data)
return result
@@ -761,8 +752,8 @@ class OsmApi:
`OsmApi.ElementDeletedApiError` is raised.
"""
uri = "/api/0.6/way/%s/full" % (WayId)
- data = self._get(uri)
- return self.ParseOsm(data)
+ data = self._session._get(uri)
+ return parser.ParseOsm(data)
def WaysGet(self, WayIdList):
"""
@@ -780,11 +771,11 @@ class OsmApi:
"""
way_list = ",".join([str(x) for x in WayIdList])
uri = "/api/0.6/ways?ways=%s" % way_list
- data = self._get(uri)
- ways = self._OsmResponseToDom(data, tag="way")
+ data = self._session._get(uri)
+ ways = dom.OsmResponseToDom(data, tag="way")
result = {}
for way in ways:
- data = self._DomParseWay(way)
+ data = dom.DomParseWay(way)
result[data["id"]] = data
return result
@@ -827,9 +818,9 @@ class OsmApi:
uri = "/api/0.6/relation/%s" % (RelationId)
if RelationVersion != -1:
uri += "/%s" % (RelationVersion)
- data = self._get(uri)
- relation = self._OsmResponseToDom(data, tag="relation", single=True)
- return self._DomParseRelation(relation)
+ data = self._session._get(uri)
+ relation = dom.OsmResponseToDom(data, tag="relation", single=True)
+ return dom.DomParseRelation(relation)
def RelationCreate(self, RelationData):
"""
@@ -996,11 +987,11 @@ class OsmApi:
`RelationId` is the unique identifier of a relation.
"""
uri = "/api/0.6/relation/%s/history" % (RelationId)
- data = self._get(uri)
- relations = self._OsmResponseToDom(data, tag="relation")
+ data = self._session._get(uri)
+ relations = dom.OsmResponseToDom(data, tag="relation")
result = {}
for relation in relations:
- data = self._DomParseRelation(relation)
+ data = dom.DomParseRelation(relation)
result[data["version"]] = data
return result
@@ -1038,11 +1029,11 @@ class OsmApi:
The `RelationId` is a unique identifier for a relation.
"""
uri = "/api/0.6/relation/%d/relations" % RelationId
- data = self._get(uri)
- relations = self._OsmResponseToDom(data, tag="relation", allow_empty=True)
+ data = self._session._get(uri)
+ relations = dom.OsmResponseToDom(data, tag="relation", allow_empty=True)
result = []
for relation in relations:
- data = self._DomParseRelation(relation)
+ data = dom.DomParseRelation(relation)
result.append(data)
return result
@@ -1108,8 +1099,8 @@ class OsmApi:
`OsmApi.ElementDeletedApiError` is raised.
"""
uri = "/api/0.6/relation/%s/full" % (RelationId)
- data = self._get(uri)
- return self.ParseOsm(data)
+ data = self._session._get(uri)
+ return parser.ParseOsm(data)
def RelationsGet(self, RelationIdList):
"""
@@ -1128,11 +1119,11 @@ class OsmApi:
"""
relation_list = ",".join([str(x) for x in RelationIdList])
uri = "/api/0.6/relations?relations=%s" % relation_list
- data = self._get(uri)
- relations = self._OsmResponseToDom(data, tag="relation")
+ data = self._session._get(uri)
+ relations = dom.OsmResponseToDom(data, tag="relation")
result = {}
for relation in relations:
- data = self._DomParseRelation(relation)
+ data = dom.DomParseRelation(relation)
result[data["id"]] = data
return result
@@ -1140,6 +1131,43 @@ class OsmApi:
# Changeset #
##################################################
+ @contextmanager
+ def Changeset(self, ChangesetTags={}):
+ """
+ Context manager for a Changeset.
+
+ It opens a Changeset, uploads the changes and closes the changeset
+ when used with the `with` statement:
+
+ #!python
+ import osmapi
+
+ with osmapi.Changeset({"comment": "Import script XYZ"}) as changeset_id:
+ print(f"Part of changeset {changeset_id}")
+ api.NodeCreate({u"lon":1, u"lat":1, u"tag": {}})
+
+ If `ChangesetTags` are given, this tags are applied (key/value).
+
+ Returns `ChangesetId`
+
+ If no authentication information are provided,
+ `OsmApi.UsernamePasswordMissingError` is raised.
+
+ If there is already an open changeset,
+ `OsmApi.ChangesetAlreadyOpenError` is raised.
+ """
+ # Create a new changeset
+ changeset_id = self.ChangesetCreate(ChangesetTags)
+ yield changeset_id
+
+ # upload data to changeset
+ autosize = self._changesetautosize
+ for i in range(0, len(self._changesetautodata), autosize):
+ chunk = self._changesetautodata[i:i+autosize]
+ self.ChangesetUpload(chunk)
+ self._changesetautodata = []
+ self.ChangesetClose()
+
def ChangesetGet(self, ChangesetId, include_discussion=False):
"""
Returns changeset with `ChangesetId` as a dict:
@@ -1169,9 +1197,9 @@ class OsmApi:
path = "/api/0.6/changeset/%s" % (ChangesetId)
if (include_discussion):
path += "?include_discussion=true"
- data = self._get(path)
- changeset = self._OsmResponseToDom(data, tag="changeset", single=True)
- return self._DomParseChangeset(changeset)
+ data = self._session._get(path)
+ changeset = dom.OsmResponseToDom(data, tag="changeset", single=True)
+ return dom.DomParseChangeset(changeset)
def ChangesetUpdate(self, ChangesetTags={}):
"""
@@ -1187,18 +1215,18 @@ class OsmApi:
`OsmApi.ChangesetClosedApiError` is raised.
"""
if not self._CurrentChangesetId:
- raise NoChangesetOpenError("No changeset currently opened")
+ raise errors.NoChangesetOpenError("No changeset currently opened")
if "created_by" not in ChangesetTags:
ChangesetTags["created_by"] = self._created_by
try:
- self._put(
+ self._session._put(
"/api/0.6/changeset/%s" % (self._CurrentChangesetId),
- self._XmlBuild("changeset", {"tag": ChangesetTags}),
+ xmlbuilder._XmlBuild("changeset", {"tag": ChangesetTags}, data=self),
return_value=False
)
- except ApiError as e:
+ except errors.ApiError as e:
if e.status == 409:
- raise ChangesetClosedApiError(e.status, e.reason, e.payload)
+ raise errors.ChangesetClosedApiError(e.status, e.reason, e.payload)
else:
raise
return self._CurrentChangesetId
@@ -1218,12 +1246,12 @@ class OsmApi:
`OsmApi.ChangesetAlreadyOpenError` is raised.
"""
if self._CurrentChangesetId:
- raise ChangesetAlreadyOpenError("Changeset already opened")
+ raise errors.ChangesetAlreadyOpenError("Changeset already opened")
if "created_by" not in ChangesetTags:
ChangesetTags["created_by"] = self._created_by
- result = self._put(
+ result = self._session._put(
"/api/0.6/changeset/create",
- self._XmlBuild("changeset", {"tag": ChangesetTags})
+ xmlbuilder._XmlBuild("changeset", {"tag": ChangesetTags}, data=self)
)
self._CurrentChangesetId = int(result)
return self._CurrentChangesetId
@@ -1244,18 +1272,18 @@ class OsmApi:
`OsmApi.ChangesetClosedApiError` is raised.
"""
if not self._CurrentChangesetId:
- raise NoChangesetOpenError("No changeset currently opened")
+ raise errors.NoChangesetOpenError("No changeset currently opened")
try:
- self._put(
+ self._session._put(
"/api/0.6/changeset/%s/close" % (self._CurrentChangesetId),
"",
return_value=False
)
CurrentChangesetId = self._CurrentChangesetId
self._CurrentChangesetId = 0
- except ApiError as e:
+ except errors.ApiError as e:
if e.status == 409:
- raise ChangesetClosedApiError(e.status, e.reason, e.payload)
+ raise errors.ChangesetClosedApiError(e.status, e.reason, e.payload)
else:
raise
return CurrentChangesetId
@@ -1286,21 +1314,23 @@ class OsmApi:
for change in ChangesData:
data += "<" + change["action"] + ">\n"
change["data"]["changeset"] = self._CurrentChangesetId
- data += self._XmlBuild(
+ data += xmlbuilder._XmlBuild(
change["type"],
change["data"],
- False
+ False,
+ data=self
).decode("utf-8")
data += "</" + change["action"] + ">\n"
data += "</osmChange>"
try:
- data = self._post(
+ data = self._session._post(
"/api/0.6/changeset/%s/upload" % (self._CurrentChangesetId),
- data.encode("utf-8")
+ data.encode("utf-8"),
+ forceAuth=True
)
- except ApiError as e:
+ except errors.ApiError as e:
if e.status == 409 and re.search(r"The changeset .* was closed at .*", e.payload):
- raise ChangesetClosedApiError(e.status, e.reason, e.payload)
+ raise errors.ChangesetClosedApiError(e.status, e.reason, e.payload)
else:
raise
try:
@@ -1308,7 +1338,7 @@ class OsmApi:
data = data.getElementsByTagName("diffResult")[0]
data = [x for x in data.childNodes if x.nodeType == x.ELEMENT_NODE]
except (xml.parsers.expat.ExpatError, IndexError) as e:
- raise XmlResponseInvalidError(
+ raise errors.XmlResponseInvalidError(
"The XML response from the OSM API is invalid: %r" % e
)
@@ -1336,8 +1366,8 @@ class OsmApi:
}
"""
uri = "/api/0.6/changeset/%s/download" % (ChangesetId)
- data = self._get(uri)
- return self.ParseOsc(data)
+ data = self._session._get(uri)
+ return parser.ParseOsc(data)
def ChangesetsGet( # noqa
self,
@@ -1394,11 +1424,11 @@ class OsmApi:
if params:
uri += "?" + urllib.parse.urlencode(params)
- data = self._get(uri)
- changesets = self._OsmResponseToDom(data, tag="changeset")
+ data = self._session._get(uri)
+ changesets = dom.OsmResponseToDom(data, tag="changeset")
result = {}
for curChangeset in changesets:
- tmpCS = self._DomParseChangeset(curChangeset)
+ tmpCS = dom.DomParseChangeset(curChangeset)
result[tmpCS["id"]] = tmpCS
return result
@@ -1435,17 +1465,18 @@ class OsmApi:
"""
params = urllib.parse.urlencode({'text': comment})
try:
- data = self._post(
+ data = self._session._post(
"/api/0.6/changeset/%s/comment" % (ChangesetId),
- params
+ params,
+ forceAuth=True
)
- except ApiError as e:
+ except errors.ApiError as e:
if e.status == 409:
- raise ChangesetClosedApiError(e.status, e.reason, e.payload)
+ raise errors.ChangesetClosedApiError(e.status, e.reason, e.payload)
else:
raise
- changeset = self._OsmResponseToDom(data, tag="changeset", single=True)
- return self._DomParseChangeset(changeset)
+ changeset = dom.OsmResponseToDom(data, tag="changeset", single=True)
+ return dom.DomParseChangeset(changeset)
def ChangesetSubscribe(self, ChangesetId):
"""
@@ -1475,17 +1506,18 @@ class OsmApi:
`OsmApi.UsernamePasswordMissingError` is raised.
"""
try:
- data = self._post(
+ data = self._session._post(
"/api/0.6/changeset/%s/subscribe" % (ChangesetId),
- None
+ None,
+ forceAuth=True
)
- except ApiError as e:
+ except errors.ApiError as e:
if e.status == 409:
- raise AlreadySubscribedApiError(e.status, e.reason, e.payload)
+ raise errors.AlreadySubscribedApiError(e.status, e.reason, e.payload)
else:
raise
- changeset = self._OsmResponseToDom(data, tag="changeset", single=True)
- return self._DomParseChangeset(changeset)
+ changeset = dom.OsmResponseToDom(data, tag="changeset", single=True)
+ return dom.DomParseChangeset(changeset)
def ChangesetUnsubscribe(self, ChangesetId):
"""
@@ -1515,17 +1547,18 @@ class OsmApi:
`OsmApi.UsernamePasswordMissingError` is raised.
"""
try:
- data = self._post(
+ data = self._session._post(
"/api/0.6/changeset/%s/unsubscribe" % (ChangesetId),
- None
+ None,
+ forceAuth=True
)
- except ApiError as e:
+ except errors.ApiError as e:
if e.status == 404:
- raise NotSubscribedApiError(e.status, e.reason, e.payload)
+ raise errors.NotSubscribedApiError(e.status, e.reason, e.payload)
else:
raise
- changeset = self._OsmResponseToDom(data, tag="changeset", single=True)
- return self._DomParseChangeset(changeset)
+ changeset = dom.OsmResponseToDom(data, tag="changeset", single=True)
+ return dom.DomParseChangeset(changeset)
##################################################
# Notes #
@@ -1570,8 +1603,8 @@ class OsmApi:
"/api/0.6/notes?bbox=%f,%f,%f,%f&limit=%d&closed=%d"
% (min_lon, min_lat, max_lon, max_lat, limit, closed)
)
- data = self._get(uri)
- return self.ParseNotes(data)
+ data = self._session._get(uri)
+ return parser.ParseNotes(data)
def NoteGet(self, id):
"""
@@ -1592,9 +1625,9 @@ class OsmApi:
`id` is the unique identifier of the note.
"""
uri = "/api/0.6/notes/%s" % (id)
- data = self._get(uri)
- noteElement = self._OsmResponseToDom(data, tag="note", single=True)
- return self._DomParseNote(noteElement)
+ data = self._session._get(uri)
+ noteElement = dom.OsmResponseToDom(data, tag="note", single=True)
+ return dom.DomParseNote(noteElement)
def NoteCreate(self, NoteData):
"""
@@ -1659,9 +1692,9 @@ class OsmApi:
params['limit'] = limit
params['closed'] = closed
uri += "?" + urllib.parse.urlencode(params)
- data = self._get(uri)
+ data = self._session._get(uri)
- return self.ParseNotes(data)
+ return parser.ParseNotes(data)
def _NoteAction(self, path, comment=None, optionalAuth=True):
"""
@@ -1675,16 +1708,16 @@ class OsmApi:
params['text'] = comment
uri += "?" + urllib.parse.urlencode(params)
try:
- result = self._post(uri, None, optionalAuth=optionalAuth)
- except ApiError as e:
+ result = self._session._post(uri, None, optionalAuth=optionalAuth)
+ except errors.ApiError as e:
if e.status == 404:
- raise NoteClosedApiError(e.status, e.reason, e.payload)
+ raise errors.NoteClosedApiError(e.status, e.reason, e.payload)
else:
raise
# parse the result
- noteElement = self._OsmResponseToDom(result, tag="note", single=True)
- return self._DomParseNote(noteElement)
+ noteElement = dom.OsmResponseToDom(result, tag="note", single=True)
+ return dom.DomParseNote(noteElement)
##################################################
# Other #
@@ -1706,128 +1739,26 @@ class OsmApi:
"/api/0.6/map?bbox=%f,%f,%f,%f"
% (min_lon, min_lat, max_lon, max_lat)
)
- data = self._get(uri)
- return self.ParseOsm(data)
-
- ##################################################
- # Data parser #
- ##################################################
-
- def ParseOsm(self, data):
- """
- Parse osm data.
-
- Returns list of dict:
-
- #!python
- {
- type: node|way|relation,
- data: {}
- }
- """
- try:
- data = xml.dom.minidom.parseString(data)
- data = data.getElementsByTagName("osm")[0]
- except (xml.parsers.expat.ExpatError, IndexError) as e:
- raise XmlResponseInvalidError(
- "The XML response from the OSM API is invalid: %r" % e
- )
-
- result = []
- for elem in data.childNodes:
- if elem.nodeName == "node":
- result.append({
- "type": elem.nodeName,
- "data": self._DomParseNode(elem)
- })
- elif elem.nodeName == "way":
- result.append({
- "type": elem.nodeName,
- "data": self._DomParseWay(elem)
- })
- elif elem.nodeName == "relation":
- result.append({
- "type": elem.nodeName,
- "data": self._DomParseRelation(elem)
- })
- return result
+ data = self._session._get(uri)
+ return parser.ParseOsm(data)
- def ParseOsc(self, data):
- """
- Parse osc data.
-
- Returns list of dict:
-
- #!python
- {
- type: node|way|relation,
- action: create|delete|modify,
- data: {}
- }
+ def flush(self):
"""
- try:
- data = xml.dom.minidom.parseString(data)
- data = data.getElementsByTagName("osmChange")[0]
- except (xml.parsers.expat.ExpatError, IndexError) as e:
- raise XmlResponseInvalidError(
- "The XML response from the OSM API is invalid: %r" % e
- )
-
- result = []
- for action in data.childNodes:
- if action.nodeName == "#text":
- continue
- for elem in action.childNodes:
- if elem.nodeName == "node":
- result.append({
- "action": action.nodeName,
- "type": elem.nodeName,
- "data": self._DomParseNode(elem)
- })
- elif elem.nodeName == "way":
- result.append({
- "action": action.nodeName,
- "type": elem.nodeName,
- "data": self._DomParseWay(elem)
- })
- elif elem.nodeName == "relation":
- result.append({
- "action": action.nodeName,
- "type": elem.nodeName,
- "data": self._DomParseRelation(elem)
- })
- return result
+ Force the changes to be uploaded to OSM and the changeset to be closed
- def ParseNotes(self, data):
- """
- Parse notes data.
+ If no authentication information are provided,
+ `OsmApi.UsernamePasswordMissingError` is raised.
- Returns a list of dict:
+ If there is no open changeset,
+ `OsmApi.NoChangesetOpenError` is raised.
- #!python
- [
- {
- 'id': integer,
- 'action': opened|commented|closed,
- 'status': open|closed
- 'date_created': creation date
- 'date_closed': closing data|None
- 'uid': User ID|None
- 'user': User name|None
- 'comments': {}
- },
- { ... }
- ]
+ If there is already an open changeset,
+ `OsmApi.ChangesetAlreadyOpenError` is raised.
"""
- noteElements = self._OsmResponseToDom(data, tag="note")
- result = []
- for noteElement in noteElements:
- note = self._DomParseNote(noteElement)
- result.append(note)
- return result
+ return self._changesetautoflush(True)
##################################################
- # Internal http function #
+ # Internal method #
##################################################
def _do(self, action, OsmType, OsmData):
@@ -1844,7 +1775,7 @@ class OsmApi:
def _do_manu(self, action, OsmType, OsmData): # noqa
if not self._CurrentChangesetId:
- raise NoChangesetOpenError(
+ raise errors.NoChangesetOpenError(
"You need to open a changeset before uploading data"
)
if "timestamp" in OsmData:
@@ -1852,21 +1783,21 @@ class OsmApi:
OsmData["changeset"] = self._CurrentChangesetId
if action == "create":
if OsmData.get("id", -1) > 0:
- raise OsmTypeAlreadyExistsError(
+ raise errors.OsmTypeAlreadyExistsError(
"This %s already exists" % OsmType
)
try:
- result = self._put(
+ result = self._session._put(
"/api/0.6/%s/create" % OsmType,
- self._XmlBuild(OsmType, OsmData)
+ xmlbuilder._XmlBuild(OsmType, OsmData, data=self)
)
- except ApiError as e:
+ except errors.ApiError as e:
if e.status == 409 and re.search(r"The changeset .* was closed at .*", e.payload):
- raise ChangesetClosedApiError(e.status, e.reason, e.payload)
+ raise errors.ChangesetClosedApiError(e.status, e.reason, e.payload)
elif e.status == 409:
- raise VersionMismatchApiError(e.status, e.reason, e.payload)
+ raise errors.VersionMismatchApiError(e.status, e.reason, e.payload)
elif e.status == 412:
- raise PreconditionFailedApiError(e.status, e.reason, e.payload)
+ raise errors.PreconditionFailedApiError(e.status, e.reason, e.payload)
else:
raise
OsmData["id"] = int(result.strip())
@@ -1874,56 +1805,41 @@ class OsmApi:
return OsmData
elif action == "modify":
try:
- result = self._put(
+ result = self._session._put(
"/api/0.6/%s/%s" % (OsmType, OsmData["id"]),
- self._XmlBuild(OsmType, OsmData)
+ xmlbuilder._XmlBuild(OsmType, OsmData, data=self)
)
- except ApiError as e:
- print(e.reason)
+ except errors.ApiError as e:
+ logger.error(e.reason)
if e.status == 409 and re.search(r"The changeset .* was closed at .*", e.payload):
- raise ChangesetClosedApiError(e.status, e.reason, e.payload)
+ raise errors.ChangesetClosedApiError(e.status, e.reason, e.payload)
elif e.status == 409:
- raise VersionMismatchApiError(e.status, e.reason, e.payload)
+ raise errors.VersionMismatchApiError(e.status, e.reason, e.payload)
elif e.status == 412:
- raise PreconditionFailedApiError(e.status, e.reason, e.payload)
+ raise errors.PreconditionFailedApiError(e.status, e.reason, e.payload)
else:
raise
OsmData["version"] = int(result.strip())
return OsmData
elif action == "delete":
try:
- result = self._delete(
+ result = self._session._delete(
"/api/0.6/%s/%s" % (OsmType, OsmData["id"]),
- self._XmlBuild(OsmType, OsmData)
+ xmlbuilder._XmlBuild(OsmType, OsmData, data=self)
)
- except ApiError as e:
+ except errors.ApiError as e:
if e.status == 409 and re.search(r"The changeset .* was closed at .*", e.payload):
- raise ChangesetClosedApiError(e.status, e.reason, e.payload)
+ raise errors.ChangesetClosedApiError(e.status, e.reason, e.payload)
elif e.status == 409:
- raise VersionMismatchApiError(e.status, e.reason, e.payload)
+ raise errors.VersionMismatchApiError(e.status, e.reason, e.payload)
elif e.status == 412:
- raise PreconditionFailedApiError(e.status, e.reason, e.payload)
+ raise errors.PreconditionFailedApiError(e.status, e.reason, e.payload)
else:
raise
OsmData["version"] = int(result.strip())
OsmData["visible"] = False
return OsmData
- def flush(self):
- """
- Force the changes to be uploaded to OSM and the changeset to be closed
-
- If no authentication information are provided,
- `OsmApi.UsernamePasswordMissingError` is raised.
-
- If there is no open changeset,
- `OsmApi.NoChangesetOpenError` is raised.
-
- If there is already an open changeset,
- `OsmApi.ChangesetAlreadyOpenError` is raised.
- """
- return self._changesetautoflush(True)
-
def _changesetautoflush(self, force=False):
autosize = self._changesetautosize
while ((len(self._changesetautodata) >= autosize) or
@@ -1942,392 +1858,3 @@ class OsmApi:
self.ChangesetClose()
self._changesetautocpt = 0
return None
-
- def _http_request(self, method, path, auth, send, return_value=True): # noqa
- """
- Returns the response generated by an HTTP request.
-
- `method` is a HTTP method to be executed
- with the request data. For example: 'GET' or 'POST'.
- `path` is the path to the requested resource relative to the
- base API address stored in self._api. Should start with a
- slash character to separate the URL.
- `auth` is a boolean indicating whether authentication should
- be preformed on this request.
- `send` contains additional data that might be sent in a
- request.
- `return_value` indicates wheter this request should return
- any data or not.
-
- If the username or password is missing,
- `OsmApi.UsernamePasswordMissingError` is raised.
-
- If the requested element has been deleted,
- `OsmApi.ElementDeletedApiError` is raised.
-
- If the response status code indicates an error,
- `OsmApi.ApiError` is raised.
- """
- if self._debug:
- error_msg = (
- "%s %s %s"
- % (time.strftime("%Y-%m-%d %H:%M:%S"), method, path)
- )
- print(error_msg, file=sys.stderr)
-
- # Add API base URL to path
- path = self._api + path
-
- user_pass = None
- if auth:
- try:
- user_pass = (self._username, self._password)
- except AttributeError:
- raise UsernamePasswordMissingError("Username/Password missing")
-
- response = self._session.request(method, path, auth=user_pass,
- data=send)
- if response.status_code != 200:
- payload = response.content.strip()
- if response.status_code == 410:
- raise ElementDeletedApiError(
- response.status_code,
- response.reason,
- payload
- )
- raise ApiError(response.status_code, response.reason, payload)
- if return_value and not response.content:
- raise ResponseEmptyApiError(
- response.status_code,
- response.reason,
- ''
- )
-
- if self._debug:
- error_msg = (
- "%s %s %s"
- % (time.strftime("%Y-%m-%d %H:%M:%S"), method, path)
- )
- print(error_msg, file=sys.stderr)
- return response.content
-
- def _http(self, cmd, path, auth, send, return_value=True): # noqa
- i = 0
- while True:
- i += 1
- try:
- return self._http_request(
- cmd,
- path,
- auth,
- send,
- return_value=return_value
- )
- except ApiError as e:
- if e.status >= 500:
- if i == self.MAX_RETRY_LIMIT:
- raise
- if i != 1:
- self._sleep()
- self._session = self._get_http_session()
- else:
- raise
- except Exception as e:
- print(e)
- if i == self.MAX_RETRY_LIMIT:
- if isinstance(e, OsmApiError):
- raise
- raise MaximumRetryLimitReachedError(
- "Give up after %s retries" % i
- )
- if i != 1:
- self._sleep()
- self._session = self._get_http_session()
-
- def _get_http_session(self):
- """
- Creates a requests session for connection pooling.
- """
- session = requests.Session()
- session.headers.update({
- 'user-agent': self._created_by
- })
- return session
-
- def _sleep(self):
- time.sleep(5)
-
- def _get(self, path):
- return self._http('GET', path, False, None)
-
- def _put(self, path, data, return_value=True):
- return self._http('PUT', path, True, data, return_value=return_value)
-
- def _post(self, path, data, optionalAuth=False):
- auth = True
- # the Notes API allows certain POSTs by non-authenticated users
- if optionalAuth:
- auth = hasattr(self, '_username')
- return self._http('POST', path, auth, data)
-
- def _delete(self, path, data):
- return self._http('DELETE', path, True, data)
-
- ##################################################
- # Internal dom function #
- ##################################################
-
- def _OsmResponseToDom(self, response, tag, single=False, allow_empty=False):
- """
- Returns the (sub-) DOM parsed from an OSM response
- """
- try:
- dom = xml.dom.minidom.parseString(response)
- osm_dom = dom.getElementsByTagName("osm")[0]
- all_data = osm_dom.getElementsByTagName(tag)
- first_element = all_data[0]
- except (IndexError) as e:
- if allow_empty:
- return []
- raise XmlResponseInvalidError(
- "The XML response from the OSM API is invalid: %r" % e
- )
- except (xml.parsers.expat.ExpatError) as e:
- raise XmlResponseInvalidError(
- "The XML response from the OSM API is invalid: %r" % e
- )
-
- if single:
- return first_element
- return all_data
-
- def _DomGetAttributes(self, DomElement): # noqa
- """
- Returns a formated dictionnary of attributes of a DomElement.
- """
- result = {}
- for k, v in DomElement.attributes.items():
- if k == "uid":
- v = int(v)
- elif k == "changeset":
- v = int(v)
- elif k == "version":
- v = int(v)
- elif k == "id":
- v = int(v)
- elif k == "lat":
- v = float(v)
- elif k == "lon":
- v = float(v)
- elif k == "open":
- v = (v == "true")
- elif k == "visible":
- v = (v == "true")
- elif k == "ref":
- v = int(v)
- elif k == "comments_count":
- v = int(v)
- elif k == "timestamp":
- v = self._ParseDate(v)
- elif k == "created_at":
- v = self._ParseDate(v)
- elif k == "closed_at":
- v = self._ParseDate(v)
- elif k == "date":
- v = self._ParseDate(v)
- result[k] = v
- return result
-
- def _DomGetTag(self, DomElement):
- """
- Returns the dictionnary of tags of a DomElement.
- """
- result = {}
- for t in DomElement.getElementsByTagName("tag"):
- k = t.attributes["k"].value
- v = t.attributes["v"].value
- result[k] = v
- return result
-
- def _DomGetNd(self, DomElement):
- """
- Returns the list of nodes of a DomElement.
- """
- result = []
- for t in DomElement.getElementsByTagName("nd"):
- result.append(int(int(t.attributes["ref"].value)))
- return result
-
- def _DomGetDiscussion(self, DomElement):
- """
- Returns the dictionnary of comments of a DomElement.
- """
- result = []
- try:
- discussion = DomElement.getElementsByTagName("discussion")[0]
- for t in discussion.getElementsByTagName("comment"):
- comment = self._DomGetAttributes(t)
- comment['text'] = self._GetXmlValue(t, "text")
- result.append(comment)
- except IndexError:
- pass
- return result
-
- def _DomGetComments(self, DomElement):
- """
- Returns the list of comments of a DomElement.
- """
- result = []
- for t in DomElement.getElementsByTagName("comment"):
- comment = {}
- comment['date'] = self._ParseDate(self._GetXmlValue(t, "date"))
- comment['action'] = self._GetXmlValue(t, "action")
- comment['text'] = self._GetXmlValue(t, "text")
- comment['html'] = self._GetXmlValue(t, "html")
- comment['uid'] = self._GetXmlValue(t, "uid")
- comment['user'] = self._GetXmlValue(t, "user")
- result.append(comment)
- return result
-
- def _DomGetMember(self, DomElement):
- """
- Returns a list of relation members.
- """
- result = []
- for m in DomElement.getElementsByTagName("member"):
- result.append(self._DomGetAttributes(m))
- return result
-
- def _DomParseNode(self, DomElement):
- """
- Returns NodeData for the node.
- """
- result = self._DomGetAttributes(DomElement)
- result["tag"] = self._DomGetTag(DomElement)
- return result
-
- def _DomParseWay(self, DomElement):
- """
- Returns WayData for the way.
- """
- result = self._DomGetAttributes(DomElement)
- result["tag"] = self._DomGetTag(DomElement)
- result["nd"] = self._DomGetNd(DomElement)
- return result
-
- def _DomParseRelation(self, DomElement):
- """
- Returns RelationData for the relation.
- """
- result = self._DomGetAttributes(DomElement)
- result["tag"] = self._DomGetTag(DomElement)
- result["member"] = self._DomGetMember(DomElement)
- return result
-
- def _DomParseChangeset(self, DomElement):
- """
- Returns ChangesetData for the changeset.
- """
- result = self._DomGetAttributes(DomElement)
- result["tag"] = self._DomGetTag(DomElement)
- result["discussion"] = self._DomGetDiscussion(DomElement)
-
- return result
-
- def _DomParseNote(self, DomElement):
- """
- Returns NoteData for the note.
- """
- result = self._DomGetAttributes(DomElement)
- result["id"] = self._GetXmlValue(DomElement, "id")
- result["status"] = self._GetXmlValue(DomElement, "status")
-
- result["date_created"] = self._ParseDate(
- self._GetXmlValue(DomElement, "date_created")
- )
- result["date_closed"] = self._ParseDate(
- self._GetXmlValue(DomElement, "date_closed")
- )
- result["comments"] = self._DomGetComments(DomElement)
-
- return result
-
- def _ParseDate(self, DateString):
- result = DateString
- try:
- result = datetime.strptime(DateString, "%Y-%m-%d %H:%M:%S UTC")
- except Exception:
- try:
- result = datetime.strptime(DateString, "%Y-%m-%dT%H:%M:%SZ")
- except Exception:
- pass
-
- return result
-
- ##################################################
- # Internal xml builder #
- ##################################################
-
- def _XmlBuild(self, ElementType, ElementData, WithHeaders=True): # noqa
-
- xml = ""
- if WithHeaders:
- xml += "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
- xml += "<osm version=\"0.6\" generator=\""
- xml += self._created_by + "\">\n"
-
- # <element attr="val">
- xml += " <" + ElementType
- if "id" in ElementData:
- xml += " id=\"" + str(ElementData["id"]) + "\""
- if "lat" in ElementData:
- xml += " lat=\"" + str(ElementData["lat"]) + "\""
- if "lon" in ElementData:
- xml += " lon=\"" + str(ElementData["lon"]) + "\""
- if "version" in ElementData:
- xml += " version=\"" + str(ElementData["version"]) + "\""
- visible_str = str(ElementData.get("visible", True)).lower()
- xml += " visible=\"" + visible_str + "\""
- if ElementType in ["node", "way", "relation"]:
- xml += " changeset=\"" + str(self._CurrentChangesetId) + "\""
- xml += ">\n"
-
- # <tag... />
- for k, v in ElementData.get("tag", {}).items():
- xml += " <tag k=\"" + self._XmlEncode(k)
- xml += "\" v=\"" + self._XmlEncode(v) + "\"/>\n"
-
- # <member... />
- for member in ElementData.get("member", []):
- xml += " <member type=\"" + member["type"]
- xml += "\" ref=\"" + str(member["ref"])
- xml += "\" role=\"" + self._XmlEncode(member["role"])
- xml += "\"/>\n"
-
- # <nd... />
- for ref in ElementData.get("nd", []):
- xml += " <nd ref=\"" + str(ref) + "\"/>\n"
-
- # </element>
- xml += " </" + ElementType + ">\n"
-
- if WithHeaders:
- xml += "</osm>\n"
-
- return xml.encode("utf8")
-
- def _XmlEncode(self, text):
- return (
- text
- .replace("&", "&")
- .replace("\"", """)
- .replace("<", "<")
- .replace(">", ">")
- )
-
- def _GetXmlValue(self, DomElement, tag):
- try:
- elem = DomElement.getElementsByTagName(tag)[0]
- return elem.firstChild.nodeValue
- except Exception:
- return None
=====================================
osmapi/__init__.py
=====================================
@@ -1,4 +1,4 @@
-__version__ = '2.0.2'
+__version__ = '3.0.0'
from .OsmApi import * # noqa
from .errors import * # noqa
=====================================
osmapi/dom.py
=====================================
@@ -0,0 +1,204 @@
+from datetime import datetime
+import xml.dom.minidom
+import xml.parsers.expat
+import logging
+
+from . import errors
+from . import xmlbuilder
+
+
+logger = logging.getLogger(__name__)
+
+
+def OsmResponseToDom(response, tag, single=False, allow_empty=False):
+ """
+ Returns the (sub-) DOM parsed from an OSM response
+ """
+ try:
+ dom = xml.dom.minidom.parseString(response)
+ osm_dom = dom.getElementsByTagName("osm")[0]
+ all_data = osm_dom.getElementsByTagName(tag)
+ first_element = all_data[0]
+ except (IndexError) as e:
+ if allow_empty:
+ return []
+ raise errors.XmlResponseInvalidError(
+ "The XML response from the OSM API is invalid: %r" % e
+ )
+ except (xml.parsers.expat.ExpatError) as e:
+ raise errors.XmlResponseInvalidError(
+ "The XML response from the OSM API is invalid: %r" % e
+ )
+
+ if single:
+ return first_element
+ return all_data
+
+
+def DomParseNode(DomElement):
+ """
+ Returns NodeData for the node.
+ """
+ result = _DomGetAttributes(DomElement)
+ result["tag"] = _DomGetTag(DomElement)
+ return result
+
+
+def DomParseWay(DomElement):
+ """
+ Returns WayData for the way.
+ """
+ result = _DomGetAttributes(DomElement)
+ result["tag"] = _DomGetTag(DomElement)
+ result["nd"] = _DomGetNd(DomElement)
+ return result
+
+
+def DomParseRelation(DomElement):
+ """
+ Returns RelationData for the relation.
+ """
+ result = _DomGetAttributes(DomElement)
+ result["tag"] = _DomGetTag(DomElement)
+ result["member"] = _DomGetMember(DomElement)
+ return result
+
+
+def DomParseChangeset(DomElement):
+ """
+ Returns ChangesetData for the changeset.
+ """
+ result = _DomGetAttributes(DomElement)
+ result["tag"] = _DomGetTag(DomElement)
+ result["discussion"] = _DomGetDiscussion(DomElement)
+
+ return result
+
+
+def DomParseNote(DomElement):
+ """
+ Returns NoteData for the note.
+ """
+ result = _DomGetAttributes(DomElement)
+ result["id"] = xmlbuilder._GetXmlValue(DomElement, "id")
+ result["status"] = xmlbuilder._GetXmlValue(DomElement, "status")
+
+ result["date_created"] = _ParseDate(
+ xmlbuilder._GetXmlValue(DomElement, "date_created")
+ )
+ result["date_closed"] = _ParseDate(
+ xmlbuilder._GetXmlValue(DomElement, "date_closed")
+ )
+ result["comments"] = _DomGetComments(DomElement)
+
+ return result
+
+
+def _DomGetAttributes(DomElement):
+ """
+ Returns a formated dictionnary of attributes of a DomElement.
+ """
+
+ def is_true(v):
+ return (v == "true")
+
+ attribute_mapping = {
+ 'uid': int,
+ 'changeset': int,
+ 'version': int,
+ 'id': int,
+ 'lat': float,
+ 'lon': float,
+ 'open': is_true,
+ 'visible': is_true,
+ 'ref': int,
+ 'comments_count': int,
+ 'timestamp': _ParseDate,
+ 'created_at': _ParseDate,
+ 'closed_at': _ParseDate,
+ 'date': _ParseDate,
+ }
+ result = {}
+ for k, v in DomElement.attributes.items():
+ try:
+ result[k] = attribute_mapping[k](v)
+ except KeyError:
+ result[k] = v
+ return result
+
+
+def _DomGetTag(DomElement):
+ """
+ Returns the dictionnary of tags of a DomElement.
+ """
+ result = {}
+ for t in DomElement.getElementsByTagName("tag"):
+ k = t.attributes["k"].value
+ v = t.attributes["v"].value
+ result[k] = v
+ return result
+
+
+def _DomGetNd(DomElement):
+ """
+ Returns the list of nodes of a DomElement.
+ """
+ result = []
+ for t in DomElement.getElementsByTagName("nd"):
+ result.append(int(int(t.attributes["ref"].value)))
+ return result
+
+
+def _DomGetDiscussion(DomElement):
+ """
+ Returns the dictionnary of comments of a DomElement.
+ """
+ result = []
+ try:
+ discussion = DomElement.getElementsByTagName("discussion")[0]
+ for t in discussion.getElementsByTagName("comment"):
+ comment = _DomGetAttributes(t)
+ comment['text'] = xmlbuilder._GetXmlValue(t, "text")
+ result.append(comment)
+ except IndexError:
+ pass
+ return result
+
+
+def _DomGetComments(DomElement):
+ """
+ Returns the list of comments of a DomElement.
+ """
+ result = []
+ for t in DomElement.getElementsByTagName("comment"):
+ comment = {}
+ comment['date'] = _ParseDate(xmlbuilder._GetXmlValue(t, "date"))
+ comment['action'] = xmlbuilder._GetXmlValue(t, "action")
+ comment['text'] = xmlbuilder._GetXmlValue(t, "text")
+ comment['html'] = xmlbuilder._GetXmlValue(t, "html")
+ comment['uid'] = xmlbuilder._GetXmlValue(t, "uid")
+ comment['user'] = xmlbuilder._GetXmlValue(t, "user")
+ result.append(comment)
+ return result
+
+
+def _DomGetMember(DomElement):
+ """
+ Returns a list of relation members.
+ """
+ result = []
+ for m in DomElement.getElementsByTagName("member"):
+ result.append(_DomGetAttributes(m))
+ return result
+
+
+def _ParseDate(DateString):
+ date_formats = ["%Y-%m-%d %H:%M:%S UTC", "%Y-%m-%dT%H:%M:%SZ"]
+ for date_format in date_formats:
+ try:
+ result = datetime.strptime(DateString, date_format)
+ return result
+ except (ValueError, TypeError):
+ logger.debug(f"{DateString} does not match {date_format}")
+
+ return DateString
=====================================
osmapi/http.py
=====================================
@@ -0,0 +1,158 @@
+import time
+import logging
+import requests
+
+from . import errors
+
+
+logger = logging.getLogger(__name__)
+
+
+class OsmApiSession:
+
+ MAX_RETRY_LIMIT = 5
+ """Maximum retries if a call to the remote API fails (default: 5)"""
+
+ def __init__(self, base_url, created_by, auth=None, session=None):
+ self._api = base_url
+ self._created_by = created_by
+ self._auth = auth
+
+ self._http_session = session
+ self._session = self._get_http_session()
+
+ def close(self):
+ if self._session:
+ self._session.close()
+
+ def _http_request(self, method, path, auth, send, return_value=True): # noqa
+ """
+ Returns the response generated by an HTTP request.
+
+ `method` is a HTTP method to be executed
+ with the request data. For example: 'GET' or 'POST'.
+ `path` is the path to the requested resource relative to the
+ base API address stored in self._api. Should start with a
+ slash character to separate the URL.
+ `auth` is a boolean indicating whether authentication should
+ be preformed on this request.
+ `send` contains additional data that might be sent in a
+ request.
+ `return_value` indicates wheter this request should return
+ any data or not.
+
+ If the username or password is missing,
+ `OsmApi.UsernamePasswordMissingError` is raised.
+
+ If the requested element has been deleted,
+ `OsmApi.ElementDeletedApiError` is raised.
+
+ If the response status code indicates an error,
+ `OsmApi.ApiError` is raised.
+ """
+ msg = (
+ "%s %s %s"
+ % (time.strftime("%Y-%m-%d %H:%M:%S"), method, path)
+ )
+ logger.debug(msg)
+
+ # Add API base URL to path
+ path = self._api + path
+
+ if auth and not self._auth:
+ raise errors.UsernamePasswordMissingError("Username/Password missing")
+
+ response = self._session.request(
+ method,
+ path,
+ data=send
+ )
+ if response.status_code != 200:
+ payload = response.content.strip()
+ if response.status_code == 410:
+ raise errors.ElementDeletedApiError(
+ response.status_code,
+ response.reason,
+ payload
+ )
+ raise errors.ApiError(response.status_code, response.reason, payload)
+ if return_value and not response.content:
+ raise errors.ResponseEmptyApiError(
+ response.status_code,
+ response.reason,
+ ''
+ )
+
+ msg = (
+ "%s %s %s"
+ % (time.strftime("%Y-%m-%d %H:%M:%S"), method, path)
+ )
+ logger.debug(msg)
+ return response.content
+
+ def _http(self, cmd, path, auth, send, return_value=True): # noqa
+ i = 0
+ while True:
+ i += 1
+ try:
+ return self._http_request(
+ cmd,
+ path,
+ auth,
+ send,
+ return_value=return_value
+ )
+ except errors.ApiError as e:
+ if e.status >= 500:
+ if i == self.MAX_RETRY_LIMIT:
+ raise
+ if i != 1:
+ self._sleep()
+ self._session = self._get_http_session()
+ else:
+ raise
+ except Exception as e:
+ logger.error(e)
+ if i == self.MAX_RETRY_LIMIT:
+ if isinstance(e, errors.OsmApiError):
+ raise
+ raise errors.MaximumRetryLimitReachedError(
+ "Give up after %s retries" % i
+ )
+ if i != 1:
+ self._sleep()
+ self._session = self._get_http_session()
+
+ def _get_http_session(self):
+ """
+ Creates a requests session for connection pooling.
+ """
+ if self._http_session:
+ session = self._http_session
+ else:
+ session = requests.Session()
+
+ session.auth = self._auth
+ session.headers.update({
+ 'user-agent': self._created_by
+ })
+ return session
+
+ def _sleep(self):
+ time.sleep(5)
+
+ def _get(self, path):
+ return self._http('GET', path, False, None)
+
+ def _put(self, path, data, return_value=True):
+ return self._http('PUT', path, True, data, return_value=return_value)
+
+ def _post(self, path, data, optionalAuth=False, forceAuth=False):
+ # the Notes API allows certain POSTs by non-authenticated users
+ auth = (optionalAuth and self._auth)
+ if forceAuth:
+ auth = True
+ return self._http('POST', path, auth, data)
+
+ def _delete(self, path, data):
+ return self._http('DELETE', path, True, data)
=====================================
osmapi/parser.py
=====================================
@@ -0,0 +1,121 @@
+import xml.dom.minidom
+import xml.parsers.expat
+
+from . import errors
+from . import dom
+
+
+def ParseOsm(data):
+ """
+ Parse osm data.
+
+ Returns list of dict:
+
+ #!python
+ {
+ type: node|way|relation,
+ data: {}
+ }
+ """
+ try:
+ data = xml.dom.minidom.parseString(data)
+ data = data.getElementsByTagName("osm")[0]
+ except (xml.parsers.expat.ExpatError, IndexError) as e:
+ raise errors.XmlResponseInvalidError(
+ "The XML response from the OSM API is invalid: %r" % e
+ )
+
+ result = []
+ for elem in data.childNodes:
+ if elem.nodeName == "node":
+ result.append({
+ "type": elem.nodeName,
+ "data": dom.DomParseNode(elem)
+ })
+ elif elem.nodeName == "way":
+ result.append({
+ "type": elem.nodeName,
+ "data": dom.DomParseWay(elem)
+ })
+ elif elem.nodeName == "relation":
+ result.append({
+ "type": elem.nodeName,
+ "data": dom.DomParseRelation(elem)
+ })
+ return result
+
+
+def ParseOsc(data):
+ """
+ Parse osc data.
+
+ Returns list of dict:
+
+ #!python
+ {
+ type: node|way|relation,
+ action: create|delete|modify,
+ data: {}
+ }
+ """
+ try:
+ data = xml.dom.minidom.parseString(data)
+ data = data.getElementsByTagName("osmChange")[0]
+ except (xml.parsers.expat.ExpatError, IndexError) as e:
+ raise errors.XmlResponseInvalidError(
+ "The XML response from the OSM API is invalid: %r" % e
+ )
+
+ result = []
+ for action in data.childNodes:
+ if action.nodeName == "#text":
+ continue
+ for elem in action.childNodes:
+ if elem.nodeName == "node":
+ result.append({
+ "action": action.nodeName,
+ "type": elem.nodeName,
+ "data": dom.DomParseNode(elem)
+ })
+ elif elem.nodeName == "way":
+ result.append({
+ "action": action.nodeName,
+ "type": elem.nodeName,
+ "data": dom.DomParseWay(elem)
+ })
+ elif elem.nodeName == "relation":
+ result.append({
+ "action": action.nodeName,
+ "type": elem.nodeName,
+ "data": dom.DomParseRelation(elem)
+ })
+ return result
+
+
+def ParseNotes(data):
+ """
+ Parse notes data.
+
+ Returns a list of dict:
+
+ #!python
+ [
+ {
+ 'id': integer,
+ 'action': opened|commented|closed,
+ 'status': open|closed
+ 'date_created': creation date
+ 'date_closed': closing data|None
+ 'uid': User ID|None
+ 'user': User name|None
+ 'comments': {}
+ },
+ { ... }
+ ]
+ """
+ noteElements = dom.OsmResponseToDom(data, tag="note")
+ result = []
+ for noteElement in noteElements:
+ note = dom.DomParseNote(noteElement)
+ result.append(note)
+ return result
=====================================
osmapi/xmlbuilder.py
=====================================
@@ -0,0 +1,64 @@
+def _XmlBuild(ElementType, ElementData, WithHeaders=True, data=None): # noqa
+ xml = ""
+ if WithHeaders:
+ xml += "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
+ xml += "<osm version=\"0.6\" generator=\""
+ xml += data._created_by + "\">\n"
+
+ # <element attr="val">
+ xml += " <" + ElementType
+ if "id" in ElementData:
+ xml += " id=\"" + str(ElementData["id"]) + "\""
+ if "lat" in ElementData:
+ xml += " lat=\"" + str(ElementData["lat"]) + "\""
+ if "lon" in ElementData:
+ xml += " lon=\"" + str(ElementData["lon"]) + "\""
+ if "version" in ElementData:
+ xml += " version=\"" + str(ElementData["version"]) + "\""
+ visible_str = str(ElementData.get("visible", True)).lower()
+ xml += " visible=\"" + visible_str + "\""
+ if ElementType in ["node", "way", "relation"]:
+ xml += " changeset=\"" + str(data._CurrentChangesetId) + "\""
+ xml += ">\n"
+
+ # <tag... />
+ for k, v in ElementData.get("tag", {}).items():
+ xml += " <tag k=\"" + _XmlEncode(k)
+ xml += "\" v=\"" + _XmlEncode(v) + "\"/>\n"
+
+ # <member... />
+ for member in ElementData.get("member", []):
+ xml += " <member type=\"" + member["type"]
+ xml += "\" ref=\"" + str(member["ref"])
+ xml += "\" role=\"" + _XmlEncode(member["role"])
+ xml += "\"/>\n"
+
+ # <nd... />
+ for ref in ElementData.get("nd", []):
+ xml += " <nd ref=\"" + str(ref) + "\"/>\n"
+
+ # </element>
+ xml += " </" + ElementType + ">\n"
+
+ if WithHeaders:
+ xml += "</osm>\n"
+
+ return xml.encode("utf8")
+
+
+def _XmlEncode(text):
+ return (
+ text
+ .replace("&", "&")
+ .replace("\"", """)
+ .replace("<", "<")
+ .replace(">", ">")
+ )
+
+
+def _GetXmlValue(DomElement, tag):
+ try:
+ elem = DomElement.getElementsByTagName(tag)[0]
+ return elem.firstChild.nodeValue
+ except Exception:
+ return None
=====================================
requirements.txt
=====================================
@@ -1,3 +1,4 @@
pdoc==8.0.1
Pygments==2.10.0
requests==2.26.0
+python-dotenv
=====================================
setup.py
=====================================
@@ -19,6 +19,7 @@ setup(
packages=find_packages(),
version=version,
install_requires=['requests'],
+ python_requires='>=3.7',
description='Python wrapper for the OSM API',
long_description=long_description,
long_description_content_type='text/markdown',
@@ -40,5 +41,6 @@ setup(
'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.9',
+ 'Programming Language :: Python :: 3.10',
],
)
=====================================
test-requirements.txt
=====================================
@@ -4,4 +4,5 @@ virtualenv
xmltodict
pytest
pytest-cov
+responses
coverage
=====================================
tests/changeset_test.py
=====================================
@@ -1,786 +1,721 @@
-from __future__ import (unicode_literals, absolute_import)
-from . import osmapi_test
import osmapi
-import mock
import xmltodict
import datetime
-try:
- import urlparse
-except Exception:
- import urllib
- urlparse = urllib.parse
-
-
-def recursive_sort(col): # noqa
- """
- Function to recursive sort a collection
- that might contain lists, dicts etc.
- In Python 3.x a list of dicts is sorted by it's hash
- """
- if hasattr(col, '__iter__'):
- if isinstance(col, list):
- try:
- col = sorted(col)
- except TypeError: # in Python 3.x: lists of dicts are not sortable
- col = sorted(col, key=lambda k: hash(frozenset(k.items())))
- except Exception:
- pass
-
- for idx, elem in enumerate(col):
- col[idx] = recursive_sort(elem)
- elif isinstance(col, dict):
- for elem in col:
- try:
- col[elem] = recursive_sort(col[elem])
- except IndexError:
- pass
- return col
+import pytest
+from responses import GET, PUT, POST
def xmltosorteddict(xml):
xml_dict = xmltodict.parse(xml, dict_constructor=dict)
- return recursive_sort(xml_dict)
-
-
-class TestOsmApiChangeset(osmapi_test.TestOsmApi):
- def test_ChangesetGet(self):
- self._session_mock()
-
- result = self.api.ChangesetGet(123)
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'GET')
- self.assertEqual(args[1], self.api_base + '/api/0.6/changeset/123')
-
- self.assertEqual(result, {
- 'id': 123,
- 'closed_at': datetime.datetime(2009, 9, 7, 22, 57, 37),
- 'created_at': datetime.datetime(2009, 9, 7, 21, 57, 36),
- 'discussion': [],
- 'max_lat': '52.4710193',
- 'max_lon': '-1.4831815',
- 'min_lat': '45.9667901',
- 'min_lon': '-1.4998534',
- 'open': False,
- 'user': 'randomjunk',
- 'uid': 3,
- 'tag': {
- 'comment': 'correct node bug',
- 'created_by': 'Potlatch 1.2a',
- },
- })
-
- def test_ChangesetUpdate(self):
- self._session_mock(auth=True)
-
- # setup mock
- self.api.ChangesetCreate = mock.Mock(
- return_value=4444
- )
- self.api._CurrentChangesetId = 4444
-
- result = self.api.ChangesetUpdate(
- {
- 'test': 'foobar'
- }
- )
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'PUT')
- self.assertEqual(args[1], self.api_base + '/api/0.6/changeset/4444')
- self.assertEqual(
- xmltosorteddict(kwargs['data']),
- xmltosorteddict(
- b'<?xml version="1.0" encoding="UTF-8"?>\n'
- b'<osm version="0.6" generator="osmapi/2.0.2">\n'
- b' <changeset visible="true">\n'
- b' <tag k="test" v="foobar"/>\n'
- b' <tag k="created_by" v="osmapi/2.0.2"/>\n'
- b' </changeset>\n'
- b'</osm>\n'
- )
- )
- self.assertEqual(result, 4444)
-
- def test_ChangesetUpdate_with_created_by(self):
- self._session_mock(auth=True)
-
- # setup mock
- self.api.ChangesetCreate = mock.Mock(
- return_value=4444
- )
- self.api._CurrentChangesetId = 4444
-
- result = self.api.ChangesetUpdate(
- {
- 'test': 'foobar',
- 'created_by': 'MyTestOSMApp'
- }
- )
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'PUT')
- self.assertEqual(args[1], self.api_base + '/api/0.6/changeset/4444')
- self.assertEqual(
- xmltosorteddict(kwargs['data']),
- xmltosorteddict(
- b'<?xml version="1.0" encoding="UTF-8"?>\n'
- b'<osm version="0.6" generator="osmapi/2.0.2">\n'
- b' <changeset visible="true">\n'
- b' <tag k="test" v="foobar"/>\n'
- b' <tag k="created_by" v="MyTestOSMApp"/>\n'
- b' </changeset>\n'
- b'</osm>\n'
- )
- )
- self.assertEqual(result, 4444)
-
- def test_ChangesetUpdate_wo_changeset(self):
- self._session_mock()
-
- with self.assertRaisesRegex(
- osmapi.NoChangesetOpenError,
- 'No changeset currently opened'):
- self.api.ChangesetUpdate(
- {
- 'test': 'foobar'
+ return xml_dict
+
+
+def test_Changeset_contextmanager(auth_api, add_response):
+ # Setup mock
+ resp = add_response(PUT, '/changeset/create', filename='test_Changeset_create.xml')
+ resp = add_response(PUT, '/node/create', filename='test_Changeset_create_node.xml')
+ resp = add_response(PUT, '/changeset/1414/close', filename='test_Changeset_close.xml')
+
+ test_node = {
+ 'lat': 47.123,
+ 'lon': 8.555,
+ 'tag': {
+ 'amenity': 'place_of_worship',
+ 'religion': 'pastafarian'
+ }
+ }
+
+ # use context manager
+ with auth_api.Changeset() as changeset_id:
+ assert changeset_id == 1414
+
+ # add test node
+ node = auth_api.NodeCreate(test_node)
+ assert node['id'] == 7272
+
+ # check requests
+ assert len(resp.calls) == 3
+
+
+def test_ChangesetGet(api, add_response):
+ # Setup mock
+ add_response(GET, '/changeset/123')
+
+ # Call
+ result = api.ChangesetGet(123)
+
+ test_changeset = {
+ 'id': 123,
+ 'closed_at': datetime.datetime(2009, 9, 7, 22, 57, 37),
+ 'created_at': datetime.datetime(2009, 9, 7, 21, 57, 36),
+ 'discussion': [],
+ 'max_lat': '52.4710193',
+ 'max_lon': '-1.4831815',
+ 'min_lat': '45.9667901',
+ 'min_lon': '-1.4998534',
+ 'open': False,
+ 'user': 'randomjunk',
+ 'uid': 3,
+ 'tag': {
+ 'comment': 'correct node bug',
+ 'created_by': 'Potlatch 1.2a',
+ },
+ }
+ assert result == test_changeset
+
+
+def test_ChangesetUpdate(auth_api, add_response):
+ # Setup mock
+ resp = add_response(PUT, '/changeset/create', filename='test_ChangesetCreate.xml')
+ resp = add_response(PUT, '/changeset/4321', filename='test_ChangesetUpdate.xml')
+
+ # Call
+ result = auth_api.ChangesetCreate()
+ assert result == 4321
+
+ result = auth_api.ChangesetUpdate({'test': 'foobar'})
+ changeset_xml = xmltosorteddict(
+ b'<?xml version="1.0" encoding="UTF-8"?>\n'
+ b'<osm version="0.6" generator="osmapi/3.0.0">\n'
+ b' <changeset visible="true">\n'
+ b' <tag k="test" v="foobar"/>\n'
+ b' <tag k="created_by" v="osmapi/3.0.0"/>\n'
+ b' </changeset>\n'
+ b'</osm>\n'
+ )
+ assert xmltosorteddict(resp.calls[1].request.body) == changeset_xml
+ assert result == 4321
+
+
+def test_ChangesetUpdate_with_created_by(auth_api, add_response):
+ # Setup mock
+ resp = add_response(PUT, '/changeset/create', filename='test_ChangesetCreate.xml')
+ resp = add_response(PUT, '/changeset/4321', filename='test_ChangesetUpdate.xml')
+
+ # Call
+ result = auth_api.ChangesetCreate()
+ assert result == 4321
+
+ result = auth_api.ChangesetUpdate(
+ {
+ 'test': 'foobar',
+ 'created_by': 'MyTestOSMApp'
+ }
+ )
+ changeset_xml = xmltosorteddict(
+ b'<?xml version="1.0" encoding="UTF-8"?>\n'
+ b'<osm version="0.6" generator="osmapi/3.0.0">\n'
+ b' <changeset visible="true">\n'
+ b' <tag k="test" v="foobar"/>\n'
+ b' <tag k="created_by" v="MyTestOSMApp"/>\n'
+ b' </changeset>\n'
+ b'</osm>\n'
+ )
+ assert xmltosorteddict(resp.calls[1].request.body) == changeset_xml
+ assert result == 4321
+
+
+def test_ChangesetUpdate_wo_changeset(auth_api):
+ with pytest.raises(osmapi.NoChangesetOpenError) as execinfo:
+ auth_api.ChangesetUpdate({'test': 'foobar'})
+ assert str(execinfo.value) == 'No changeset currently opened'
+
+
+def test_ChangesetCreate(auth_api, add_response):
+ resp = add_response(PUT, '/changeset/create')
+ result = auth_api.ChangesetCreate(
+ {
+ 'foobar': 'A new test changeset'
+ }
+ )
+ assert result == 4321
+
+ changeset_xml = xmltosorteddict(
+ b'<?xml version="1.0" encoding="UTF-8"?>\n'
+ b'<osm version="0.6" generator="osmapi/3.0.0">\n'
+ b' <changeset visible="true">\n'
+ b' <tag k="foobar" v="A new test changeset"/>\n'
+ b' <tag k="created_by" v="osmapi/3.0.0"/>\n'
+ b' </changeset>\n'
+ b'</osm>\n'
+ )
+ assert xmltosorteddict(resp.calls[0].request.body) == changeset_xml
+
+
+def test_ChangesetCreate_with_created_by(auth_api, add_response):
+ resp = add_response(PUT, '/changeset/create')
+
+ result = auth_api.ChangesetCreate(
+ {
+ 'foobar': 'A new test changeset',
+ 'created_by': 'CoolTestApp',
+ }
+ )
+ assert result == 1234
+
+ changeset_xml = xmltosorteddict(
+ b'<?xml version="1.0" encoding="UTF-8"?>\n'
+ b'<osm version="0.6" generator="osmapi/3.0.0">\n'
+ b' <changeset visible="true">\n'
+ b' <tag k="foobar" v="A new test changeset"/>\n'
+ b' <tag k="created_by" v="CoolTestApp"/>\n'
+ b' </changeset>\n'
+ b'</osm>\n'
+ )
+ assert xmltosorteddict(resp.calls[0].request.body) == changeset_xml
+
+
+def test_ChangesetCreate_with_open_changeset(auth_api, add_response):
+ add_response(PUT, '/changeset/create')
+
+ auth_api.ChangesetCreate(
+ {
+ 'test': 'an already open changeset',
+ }
+ )
+
+ with pytest.raises(osmapi.ChangesetAlreadyOpenError) as execinfo:
+ auth_api.ChangesetCreate({'test': 'foobar'})
+ assert str(execinfo.value) == 'Changeset already opened'
+
+
+def test_ChangesetClose(auth_api, add_response):
+ # setup mock
+ resp = add_response(PUT, '/changeset/create', filename='test_Changeset_create.xml')
+ resp = add_response(PUT, '/changeset/1414/close')
+
+ # Call
+ auth_api.ChangesetCreate()
+ auth_api.ChangesetClose()
+
+ assert '/api/0.6/changeset/1414/close' in resp.calls[1].request.url
+
+
+def test_ChangesetClose_with_no_changeset(auth_api):
+ with pytest.raises(osmapi.NoChangesetOpenError) as execinfo:
+ auth_api.ChangesetClose()
+ assert str(execinfo.value) == 'No changeset currently opened'
+
+
+def test_ChangesetUpload_create_node(auth_api, add_response):
+ # Setup
+ resp = add_response(PUT, '/changeset/create', body='4444')
+ resp = add_response(POST, '/changeset/4444/upload')
+
+ changesdata = [
+ {
+ 'type': 'node',
+ 'action': 'create',
+ 'data': {
+ 'lat': 47.123,
+ 'lon': 8.555,
+ 'tag': {
+ 'amenity': 'place_of_worship',
+ 'religion': 'pastafarian'
}
- )
-
- def test_ChangesetCreate(self):
- self._session_mock(auth=True)
-
- result = self.api.ChangesetCreate(
- {
- 'foobar': 'A new test changeset'
- }
- )
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'PUT')
- self.assertEqual(args[1], self.api_base + '/api/0.6/changeset/create')
- self.assertEqual(
- xmltosorteddict(kwargs['data']),
- xmltosorteddict(
- b'<?xml version="1.0" encoding="UTF-8"?>\n'
- b'<osm version="0.6" generator="osmapi/2.0.2">\n'
- b' <changeset visible="true">\n'
- b' <tag k="foobar" v="A new test changeset"/>\n'
- b' <tag k="created_by" v="osmapi/2.0.2"/>\n'
- b' </changeset>\n'
- b'</osm>\n'
- )
- )
- self.assertEqual(result, 4321)
-
- def test_ChangesetCreate_with_created_by(self):
- self._session_mock(auth=True)
-
- result = self.api.ChangesetCreate(
- {
- 'foobar': 'A new test changeset',
- 'created_by': 'CoolTestApp',
- }
- )
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'PUT')
- self.assertEqual(args[1], self.api_base + '/api/0.6/changeset/create')
- self.assertEqual(
- xmltosorteddict(kwargs['data']),
- xmltosorteddict(
- b'<?xml version="1.0" encoding="UTF-8"?>\n'
- b'<osm version="0.6" generator="osmapi/2.0.2">\n'
- b' <changeset visible="true">\n'
- b' <tag k="foobar" v="A new test changeset"/>\n'
- b' <tag k="created_by" v="CoolTestApp"/>\n'
- b' </changeset>\n'
- b'</osm>\n'
- )
- )
- self.assertEqual(result, 1234)
-
- def test_ChangesetCreate_with_open_changeset(self):
- self._session_mock(auth=True)
-
- self.api.ChangesetCreate(
- {
- 'test': 'an already open changeset',
}
- )
-
- with self.assertRaisesRegex(
- osmapi.ChangesetAlreadyOpenError,
- 'Changeset already opened'):
- self.api.ChangesetCreate(
- {
- 'test': 'foobar'
- }
- )
-
- def test_ChangesetClose(self):
- self._session_mock(auth=True)
-
- # setup mock
- self.api.ChangesetCreate = mock.Mock(
- return_value=4444
- )
- self.api._CurrentChangesetId = 4444
-
- self.api.ChangesetClose()
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'PUT')
- self.assertEqual(
- args[1],
- f'{self.api_base}/api/0.6/changeset/4444/close'
- )
-
- def test_ChangesetClose_with_no_changeset(self):
- self._session_mock()
-
- with self.assertRaisesRegex(
- osmapi.NoChangesetOpenError,
- 'No changeset currently opened'):
- self.api.ChangesetClose()
-
- def test_ChangesetUpload_create_node(self):
- self._session_mock(auth=True)
-
- # setup mock
- self.api.ChangesetCreate = mock.Mock(
- return_value=4444
- )
- self.api._CurrentChangesetId = 4444
-
- changesdata = [
- {
- 'type': 'node',
- 'action': 'create',
- 'data': {
- 'lat': 47.123,
- 'lon': 8.555,
- 'tag': {
- 'amenity': 'place_of_worship',
- 'religion': 'pastafarian'
- }
+ }
+ ]
+
+ upload_xml = xmltosorteddict(
+ b'<?xml version="1.0" encoding="UTF-8"?>\n'
+ b'<osmChange version="0.6" generator="osmapi/3.0.0">\n'
+ b'<create>\n'
+ b' <node lat="47.123" lon="8.555" visible="true" '
+ b'changeset="4444">\n'
+ b' <tag k="amenity" v="place_of_worship"/>\n'
+ b' <tag k="religion" v="pastafarian"/>\n'
+ b' </node>\n'
+ b'</create>\n'
+ b'</osmChange>'
+ )
+
+ # Call
+ auth_api.ChangesetCreate()
+ result = auth_api.ChangesetUpload(changesdata)
+
+ # Assert
+ assert xmltosorteddict(resp.calls[1].request.body) == upload_xml
+ assert result[0]['type'] == changesdata[0]['type']
+ assert result[0]['action'] == changesdata[0]['action']
+
+ data = result[0]['data']
+ assert data['lat'] == changesdata[0]['data']['lat']
+ assert data['lon'] == changesdata[0]['data']['lon']
+ assert data['tag'] == changesdata[0]['data']['tag']
+ assert data['id'] == 4295832900
+ assert result[0]['data']['version'] == 1
+
+
+def test_ChangesetUpload_modify_way(auth_api, add_response):
+ # setup mock
+ resp = add_response(PUT, '/changeset/create', body='4444')
+ resp = add_response(POST, '/changeset/4444/upload')
+
+ changesdata = [
+ {
+ 'type': 'way',
+ 'action': 'modify',
+ 'data': {
+ 'id': 4294967296,
+ 'version': 2,
+ 'nd': [
+ 4295832773,
+ 4295832773,
+ 4294967304,
+ 4294967303,
+ 4294967300,
+ 4608751,
+ 4294967305,
+ 4294967302,
+ 8548430,
+ 4294967296,
+ 4294967301,
+ 4294967298,
+ 4294967306,
+ 7855737,
+ 4294967297,
+ 4294967299
+ ],
+ 'tag': {
+ 'highway': 'secondary',
+ 'name': 'Stansted Road'
}
}
- ]
-
- result = self.api.ChangesetUpload(changesdata)
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'POST')
- self.assertEqual(
- args[1],
- f'{self.api_base}/api/0.6/changeset/4444/upload'
- )
- self.assertEqual(
- xmltosorteddict(kwargs['data']),
- xmltosorteddict(
- b'<?xml version="1.0" encoding="UTF-8"?>\n'
- b'<osmChange version="0.6" generator="osmapi/2.0.2">\n'
- b'<create>\n'
- b' <node lat="47.123" lon="8.555" visible="true" '
- b'changeset="4444">\n'
- b' <tag k="religion" v="pastafarian"/>\n'
- b' <tag k="amenity" v="place_of_worship"/>\n'
- b' </node>\n'
- b'</create>\n'
- b'</osmChange>'
- )
- )
-
- self.assertEqual(result[0]['type'], changesdata[0]['type'])
- self.assertEqual(result[0]['action'], changesdata[0]['action'])
-
- data = result[0]['data']
- self.assertEqual(data['lat'], changesdata[0]['data']['lat'])
- self.assertEqual(data['lon'], changesdata[0]['data']['lon'])
- self.assertEqual(data['tag'], changesdata[0]['data']['tag'])
- self.assertEqual(data['id'], 4295832900)
- self.assertEqual(result[0]['data']['version'], 1)
-
- def test_ChangesetUpload_modify_way(self):
- self._session_mock(auth=True)
-
- # setup mock
- self.api.ChangesetCreate = mock.Mock(
- return_value=4444
- )
- self.api._CurrentChangesetId = 4444
-
- changesdata = [
- {
- 'type': 'way',
- 'action': 'modify',
- 'data': {
- 'id': 4294967296,
- 'version': 2,
- 'nd': [
- 4295832773,
- 4295832773,
- 4294967304,
- 4294967303,
- 4294967300,
- 4608751,
- 4294967305,
- 4294967302,
- 8548430,
- 4294967296,
- 4294967301,
- 4294967298,
- 4294967306,
- 7855737,
- 4294967297,
- 4294967299
- ],
- 'tag': {
- 'highway': 'secondary',
- 'name': 'Stansted Road'
- }
+ }
+ ]
+
+ upload_xml = xmltosorteddict(
+ b'<?xml version="1.0" encoding="UTF-8"?>\n'
+ b'<osmChange version="0.6" generator="osmapi/3.0.0">\n'
+ b'<modify>\n'
+ b' <way id="4294967296" version="2" visible="true" '
+ b'changeset="4444">\n'
+ b' <tag k="highway" v="secondary"/>\n'
+ b' <tag k="name" v="Stansted Road"/>\n'
+ b' <nd ref="4295832773"/>\n'
+ b' <nd ref="4295832773"/>\n'
+ b' <nd ref="4294967304"/>\n'
+ b' <nd ref="4294967303"/>\n'
+ b' <nd ref="4294967300"/>\n'
+ b' <nd ref="4608751"/>\n'
+ b' <nd ref="4294967305"/>\n'
+ b' <nd ref="4294967302"/>\n'
+ b' <nd ref="8548430"/>\n'
+ b' <nd ref="4294967296"/>\n'
+ b' <nd ref="4294967301"/>\n'
+ b' <nd ref="4294967298"/>\n'
+ b' <nd ref="4294967306"/>\n'
+ b' <nd ref="7855737"/>\n'
+ b' <nd ref="4294967297"/>\n'
+ b' <nd ref="4294967299"/>\n'
+ b' </way>\n'
+ b'</modify>\n'
+ b'</osmChange>'
+ )
+
+ # Call
+ auth_api.ChangesetCreate()
+ result = auth_api.ChangesetUpload(changesdata)
+
+ # Assert
+ assert xmltosorteddict(resp.calls[1].request.body) == upload_xml
+
+ assert result[0]['type'] == changesdata[0]['type']
+ assert result[0]['action'] == changesdata[0]['action']
+
+ data = result[0]['data']
+ assert data['nd'] == changesdata[0]['data']['nd']
+ assert data['tag'] == changesdata[0]['data']['tag']
+ assert data['id'] == 4294967296
+ assert data['version'] == 3
+
+
+def test_ChangesetUpload_delete_relation(auth_api, add_response):
+ # setup mock
+ resp = add_response(PUT, '/changeset/create', body='4444')
+ resp = add_response(POST, '/changeset/4444/upload')
+
+ changesdata = [
+ {
+ 'type': 'relation',
+ 'action': 'delete',
+ 'data': {
+ 'id': 676,
+ 'version': 2,
+ 'member': [
+ {
+ 'ref': 4799,
+ 'role': 'outer',
+ 'type': 'way'
+ },
+ {
+ 'ref': 9391,
+ 'role': 'outer',
+ 'type': 'way'
+ },
+ ],
+ 'tag': {
+ 'admin_level': '9',
+ 'boundary': 'administrative',
+ 'type': 'multipolygon'
}
}
- ]
-
- result = self.api.ChangesetUpload(changesdata)
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'POST')
- self.assertEqual(
- args[1],
- f'{self.api_base}/api/0.6/changeset/4444/upload'
- )
- self.assertEqual(
- xmltosorteddict(kwargs['data']),
- xmltosorteddict(
- b'<?xml version="1.0" encoding="UTF-8"?>\n'
- b'<osmChange version="0.6" generator="osmapi/2.0.2">\n'
- b'<modify>\n'
- b' <way id="4294967296" version="2" visible="true" '
- b'changeset="4444">\n'
- b' <tag k="name" v="Stansted Road"/>\n'
- b' <tag k="highway" v="secondary"/>\n'
- b' <nd ref="4295832773"/>\n'
- b' <nd ref="4295832773"/>\n'
- b' <nd ref="4294967304"/>\n'
- b' <nd ref="4294967303"/>\n'
- b' <nd ref="4294967300"/>\n'
- b' <nd ref="4608751"/>\n'
- b' <nd ref="4294967305"/>\n'
- b' <nd ref="4294967302"/>\n'
- b' <nd ref="8548430"/>\n'
- b' <nd ref="4294967296"/>\n'
- b' <nd ref="4294967301"/>\n'
- b' <nd ref="4294967298"/>\n'
- b' <nd ref="4294967306"/>\n'
- b' <nd ref="7855737"/>\n'
- b' <nd ref="4294967297"/>\n'
- b' <nd ref="4294967299"/>\n'
- b' </way>\n'
- b'</modify>\n'
- b'</osmChange>'
- )
- )
-
- self.assertEqual(result[0]['type'], changesdata[0]['type'])
- self.assertEqual(result[0]['action'], changesdata[0]['action'])
-
- data = result[0]['data']
- self.assertEqual(data['nd'], changesdata[0]['data']['nd'])
- self.assertEqual(data['tag'], changesdata[0]['data']['tag'])
- self.assertEqual(data['id'], 4294967296)
- self.assertEqual(data['version'], 3)
-
- def test_ChangesetUpload_delete_relation(self):
- self._session_mock(auth=True)
-
- # setup mock
- self.api.ChangesetCreate = mock.Mock(
- return_value=4444
- )
- self.api._CurrentChangesetId = 4444
-
- changesdata = [
- {
- 'type': 'relation',
- 'action': 'delete',
- 'data': {
- 'id': 676,
- 'version': 2,
- 'member': [
- {
- 'ref': 4799,
- 'role': 'outer',
- 'type': 'way'
- },
- {
- 'ref': 9391,
- 'role': 'outer',
- 'type': 'way'
- },
- ],
- 'tag': {
- 'admin_level': '9',
- 'boundary': 'administrative',
- 'type': 'multipolygon'
- }
+ }
+ ]
+
+ upload_xml = xmltosorteddict(
+ b'<?xml version="1.0" encoding="UTF-8"?>\n'
+ b'<osmChange version="0.6" generator="osmapi/3.0.0">\n'
+ b'<delete>\n'
+ b' <relation id="676" version="2" visible="true" '
+ b'changeset="4444">\n'
+ b' <tag k="admin_level" v="9"/>\n'
+ b' <tag k="boundary" v="administrative"/>\n'
+ b' <tag k="type" v="multipolygon"/>\n'
+ b' <member type="way" ref="4799" role="outer"/>\n'
+ b' <member type="way" ref="9391" role="outer"/>\n'
+ b' </relation>\n'
+ b'</delete>\n'
+ b'</osmChange>'
+ )
+
+ # Call
+ auth_api.ChangesetCreate()
+ result = auth_api.ChangesetUpload(changesdata)
+
+ # Assert
+ assert xmltosorteddict(resp.calls[1].request.body) == upload_xml
+ assert result[0]['type'] == changesdata[0]['type']
+ assert result[0]['action'] == changesdata[0]['action']
+
+ data = result[0]['data']
+ assert data['member'], changesdata[0]['data']['member']
+ assert data['tag'] == changesdata[0]['data']['tag']
+ assert data['id'] == 676
+ assert 'version' not in data
+
+
+def test_ChangesetUpload_invalid_response(auth_api, add_response):
+ # setup mock
+ add_response(PUT, '/changeset/create', body='4444')
+ add_response(POST, '/changeset/4444/upload', body='4444')
+
+ changesdata = [
+ {
+ 'type': 'relation',
+ 'action': 'delete',
+ 'data': {
+ 'id': 676,
+ 'version': 2,
+ 'member': [
+ {
+ 'ref': 4799,
+ 'role': 'outer',
+ 'type': 'way'
+ },
+ {
+ 'ref': 9391,
+ 'role': 'outer',
+ 'type': 'way'
+ },
+ ],
+ 'tag': {
+ 'admin_level': '9',
+ 'boundary': 'administrative',
+ 'type': 'multipolygon'
}
}
- ]
-
- result = self.api.ChangesetUpload(changesdata)
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'POST')
- self.assertEqual(
- args[1],
- f'{self.api_base}/api/0.6/changeset/4444/upload'
- )
- self.assertEqual(
- xmltosorteddict(kwargs['data']),
- xmltosorteddict(
- b'<?xml version="1.0" encoding="UTF-8"?>\n'
- b'<osmChange version="0.6" generator="osmapi/2.0.2">\n'
- b'<delete>\n'
- b' <relation id="676" version="2" visible="true" '
- b'changeset="4444">\n'
- b' <tag k="admin_level" v="9"/>\n'
- b' <tag k="boundary" v="administrative"/>\n'
- b' <tag k="type" v="multipolygon"/>\n'
- b' <member type="way" ref="4799" role="outer"/>\n'
- b' <member type="way" ref="9391" role="outer"/>\n'
- b' </relation>\n'
- b'</delete>\n'
- b'</osmChange>'
- )
- )
-
- self.assertEqual(result[0]['type'], changesdata[0]['type'])
- self.assertEqual(result[0]['action'], changesdata[0]['action'])
-
- data = result[0]['data']
- self.assertEqual(data['member'], changesdata[0]['data']['member'])
- self.assertEqual(data['tag'], changesdata[0]['data']['tag'])
- self.assertEqual(data['id'], 676)
- self.assertNotIn('version', data)
-
- def test_ChangesetUpload_invalid_response(self):
- self._session_mock(auth=True)
-
- # setup mock
- self.api.ChangesetCreate = mock.Mock(
- return_value=4444
- )
- self.api._CurrentChangesetId = 4444
-
- changesdata = [
- {
- 'type': 'relation',
- 'action': 'delete',
- 'data': {
- 'id': 676,
- 'version': 2,
- 'member': [
- {
- 'ref': 4799,
- 'role': 'outer',
- 'type': 'way'
- },
- {
- 'ref': 9391,
- 'role': 'outer',
- 'type': 'way'
- },
- ],
- 'tag': {
- 'admin_level': '9',
- 'boundary': 'administrative',
- 'type': 'multipolygon'
- }
+ }
+ ]
+
+ # Call + assert
+ auth_api.ChangesetCreate()
+ with pytest.raises(osmapi.XmlResponseInvalidError) as execinfo:
+ auth_api.ChangesetUpload(changesdata)
+ assert 'The XML response from the OSM API is invalid' in str(execinfo.value)
+
+
+def test_ChangesetUpload_no_auth(api):
+ changesdata = [
+ {
+ 'type': 'node',
+ 'action': 'create',
+ 'data': {
+ 'lat': 47.123,
+ 'lon': 8.555,
+ 'tag': {
+ 'amenity': 'place_of_worship',
+ 'religion': 'pastafarian'
}
}
- ]
-
- with self.assertRaises(osmapi.XmlResponseInvalidError):
- self.api.ChangesetUpload(changesdata)
-
- def test_ChangesetDownload(self):
- self._session_mock()
-
- result = self.api.ChangesetDownload(23123)
-
- args, _ = self.api._session.request.call_args
- self.assertEqual(args[0], 'GET')
- self.assertEqual(
- args[1],
- f'{self.api_base}/api/0.6/changeset/23123/download'
- )
-
- self.assertEqual(len(result), 16)
- self.assertEqual(
- result[1],
- {
- 'action': 'create',
- 'type': 'node',
- 'data': {
- 'changeset': 23123,
- 'id': 4295668171,
- 'lat': 46.4909781,
- 'lon': 11.2743295,
- 'tag': {
- 'highway': 'traffic_signals'
- },
- 'timestamp': datetime.datetime(2013, 5, 14, 10, 33, 4),
- 'uid': 1178,
- 'user': 'tyrTester06',
- 'version': 1,
- 'visible': True
- }
+ }
+ ]
+
+ with pytest.raises(osmapi.UsernamePasswordMissingError) as execinfo:
+ api.ChangesetUpload(changesdata)
+ assert str(execinfo.value) == "Username/Password missing"
+
+
+def test_ChangesetDownload(api, add_response):
+ # Setup mock
+ add_response(GET, '/changeset/23123/download')
+
+ # Call
+ result = api.ChangesetDownload(23123)
+
+ # Assertion
+ assert len(result) == 16
+ assert result[1] == (
+ {
+ 'action': 'create',
+ 'type': 'node',
+ 'data': {
+ 'changeset': 23123,
+ 'id': 4295668171,
+ 'lat': 46.4909781,
+ 'lon': 11.2743295,
+ 'tag': {
+ 'highway': 'traffic_signals'
+ },
+ 'timestamp': datetime.datetime(2013, 5, 14, 10, 33, 4),
+ 'uid': 1178,
+ 'user': 'tyrTester06',
+ 'version': 1,
+ 'visible': True
}
- )
-
- def test_ChangesetDownload_invalid_response(self):
- self._session_mock()
- with self.assertRaises(osmapi.XmlResponseInvalidError):
- self.api.ChangesetDownload(23123)
-
- def test_ChangesetDownloadContainingUnicode(self):
- self._session_mock()
-
- # This changeset contains unicode tag values
- # Note that the fixture data has been reduced from the
- # original from openstreetmap.org
- result = self.api.ChangesetDownload(37393499)
-
- self.assertEqual(len(result), 2)
- self.assertEqual(
- result[1],
- {
- 'action': 'create',
- 'type': 'way',
- 'data': {
- 'changeset': 37393499,
- 'id': 399491497,
- 'nd': [4022271571, 4022271567, 4022271565],
- 'tag': {'highway': 'service',
- # UTF-8 encoded 'LATIN SMALL LETTER O WITH STROKE'
- # Aka. 0xf8 in latin-1/ISO 8859-1
- 'name': b'S\xc3\xb8nderskovvej'.decode('utf-8'),
- 'service': 'driveway'},
- 'timestamp': datetime.datetime(2016, 2, 23, 16, 55, 35),
- 'uid': 328556,
- 'user': 'InternationalUser',
- 'version': 1,
- 'visible': True
- }
+ }
+ )
+
+
+def test_ChangesetDownload_invalid_response(api, add_response):
+ add_response(GET, '/changeset/23123/download')
+ with pytest.raises(osmapi.XmlResponseInvalidError) as execinfo:
+ api.ChangesetDownload(23123)
+ assert 'The XML response from the OSM API is invalid' in str(execinfo.value)
+
+
+def test_ChangesetDownloadContainingUnicode(api, add_response):
+ add_response(GET, '/changeset/37393499/download')
+
+ # This changeset contains unicode tag values
+ # Note that the fixture data has been reduced from the
+ # original from openstreetmap.org
+ result = api.ChangesetDownload(37393499)
+
+ assert len(result) == 2
+ assert result[1] == (
+ {
+ 'action': 'create',
+ 'type': 'way',
+ 'data': {
+ 'changeset': 37393499,
+ 'id': 399491497,
+ 'nd': [4022271571, 4022271567, 4022271565],
+ 'tag': {'highway': 'service',
+ # UTF-8 encoded 'LATIN SMALL LETTER O WITH STROKE'
+ # Aka. 0xf8 in latin-1/ISO 8859-1
+ 'name': b'S\xc3\xb8nderskovvej'.decode('utf-8'),
+ 'service': 'driveway'},
+ 'timestamp': datetime.datetime(2016, 2, 23, 16, 55, 35),
+ 'uid': 328556,
+ 'user': 'InternationalUser',
+ 'version': 1,
+ 'visible': True
}
- )
-
- def test_ChangesetsGet(self):
- self._session_mock()
-
- result = self.api.ChangesetsGet(
- only_closed=True,
- username='metaodi'
- )
-
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'GET')
- self.assertEqual(
- dict(urlparse.parse_qsl(urlparse.urlparse(args[1])[4])),
+ }
+ )
+
+
+def test_ChangesetsGet(api, add_response):
+ resp = add_response(GET, '/changesets')
+
+ result = api.ChangesetsGet(
+ only_closed=True,
+ username='metaodi'
+ )
+
+ assert resp.calls[0].request.params == {'display_name': 'metaodi', 'closed': '1'}
+ assert len(result) == 10
+ assert result[41417] == ({
+ 'closed_at': datetime.datetime(2014, 4, 29, 20, 25, 1),
+ 'created_at': datetime.datetime(2014, 4, 29, 20, 25, 1),
+ 'id': 41417,
+ 'discussion': [],
+ 'max_lat': '58.8997467',
+ 'max_lon': '22.7364427',
+ 'min_lat': '58.8501594',
+ 'min_lon': '22.6984333',
+ 'open': False,
+ 'tag': {
+ 'comment': 'Test delete of relation',
+ 'created_by': 'iD 1.3.9',
+ 'imagery_used': 'Bing'
+ },
+ 'uid': 1841,
+ 'user': 'metaodi'
+ })
+
+
+def test_ChangesetGetWithComment(api, add_response):
+ resp = add_response(GET, '/changeset/52924')
+
+ result = api.ChangesetGet(52924, include_discussion=True)
+
+ assert resp.calls[0].request.params == {'include_discussion': 'true'}
+ assert result == {
+ 'id': 52924,
+ 'closed_at': datetime.datetime(2015, 1, 1, 14, 54, 2),
+ 'created_at': datetime.datetime(2015, 1, 1, 14, 54, 1),
+ 'comments_count': 3,
+ 'max_lat': '58.3369242',
+ 'max_lon': '25.8829107',
+ 'min_lat': '58.336813',
+ 'min_lon': '25.8823273',
+ 'discussion': [
{
- 'display_name': 'metaodi',
- 'closed': '1'
- }
+ 'date': datetime.datetime(2015, 1, 1, 18, 56, 48),
+ 'text': 'test',
+ 'uid': 1841,
+ 'user': 'metaodi',
+ },
+ {
+ 'date': datetime.datetime(2015, 1, 1, 18, 58, 3),
+ 'text': 'another comment',
+ 'uid': 1841,
+ 'user': 'metaodi',
+ },
+ {
+ 'date': datetime.datetime(2015, 1, 1, 19, 16, 5),
+ 'text': 'hello',
+ 'uid': 1841,
+ 'user': 'metaodi',
+ },
+ ],
+ 'open': False,
+ 'user': 'metaodi',
+ 'uid': 1841,
+ 'tag': {
+ 'comment': 'My test',
+ 'created_by': 'osmapi/0.4.1',
+ },
+ }
+
+
+def test_ChangesetComment(auth_api, add_response):
+ resp = add_response(POST, '/changeset/123/comment')
+
+ result = auth_api.ChangesetComment(
+ 123,
+ comment="test comment"
+ )
+
+ assert resp.calls[0].request.body == "text=test+comment"
+ assert result == {
+ 'id': 123,
+ 'closed_at': datetime.datetime(2009, 9, 7, 22, 57, 37),
+ 'created_at': datetime.datetime(2009, 9, 7, 21, 57, 36),
+ 'discussion': [],
+ 'max_lat': '52.4710193',
+ 'max_lon': '-1.4831815',
+ 'min_lat': '45.9667901',
+ 'min_lon': '-1.4998534',
+ 'open': False,
+ 'user': 'randomjunk',
+ 'uid': 3,
+ 'tag': {
+ 'comment': 'correct node bug',
+ 'created_by': 'Potlatch 1.2a',
+ },
+ }
+
+
+def test_ChangesetComment_no_auth(api):
+ with pytest.raises(osmapi.UsernamePasswordMissingError) as execinfo:
+ api.ChangesetComment(
+ 123,
+ comment="test comment"
)
+ assert str(execinfo.value) == "Username/Password missing"
- self.assertEqual(len(result), 10)
-
- self.assertEqual(result[41417], {
- 'closed_at': datetime.datetime(2014, 4, 29, 20, 25, 1),
- 'created_at': datetime.datetime(2014, 4, 29, 20, 25, 1),
- 'id': 41417,
- 'discussion': [],
- 'max_lat': '58.8997467',
- 'max_lon': '22.7364427',
- 'min_lat': '58.8501594',
- 'min_lon': '22.6984333',
- 'open': False,
- 'tag': {
- 'comment': 'Test delete of relation',
- 'created_by': 'iD 1.3.9',
- 'imagery_used': 'Bing'
- },
- 'uid': 1841,
- 'user': 'metaodi'
- })
- def test_ChangesetGetWithComment(self):
- self._session_mock()
+def test_ChangesetSubscribe(auth_api, add_response):
+ add_response(POST, '/changeset/123/subscribe')
- result = self.api.ChangesetGet(52924, include_discussion=True)
+ result = auth_api.ChangesetSubscribe(123)
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'GET')
- self.assertEqual(
- args[1],
- self.api_base + '/api/0.6/changeset/52924?include_discussion=true'
- )
+ assert result == {
+ 'id': 123,
+ 'closed_at': datetime.datetime(2009, 9, 7, 22, 57, 37),
+ 'created_at': datetime.datetime(2009, 9, 7, 21, 57, 36),
+ 'discussion': [],
+ 'max_lat': '52.4710193',
+ 'max_lon': '-1.4831815',
+ 'min_lat': '45.9667901',
+ 'min_lon': '-1.4998534',
+ 'open': False,
+ 'user': 'randomjunk',
+ 'uid': 3,
+ 'tag': {
+ 'comment': 'correct node bug',
+ 'created_by': 'Potlatch 1.2a',
+ },
+ }
- self.assertEqual(result, {
- 'id': 52924,
- 'closed_at': datetime.datetime(2015, 1, 1, 14, 54, 2),
- 'created_at': datetime.datetime(2015, 1, 1, 14, 54, 1),
- 'comments_count': 3,
- 'max_lat': '58.3369242',
- 'max_lon': '25.8829107',
- 'min_lat': '58.336813',
- 'min_lon': '25.8823273',
- 'discussion': [
- {
- 'date': datetime.datetime(2015, 1, 1, 18, 56, 48),
- 'text': 'test',
- 'uid': 1841,
- 'user': 'metaodi',
- },
- {
- 'date': datetime.datetime(2015, 1, 1, 18, 58, 3),
- 'text': 'another comment',
- 'uid': 1841,
- 'user': 'metaodi',
- },
- {
- 'date': datetime.datetime(2015, 1, 1, 19, 16, 5),
- 'text': 'hello',
- 'uid': 1841,
- 'user': 'metaodi',
- },
- ],
- 'open': False,
- 'user': 'metaodi',
- 'uid': 1841,
- 'tag': {
- 'comment': 'My test',
- 'created_by': 'osmapi/0.4.1',
- },
- })
- def test_ChangesetComment(self):
- self._session_mock(auth=True)
+def test_ChangesetSubscribeWhenAlreadySubscribed(auth_api, add_response):
+ add_response(POST, '/changeset/52924/subscribe', status=409)
- result = self.api.ChangesetComment(
- 123,
- comment="test comment"
- )
+ with pytest.raises(osmapi.AlreadySubscribedApiError) as execinfo:
+ auth_api.ChangesetSubscribe(52924)
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'POST')
- self.assertEqual(
- args[1],
- f'{self.api_base}/api/0.6/changeset/123/comment'
- )
- self.assertEqual(
- kwargs['data'],
- "text=test+comment"
- )
- self.assertEqual(result, {
- 'id': 123,
- 'closed_at': datetime.datetime(2009, 9, 7, 22, 57, 37),
- 'created_at': datetime.datetime(2009, 9, 7, 21, 57, 36),
- 'discussion': [],
- 'max_lat': '52.4710193',
- 'max_lon': '-1.4831815',
- 'min_lat': '45.9667901',
- 'min_lon': '-1.4998534',
- 'open': False,
- 'user': 'randomjunk',
- 'uid': 3,
- 'tag': {
- 'comment': 'correct node bug',
- 'created_by': 'Potlatch 1.2a',
- },
- })
+ assert execinfo.value.payload == b"You are already subscribed to changeset 52924."
+ assert execinfo.value.reason == 'Conflict'
+ assert execinfo.value.status == 409
- def test_ChangesetSubscribe(self):
- self._session_mock(auth=True)
- result = self.api.ChangesetSubscribe(123)
+def test_ChangesetSubscribe_no_auth(api):
+ with pytest.raises(osmapi.UsernamePasswordMissingError) as execinfo:
+ api.ChangesetSubscribe(45627)
+ assert str(execinfo.value) == "Username/Password missing"
- args, _ = self.api._session.request.call_args
- self.assertEqual(args[0], 'POST')
- self.assertEqual(
- args[1],
- f'{self.api_base}/api/0.6/changeset/123/subscribe'
- )
- self.assertEqual(result, {
- 'id': 123,
- 'closed_at': datetime.datetime(2009, 9, 7, 22, 57, 37),
- 'created_at': datetime.datetime(2009, 9, 7, 21, 57, 36),
- 'discussion': [],
- 'max_lat': '52.4710193',
- 'max_lon': '-1.4831815',
- 'min_lat': '45.9667901',
- 'min_lon': '-1.4998534',
- 'open': False,
- 'user': 'randomjunk',
- 'uid': 3,
- 'tag': {
- 'comment': 'correct node bug',
- 'created_by': 'Potlatch 1.2a',
- },
- })
- def test_ChangesetSubscribeWhenAlreadySubscribed(self):
- self._session_mock(auth=True, status=409)
+def test_ChangesetUnsubscribe(auth_api, add_response):
+ add_response(POST, '/changeset/123/unsubscribe')
- with self.assertRaises(osmapi.AlreadySubscribedApiError) as cm:
- self.api.ChangesetSubscribe(52924)
+ result = auth_api.ChangesetUnsubscribe(123)
- self.assertEqual(cm.exception.status, 409)
- self.assertEqual(
- cm.exception.payload,
- "You are already subscribed to changeset 52924."
- )
+ assert result == {
+ 'id': 123,
+ 'closed_at': datetime.datetime(2009, 9, 7, 22, 57, 37),
+ 'created_at': datetime.datetime(2009, 9, 7, 21, 57, 36),
+ 'discussion': [],
+ 'max_lat': '52.4710193',
+ 'max_lon': '-1.4831815',
+ 'min_lat': '45.9667901',
+ 'min_lon': '-1.4998534',
+ 'open': False,
+ 'user': 'randomjunk',
+ 'uid': 3,
+ 'tag': {
+ 'comment': 'correct node bug',
+ 'created_by': 'Potlatch 1.2a',
+ },
+ }
- def test_ChangesetUnsubscribe(self):
- self._session_mock(auth=True)
- result = self.api.ChangesetUnsubscribe(123)
+def test_ChangesetUnsubscribeWhenNotSubscribed(auth_api, add_response):
+ add_response(POST, '/changeset/52924/unsubscribe', status=404)
- args, kwargs = self.api._session.request.call_args
- self.assertEqual(args[0], 'POST')
- self.assertEqual(
- args[1],
- f'{self.api_base}/api/0.6/changeset/123/unsubscribe'
- )
- self.assertEqual(result, {
- 'id': 123,
- 'closed_at': datetime.datetime(2009, 9, 7, 22, 57, 37),
- 'created_at': datetime.datetime(2009, 9, 7, 21, 57, 36),
- 'discussion': [],
- 'max_lat': '52.4710193',
- 'max_lon': '-1.4831815',
- 'min_lat': '45.9667901',
- 'min_lon': '-1.4998534',
- 'open': False,
- 'user': 'randomjunk',
- 'uid': 3,
- 'tag': {
- 'comment': 'correct node bug',
- 'created_by': 'Potlatch 1.2a',
- },
- })
+ with pytest.raises(osmapi.NotSubscribedApiError) as execinfo:
+ auth_api.ChangesetUnsubscribe(52924)
- def test_ChangesetUnsubscribeWhenNotSubscribed(self):
- self._session_mock(auth=True, status=404)
+ assert execinfo.value.payload == b"You are not subscribed to changeset 52924."
+ assert execinfo.value.reason == 'Not Found'
+ assert execinfo.value.status == 404
- with self.assertRaises(osmapi.NotSubscribedApiError) as cm:
- self.api.ChangesetUnsubscribe(52924)
- self.assertEqual(cm.exception.status, 404)
- self.assertEqual(
- cm.exception.payload,
- "You are not subscribed to changeset 52924."
- )
+def test_ChangesetUnsubscribe_no_auth(api):
+ with pytest.raises(osmapi.UsernamePasswordMissingError) as execinfo:
+ api.ChangesetUnsubscribe(45627)
+ assert str(execinfo.value) == "Username/Password missing"
=====================================
tests/conftest.py
=====================================
@@ -0,0 +1,81 @@
+import osmapi
+import pytest
+import mock
+import responses
+import os
+import re
+
+__location__ = os.path.realpath(
+ os.path.join(
+ os.getcwd(),
+ os.path.dirname(__file__)
+ )
+)
+
+
+ at pytest.fixture
+def file_content():
+ def _file_content(filename):
+ path = os.path.join(__location__, 'fixtures', filename)
+ if not os.path.exists(path):
+ return ""
+ with open(path) as f:
+ return f.read()
+
+ return _file_content
+
+
+ at pytest.fixture
+def api():
+ api_base = "http://api06.dev.openstreetmap.org"
+ api = osmapi.OsmApi(
+ api=api_base
+ )
+ api._session._sleep = mock.Mock()
+
+ yield api
+ api.close()
+
+
+ at pytest.fixture
+def auth_api():
+ api_base = "http://api06.dev.openstreetmap.org"
+ api = osmapi.OsmApi(
+ api=api_base,
+ username='testuser',
+ password='testpassword'
+ )
+ api._session._sleep = mock.Mock()
+
+ yield api
+ api.close()
+
+
+ at pytest.fixture
+def mocked_responses():
+ with responses.RequestsMock() as rsps:
+ yield rsps
+
+
+ at pytest.fixture
+def add_response(mocked_responses, file_content, request):
+ def _add_response(method, path=None, filename=None, body=None, status=200):
+ if not filename:
+ # use testname by default
+ filename = f"{request.node.originalname}.xml"
+
+ if path:
+ url = f'http://api06.dev.openstreetmap.org/api/0.6{path}'
+ else:
+ url = re.compile(r'http:\/\/api06\.dev\.openstreetmap\.org.*')
+
+ if not body:
+ body = file_content(filename)
+ mocked_responses.add(
+ method,
+ url=url,
+ body=body,
+ status=status
+ )
+ return mocked_responses
+ return _add_response
=====================================
tests/dom_test.py
=====================================
@@ -0,0 +1,60 @@
+from . import osmapi_test
+import osmapi
+import mock
+import datetime
+
+
+class TestOsmApiDom(osmapi_test.TestOsmApi):
+ def test_DomGetAttributes(self):
+ mock_domelement = mock.Mock()
+ mock_domelement.attributes = {
+ 'uid': '12345',
+ 'open': 'false',
+ 'visible': 'true',
+ 'lat': '47.1234',
+ 'date': '2021-12-10T21:28:03Z',
+ 'new_attribute': 'Test 123',
+ }
+
+ result = osmapi.dom._DomGetAttributes(mock_domelement)
+
+ self.assertIsInstance(result, dict)
+ self.assertEqual(result['uid'], 12345)
+ self.assertEqual(result['open'], False)
+ self.assertEqual(result['visible'], True)
+ self.assertEqual(result['lat'], 47.1234)
+ self.assertEqual(result['date'], datetime.datetime(2021, 12, 10, 21, 28, 3))
+ self.assertEqual(result['new_attribute'], 'Test 123')
+
+ def test_ParseDate(self):
+ self.assertEqual(
+ osmapi.dom._ParseDate('2021-02-25T09:49:33Z'),
+ datetime.datetime(2021, 2, 25, 9, 49, 33)
+ )
+ self.assertEqual(
+ osmapi.dom._ParseDate('2021-02-25 09:49:33 UTC'),
+ datetime.datetime(2021, 2, 25, 9, 49, 33)
+ )
+ with self.assertLogs('osmapi.dom', level='DEBUG') as cm:
+ self.assertEqual(
+ osmapi.dom._ParseDate('2021-02-25'),
+ '2021-02-25'
+ )
+ self.assertEqual(
+ osmapi.dom._ParseDate(''),
+ ''
+ )
+ self.assertIsNone(osmapi.dom._ParseDate(None))
+
+ # test logging output
+ self.assertEqual(
+ cm.output,
+ [
+ 'DEBUG:osmapi.dom:2021-02-25 does not match %Y-%m-%d %H:%M:%S UTC',
+ 'DEBUG:osmapi.dom:2021-02-25 does not match %Y-%m-%dT%H:%M:%SZ',
+ 'DEBUG:osmapi.dom: does not match %Y-%m-%d %H:%M:%S UTC',
+ 'DEBUG:osmapi.dom: does not match %Y-%m-%dT%H:%M:%SZ',
+ 'DEBUG:osmapi.dom:None does not match %Y-%m-%d %H:%M:%S UTC',
+ 'DEBUG:osmapi.dom:None does not match %Y-%m-%dT%H:%M:%SZ',
+ ]
+ )
=====================================
tests/fixtures/test_Changeset_create.xml
=====================================
@@ -0,0 +1 @@
+1414
=====================================
tests/fixtures/test_Changeset_create_node.xml
=====================================
@@ -0,0 +1 @@
+7272
=====================================
tests/fixtures/test_Changeset_upload.xml
=====================================
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<diffResult version="0.6" generator="OpenStreetMap server" copyright="OpenStreetMap and contributors" attribution="http://www.openstreetmap.org/copyright" license="http://opendatacommons.org/licenses/odbl/1-0/">
+ <node old_id="0" new_id="7272" new_version="1"/>
+</diffResult>
+
=====================================
tests/helper_test.py
=====================================
@@ -22,10 +22,18 @@ class TestOsmApiHelper(osmapi_test.TestOsmApi):
mock_response.status_code = status
mock_response.reason = "test reason"
mock_response.content = 'test response'
- self.api._session.request = mock.Mock(return_value=mock_response)
- self.api._session.close = mock.Mock()
- self.api._username = 'testuser'
- self.api._password = 'testpassword'
+
+ self.mock_session = mock.Mock()
+ self.mock_session.request = mock.Mock(return_value=mock_response)
+ self.mock_session.close = mock.Mock()
+ self.mock_session.auth = ('testuser', 'testpassword')
+
+ self.api = osmapi.OsmApi(
+ api=self.api_base,
+ session=self.mock_session,
+ username='testuser',
+ password='testpassword'
+ )
def test_passwordfile_only(self):
path = os.path.join(
@@ -59,7 +67,7 @@ class TestOsmApiHelper(osmapi_test.TestOsmApi):
def test_close_call(self):
self.api.close()
- self.assertEqual(self.api._session.close.call_count, 1)
+ self.assertEqual(self.api._session._session.close.call_count, 1)
def test_close_context_manager(self):
with osmapi.OsmApi() as my_api:
@@ -67,72 +75,69 @@ class TestOsmApiHelper(osmapi_test.TestOsmApi):
self.assertEqual(my_api._session.close.call_count, 1)
def test_http_request_get(self):
- response = self.api._http_request(
+ response = self.api._session._http_request(
'GET',
'/api/0.6/test',
False,
None
)
- self.api._session.request.assert_called_with(
+ self.mock_session.request.assert_called_with(
'GET',
self.api_base + '/api/0.6/test',
- auth=None,
data=None
)
self.assertEqual(response, "test response")
- self.assertEqual(self.api._session.request.call_count, 1)
+ self.assertEqual(self.mock_session.request.call_count, 1)
def test_http_request_put(self):
data = "data"
- response = self.api._http_request(
+ response = self.api._session._http_request(
'PUT',
'/api/0.6/testput',
False,
data
)
- self.api._session.request.assert_called_with(
+ self.mock_session.request.assert_called_with(
'PUT',
self.api_base + '/api/0.6/testput',
- data="data",
- auth=None
+ data="data"
)
self.assertEqual(response, "test response")
def test_http_request_delete(self):
data = "delete data"
- response = self.api._http_request(
+ response = self.api._session._http_request(
'PUT',
'/api/0.6/testdelete',
False,
data
)
- self.api._session.request.assert_called_with(
+ self.mock_session.request.assert_called_with(
'PUT',
self.api_base + '/api/0.6/testdelete',
- data="delete data",
- auth=None
+ data="delete data"
)
self.assertEqual(response, "test response")
def test_http_request_auth(self):
- response = self.api._http_request(
+ response = self.api._session._http_request(
'PUT',
'/api/0.6/testauth',
True,
None
)
- self.api._session.request.assert_called_with(
+ self.mock_session.request.assert_called_with(
'PUT',
self.api_base + '/api/0.6/testauth',
- auth=('testuser', 'testpassword'),
data=None
)
+ self.assertEqual(self.mock_session.auth, ('testuser', 'testpassword'))
self.assertEqual(response, "test response")
def test_http_request_410_response(self):
self.setupMock(410)
with self.assertRaises(osmapi.ElementDeletedApiError) as cm:
- self.api._http_request(
+ self.api._session._http_request(
'GET',
'/api/0.6/test410',
False,
@@ -145,7 +150,7 @@ class TestOsmApiHelper(osmapi_test.TestOsmApi):
def test_http_request_500_response(self):
self.setupMock(500)
with self.assertRaises(osmapi.ApiError) as cm:
- self.api._http_request(
+ self.api._session._http_request(
'GET',
self.api_base + '/api/0.6/test500',
False,
=====================================
tests/node_test.py
=====================================
@@ -11,7 +11,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
result = self.api.NodeGet(123)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], self.api_base + '/api/0.6/node/123')
@@ -37,7 +37,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
result = self.api.NodeGet(123, NodeVersion=2)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], self.api_base + '/api/0.6/node/123/2')
@@ -63,15 +63,17 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
self.api.NodeGet(987)
def test_NodeCreate_changesetauto(self):
- # setup mock
- self.api = osmapi.OsmApi(
- api="api06.dev.openstreetmap.org",
- changesetauto=True
- )
for filename in ['test_NodeCreate_changesetauto.xml',
'test_ChangesetUpload_create_node.xml',
'test_ChangesetClose.xml']:
+ # setup mock
self._session_mock(auth=True, filenames=[filename])
+ self.api = osmapi.OsmApi(
+ api="api06.dev.openstreetmap.org",
+ changesetauto=True,
+ session=self.session_mock
+ )
+ self.api._session._sleep = mock.Mock()
test_node = {
'lat': 47.123,
@@ -108,7 +110,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
self.assertEqual(cs, 1111)
result = self.api.NodeCreate(test_node)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'PUT')
self.assertEqual(args[1], self.api_base + '/api/0.6/node/create')
@@ -178,7 +180,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
def test_NodeCreate_with_exception(self):
self._session_mock(auth=True)
- self.api._http_request = mock.Mock(side_effect=Exception)
+ self.api._session._http_request = mock.Mock(side_effect=Exception)
# setup mock
self.api.ChangesetCreate = mock.Mock(
@@ -224,7 +226,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
self.assertEqual(cs, 1111)
result = self.api.NodeUpdate(test_node)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'PUT')
self.assertEqual(args[1], self.api_base + '/api/0.6/node/7676')
@@ -316,7 +318,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
result = self.api.NodeDelete(test_node)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'DELETE')
self.assertEqual(args[1], self.api_base + '/api/0.6/node/7676')
self.assertEqual(result['id'], 7676)
@@ -327,7 +329,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
result = self.api.NodeHistory(123)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], self.api_base + '/api/0.6/node/123/history')
@@ -348,7 +350,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
result = self.api.NodeWays(234)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], self.api_base + '/api/0.6/node/234/ways')
@@ -368,7 +370,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
result = self.api.NodeWays(404)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], f'{self.api_base}/api/0.6/node/404/ways')
@@ -380,7 +382,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
result = self.api.NodeRelations(4295668179)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -410,7 +412,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
result = self.api.NodeRelations(4295668179)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -425,7 +427,7 @@ class TestOsmApiNode(osmapi_test.TestOsmApi):
result = self.api.NodesGet([123, 345])
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
=====================================
tests/notes_test.py
=====================================
@@ -20,7 +20,7 @@ class TestOsmApiNotes(osmapi_test.TestOsmApi):
52.4710193
)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
urlParts = urlparse.urlparse(args[1])
params = urlparse.parse_qs(urlParts.query)
@@ -68,7 +68,7 @@ class TestOsmApiNotes(osmapi_test.TestOsmApi):
result = self.api.NoteGet(1111)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], self.api_base + '/api/0.6/notes/1111')
@@ -115,7 +115,7 @@ class TestOsmApiNotes(osmapi_test.TestOsmApi):
}
result = self.api.NoteCreate(note)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'POST')
urlParts = urlparse.urlparse(args[1])
params = urlparse.parse_qs(urlParts.query)
@@ -152,7 +152,7 @@ class TestOsmApiNotes(osmapi_test.TestOsmApi):
}
result = self.api.NoteCreate(note)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'POST')
urlParts = urlparse.urlparse(args[1])
params = urlparse.parse_qs(urlParts.query)
@@ -184,7 +184,7 @@ class TestOsmApiNotes(osmapi_test.TestOsmApi):
result = self.api.NoteComment(812, 'This is a comment')
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'POST')
self.assertEqual(
args[1],
@@ -223,7 +223,7 @@ class TestOsmApiNotes(osmapi_test.TestOsmApi):
result = self.api.NoteComment(842, 'blubb')
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'POST')
self.assertEqual(
args[1],
@@ -262,7 +262,7 @@ class TestOsmApiNotes(osmapi_test.TestOsmApi):
result = self.api.NoteClose(814, 'Close this note!')
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'POST')
self.assertEqual(
args[1],
@@ -301,7 +301,7 @@ class TestOsmApiNotes(osmapi_test.TestOsmApi):
result = self.api.NoteReopen(815, 'Reopen this note!')
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'POST')
self.assertEqual(
args[1],
@@ -349,7 +349,7 @@ class TestOsmApiNotes(osmapi_test.TestOsmApi):
result = self.api.NotesSearch('street')
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
urlParts = urlparse.urlparse(args[1])
params = urlparse.parse_qs(urlParts.query)
=====================================
tests/osmapi_test.py
=====================================
@@ -2,12 +2,7 @@ from __future__ import unicode_literals
from osmapi import OsmApi
import mock
import os
-import sys
-
-if sys.version_info < (2, 7):
- import unittest2 as unittest
-else:
- import unittest
+import unittest
__location__ = os.path.realpath(
os.path.join(
@@ -28,10 +23,6 @@ class TestOsmApi(unittest.TestCase):
print(self.api)
def _session_mock(self, auth=False, filenames=None, status=200):
- if auth:
- self.api._username = 'testuser'
- self.api._password = 'testpassword'
-
response_mock = mock.Mock()
response_mock.status_code = status
return_values = self._return_values(filenames)
@@ -41,13 +32,24 @@ class TestOsmApi(unittest.TestCase):
if return_values:
response_mock.content = return_values[0]
- session_mock = mock.Mock()
- session_mock.request = mock.Mock(return_value=response_mock)
+ self.session_mock = mock.Mock()
+ self.session_mock.request = mock.Mock(return_value=response_mock)
- self.api._get_http_session = mock.Mock(return_value=session_mock)
- self.api._session = session_mock
+ if auth:
+ self.api = OsmApi(
+ api=self.api_base,
+ username='testuser',
+ password='testpassword',
+ session=self.session_mock
+ )
+ else:
+ self.api = OsmApi(
+ api=self.api_base,
+ session=self.session_mock
+ )
- self.api._sleep = mock.Mock()
+ self.api._get_http_session = mock.Mock(return_value=self.session_mock)
+ self.api._session._sleep = mock.Mock()
def _return_values(self, filenames):
if filenames is None:
=====================================
tests/relation_test.py
=====================================
@@ -11,7 +11,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationGet(321)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], self.api_base + '/api/0.6/relation/321')
@@ -82,7 +82,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationGet(765, 2)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], self.api_base + '/api/0.6/relation/765/2')
@@ -125,7 +125,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationCreate(test_relation)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'PUT')
self.assertEqual(args[1], self.api_base + '/api/0.6/relation/create')
@@ -195,7 +195,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationUpdate(test_relation)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'PUT')
self.assertEqual(args[1], self.api_base + '/api/0.6/relation/8989')
@@ -224,7 +224,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationDelete(test_relation)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'DELETE')
self.assertEqual(args[1], self.api_base + '/api/0.6/relation/8989')
@@ -236,7 +236,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationHistory(2470397)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -259,7 +259,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationRelations(1532552)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -281,7 +281,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationRelations(1532552)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -296,7 +296,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationFull(2470397)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -316,7 +316,7 @@ class TestOsmApiRelation(osmapi_test.TestOsmApi):
result = self.api.RelationsGet([1532552, 1532553])
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
=====================================
tests/way_test.py
=====================================
@@ -11,7 +11,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WayGet(321)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], self.api_base + '/api/0.6/way/321')
@@ -53,7 +53,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WayGet(4294967296, 2)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -94,7 +94,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WayCreate(test_way)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'PUT')
self.assertEqual(args[1], self.api_base + '/api/0.6/way/create')
@@ -148,7 +148,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WayUpdate(test_way)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'PUT')
self.assertEqual(args[1], self.api_base + '/api/0.6/way/876')
@@ -210,7 +210,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WayDelete(test_way)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'DELETE')
self.assertEqual(args[1], self.api_base + '/api/0.6/way/876')
self.assertEqual(result['id'], 876)
@@ -221,7 +221,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WayHistory(4294967296)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -243,7 +243,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WayRelations(4295032193)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -273,7 +273,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WayRelations(4295032193)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
@@ -288,7 +288,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WayFull(321)
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(args[1], self.api_base + '/api/0.6/way/321/full')
@@ -311,7 +311,7 @@ class TestOsmApiWay(osmapi_test.TestOsmApi):
result = self.api.WaysGet([456, 678])
- args, kwargs = self.api._session.request.call_args
+ args, kwargs = self.session_mock.request.call_args
self.assertEqual(args[0], 'GET')
self.assertEqual(
args[1],
View it on GitLab: https://salsa.debian.org/debian-gis-team/python-osmapi/-/compare/60da301b36a38c7f73379294cae44694c4bc4e3a...1aedd42229b3a6cc3f6cf4e6941799b726c4489d
--
View it on GitLab: https://salsa.debian.org/debian-gis-team/python-osmapi/-/compare/60da301b36a38c7f73379294cae44694c4bc4e3a...1aedd42229b3a6cc3f6cf4e6941799b726c4489d
You're receiving this email because of your account on salsa.debian.org.
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://alioth-lists.debian.net/pipermail/pkg-grass-devel/attachments/20220213/dda11661/attachment-0001.htm>
More information about the Pkg-grass-devel
mailing list