diff --git a/CHANGELOG.md b/CHANGELOG.md index 73f6da7..57be826 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/config.ini.example b/config.ini.example index 95c759d..f1d19b9 100644 --- a/config.ini.example +++ b/config.ini.example @@ -644,6 +644,8 @@ log_max_bytes = 5242880 # log_backup_count: number of rotated backup files to keep (e.g. meshcore_bot.log.1 โ€ฆ .3) log_backup_count = 3 [External_Data] +# URL shortener Service (gd,shlink). Default: gd +short_url_website_service = gd # URL shortener API base (v.gd / is.gd-compatible create.php). Default: https://v.gd # See https://v.gd/apishorteningreference.php short_url_website = https://v.gd diff --git a/docs/path-command-config.md b/docs/path-command-config.md index 4688582..c0324d6 100644 --- a/docs/path-command-config.md +++ b/docs/path-command-config.md @@ -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`) diff --git a/modules/commands/path_command.py b/modules/commands/path_command.py index d23353f..b390e6c 100644 --- a/modules/commands/path_command.py +++ b/modules/commands/path_command.py @@ -425,6 +425,7 @@ class PathCommand(BaseCommand): {k: str(v) for k, v in fields.items()}, message=message, logger=self.logger, + config=self.bot.config, prefix_hex_chars=getattr(self.bot, 'prefix_hex_chars', 2), ).rstrip() if not formatted: diff --git a/modules/commands/test_command.py b/modules/commands/test_command.py index 0a1529a..9f2d320 100644 --- a/modules/commands/test_command.py +++ b/modules/commands/test_command.py @@ -674,6 +674,7 @@ class TestCommand(BaseCommand): fields, message=message, logger=self.logger, + config=self.bot.config, prefix_hex_chars=getattr(self.bot, 'prefix_hex_chars', 2), ) except (KeyError, ValueError) as e: diff --git a/modules/response_template.py b/modules/response_template.py index 66a4a07..4dbc691 100644 --- a/modules/response_template.py +++ b/modules/response_template.py @@ -2,14 +2,18 @@ """Piped placeholders for command response templates (feed-style ``{field|filter:args}``). Used by :class:`~modules.commands.test_command.TestCommand` and extensible for other -commands. Same brace limitation as feed formatting: no nested ``{}`` inside a placeholder. +commands. A placeholder holds either a bare field name (``{sender}``) or a +double-quoted string literal (``{"Hello {sender}!"}``); a quoted literal may embed +further ``{...}`` placeholders, which are expanded first and substituted into the +literal before any filters run. Either form may be followed by a ``|filter:arg`` +chain, evaluated left to right. """ from __future__ import annotations -import re from typing import Any, Callable +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] @@ -66,48 +70,184 @@ def _filter_prefix_if_nonempty(value: str, ctx: dict[str, Any], args: str) -> st return '' return args + value +def _filter_shorten_url(value: str, ctx: dict[str, Any], args: str) -> str: + """Shorten *value* URL using configured URL shortener (v.gd / is.gd compatible).""" + logger = ctx.get('logger') + config = ctx.get('config') + if logger is not None: + logger.debug("Shortening URL %r", value) + if config is None or value == '': + if logger is not None: + logger.debug("Abandoning shorten url due to empty value or config") + return value + return shorten_url_sync(value, config=config, logger=logger) or value + +def _filter_if_notempty(value: str, ctx: dict[str, Any], args: str) -> str: + """Return *args* literal only when *value* is non-empty after prior filters.""" + if not value: + return '' + return args RESPONSE_TEMPLATE_FILTERS: dict[str, FilterFn] = { 'pathbytes_min': _filter_pathbytes_min, 'pathbytes': _filter_pathbytes_min, 'hops_min': _filter_hops_min, 'prefix_if_nonempty': _filter_prefix_if_nonempty, + 'if_notempty': _filter_if_notempty, + 'shorten_url': _filter_shorten_url, } +# prefix_if_nonempty's literal argument may itself contain '|', so once the parser +# sees this filter name it stops splitting on '|' and takes everything up to the +# placeholder's closing '}' as one literal argument. It must therefore be last in +# a chain whenever its literal needs a pipe. +_GREEDY_ARG_FILTERS = frozenset({'prefix_if_nonempty'}) -def _field_and_filter_specs(inner: str) -> tuple[str, list[tuple[str, str]]]: - """Split ``inner`` into field name and ``(filter_name, args)`` pairs. - Pipe ``|`` separates filters. ``prefix_if_nonempty`` is special: its argument may - contain ``|`` (e.g. `` | Path Dist: ``), so once that filter is reached we merge - all remaining segments and treat the rest as its args. ``prefix_if_nonempty`` must - be last in the chain if the literal includes a pipe. +class _TemplateParser: + """Finite-state parser for ``{field|filter:arg|...}``-style placeholders. + + Walks the template left to right, character by character, alternating between + plain text and placeholder spans. A placeholder's base value is either a bare + field name or a ``"..."`` string literal; a literal may contain nested + ``{...}`` placeholders (parsed recursively, same grammar) which are expanded + before the literal is used as the base value for any following filters. """ - raw_parts = inner.split('|') - field_name = raw_parts[0].strip() - if len(raw_parts) < 2: - return field_name, [] - specs: list[tuple[str, str]] = [] - i = 1 - while i < len(raw_parts): - if raw_parts[i].lstrip().startswith('prefix_if_nonempty'): - merged = '|'.join(raw_parts[i:]) - if re.match(r'^\s*prefix_if_nonempty\s*$', merged): - specs.append(('prefix_if_nonempty', '')) + + def __init__(self, template: str, fields: dict[str, Any], ctx: dict[str, Any], logger: Any): + self.s = template + self.n = len(template) + self.fields = fields + self.ctx = ctx + self.logger = logger + + def render(self) -> str: + out: list[str] = [] + i = 0 + while i < self.n: + j = self.s.find('{', i) + if j == -1: + out.append(self.s[i:]) break - m = re.match(r'^\s*prefix_if_nonempty\s*:(.*)$', merged, flags=re.DOTALL) - if m: - specs.append(('prefix_if_nonempty', m.group(1))) + out.append(self.s[i:j]) + value, end = self._parse_placeholder(j) + if value is None: + out.append('{') + i = j + 1 else: - specs.append(('prefix_if_nonempty', '')) - break - segment = raw_parts[i].strip() - name, sep, arg = segment.partition(':') - name = name.strip() - arg = arg if sep else '' - specs.append((name, arg)) + out.append(value) + i = end + return ''.join(out) + + def _skip_ws(self, i: int) -> int: + while i < self.n and self.s[i].isspace(): + i += 1 + return i + + def _parse_placeholder(self, start: int) -> tuple[str | None, int]: + """Parse the placeholder beginning at ``self.s[start] == '{'``. + + Returns ``(expanded_value, index_after_closing_brace)``, or ``(None, start)`` + if there is no well-formed placeholder here (left as literal text). + """ + i = start + 1 + if i < self.n and self.s[i] == '}': + return None, start # `{}` has no content + i = self._skip_ws(i) + if i >= self.n: + return None, start + if self.s[i] == '"': + value, i = self._parse_quoted_string(i) + else: + value, i = self._parse_field_name(i) + if value is None: + return None, start + + i = self._skip_ws(i) + filter_specs: list[tuple[str, str]] = [] + while i < self.n and self.s[i] == '|': + i += 1 + name, args, i = self._parse_filter_spec(i) + if name is None: + return None, start + filter_specs.append((name, args)) + i = self._skip_ws(i) + + if i >= self.n or self.s[i] != '}': + return None, start + raw_inner = self.s[start + 1:i] + for name, args in filter_specs: + value = self._apply_filter(name, args, value, raw_inner) + return value, i + 1 + + def _parse_field_name(self, i: int) -> tuple[str | None, int]: + start = i + while i < self.n and self.s[i] not in '|}': + i += 1 + if i >= self.n: + return None, i + name = self.s[start:i].strip() + return str(self.fields.get(name, '')), i + + def _parse_quoted_string(self, i: int) -> tuple[str | None, int]: + """Parse a ``"..."`` literal starting at the opening quote. + + ``\\"`` and ``\\\\`` are recognized escapes; any ``{...}`` inside the + literal is expanded recursively and substituted in place. + """ i += 1 - return field_name, specs + parts: list[str] = [] + while i < self.n: + ch = self.s[i] + if ch == '\\' and i + 1 < self.n and self.s[i + 1] in ('"', '\\'): + parts.append(self.s[i + 1]) + i += 2 + continue + if ch == '"': + return ''.join(parts), i + 1 + if ch == '{': + value, i = self._parse_placeholder(i) + if value is None: + return None, i + parts.append(value) + continue + parts.append(ch) + i += 1 + return None, i # unterminated string literal + + def _parse_filter_spec(self, i: int) -> tuple[str | None, str, int]: + start = i + while i < self.n and self.s[i] not in ':|}': + i += 1 + if i >= self.n: + return None, '', i + name = self.s[start:i].strip() + if i >= self.n or self.s[i] != ':': + return name, '', i + i += 1 # consume ':' + if name in _GREEDY_ARG_FILTERS: + close = self.s.find('}', i) + if close == -1: + return None, '', self.n + return name, self.s[i:close], close + quoted_at = self._skip_ws(i) + if quoted_at < self.n and self.s[quoted_at] == '"': + value, j = self._parse_quoted_string(quoted_at) + if value is None: + return None, '', j + return name, value, j + arg_start = i + while i < self.n and self.s[i] not in '|}': + i += 1 + return name, self.s[arg_start:i], i + + def _apply_filter(self, name: str, args: str, value: str, raw_inner: str) -> str: + fn = RESPONSE_TEMPLATE_FILTERS.get(name) + if fn is None: + if self.logger is not None: + self.logger.warning(f"Unknown response template filter {name!r} in {{{raw_inner}}}") + return value + return fn(value, self.ctx, args) def format_piped_template( @@ -116,9 +256,10 @@ def format_piped_template( *, message: Any = None, logger: Any = None, + config: Any = None, prefix_hex_chars: int = 2, ) -> str: - """Replace ``{field}`` and ``{field|filter:arg|...}`` using *fields* and optional *message*. + """Replace ``{field}``, ``{"literal {field}"}``, and their piped filter chains. Args: template: Raw template string from config. @@ -136,21 +277,8 @@ def format_piped_template( 'message': message, 'logger': logger, 'prefix_hex_chars': prefix_hex_chars, + 'config': config, } - - def replace_placeholder(match: re.Match[str]) -> str: - inner_raw = match.group(1) - if '|' not in inner_raw: - return str(fields.get(inner_raw.strip(), '')) - field_name, filter_specs = _field_and_filter_specs(inner_raw) - value = str(fields.get(field_name, '')) - for name, arg in filter_specs: - fn = RESPONSE_TEMPLATE_FILTERS.get(name) - if fn is None: - if logger is not None: - logger.warning(f"Unknown response template filter {name!r} in {{{inner_raw}}}") - continue - value = fn(value, ctx, arg) - return value - - return re.sub(r"\{([^}]+)\}", replace_placeholder, template) + if (logger is not None) and (config is not None): + logger.debug("Rendering response template %r with fields %r", template, fields) + return _TemplateParser(template, fields, ctx, logger).render() diff --git a/modules/url_shortener.py b/modules/url_shortener.py index b815e8b..b76de44 100644 --- a/modules/url_shortener.py +++ b/modules/url_shortener.py @@ -47,6 +47,7 @@ def _safe_config_get(config: Any, section: str, option: str, fallback: str = "") except Exception: return fallback + DEFAULT_SHORT_URL_BASE = "https://v.gd" # Hostnames that use the public create.php API without an API key query param. @@ -82,7 +83,7 @@ def _parse_simple_response(body: str) -> str | None: return None -def _build_create_url(long_url: str, base: str, api_key: str) -> str: +def _build_create_gd_url(long_url: str, base: str, api_key: str) -> str: from urllib.parse import urlparse, urlunparse encoded = quote(long_url, safe="") @@ -99,12 +100,82 @@ def _build_create_url(long_url: str, base: str, api_key: str) -> str: query = f"format=simple&url={encoded}" if api_key and _host_allows_key_in_query(parsed.hostname or ""): query = f"{query}&key={quote(api_key, safe='')}" - rebuilt = urlunparse( - (parsed.scheme or "https", netloc, path, "", query, "") - ) + rebuilt = urlunparse((parsed.scheme or "https", netloc, path, "", query, "")) return rebuilt +def _build_create_shlink_url(long_url: str, base: str, api_key: str) -> str: + from urllib.parse import urlparse, urlunparse + + root = _normalize_base(base) + if "://" not in root: + root = f"https://{root}" + parsed = urlparse(root) + netloc = parsed.netloc + if not netloc and parsed.path: + netloc = parsed.path.split("/")[0] + path = (parsed.path or "").rstrip("/") + "/rest/v3/short-urls" + if not path.startswith("/"): + path = "/" + path + query = "" + rebuilt = urlunparse((parsed.scheme or "https", netloc, path, "", query, "")) + return rebuilt + + +def _shorten_url_with_shlink( + long_url: str, + base: str, + api_key: str, + session: requests.Session | None = None, + timeout: float = 5.0, + logger: logging.Logger | None = None, +) -> str: + """Shorten a URL using Shlink API.""" + import json + + shortener_url = _build_create_shlink_url(long_url, base, api_key) + headers = { + "Content-Type": "application/json", + "X-Api-Key": api_key, + } + payload = json.dumps( + {"longUrl": long_url, "findIfExists": True, "tags": ["meshcore-bot"]} + ) + + get = session.post if session is not None else requests.post + response = get(shortener_url, headers=headers, data=payload, timeout=timeout) + if logger: + logger.debug("Shlink response: %s", response.text) + data = response.json() + short_url = data.get("shortUrl") or data.get("shortUrlSlug") + + if short_url: + return short_url + + return "" + + +def _shorten_url_with_gd( + long_url: str, + base: str, + api_key: str, + session: requests.Session | None = None, + timeout: float = 5.0, + logger: logging.Logger | None = None, +) -> str: + """Shorten a URL using v.gd / is.gd API.""" + shortener_url = _build_create_gd_url(long_url, base, api_key) + + get = session.get if session is not None else requests.get + + response = get(shortener_url, timeout=timeout) + short = _parse_simple_response(response.text) + if short: + return short + + return "" + + def shorten_url_sync( url: Any, *, @@ -123,37 +194,47 @@ def shorten_url_sync( return "" base = _safe_config_get(config, "External_Data", "short_url_website", "") - api_key = (_safe_config_get(config, "External_Data", "short_url_website_api_key", "") or "").strip() + service = ( + _safe_config_get(config, "External_Data", "short_url_website_service", "gd") + .strip() + .lower() + ) + api_key = ( + _safe_config_get(config, "External_Data", "short_url_website_api_key", "") + or "" + ).strip() base = _normalize_base(base) - shortener_url = _build_create_url(url_str, base, api_key) - get = session.get if session is not None else requests.get + 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, + ) - try: - response = get(shortener_url, timeout=timeout) - except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: - if logger: - logger.debug("Error shortening URL: %s", e) - return "" - except Exception as e: - if logger: - logger.debug("Unexpected error shortening URL: %s", e) - return "" + # 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, + ) - if not response.ok: - if logger: - logger.debug("Error shortening URL: HTTP %s", response.status_code) - return "" - - short = _parse_simple_response(response.text) - if short: - return short - if logger: - logger.debug("URL shortener returned error: %s", response.text.strip()[:200]) - return "" except Exception as e: if logger: - logger.debug("shorten_url_sync failed: %s", e) + logger.error("Unexpected error shortening URL: %s", e) return "" diff --git a/tests/test_url_shortener.py b/tests/test_url_shortener.py index 275cfdd..8829a91 100644 --- a/tests/test_url_shortener.py +++ b/tests/test_url_shortener.py @@ -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" diff --git a/tests/unit/test_response_template.py b/tests/unit/test_response_template.py index 3e965d6..56a9e9b 100644 --- a/tests/unit/test_response_template.py +++ b/tests/unit/test_response_template.py @@ -324,3 +324,81 @@ def test_hops_min_with_an_unusable_argument_passes_the_value_through(): msg = _msg(path="Direct", hops=0) assert format_piped_template("{d|hops_min:abc}", {"d": "x"}, message=msg) == "x" assert format_piped_template("{d|hops_min:-1}", {"d": "x"}, message=msg) == "x" + + +@pytest.mark.unit +def test_unknown_filter_passes_value_through_and_warns(): + logger = Mock() + out = format_piped_template("{x|nope:1}", {"x": "hi"}, logger=logger) + assert out == "hi" + logger.warning.assert_called_once() + + +@pytest.mark.unit +def test_empty_braces_are_left_literal(): + assert format_piped_template("a{}b", {}) == "a{}b" + + +@pytest.mark.unit +def test_unterminated_placeholder_is_left_literal(): + assert format_piped_template("a {oops no close", {"x": "hi"}) == "a {oops no close" + + +@pytest.mark.unit +def test_a_malformed_placeholder_does_not_block_a_later_valid_one(): + assert format_piped_template("{} then {x}", {"x": "hi"}) == "{} then hi" + + +@pytest.mark.unit +def test_quoted_string_literal_is_used_verbatim(): + assert format_piped_template('{"hello world"}', {}) == "hello world" + + +@pytest.mark.unit +def test_quoted_string_literal_substitutes_a_nested_field(): + assert format_piped_template('{"Hello {name}!"}', {"name": "Alice"}) == "Hello Alice!" + + +@pytest.mark.unit +def test_quoted_string_literal_substitutes_multiple_nested_fields(): + assert format_piped_template('{"{a}-{b}"}', {"a": "x", "b": "y"}) == "x-y" + + +@pytest.mark.unit +def test_quoted_string_literal_nested_field_missing_renders_empty(): + assert format_piped_template('{"Hi {ghost}"}', {}) == "Hi " + + +@pytest.mark.unit +def test_quoted_string_literal_supports_escaped_quotes_and_backslashes(): + assert format_piped_template('{"She said \\"hi\\""}', {}) == 'She said "hi"' + assert format_piped_template('{"a\\\\b"}', {}) == "a\\b" + + +@pytest.mark.unit +def test_quoted_string_literal_can_be_filtered(): + msg = _msg(path="01,02 (2 hops)", hops=2) + assert format_piped_template('{"{d}"|hops_min:1}', {"d": "12.4km"}, message=msg) == "12.4km" + assert format_piped_template('{"{d}"|hops_min:5}', {"d": "12.4km"}, message=msg) == "" + + +@pytest.mark.unit +def test_nested_placeholder_inside_a_quoted_literal_can_carry_its_own_filter(): + msg = _msg(path="01,02 (2 hops)", hops=2) + assert format_piped_template('{"Dist: {d|hops_min:1}"}', {"d": "12.4km"}, message=msg) == "Dist: 12.4km" + assert format_piped_template('{"Dist: {d|hops_min:5}"}', {"d": "12.4km"}, message=msg) == "Dist: " + + +@pytest.mark.unit +def test_quoted_filter_argument_with_a_nested_placeholder_does_not_close_early(): + """Regression: a quoted filter arg's own '}' (from a nested {field}) must not be + mistaken for the placeholder's closing brace and truncate the rest of the chain.""" + template = ( + '{packet_hash | if_notempty: ' + '"https://analyzer.example.net/#/packets/{packet_hash}?obs=1620457" ' + '| shorten_url}' + ) + assert format_piped_template(template, {"packet_hash": ""}) == "" + assert format_piped_template(template, {"packet_hash": "ABCDEF12"}) == ( + "https://analyzer.example.net/#/packets/ABCDEF12?obs=1620457" + )