e3fa6b8d6e
The stock ``base_geocoder._call_openstreetmap`` sends a generic Odoo ``User-Agent`` shared across thousands of instances, which Nominatim rate-limits and blocks (HTTP 429 "Too many requests"), so mass geolocation fails regardless of client-side throttling. Override ``_call_openstreetmap`` to: - send a deployment-specific ``User-Agent`` read from a new system parameter ``partner_geolocalize_usability.user_agent`` (shipped with an obvious placeholder so an unconfigured instance fails loudly instead of silently reusing the banned generic agent); - throttle requests to one per second (``time.sleep``) before each call, complying with the Nominatim usage policy even when callers loop over many partners; - return ``None`` instead of raising ``IndexError`` when Nominatim returns an empty result for an unknown address. Document the required system parameter in the README and add tests covering the User-Agent selection, throttling and empty/blank results.
93 lines
3.3 KiB
Python
93 lines
3.3 KiB
Python
from unittest.mock import MagicMock, patch
|
|
|
|
from odoo.tests.common import TransactionCase
|
|
|
|
from ..models.base_geocoder import (
|
|
NOMINATIM_MIN_INTERVAL,
|
|
USER_AGENT_PARAM,
|
|
USER_AGENT_PLACEHOLDER,
|
|
)
|
|
|
|
_MODULE = "odoo.addons.partner_geolocalize_usability.models.base_geocoder"
|
|
|
|
|
|
def _fake_response(payload, status_code=200):
|
|
response = MagicMock()
|
|
response.status_code = status_code
|
|
response.json.return_value = payload
|
|
return response
|
|
|
|
|
|
class TestBaseGeocoder(TransactionCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
cls.geocoder = cls.env["base.geocoder"]
|
|
cls.Param = cls.env["ir.config_parameter"].sudo()
|
|
|
|
def test_uses_configured_user_agent(self):
|
|
# A deployment-specific User-Agent must be sent to Nominatim.
|
|
self.Param.set_param(USER_AGENT_PARAM, "acme/1.0 (contact: x@acme.test)")
|
|
|
|
with patch(f"{_MODULE}.time.sleep"), patch(
|
|
f"{_MODULE}.requests.get",
|
|
return_value=_fake_response([{"lat": "1.5", "lon": "2.5"}]),
|
|
) as mocked_get:
|
|
result = self.geocoder._call_openstreetmap("some address")
|
|
|
|
self.assertEqual(result, (1.5, 2.5))
|
|
_, kwargs = mocked_get.call_args
|
|
self.assertEqual(
|
|
kwargs["headers"]["User-Agent"], "acme/1.0 (contact: x@acme.test)"
|
|
)
|
|
|
|
def test_falls_back_to_placeholder_user_agent(self):
|
|
# Without configuration, the obvious placeholder is used (never Odoo's
|
|
# banned generic agent).
|
|
self.Param.set_param(USER_AGENT_PARAM, False)
|
|
|
|
with patch(f"{_MODULE}.time.sleep"), patch(
|
|
f"{_MODULE}.requests.get",
|
|
return_value=_fake_response([{"lat": "1.0", "lon": "2.0"}]),
|
|
) as mocked_get:
|
|
self.geocoder._call_openstreetmap("addr")
|
|
|
|
_, kwargs = mocked_get.call_args
|
|
self.assertEqual(
|
|
kwargs["headers"]["User-Agent"], USER_AGENT_PLACEHOLDER
|
|
)
|
|
|
|
def test_throttles_before_request(self):
|
|
# The 1 req/s policy is enforced by sleeping before each call.
|
|
self.Param.set_param(USER_AGENT_PARAM, "acme/1.0 (contact: x@acme.test)")
|
|
|
|
with patch(f"{_MODULE}.time.sleep") as mocked_sleep, patch(
|
|
f"{_MODULE}.requests.get",
|
|
return_value=_fake_response([{"lat": "1.0", "lon": "2.0"}]),
|
|
):
|
|
self.geocoder._call_openstreetmap("addr")
|
|
|
|
mocked_sleep.assert_called_once_with(NOMINATIM_MIN_INTERVAL)
|
|
|
|
def test_empty_result_returns_none(self):
|
|
# An unknown address (empty Nominatim result) must not raise IndexError.
|
|
self.Param.set_param(USER_AGENT_PARAM, "acme/1.0 (contact: x@acme.test)")
|
|
|
|
with patch(f"{_MODULE}.time.sleep"), patch(
|
|
f"{_MODULE}.requests.get", return_value=_fake_response([])
|
|
):
|
|
result = self.geocoder._call_openstreetmap("nowhere at all")
|
|
|
|
self.assertIsNone(result)
|
|
|
|
def test_blank_address_returns_none_without_request(self):
|
|
# Empty address short-circuits: no HTTP call, no throttle.
|
|
with patch(f"{_MODULE}.time.sleep") as mocked_sleep, patch(
|
|
f"{_MODULE}.requests.get"
|
|
) as mocked_get:
|
|
result = self.geocoder._call_openstreetmap("")
|
|
|
|
self.assertIsNone(result)
|
|
mocked_get.assert_not_called()
|
|
mocked_sleep.assert_not_called()
|