mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-09-16 21:02:35 +00:00
Fixed url shortener test case and updated linting errors. Updated documentation and changelog to reflect changes.
This commit is contained in:
@@ -8,6 +8,11 @@ semantic versioning.
|
||||
|
||||
### Fixed
|
||||
|
||||
- Refactor response_template parsing to use finite state machine when processing
|
||||
templates adding additional flexibility such as the use of nested fields. Added
|
||||
support for shlink to External_Data configuration. Added url_shortener filter and
|
||||
if_notempty filter to response_template.
|
||||
|
||||
- `path` no longer answers "No path information available in current message" on a
|
||||
busy mesh (#255). Verifying a channel message against the RF cache only ever
|
||||
checked the newest row, which assumes the RF log row and the decoded CHAN event
|
||||
|
||||
@@ -27,6 +27,17 @@ These options only affect the **path** command’s reply text and whether repeat
|
||||
reply_prefix = "{path_distance|prefix_if_nonempty:📏 }\n"
|
||||
```
|
||||
- `hops_min:N` clears a field unless the message actually travelled at least N hops. `{path_distance}` renders `N/A` on a direct message, which `prefix_if_nonempty` treats as a value, so gate it first: `{path_distance|hops_min:1|prefix_if_nonempty:📏 }`. Unlike `pathbytes_min:N`, which asks how the path is *encoded*, this keeps a measurable one-byte multi-hop path.
|
||||
- `if_notempty:LITERAL` renders `LITERAL` when the value is non-empty after prior filters, and clears entirely otherwise — the opposite pairing of `prefix_if_nonempty`, useful when the whole output should be a fixed (or field-built) literal rather than the value with a label prepended. Since `{packet_hash}` is empty whenever RF correlation fails, gating on it hides the whole clause instead of printing a broken link:
|
||||
|
||||
```ini
|
||||
reply_prefix = {packet_hash|if_notempty:"https://scope.example.net/#/packets/{packet_hash}"}
|
||||
```
|
||||
The `LITERAL` argument may itself be a double-quoted string containing nested `{field}` placeholders (expanded before the filter runs), so the link above still carries the packet hash even though the field being gated on (`packet_hash`) and the field inside the literal are the same one.
|
||||
- `shorten_url` runs the value through the shared shortener configured under `[External_Data]` (`short_url_website`, `short_url_website_service` — `gd` for v.gd/is.gd-compatible or `shlink`, and `short_url_website_api_key` where required). It falls back to the original, unshortened value if shortening fails or isn't configured, so a clause never silently disappears because of a network error. Chain it after building the link so only the final URL is sent over RF:
|
||||
|
||||
```ini
|
||||
reply_prefix = {packet_hash|if_notempty:"https://scope.example.net/#/packets/{packet_hash}"|shorten_url}
|
||||
```
|
||||
|
||||
**`minimum_path_bytes`** (integer `0`–`3`, default `0`)
|
||||
|
||||
|
||||
@@ -13,8 +13,8 @@ from __future__ import annotations
|
||||
|
||||
from typing import Any, Callable
|
||||
|
||||
from .utils import message_hop_count, message_path_bytes_per_hop
|
||||
from .url_shortener import shorten_url_sync
|
||||
from .utils import message_hop_count, message_path_bytes_per_hop
|
||||
|
||||
FilterFn = Callable[[str, dict[str, Any], str], str]
|
||||
|
||||
|
||||
+26
-25
@@ -107,7 +107,6 @@ def _build_create_gd_url(long_url: str, base: str, api_key: str) -> str:
|
||||
def _build_create_shlink_url(long_url: str, base: str, api_key: str) -> str:
|
||||
from urllib.parse import urlparse, urlunparse
|
||||
|
||||
encoded = quote(long_url, safe="")
|
||||
root = _normalize_base(base)
|
||||
if "://" not in root:
|
||||
root = f"https://{root}"
|
||||
@@ -206,30 +205,32 @@ def shorten_url_sync(
|
||||
).strip()
|
||||
base = _normalize_base(base)
|
||||
|
||||
if base != "" and api_key != "":
|
||||
if service == "shlink":
|
||||
return _shorten_url_with_shlink(
|
||||
url_str,
|
||||
base,
|
||||
api_key,
|
||||
session=session,
|
||||
timeout=timeout,
|
||||
logger=logger,
|
||||
)
|
||||
else:
|
||||
return _shorten_url_with_gd(
|
||||
url_str,
|
||||
base,
|
||||
api_key,
|
||||
session=session,
|
||||
timeout=timeout,
|
||||
logger=logger,
|
||||
)
|
||||
else:
|
||||
if logger:
|
||||
logger.warning(
|
||||
"Short URL base and API key are empty; some services may reject requests."
|
||||
)
|
||||
if service == "shlink":
|
||||
if not api_key:
|
||||
if logger:
|
||||
logger.warning(
|
||||
"short_url_website_service=shlink requires short_url_website_api_key; skipping."
|
||||
)
|
||||
return ""
|
||||
return _shorten_url_with_shlink(
|
||||
url_str,
|
||||
base,
|
||||
api_key,
|
||||
session=session,
|
||||
timeout=timeout,
|
||||
logger=logger,
|
||||
)
|
||||
|
||||
# v.gd / is.gd-compatible: api_key is optional (unused for the public hosts,
|
||||
# only appended for self-hosted alternates via _host_allows_key_in_query).
|
||||
return _shorten_url_with_gd(
|
||||
url_str,
|
||||
base,
|
||||
api_key,
|
||||
session=session,
|
||||
timeout=timeout,
|
||||
logger=logger,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
if logger:
|
||||
|
||||
+167
-5
@@ -1,6 +1,7 @@
|
||||
"""Unit tests for modules.url_shortener."""
|
||||
|
||||
import configparser
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
@@ -8,7 +9,8 @@ import requests
|
||||
|
||||
from modules.url_shortener import (
|
||||
DEFAULT_SHORT_URL_BASE,
|
||||
_build_create_url,
|
||||
_build_create_gd_url,
|
||||
_build_create_shlink_url,
|
||||
_coerce_url_string,
|
||||
shorten_url_sync,
|
||||
)
|
||||
@@ -22,23 +24,42 @@ def _minimal_config(**external_data):
|
||||
return c
|
||||
|
||||
|
||||
class TestBuildCreateUrl:
|
||||
class TestBuildCreateGdUrl:
|
||||
def test_vgd_no_key_in_query(self):
|
||||
u = _build_create_url("http://example.com/path?q=1", "https://v.gd", "secret")
|
||||
u = _build_create_gd_url("http://example.com/path?q=1", "https://v.gd", "secret")
|
||||
assert "key=" not in u
|
||||
assert "format=simple" in u
|
||||
assert "url=http" in u
|
||||
|
||||
def test_custom_host_appends_key_when_set(self):
|
||||
u = _build_create_url("http://a.com", "https://short.example/api", "k1")
|
||||
u = _build_create_gd_url("http://a.com", "https://short.example/api", "k1")
|
||||
assert "key=k1" in u
|
||||
|
||||
def test_is_gd_no_key_in_query(self):
|
||||
u = _build_create_url("http://a.com", "https://is.gd", "secret")
|
||||
u = _build_create_gd_url("http://a.com", "https://is.gd", "secret")
|
||||
assert "key=" not in u
|
||||
assert "create.php" in u
|
||||
|
||||
|
||||
class TestBuildCreateShlinkUrl:
|
||||
def test_appends_rest_v3_short_urls_path(self):
|
||||
u = _build_create_shlink_url("http://a.com", "https://short.example", "k1")
|
||||
assert u == "https://short.example/rest/v3/short-urls"
|
||||
|
||||
def test_strips_trailing_slash_on_base(self):
|
||||
u = _build_create_shlink_url("http://a.com", "https://short.example/", "k1")
|
||||
assert u == "https://short.example/rest/v3/short-urls"
|
||||
|
||||
def test_bare_hostname_gets_https_scheme(self):
|
||||
u = _build_create_shlink_url("http://a.com", "short.example", "k1")
|
||||
assert u == "https://short.example/rest/v3/short-urls"
|
||||
|
||||
def test_api_key_never_appears_in_the_url(self):
|
||||
"""Shlink authenticates via the X-Api-Key header, not a query param."""
|
||||
u = _build_create_shlink_url("http://a.com", "https://short.example", "super-secret")
|
||||
assert "super-secret" not in u
|
||||
|
||||
|
||||
class TestCoerceUrlString:
|
||||
def test_dict_href(self):
|
||||
assert _coerce_url_string({"href": "https://a.com/x"}) == "https://a.com/x"
|
||||
@@ -85,6 +106,21 @@ class TestShortenUrlSync:
|
||||
assert call_url.startswith("https://v.gd/create.php")
|
||||
assert "format=simple" in call_url
|
||||
|
||||
def test_default_gd_service_needs_no_api_key(self):
|
||||
"""Regression: v.gd/is.gd are documented as keyless (config.ini.example);
|
||||
the default `gd` service must not require short_url_website_api_key."""
|
||||
cfg = _minimal_config(short_url_website="https://v.gd")
|
||||
assert cfg.get("External_Data", "short_url_website_api_key", fallback="") == ""
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.ok = True
|
||||
mock_resp.text = "https://v.gd/nokey\n"
|
||||
session = MagicMock()
|
||||
session.get.return_value = mock_resp
|
||||
|
||||
out = shorten_url_sync("http://a.com", config=cfg, session=session)
|
||||
assert out == "https://v.gd/nokey"
|
||||
session.get.assert_called_once()
|
||||
|
||||
def test_error_line_returns_empty(self):
|
||||
cfg = _minimal_config()
|
||||
mock_resp = MagicMock()
|
||||
@@ -170,6 +206,17 @@ class TestShortenUrlSync:
|
||||
call_url = session.get.call_args[0][0]
|
||||
assert call_url.startswith(DEFAULT_SHORT_URL_BASE)
|
||||
|
||||
def test_service_option_is_case_insensitive(self):
|
||||
cfg = _minimal_config(short_url_website="https://v.gd", short_url_website_service="GD")
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.ok = True
|
||||
mock_resp.text = "https://v.gd/caseok"
|
||||
session = MagicMock()
|
||||
session.get.return_value = mock_resp
|
||||
|
||||
out = shorten_url_sync("http://a.com", config=cfg, session=session)
|
||||
assert out == "https://v.gd/caseok"
|
||||
|
||||
@patch("modules.url_shortener.requests.get")
|
||||
def test_no_session_uses_requests_get(self, mock_get):
|
||||
cfg = _minimal_config(short_url_website="https://v.gd")
|
||||
@@ -183,6 +230,103 @@ class TestShortenUrlSync:
|
||||
mock_get.assert_called_once()
|
||||
|
||||
|
||||
class TestShortenUrlSyncShlink:
|
||||
def _shlink_config(self, **overrides):
|
||||
defaults = {
|
||||
"short_url_website_service": "shlink",
|
||||
"short_url_website": "https://short.example",
|
||||
"short_url_website_api_key": "test-api-key",
|
||||
}
|
||||
defaults.update(overrides)
|
||||
return _minimal_config(**defaults)
|
||||
|
||||
def test_success_returns_short_url(self):
|
||||
cfg = self._shlink_config()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"shortUrl": "https://short.example/abc123"}
|
||||
session = MagicMock()
|
||||
session.post.return_value = mock_resp
|
||||
|
||||
out = shorten_url_sync("https://example.com/long/path", config=cfg, session=session)
|
||||
|
||||
assert out == "https://short.example/abc123"
|
||||
session.post.assert_called_once()
|
||||
|
||||
def test_posts_to_rest_v3_short_urls_with_api_key_header(self):
|
||||
cfg = self._shlink_config()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"shortUrl": "https://short.example/abc123"}
|
||||
session = MagicMock()
|
||||
session.post.return_value = mock_resp
|
||||
|
||||
shorten_url_sync("https://example.com/long/path", config=cfg, session=session)
|
||||
|
||||
call = session.post.call_args
|
||||
assert call[0][0] == "https://short.example/rest/v3/short-urls"
|
||||
assert call.kwargs["headers"]["X-Api-Key"] == "test-api-key"
|
||||
assert call.kwargs["headers"]["Content-Type"] == "application/json"
|
||||
payload = json.loads(call.kwargs["data"])
|
||||
assert payload["longUrl"] == "https://example.com/long/path"
|
||||
assert payload["findIfExists"] is True
|
||||
|
||||
def test_falls_back_to_short_url_slug(self):
|
||||
cfg = self._shlink_config()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"shortUrlSlug": "abc123"}
|
||||
session = MagicMock()
|
||||
session.post.return_value = mock_resp
|
||||
|
||||
out = shorten_url_sync("http://a.com", config=cfg, session=session)
|
||||
assert out == "abc123"
|
||||
|
||||
def test_missing_short_url_in_response_returns_empty(self):
|
||||
cfg = self._shlink_config()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"unexpected": "shape"}
|
||||
session = MagicMock()
|
||||
session.post.return_value = mock_resp
|
||||
|
||||
assert shorten_url_sync("http://a.com", config=cfg, session=session) == ""
|
||||
|
||||
def test_missing_api_key_skips_the_request(self):
|
||||
"""Regression: shlink genuinely needs an API key, unlike v.gd/is.gd, so it
|
||||
must not attempt the call (and must not crash) when one isn't configured."""
|
||||
cfg = self._shlink_config(short_url_website_api_key="")
|
||||
session = MagicMock()
|
||||
|
||||
out = shorten_url_sync("http://a.com", config=cfg, session=session)
|
||||
|
||||
assert out == ""
|
||||
session.post.assert_not_called()
|
||||
|
||||
def test_request_exception_returns_empty(self):
|
||||
cfg = self._shlink_config()
|
||||
session = MagicMock()
|
||||
session.post.side_effect = requests.exceptions.ConnectionError("unreachable")
|
||||
|
||||
assert shorten_url_sync("http://a.com", config=cfg, session=session) == ""
|
||||
|
||||
def test_malformed_json_response_returns_empty(self):
|
||||
cfg = self._shlink_config()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.side_effect = ValueError("not json")
|
||||
session = MagicMock()
|
||||
session.post.return_value = mock_resp
|
||||
|
||||
assert shorten_url_sync("http://a.com", config=cfg, session=session) == ""
|
||||
|
||||
@patch("modules.url_shortener.requests.post")
|
||||
def test_no_session_uses_requests_post(self, mock_post):
|
||||
cfg = self._shlink_config()
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"shortUrl": "https://short.example/xyz"}
|
||||
mock_post.return_value = mock_resp
|
||||
|
||||
out = shorten_url_sync("http://a.com", config=cfg, session=None)
|
||||
assert out == "https://short.example/xyz"
|
||||
mock_post.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shorten_url_async():
|
||||
from modules.url_shortener import shorten_url
|
||||
@@ -196,3 +340,21 @@ async def test_shorten_url_async():
|
||||
|
||||
out = await shorten_url("http://d.com", config=cfg, session=session)
|
||||
assert out == "https://v.gd/async1"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shorten_url_async_shlink():
|
||||
from modules.url_shortener import shorten_url
|
||||
|
||||
cfg = _minimal_config(
|
||||
short_url_website_service="shlink",
|
||||
short_url_website="https://short.example",
|
||||
short_url_website_api_key="test-api-key",
|
||||
)
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"shortUrl": "https://short.example/async1"}
|
||||
session = MagicMock()
|
||||
session.post.return_value = mock_resp
|
||||
|
||||
out = await shorten_url("http://d.com", config=cfg, session=session)
|
||||
assert out == "https://short.example/async1"
|
||||
|
||||
Reference in New Issue
Block a user