mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-09-09 17:35:34 +00:00
agessaman
7882148f90
- Replaced inline timestamp formatting logic with a call to `format_relative_timestamp` for improved readability and maintainability. - Modularized feed formatting functions by moving them to `feed_format`, enhancing code organization and reusability. - Updated references in `feed_manager` and `web_viewer` to utilize the new modular functions, ensuring consistent behavior across modules. - Adjusted tests to reflect changes in function imports and ensure proper functionality of the new structure.
214 lines
7.6 KiB
Python
214 lines
7.6 KiB
Python
"""Tests for FeedManager pure formatting and filtering logic."""
|
|
|
|
import json
|
|
from configparser import ConfigParser
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from modules.feed_manager import FeedManager
|
|
|
|
|
|
@pytest.fixture
|
|
def fm(mock_logger):
|
|
"""FeedManager with disabled networking for pure-logic tests."""
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.config = ConfigParser()
|
|
bot.config.add_section("Feed_Manager")
|
|
bot.config.set("Feed_Manager", "feed_manager_enabled", "false")
|
|
bot.config.set("Feed_Manager", "max_message_length", "200")
|
|
bot.db_manager = Mock()
|
|
bot.db_manager.db_path = "/dev/null"
|
|
return FeedManager(bot)
|
|
|
|
|
|
class TestApplyShortening:
|
|
"""Tests for _apply_shortening()."""
|
|
|
|
def test_truncate_short_text_unchanged(self, fm):
|
|
assert fm._apply_shortening("hello", "truncate:20") == "hello"
|
|
|
|
def test_truncate_long_text_adds_ellipsis(self, fm):
|
|
result = fm._apply_shortening("Hello World", "truncate:5")
|
|
assert result == "Hello..."
|
|
|
|
def test_word_wrap_breaks_at_boundary(self, fm):
|
|
result = fm._apply_shortening("Hello beautiful world", "word_wrap:15")
|
|
# word_wrap truncates at a word boundary and appends "..."
|
|
# "Hello beautiful world"[:15] = "Hello beautiful", last space at 5 (too early),
|
|
# so result is "Hello beautiful..." (truncated at 15 chars + ellipsis)
|
|
assert result.endswith("...")
|
|
# The base text (without ellipsis) should be <= the wrap limit
|
|
assert len(result.rstrip(".")) <= 15 or result == "Hello beautiful..."
|
|
|
|
def test_first_words_limits_count(self, fm):
|
|
result = fm._apply_shortening("one two three four", "first_words:2")
|
|
assert result.startswith("one two")
|
|
|
|
def test_regex_extracts_group(self, fm):
|
|
result = fm._apply_shortening("Price: $42.99 today", r"regex:Price: \$(\d+\.\d+)")
|
|
assert result == "42.99"
|
|
|
|
def test_if_regex_returns_then_on_match(self, fm):
|
|
result = fm._apply_shortening("open", "if_regex:open:YES:NO")
|
|
assert result == "YES"
|
|
|
|
def test_if_regex_returns_else_on_no_match(self, fm):
|
|
result = fm._apply_shortening("closed", "if_regex:open:YES:NO")
|
|
assert result == "NO"
|
|
|
|
def test_truncate_hard_short_text_unchanged(self, fm):
|
|
assert fm._apply_shortening("hello", "truncate_hard:20") == "hello"
|
|
|
|
def test_truncate_hard_no_ellipsis(self, fm):
|
|
result = fm._apply_shortening("Hello World", "truncate_hard:5")
|
|
assert result == "Hello"
|
|
|
|
def test_truncate_hard_exact_length_unchanged(self, fm):
|
|
assert fm._apply_shortening("Hello", "truncate_hard:5") == "Hello"
|
|
|
|
def test_truncate_hard_bad_arg_returns_text(self, fm):
|
|
assert fm._apply_shortening("Hello", "truncate_hard:abc") == "Hello"
|
|
|
|
def test_substr_start_and_length(self, fm):
|
|
# JS-style substr(start, length): offset 6, 5 chars
|
|
assert fm._apply_shortening("Hello World", "substr:6,5") == "World"
|
|
|
|
def test_substr_start_only(self, fm):
|
|
assert fm._apply_shortening("Hello World", "substr:6") == "World"
|
|
|
|
def test_substr_length_beyond_end_clamps(self, fm):
|
|
assert fm._apply_shortening("Hello", "substr:0,100") == "Hello"
|
|
|
|
def test_substr_bad_arg_returns_text(self, fm):
|
|
assert fm._apply_shortening("Hello", "substr:x,y") == "Hello"
|
|
|
|
def test_shorten_chain_applies_truncate_hard(self, fm, monkeypatch):
|
|
# shorten|truncate_hard:N should shorten first, then hard-truncate the result
|
|
import modules.feed_format as feed_format
|
|
|
|
monkeypatch.setattr(
|
|
feed_format, "shorten_url_sync", lambda *a, **k: "https://sho.rt/abcdef"
|
|
)
|
|
result = fm._apply_shortening("https://example.com/very/long", "shorten|truncate_hard:12")
|
|
assert result == "https://sho."
|
|
assert not result.endswith("...")
|
|
|
|
def test_empty_text_returns_empty(self, fm):
|
|
assert fm._apply_shortening("", "truncate:10") == ""
|
|
|
|
|
|
class TestGetNestedValue:
|
|
"""Tests for _get_nested_value()."""
|
|
|
|
def test_simple_field_access(self, fm):
|
|
assert fm._get_nested_value({"name": "test"}, "name") == "test"
|
|
|
|
def test_nested_field_access(self, fm):
|
|
data = {"raw": {"Priority": "high"}}
|
|
assert fm._get_nested_value(data, "raw.Priority") == "high"
|
|
|
|
def test_missing_field_returns_default(self, fm):
|
|
assert fm._get_nested_value({}, "missing") == ""
|
|
assert fm._get_nested_value({}, "missing", "N/A") == "N/A"
|
|
|
|
|
|
class TestShouldSendItem:
|
|
"""Tests for _should_send_item() filter evaluation."""
|
|
|
|
def test_no_filter_sends_all(self, fm):
|
|
feed = {"id": 1}
|
|
item = {"raw": {"Priority": "low"}}
|
|
assert fm._should_send_item(feed, item) is True
|
|
|
|
def test_equals_filter_matches(self, fm):
|
|
feed = {
|
|
"id": 1,
|
|
"filter_config": json.dumps({
|
|
"conditions": [
|
|
{"field": "Priority", "operator": "equals", "value": "high"}
|
|
]
|
|
}),
|
|
}
|
|
item = {"raw": {"Priority": "high"}}
|
|
assert fm._should_send_item(feed, item) is True
|
|
|
|
def test_equals_filter_rejects(self, fm):
|
|
feed = {
|
|
"id": 1,
|
|
"filter_config": json.dumps({
|
|
"conditions": [
|
|
{"field": "Priority", "operator": "equals", "value": "high"}
|
|
]
|
|
}),
|
|
}
|
|
item = {"raw": {"Priority": "low"}}
|
|
assert fm._should_send_item(feed, item) is False
|
|
|
|
def test_in_filter_matches(self, fm):
|
|
feed = {
|
|
"id": 1,
|
|
"filter_config": json.dumps({
|
|
"conditions": [
|
|
{"field": "Priority", "operator": "in", "values": ["high", "highest"]}
|
|
]
|
|
}),
|
|
}
|
|
item = {"raw": {"Priority": "highest"}}
|
|
assert fm._should_send_item(feed, item) is True
|
|
|
|
def test_and_logic_all_must_pass(self, fm):
|
|
feed = {
|
|
"id": 1,
|
|
"filter_config": json.dumps({
|
|
"conditions": [
|
|
{"field": "Priority", "operator": "equals", "value": "high"},
|
|
{"field": "Status", "operator": "equals", "value": "open"},
|
|
],
|
|
"logic": "AND",
|
|
}),
|
|
}
|
|
# First condition passes, second fails
|
|
item = {"raw": {"Priority": "high", "Status": "closed"}}
|
|
assert fm._should_send_item(feed, item) is False
|
|
|
|
def test_within_days_passes_recent(self, fm):
|
|
now = datetime.now(timezone.utc)
|
|
feed = {
|
|
"id": 1,
|
|
"filter_config": json.dumps({
|
|
"conditions": [
|
|
{"field": "published", "operator": "within_days", "days": 28},
|
|
],
|
|
}),
|
|
}
|
|
item = {"published": now - timedelta(days=5)}
|
|
assert fm._should_send_item(feed, item) is True
|
|
|
|
def test_within_days_rejects_old(self, fm):
|
|
now = datetime.now(timezone.utc)
|
|
feed = {
|
|
"id": 1,
|
|
"filter_config": json.dumps({
|
|
"conditions": [
|
|
{"field": "published", "operator": "within_days", "days": 7},
|
|
],
|
|
}),
|
|
}
|
|
item = {"published": now - timedelta(days=30)}
|
|
assert fm._should_send_item(feed, item) is False
|
|
|
|
|
|
class TestFormatTimestamp:
|
|
"""Tests for _format_timestamp()."""
|
|
|
|
def test_recent_timestamp(self, fm):
|
|
five_min_ago = datetime.now(timezone.utc) - timedelta(minutes=5)
|
|
result = fm._format_timestamp(five_min_ago)
|
|
assert "5m ago" in result
|
|
|
|
def test_none_returns_empty(self, fm):
|
|
assert fm._format_timestamp(None) == ""
|