3955935da4
Follow YouTube channels via Atom feeds with !yt follow/unfollow/list/check. Resolves any YouTube URL to a channel ID, polls for new videos, and announces them in IRC channels.
1106 lines
36 KiB
Python
1106 lines
36 KiB
Python
"""Tests for the YouTube channel follow plugin."""
|
|
|
|
import asyncio
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from derp.irc import Message
|
|
|
|
# plugins/ is not a Python package -- load the module from file path
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"plugins.youtube", Path(__file__).resolve().parent.parent / "plugins" / "youtube.py",
|
|
)
|
|
_mod = importlib.util.module_from_spec(_spec)
|
|
sys.modules[_spec.name] = _mod
|
|
_spec.loader.exec_module(_mod)
|
|
|
|
from plugins.youtube import ( # noqa: E402
|
|
_MAX_ANNOUNCE,
|
|
_channels,
|
|
_delete,
|
|
_derive_name,
|
|
_errors,
|
|
_extract_channel_id,
|
|
_is_youtube_url,
|
|
_load,
|
|
_parse_feed,
|
|
_poll_once,
|
|
_pollers,
|
|
_restore,
|
|
_save,
|
|
_start_poller,
|
|
_state_key,
|
|
_stop_poller,
|
|
_truncate,
|
|
_validate_name,
|
|
cmd_yt,
|
|
on_connect,
|
|
)
|
|
|
|
# -- Fixtures ----------------------------------------------------------------
|
|
|
|
YT_ATOM_FEED = b"""\
|
|
<?xml version="1.0" encoding="UTF-8"?>
|
|
<feed xmlns:yt="http://www.youtube.com/xml/schemas/2015"
|
|
xmlns="http://www.w3.org/2005/Atom">
|
|
<title>3Blue1Brown - Videos</title>
|
|
<author><name>3Blue1Brown</name></author>
|
|
<entry>
|
|
<id>yt:video:abc123</id>
|
|
<yt:videoId>abc123</yt:videoId>
|
|
<title>Linear Algebra</title>
|
|
<link rel="alternate" href="https://www.youtube.com/watch?v=abc123"/>
|
|
</entry>
|
|
<entry>
|
|
<id>yt:video:def456</id>
|
|
<yt:videoId>def456</yt:videoId>
|
|
<title>Calculus</title>
|
|
<link rel="alternate" href="https://www.youtube.com/watch?v=def456"/>
|
|
</entry>
|
|
<entry>
|
|
<id>yt:video:ghi789</id>
|
|
<yt:videoId>ghi789</yt:videoId>
|
|
<title>Neural Networks</title>
|
|
<link rel="alternate" href="https://www.youtube.com/watch?v=ghi789"/>
|
|
</entry>
|
|
</feed>
|
|
"""
|
|
|
|
YT_ATOM_NO_VIDEOID = b"""\
|
|
<?xml version="1.0" encoding="UTF-8"?>
|
|
<feed xmlns="http://www.w3.org/2005/Atom">
|
|
<title>Test Channel</title>
|
|
<author><name>Test Channel</name></author>
|
|
<entry>
|
|
<id>yt:video:xyz</id>
|
|
<title>Fallback Link</title>
|
|
<link rel="alternate" href="https://www.youtube.com/watch?v=xyz"/>
|
|
</entry>
|
|
</feed>
|
|
"""
|
|
|
|
YT_ATOM_EMPTY = b"""\
|
|
<?xml version="1.0" encoding="UTF-8"?>
|
|
<feed xmlns:yt="http://www.youtube.com/xml/schemas/2015"
|
|
xmlns="http://www.w3.org/2005/Atom">
|
|
<title>Empty Channel</title>
|
|
<author><name>Empty Channel</name></author>
|
|
</feed>
|
|
"""
|
|
|
|
FAKE_YT_PAGE_BROWSE = b"""\
|
|
<!DOCTYPE html><html><head></head><body>
|
|
<script>var ytInitialData = {"browseId":"UCYO_jab_esuFRV4b17AJtAw"};</script>
|
|
</body></html>
|
|
"""
|
|
|
|
FAKE_YT_PAGE_CHANNELID = b"""\
|
|
<!DOCTYPE html><html><head></head><body>
|
|
<script>var ytInitialData = {"channelId":"UCsXVk37bltHxD1rDPwtNM8Q"};</script>
|
|
</body></html>
|
|
"""
|
|
|
|
FAKE_YT_PAGE_NO_ID = b"""\
|
|
<!DOCTYPE html><html><head></head><body>No channel here</body></html>
|
|
"""
|
|
|
|
|
|
# -- Helpers -----------------------------------------------------------------
|
|
|
|
class _FakeState:
|
|
"""In-memory stand-in for bot.state."""
|
|
|
|
def __init__(self):
|
|
self._store: dict[str, dict[str, str]] = {}
|
|
|
|
def get(self, plugin: str, key: str, default: str | None = None) -> str | None:
|
|
return self._store.get(plugin, {}).get(key, default)
|
|
|
|
def set(self, plugin: str, key: str, value: str) -> None:
|
|
self._store.setdefault(plugin, {})[key] = value
|
|
|
|
def delete(self, plugin: str, key: str) -> bool:
|
|
try:
|
|
del self._store[plugin][key]
|
|
return True
|
|
except KeyError:
|
|
return False
|
|
|
|
def keys(self, plugin: str) -> list[str]:
|
|
return sorted(self._store.get(plugin, {}).keys())
|
|
|
|
|
|
class _FakeBot:
|
|
"""Minimal bot stand-in that captures sent/replied messages."""
|
|
|
|
def __init__(self, *, admin: bool = False):
|
|
self.sent: list[tuple[str, str]] = []
|
|
self.replied: list[str] = []
|
|
self.state = _FakeState()
|
|
self._admin = admin
|
|
|
|
async def send(self, target: str, text: str) -> None:
|
|
self.sent.append((target, text))
|
|
|
|
async def reply(self, message, text: str) -> None:
|
|
self.replied.append(text)
|
|
|
|
def _is_admin(self, message) -> bool:
|
|
return self._admin
|
|
|
|
|
|
def _msg(text: str, nick: str = "alice", target: str = "#test") -> Message:
|
|
"""Create a channel PRIVMSG."""
|
|
return Message(
|
|
raw="", prefix=f"{nick}!~{nick}@host", nick=nick,
|
|
command="PRIVMSG", params=[target, text], tags={},
|
|
)
|
|
|
|
|
|
def _pm(text: str, nick: str = "alice") -> Message:
|
|
"""Create a private PRIVMSG."""
|
|
return Message(
|
|
raw="", prefix=f"{nick}!~{nick}@host", nick=nick,
|
|
command="PRIVMSG", params=["botname", text], tags={},
|
|
)
|
|
|
|
|
|
def _clear() -> None:
|
|
"""Reset module-level state between tests."""
|
|
for task in _pollers.values():
|
|
if task and not task.done():
|
|
task.cancel()
|
|
_pollers.clear()
|
|
_channels.clear()
|
|
_errors.clear()
|
|
|
|
|
|
def _fake_fetch_ok(url, etag="", last_modified=""):
|
|
"""Fake fetch that returns YT_ATOM_FEED."""
|
|
return {
|
|
"status": 200,
|
|
"body": YT_ATOM_FEED,
|
|
"etag": '"abc"',
|
|
"last_modified": "Sat, 15 Feb 2026 12:00:00 GMT",
|
|
"error": "",
|
|
}
|
|
|
|
|
|
def _fake_fetch_error(url, etag="", last_modified=""):
|
|
"""Fake fetch that returns an error."""
|
|
return {
|
|
"status": 0,
|
|
"body": b"",
|
|
"etag": "",
|
|
"last_modified": "",
|
|
"error": "Connection refused",
|
|
}
|
|
|
|
|
|
def _fake_fetch_304(url, etag="", last_modified=""):
|
|
"""Fake fetch that returns 304 Not Modified."""
|
|
return {
|
|
"status": 304,
|
|
"body": b"",
|
|
"etag": etag,
|
|
"last_modified": last_modified,
|
|
"error": "",
|
|
}
|
|
|
|
|
|
def _fake_resolve_ok(url):
|
|
"""Fake page scrape returning a channel ID."""
|
|
return "UCYO_jab_esuFRV4b17AJtAw"
|
|
|
|
|
|
def _fake_resolve_fail(url):
|
|
"""Fake page scrape returning None."""
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestIsYoutubeUrl
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestIsYoutubeUrl:
|
|
def test_standard(self):
|
|
assert _is_youtube_url("https://www.youtube.com/watch?v=abc") is True
|
|
|
|
def test_short(self):
|
|
assert _is_youtube_url("https://youtu.be/abc") is True
|
|
|
|
def test_mobile(self):
|
|
assert _is_youtube_url("https://m.youtube.com/watch?v=abc") is True
|
|
|
|
def test_no_www(self):
|
|
assert _is_youtube_url("https://youtube.com/@handle") is True
|
|
|
|
def test_not_youtube(self):
|
|
assert _is_youtube_url("https://example.com/video") is False
|
|
|
|
def test_invalid(self):
|
|
assert _is_youtube_url("not a url") is False
|
|
|
|
def test_empty(self):
|
|
assert _is_youtube_url("") is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestExtractChannelId
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestExtractChannelId:
|
|
def test_direct_channel_url(self):
|
|
url = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw"
|
|
assert _extract_channel_id(url) == "UCYO_jab_esuFRV4b17AJtAw"
|
|
|
|
def test_channel_url_with_path(self):
|
|
url = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw/videos"
|
|
assert _extract_channel_id(url) == "UCYO_jab_esuFRV4b17AJtAw"
|
|
|
|
def test_handle_url_no_match(self):
|
|
url = "https://www.youtube.com/@3blue1brown"
|
|
assert _extract_channel_id(url) is None
|
|
|
|
def test_video_url_no_match(self):
|
|
url = "https://www.youtube.com/watch?v=abc123"
|
|
assert _extract_channel_id(url) is None
|
|
|
|
def test_short_url_no_match(self):
|
|
url = "https://youtu.be/abc123"
|
|
assert _extract_channel_id(url) is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestDeriveName
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDeriveName:
|
|
def test_simple_title(self):
|
|
assert _derive_name("3Blue1Brown") == "3blue1brown"
|
|
|
|
def test_title_with_spaces(self):
|
|
assert _derive_name("Linus Tech Tips") == "linus-tech-tips"
|
|
|
|
def test_title_with_special(self):
|
|
assert _derive_name("Tom Scott!") == "tom-scott"
|
|
|
|
def test_empty_title(self):
|
|
assert _derive_name("") == "yt"
|
|
|
|
def test_long_title_truncated(self):
|
|
result = _derive_name("A" * 50)
|
|
assert len(result) <= 20
|
|
|
|
def test_only_special_chars(self):
|
|
assert _derive_name("!!!") == "yt"
|
|
|
|
def test_validates(self):
|
|
result = _derive_name("Some Channel Name")
|
|
assert _validate_name(result)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestTruncate
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestTruncate:
|
|
def test_short_text_unchanged(self):
|
|
assert _truncate("hello", 80) == "hello"
|
|
|
|
def test_exact_length_unchanged(self):
|
|
text = "a" * 80
|
|
assert _truncate(text, 80) == text
|
|
|
|
def test_long_text_truncated(self):
|
|
text = "a" * 100
|
|
result = _truncate(text, 80)
|
|
assert len(result) == 80
|
|
assert result.endswith("...")
|
|
|
|
def test_default_max_length(self):
|
|
text = "a" * 100
|
|
result = _truncate(text)
|
|
assert len(result) == 80
|
|
|
|
def test_trailing_space_stripped(self):
|
|
text = "word " * 20
|
|
result = _truncate(text, 20)
|
|
assert not result.endswith(" ...")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestParseFeed
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestParseFeed:
|
|
def test_parses_entries(self):
|
|
channel_name, items = _parse_feed(YT_ATOM_FEED)
|
|
assert channel_name == "3Blue1Brown"
|
|
assert len(items) == 3
|
|
assert items[0]["id"] == "yt:video:abc123"
|
|
assert items[0]["title"] == "Linear Algebra"
|
|
assert items[0]["link"] == "https://www.youtube.com/watch?v=abc123"
|
|
|
|
def test_builds_link_from_videoid(self):
|
|
_, items = _parse_feed(YT_ATOM_FEED)
|
|
assert items[1]["link"] == "https://www.youtube.com/watch?v=def456"
|
|
|
|
def test_fallback_link_no_videoid(self):
|
|
_, items = _parse_feed(YT_ATOM_NO_VIDEOID)
|
|
assert len(items) == 1
|
|
assert items[0]["link"] == "https://www.youtube.com/watch?v=xyz"
|
|
|
|
def test_empty_feed(self):
|
|
channel_name, items = _parse_feed(YT_ATOM_EMPTY)
|
|
assert channel_name == "Empty Channel"
|
|
assert items == []
|
|
|
|
def test_author_name_preferred(self):
|
|
channel_name, _ = _parse_feed(YT_ATOM_FEED)
|
|
assert channel_name == "3Blue1Brown"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestResolveChannel
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestResolveChannel:
|
|
def test_browse_id_preferred(self):
|
|
from plugins.youtube import _PAGE_BROWSE_RE
|
|
m = _PAGE_BROWSE_RE.search(FAKE_YT_PAGE_BROWSE)
|
|
assert m is not None
|
|
assert m.group(1).decode() == "UCYO_jab_esuFRV4b17AJtAw"
|
|
|
|
def test_channelid_fallback(self):
|
|
from plugins.youtube import _PAGE_CHANNEL_RE
|
|
m = _PAGE_CHANNEL_RE.search(FAKE_YT_PAGE_CHANNELID)
|
|
assert m is not None
|
|
assert m.group(1).decode() == "UCsXVk37bltHxD1rDPwtNM8Q"
|
|
|
|
def test_no_match_in_page(self):
|
|
from plugins.youtube import _PAGE_BROWSE_RE, _PAGE_CHANNEL_RE
|
|
assert _PAGE_BROWSE_RE.search(FAKE_YT_PAGE_NO_ID) is None
|
|
assert _PAGE_CHANNEL_RE.search(FAKE_YT_PAGE_NO_ID) is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestStateHelpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestStateHelpers:
|
|
def test_save_and_load(self):
|
|
bot = _FakeBot()
|
|
data = {"feed_url": "https://example.com/feed", "name": "test"}
|
|
_save(bot, "#ch:test", data)
|
|
loaded = _load(bot, "#ch:test")
|
|
assert loaded == data
|
|
|
|
def test_load_missing(self):
|
|
bot = _FakeBot()
|
|
assert _load(bot, "nonexistent") is None
|
|
|
|
def test_delete(self):
|
|
bot = _FakeBot()
|
|
_save(bot, "#ch:test", {"name": "test"})
|
|
_delete(bot, "#ch:test")
|
|
assert _load(bot, "#ch:test") is None
|
|
|
|
def test_state_key(self):
|
|
assert _state_key("#ops", "3b1b") == "#ops:3b1b"
|
|
|
|
def test_load_invalid_json(self):
|
|
bot = _FakeBot()
|
|
bot.state.set("yt", "bad", "not json{{{")
|
|
assert _load(bot, "bad") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestCmdYtFollow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCmdYtFollow:
|
|
def test_follow_with_channel_url(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with (
|
|
patch.object(_mod, "_fetch_feed", _fake_fetch_ok),
|
|
):
|
|
url = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw"
|
|
await cmd_yt(bot, _msg(f"!yt follow {url} 3b1b"))
|
|
await asyncio.sleep(0)
|
|
assert len(bot.replied) == 1
|
|
assert "Following '3b1b'" in bot.replied[0]
|
|
assert "3 existing videos" in bot.replied[0]
|
|
data = _load(bot, "#test:3b1b")
|
|
assert data is not None
|
|
assert data["channel_id"] == "UCYO_jab_esuFRV4b17AJtAw"
|
|
assert data["name"] == "3b1b"
|
|
assert data["channel"] == "#test"
|
|
assert len(data["seen"]) == 3
|
|
assert "#test:3b1b" in _pollers
|
|
_stop_poller("#test:3b1b")
|
|
await asyncio.sleep(0)
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_follow_with_handle_url(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with (
|
|
patch.object(_mod, "_resolve_channel", _fake_resolve_ok),
|
|
patch.object(_mod, "_fetch_feed", _fake_fetch_ok),
|
|
):
|
|
await cmd_yt(bot, _msg("!yt follow https://www.youtube.com/@3blue1brown"))
|
|
await asyncio.sleep(0)
|
|
assert len(bot.replied) == 1
|
|
assert "Following" in bot.replied[0]
|
|
_clear()
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_follow_derives_name(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with (
|
|
patch.object(_mod, "_fetch_feed", _fake_fetch_ok),
|
|
):
|
|
url = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw"
|
|
await cmd_yt(bot, _msg(f"!yt follow {url}"))
|
|
await asyncio.sleep(0)
|
|
# Name derived from channel title "3Blue1Brown"
|
|
assert "Following '3blue1brown'" in bot.replied[0]
|
|
_clear()
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_follow_with_video_url_no_scheme(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with (
|
|
patch.object(_mod, "_resolve_channel", _fake_resolve_ok),
|
|
patch.object(_mod, "_fetch_feed", _fake_fetch_ok),
|
|
):
|
|
await cmd_yt(bot, _msg("!yt follow youtube.com/watch?v=abc123 test"))
|
|
await asyncio.sleep(0)
|
|
assert len(bot.replied) == 1
|
|
assert "Following 'test'" in bot.replied[0]
|
|
data = _load(bot, "#test:test")
|
|
assert data is not None
|
|
assert data["channel_id"] == "UCYO_jab_esuFRV4b17AJtAw"
|
|
_clear()
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_follow_requires_admin(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=False)
|
|
asyncio.run(cmd_yt(bot, _msg("!yt follow https://www.youtube.com/@test")))
|
|
assert "Permission denied" in bot.replied[0]
|
|
|
|
def test_follow_requires_channel(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
asyncio.run(cmd_yt(bot, _pm("!yt follow https://www.youtube.com/@test")))
|
|
assert "Use this command in a channel" in bot.replied[0]
|
|
|
|
def test_follow_not_youtube_url(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
asyncio.run(cmd_yt(bot, _msg("!yt follow https://example.com/video")))
|
|
assert "Not a YouTube URL" in bot.replied[0]
|
|
|
|
def test_follow_resolve_fails(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_resolve_channel", _fake_resolve_fail):
|
|
await cmd_yt(bot, _msg("!yt follow https://www.youtube.com/@nonexistent"))
|
|
assert "Could not resolve" in bot.replied[0]
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_follow_invalid_name(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_ok):
|
|
url = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw"
|
|
await cmd_yt(bot, _msg(f"!yt follow {url} BAD!"))
|
|
assert "Invalid name" in bot.replied[0]
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_follow_duplicate(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_ok):
|
|
url = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw"
|
|
await cmd_yt(bot, _msg(f"!yt follow {url} dupe"))
|
|
await asyncio.sleep(0)
|
|
bot.replied.clear()
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_ok):
|
|
url2 = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw"
|
|
await cmd_yt(bot, _msg(f"!yt follow {url2} dupe"))
|
|
assert "already exists" in bot.replied[0]
|
|
_clear()
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_follow_fetch_error(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_error):
|
|
url = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw"
|
|
await cmd_yt(bot, _msg(f"!yt follow {url}"))
|
|
assert "Feed fetch failed" in bot.replied[0]
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_follow_no_url(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
asyncio.run(cmd_yt(bot, _msg("!yt follow")))
|
|
assert "Usage:" in bot.replied[0]
|
|
|
|
def test_follow_channel_limit(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
for i in range(20):
|
|
_save(bot, f"#test:ch{i}", {"name": f"ch{i}", "channel": "#test"})
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_ok):
|
|
url = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw"
|
|
await cmd_yt(bot, _msg(f"!yt follow {url} overflow"))
|
|
assert "limit reached" in bot.replied[0]
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_follow_prepends_https(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with (
|
|
patch.object(_mod, "_resolve_channel", _fake_resolve_ok),
|
|
patch.object(_mod, "_fetch_feed", _fake_fetch_ok),
|
|
):
|
|
await cmd_yt(bot, _msg("!yt follow youtube.com/@test yttest"))
|
|
await asyncio.sleep(0)
|
|
data = _load(bot, "#test:yttest")
|
|
assert data is not None
|
|
assert data["channel_id"] == "UCYO_jab_esuFRV4b17AJtAw"
|
|
_clear()
|
|
|
|
asyncio.run(inner())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestCmdYtUnfollow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCmdYtUnfollow:
|
|
def test_unfollow_success(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_ok):
|
|
url = "https://www.youtube.com/channel/UCYO_jab_esuFRV4b17AJtAw"
|
|
await cmd_yt(bot, _msg(f"!yt follow {url} delfeed"))
|
|
await asyncio.sleep(0)
|
|
bot.replied.clear()
|
|
await cmd_yt(bot, _msg("!yt unfollow delfeed"))
|
|
assert "Unfollowed 'delfeed'" in bot.replied[0]
|
|
assert _load(bot, "#test:delfeed") is None
|
|
assert "#test:delfeed" not in _pollers
|
|
await asyncio.sleep(0)
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_unfollow_requires_admin(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=False)
|
|
asyncio.run(cmd_yt(bot, _msg("!yt unfollow somefeed")))
|
|
assert "Permission denied" in bot.replied[0]
|
|
|
|
def test_unfollow_requires_channel(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
asyncio.run(cmd_yt(bot, _pm("!yt unfollow somefeed")))
|
|
assert "Use this command in a channel" in bot.replied[0]
|
|
|
|
def test_unfollow_nonexistent(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
asyncio.run(cmd_yt(bot, _msg("!yt unfollow nosuchfeed")))
|
|
assert "No channel" in bot.replied[0]
|
|
|
|
def test_unfollow_no_name(self):
|
|
_clear()
|
|
bot = _FakeBot(admin=True)
|
|
asyncio.run(cmd_yt(bot, _msg("!yt unfollow")))
|
|
assert "Usage:" in bot.replied[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestCmdYtList
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCmdYtList:
|
|
def test_list_empty(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
asyncio.run(cmd_yt(bot, _msg("!yt list")))
|
|
assert "No YouTube channels" in bot.replied[0]
|
|
|
|
def test_list_populated(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
_save(bot, "#test:3b1b", {
|
|
"name": "3b1b", "channel": "#test",
|
|
"feed_url": "https://yt.com/feed", "last_error": "",
|
|
})
|
|
_save(bot, "#test:veritasium", {
|
|
"name": "veritasium", "channel": "#test",
|
|
"feed_url": "https://yt.com/feed2", "last_error": "",
|
|
})
|
|
asyncio.run(cmd_yt(bot, _msg("!yt list")))
|
|
assert "YouTube:" in bot.replied[0]
|
|
assert "3b1b" in bot.replied[0]
|
|
assert "veritasium" in bot.replied[0]
|
|
|
|
def test_list_shows_error(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
_save(bot, "#test:broken", {
|
|
"name": "broken", "channel": "#test",
|
|
"feed_url": "https://yt.com/feed", "last_error": "Connection refused",
|
|
})
|
|
asyncio.run(cmd_yt(bot, _msg("!yt list")))
|
|
assert "broken (error)" in bot.replied[0]
|
|
|
|
def test_list_requires_channel(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
asyncio.run(cmd_yt(bot, _pm("!yt list")))
|
|
assert "Use this command in a channel" in bot.replied[0]
|
|
|
|
def test_list_only_this_channel(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
_save(bot, "#test:mine", {
|
|
"name": "mine", "channel": "#test",
|
|
"feed_url": "https://yt.com/feed", "last_error": "",
|
|
})
|
|
_save(bot, "#other:theirs", {
|
|
"name": "theirs", "channel": "#other",
|
|
"feed_url": "https://yt.com/feed2", "last_error": "",
|
|
})
|
|
asyncio.run(cmd_yt(bot, _msg("!yt list")))
|
|
assert "mine" in bot.replied[0]
|
|
assert "theirs" not in bot.replied[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestCmdYtCheck
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCmdYtCheck:
|
|
def test_check_success(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "chk", "channel": "#test",
|
|
"interval": 600, "seen": ["yt:video:abc123", "yt:video:def456", "yt:video:ghi789"],
|
|
"last_poll": "", "last_error": "", "etag": "", "last_modified": "",
|
|
"title": "Test", "channel_id": "UC123",
|
|
}
|
|
_save(bot, "#test:chk", data)
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_ok):
|
|
await cmd_yt(bot, _msg("!yt check chk"))
|
|
assert "chk: checked" in bot.replied[0]
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_check_nonexistent(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
asyncio.run(cmd_yt(bot, _msg("!yt check nope")))
|
|
assert "No channel" in bot.replied[0]
|
|
|
|
def test_check_requires_channel(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
asyncio.run(cmd_yt(bot, _pm("!yt check something")))
|
|
assert "Use this command in a channel" in bot.replied[0]
|
|
|
|
def test_check_shows_error(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "errfeed", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
_save(bot, "#test:errfeed", data)
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_error):
|
|
await cmd_yt(bot, _msg("!yt check errfeed"))
|
|
assert "error" in bot.replied[0].lower()
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_check_announces_new_items(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "news", "channel": "#test",
|
|
"interval": 600, "seen": ["yt:video:abc123"],
|
|
"last_poll": "", "last_error": "", "etag": "", "last_modified": "",
|
|
"title": "Test", "channel_id": "UC123",
|
|
}
|
|
_save(bot, "#test:news", data)
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_ok):
|
|
await cmd_yt(bot, _msg("!yt check news"))
|
|
announcements = [s for t, s in bot.sent if t == "#test"]
|
|
assert len(announcements) == 2
|
|
assert "[news]" in announcements[0]
|
|
assert "Calculus" in announcements[0]
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_check_no_name(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
asyncio.run(cmd_yt(bot, _msg("!yt check")))
|
|
assert "Usage:" in bot.replied[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestPollOnce
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPollOnce:
|
|
def test_poll_304_clears_error(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "f304", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "old err",
|
|
"etag": '"xyz"', "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
key = "#test:f304"
|
|
_save(bot, key, data)
|
|
_channels[key] = data
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_304):
|
|
await _poll_once(bot, key)
|
|
updated = _load(bot, key)
|
|
assert updated["last_error"] == ""
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_poll_error_increments(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "ferr", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
key = "#test:ferr"
|
|
_save(bot, key, data)
|
|
_channels[key] = data
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_error):
|
|
await _poll_once(bot, key)
|
|
await _poll_once(bot, key)
|
|
assert _errors[key] == 2
|
|
updated = _load(bot, key)
|
|
assert updated["last_error"] == "Connection refused"
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_poll_max_announce(self):
|
|
"""Only MAX_ANNOUNCE items are individually announced."""
|
|
_clear()
|
|
bot = _FakeBot()
|
|
entries_xml = ""
|
|
for i in range(8):
|
|
entries_xml += f"""
|
|
<entry>
|
|
<id>yt:video:vid{i}</id>
|
|
<yt:videoId>vid{i}</yt:videoId>
|
|
<title>Video {i}</title>
|
|
</entry>"""
|
|
big_feed = f"""\
|
|
<?xml version="1.0" encoding="UTF-8"?>
|
|
<feed xmlns:yt="http://www.youtube.com/xml/schemas/2015"
|
|
xmlns="http://www.w3.org/2005/Atom">
|
|
<title>Big Channel</title>
|
|
<author><name>Big Channel</name></author>
|
|
{entries_xml}
|
|
</feed>""".encode()
|
|
|
|
def fake_big(url, etag="", lm=""):
|
|
return {"status": 200, "body": big_feed, "etag": "", "last_modified": "", "error": ""}
|
|
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "big", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
key = "#test:big"
|
|
_save(bot, key, data)
|
|
_channels[key] = data
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", fake_big):
|
|
await _poll_once(bot, key, announce=True)
|
|
messages = [s for t, s in bot.sent if t == "#test"]
|
|
# 5 individual + 1 "... and N more"
|
|
assert len(messages) == _MAX_ANNOUNCE + 1
|
|
assert "... and 3 more" in messages[-1]
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_poll_no_announce_flag(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "quiet", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
key = "#test:quiet"
|
|
_save(bot, key, data)
|
|
_channels[key] = data
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_ok):
|
|
await _poll_once(bot, key, announce=False)
|
|
assert len(bot.sent) == 0
|
|
updated = _load(bot, key)
|
|
assert len(updated["seen"]) == 3
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_poll_updates_etag(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "etag", "channel": "#test",
|
|
"interval": 600,
|
|
"seen": ["yt:video:abc123", "yt:video:def456", "yt:video:ghi789"],
|
|
"last_poll": "", "last_error": "", "etag": "", "last_modified": "",
|
|
"title": "", "channel_id": "UC123",
|
|
}
|
|
key = "#test:etag"
|
|
_save(bot, key, data)
|
|
_channels[key] = data
|
|
|
|
async def inner():
|
|
with patch.object(_mod, "_fetch_feed", _fake_fetch_ok):
|
|
await _poll_once(bot, key)
|
|
updated = _load(bot, key)
|
|
assert updated["etag"] == '"abc"'
|
|
|
|
asyncio.run(inner())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestRestore
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRestore:
|
|
def test_restore_spawns_pollers(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "restored", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
_save(bot, "#test:restored", data)
|
|
|
|
async def inner():
|
|
_restore(bot)
|
|
assert "#test:restored" in _pollers
|
|
task = _pollers["#test:restored"]
|
|
assert not task.done()
|
|
_stop_poller("#test:restored")
|
|
await asyncio.sleep(0)
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_restore_skips_active(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "active", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
_save(bot, "#test:active", data)
|
|
|
|
async def inner():
|
|
dummy = asyncio.create_task(asyncio.sleep(9999))
|
|
_pollers["#test:active"] = dummy
|
|
_restore(bot)
|
|
assert _pollers["#test:active"] is dummy
|
|
dummy.cancel()
|
|
await asyncio.sleep(0)
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_restore_replaces_done_task(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "done", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
_save(bot, "#test:done", data)
|
|
|
|
async def inner():
|
|
done_task = asyncio.create_task(asyncio.sleep(0))
|
|
await done_task
|
|
_pollers["#test:done"] = done_task
|
|
_restore(bot)
|
|
new_task = _pollers["#test:done"]
|
|
assert new_task is not done_task
|
|
assert not new_task.done()
|
|
_stop_poller("#test:done")
|
|
await asyncio.sleep(0)
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_restore_skips_bad_json(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
bot.state.set("yt", "#test:bad", "not json{{{")
|
|
|
|
async def inner():
|
|
_restore(bot)
|
|
assert "#test:bad" not in _pollers
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_on_connect_calls_restore(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "conn", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
_save(bot, "#test:conn", data)
|
|
|
|
async def inner():
|
|
msg = _msg("", target="botname")
|
|
await on_connect(bot, msg)
|
|
assert "#test:conn" in _pollers
|
|
_stop_poller("#test:conn")
|
|
await asyncio.sleep(0)
|
|
|
|
asyncio.run(inner())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestPollerManagement
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPollerManagement:
|
|
def test_start_and_stop(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "mgmt", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
key = "#test:mgmt"
|
|
_save(bot, key, data)
|
|
_channels[key] = data
|
|
|
|
async def inner():
|
|
_start_poller(bot, key)
|
|
assert key in _pollers
|
|
assert not _pollers[key].done()
|
|
_stop_poller(key)
|
|
await asyncio.sleep(0)
|
|
assert key not in _pollers
|
|
assert key not in _channels
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_start_idempotent(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
data = {
|
|
"feed_url": "https://yt.com/feed", "name": "idem", "channel": "#test",
|
|
"interval": 600, "seen": [], "last_poll": "", "last_error": "",
|
|
"etag": "", "last_modified": "", "title": "", "channel_id": "UC123",
|
|
}
|
|
key = "#test:idem"
|
|
_save(bot, key, data)
|
|
_channels[key] = data
|
|
|
|
async def inner():
|
|
_start_poller(bot, key)
|
|
first = _pollers[key]
|
|
_start_poller(bot, key)
|
|
assert _pollers[key] is first
|
|
_stop_poller(key)
|
|
await asyncio.sleep(0)
|
|
|
|
asyncio.run(inner())
|
|
|
|
def test_stop_nonexistent(self):
|
|
_clear()
|
|
_stop_poller("#test:nonexistent")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestCmdYtUsage
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCmdYtUsage:
|
|
def test_no_args(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
asyncio.run(cmd_yt(bot, _msg("!yt")))
|
|
assert "Usage:" in bot.replied[0]
|
|
|
|
def test_unknown_subcommand(self):
|
|
_clear()
|
|
bot = _FakeBot()
|
|
asyncio.run(cmd_yt(bot, _msg("!yt foobar")))
|
|
assert "Usage:" in bot.replied[0]
|