From 8a7a85174a13e5e39416b79988bc0466fe1db93a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cristi=C3=A1n=20Maureira-Fredes?= Date: Wed, 12 Aug 2026 11:13:55 +0200 Subject: [PATCH] Add a pytest test suite and GitHub Actions CI Covers every comandos/*.py cog plus utils.py and configuration.py (103 tests), using lightweight fakes for discord.py objects instead of a full test harness like dpytest: - conftest.py (root): sandboxes a throwaway config.toml + logs/ dir and chdir()s into it before any comandos module is imported, since every such module instantiates the Config() singleton at import time and config.toml is git-ignored (real secrets). - tests/factories.py: fakes for members/roles/channels/messages/ attachments/bots. Uses MagicMock(spec=...) specifically where production code runs isinstance() checks (discord.Member, discord.TextChannel, discord.Interaction), plain objects elsewhere. - tests/conftest.py: fixtures wiring FloodSpam/Moderacion cogs the way on_ready() would, without needing a real discord.Client/Guild. - .github/workflows/tests.yml: runs the suite on push to main and on every pull request. Along the way this surfaced a few pre-existing bugs, documented as tests/comments rather than fixed here (out of scope for "add tests"): - archivar.archivar_canal() returns (False, None) on a write failure; since a non-empty tuple is always truthy, archivar()'s `if status:` treats that as success (test_failure_tuple_is_still_truthy). - FloodSpam.on_message re-fetches the channel via bot.get_channel(message.channel.id) instead of using message.channel directly; a cache miss makes self._msg_channel None and crashes the first .send() call downstream. - Moderacion._aceptar_mensaje's jump_url is built from self._msg_id, a single field shared across all pending submissions, rather than the specific post being accepted (vp.post_id) - a second submission arriving before the first is moderated could produce a jump_url pointing at the wrong message. --- .github/workflows/tests.yml | 28 +++ .gitignore | 1 + conftest.py | 120 +++++++++ pytest.ini | 4 + requirements-dev.txt | 4 + tests/__init__.py | 0 tests/conftest.py | 59 +++++ tests/factories.py | 204 ++++++++++++++++ tests/test_archivar.py | 116 +++++++++ tests/test_ayuda.py | 61 +++++ tests/test_configuration.py | 61 +++++ tests/test_enviar.py | 32 +++ tests/test_flood.py | 469 ++++++++++++++++++++++++++++++++++++ tests/test_limpia.py | 50 ++++ tests/test_moderacion.py | 320 ++++++++++++++++++++++++ tests/test_ping.py | 63 +++++ tests/test_utils.py | 51 ++++ utils.py | 4 +- 18 files changed, 1645 insertions(+), 2 deletions(-) create mode 100644 .github/workflows/tests.yml create mode 100644 conftest.py create mode 100644 pytest.ini create mode 100644 requirements-dev.txt create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/factories.py create mode 100644 tests/test_archivar.py create mode 100644 tests/test_ayuda.py create mode 100644 tests/test_configuration.py create mode 100644 tests/test_enviar.py create mode 100644 tests/test_flood.py create mode 100644 tests/test_limpia.py create mode 100644 tests/test_moderacion.py create mode 100644 tests/test_ping.py create mode 100644 tests/test_utils.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..5027080 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,28 @@ +name: Tests + +on: + push: + branches: [main] + pull_request: + +jobs: + pytest: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.11"] + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + + - name: Install dependencies + run: pip install -r requirements-dev.txt + + - name: Run tests + run: pytest -v diff --git a/.gitignore b/.gitignore index c857a09..7daa5fe 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ config.toml __pycache__ bot.log +.pytest_cache diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..2fc48a8 --- /dev/null +++ b/conftest.py @@ -0,0 +1,120 @@ +"""Root conftest.py. + +Every module under ``comandos/`` (and ``configuration.py`` itself) calls +``Config()`` at *import time*, and ``Config.__init__`` reads ``config.toml`` +from the current working directory, exiting the process if it's missing or +incomplete. ``config.toml`` is git-ignored (it holds real bot secrets), so it +won't exist in a fresh checkout or in CI. + +To make the test modules importable at all, we build a throwaway +``config.toml`` plus the ``logs/`` directory it expects, and ``chdir`` into +that sandbox *before* pytest imports any test module. This has to happen as +plain module-level code (not inside a fixture) because pytest imports +conftest.py files before it collects/imports the test modules that in turn +``import comandos.flood`` etc. +""" +import os +import shutil +import tempfile +from pathlib import Path + +import pytest + +_REPO_ROOT = Path(__file__).resolve().parent + +_SANDBOX = Path(tempfile.mkdtemp(prefix="pyes-bot-tests-")) +(_SANDBOX / "logs").mkdir() +(_SANDBOX / "config.toml").write_text( + """ +[bot] +token = "test-token" +id = 111111111111111111 +log_file = "logs/bot_log.csv" + +[moderation] +log_file = "logs/mod_log.csv" +channel_id = 222222222222222222 +role = "Coordinacion" +muted_role = "Muted" + +[server] +guild = 333333333333333333 + +[channels] + [channels.eventos] + main = 444444444444444444 + moderation = 555555555555555555 + submission = 666666666666666666 +""" +) + +# A couple of commands read files via relative paths at call time (not +# import time) - e.g. ping.py's "resources/llama.gif". Make sure those are +# still reachable from the sandboxed working directory. +if (_REPO_ROOT / "resources").is_dir(): + shutil.copytree(_REPO_ROOT / "resources", _SANDBOX / "resources") + +os.chdir(_SANDBOX) + + +@pytest.fixture(scope="session") +def sandbox_dir(): + """The temporary directory tests are running from.""" + return _SANDBOX + + +@pytest.fixture(scope="session") +def repo_root(): + """The actual repository root, for tests that need real project files.""" + return _REPO_ROOT + + +@pytest.fixture +def config(): + """The (singleton) Config instance, backed by the sandboxed config.toml.""" + from configuration import Config + + return Config() + + +@pytest.fixture +def isolated_logs(config, tmp_path, monkeypatch): + """Point every log file Config knows about at a fresh, empty file. + + Config is a singleton shared across the whole test session, and several + of its log paths are appended to by the code under test (e.g. + ``add_spam_message``). Without this, one test's writes would leak into + the next test that also touches those files. + """ + logs_dir = tmp_path / "logs" + logs_dir.mkdir() + + for attr, filename in [ + ("log_spam_file", "spam_log.csv"), + ("log_image_spam_file", "image_spam_log.csv"), + ("log_mod_file", "mod_log.csv"), + ("log_accepted_file", "mod_log_accepted.csv"), + ("log_rejected_file", "mod_log_rejected.csv"), + ("log_main_file", "main_log.csv"), + ("log_file", "bot_log.csv"), + ]: + path = logs_dir / filename + path.write_text("\n") + monkeypatch.setattr(config, attr, path) + + return config + + +@pytest.fixture(autouse=True) +def patched_message_delete(monkeypatch): + """The production code deletes messages via ``discord.Message.delete(message)`` + (an unbound call on the real class) rather than ``message.delete()``, so + fakes that aren't real ``discord.Message`` instances need this patched to + avoid hitting real discord.py internals/HTTP calls. + """ + import discord + from unittest.mock import AsyncMock + + mock_delete = AsyncMock() + monkeypatch.setattr(discord.Message, "delete", mock_delete) + return mock_delete diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..01d0c51 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,4 @@ +[pytest] +asyncio_mode = auto +filterwarnings = + ignore:'asyncio.iscoroutinefunction' is deprecated:DeprecationWarning:discord.ext.commands.core diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..83b2c4b --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,4 @@ +-r requirements.txt + +pytest +pytest-asyncio diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..8b07dba --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,59 @@ +import pandas as pd +import pytest + +from comandos.flood import FloodSpam +from comandos.moderacion import Moderacion +from tests.factories import make_bot, make_role, make_text_channel + + +@pytest.fixture +def coord_role(): + return make_role(name="Coordinacion", id=1) + + +@pytest.fixture +def muted_role(): + return make_role(name="Muted", id=2) + + +@pytest.fixture +def mod_channel(): + return make_text_channel(id=999, name="mod-general") + + +@pytest.fixture +def flood_cog(isolated_logs, coord_role, muted_role, mod_channel): + """A FloodSpam cog wired up the way on_ready() would, without needing a + real discord.Client/Guild.""" + bot = make_bot(channels={mod_channel.id: mod_channel}) + cog = FloodSpam(bot) + cog._coord_role = coord_role + cog._muted_role = muted_role + cog._main_mod_channel = mod_channel + return cog + + +@pytest.fixture +def moderacion_channels(config): + ids = config.CHANNELS["eventos"] + return { + "main": make_text_channel(id=ids["main"], name="eventos"), + "mod": make_text_channel(id=ids["moderation"], name="eventos-mod"), + "sub": make_text_channel(id=ids["submission"], name="envio-eventos"), + } + + +@pytest.fixture +async def moderacion_cog(isolated_logs, moderacion_channels): + """A Moderacion cog wired up the way on_ready() would (channel mapping + populated from config.CHANNELS), with an empty ``data_mod`` table.""" + from types import SimpleNamespace + + channels = {c.id: c for c in moderacion_channels.values()} + bot = make_bot(channels=channels, guild=SimpleNamespace(id=333333333333333333)) + bot.data_mod = pd.DataFrame( + columns=["date", "message_id", "channel", "author_id", "author", "message"] + ) + cog = Moderacion(bot) + await cog.on_ready() + return cog diff --git a/tests/factories.py b/tests/factories.py new file mode 100644 index 0000000..4b08e0f --- /dev/null +++ b/tests/factories.py @@ -0,0 +1,204 @@ +"""Lightweight fakes for discord.py objects used across the test suite. + +We deliberately avoid a full discord.py test harness (e.g. dpytest) and +instead build just-enough fakes: + +- Plain objects (``SimpleNamespace``/small classes) for anything the + production code only reads attributes off of or calls async methods on. +- ``unittest.mock.MagicMock(spec=...)`` for anything the code runs + ``isinstance()`` checks against - Mock's ``spec=`` makes + ``isinstance(mock, SpecClass)`` return True, which a plain fake can't do. +""" +from io import BytesIO +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import discord +from PIL import Image + +from utils import strip_message + + +def make_role(name="Coordinacion", id=1000): + return SimpleNamespace(name=name, id=id, mention=f"@{name}") + + +def make_member(name="usuario", id=123, roles=None, bot=False): + """A fake author. Uses ``spec=discord.Member`` so that the + ``isinstance(author, discord.Member)`` check in ``spam_check`` passes.""" + member = MagicMock(spec=discord.Member) + member.name = name + member.id = id + member.bot = bot + member.discriminator = "0" + member.mention = f"<@{id}>" + member.roles = roles if roles is not None else [] + member.add_roles = AsyncMock() + member.remove_roles = AsyncMock() + member.__str__.return_value = name + return member + + +def make_attachment(content_type="image/png", data=b"fake-image-bytes", filename="image.png"): + return SimpleNamespace( + content_type=content_type, + filename=filename, + read=AsyncMock(return_value=data), + ) + + +def make_thread(): + return SimpleNamespace(send=AsyncMock()) + + +async def _async_history(messages): + for m in messages: + yield m + + +def make_text_channel(id=555, name="general", thread=None, history_messages=None): + """Uses ``spec=discord.TextChannel`` for the isinstance checks in + ``archivar.py``/``limpia.py``.""" + channel = MagicMock(spec=discord.TextChannel) + channel.id = id + channel.name = name + channel.mention = f"#{name}" + channel.category_id = None + channel.send = AsyncMock() + channel.create_thread = AsyncMock(return_value=thread or make_thread()) + channel.purge = AsyncMock() + channel.delete_messages = AsyncMock() + channel.history = MagicMock( + side_effect=lambda *a, **kw: _async_history(history_messages or []) + ) + return channel + + +def make_dm_channel(): + """A channel type NOT accepted by archivar.py/limpia.py's isinstance checks.""" + return MagicMock(spec=discord.DMChannel) + + +def make_category(name="categoria", channels=None): + category = MagicMock(spec=discord.CategoryChannel) + category.name = name + category.channels = channels or [] + return category + + +def make_message( + content="", + author=None, + channel=None, + attachments=None, + mentions=None, + role_mentions=None, + id=999, +): + return SimpleNamespace( + id=id, + content=content, + author=author if author is not None else make_member(), + channel=channel if channel is not None else make_text_channel(), + attachments=attachments if attachments is not None else [], + mentions=mentions if mentions is not None else [], + role_mentions=role_mentions if role_mentions is not None else [], + ) + + +def make_bot(channels=None, guild=None, user=None, users=None, guilds=None): + """``channels``/``users`` are kept as live dicts on the returned bot (as + ``bot.channels``/``bot.users_by_id``) so tests can register more entries + later, e.g. when a message arrives on a channel ``on_message`` looks up + via ``bot.get_channel(message.channel.id)``.""" + channels = dict(channels or {}) + users = dict(users or {}) + bot = SimpleNamespace( + channels=channels, + users_by_id=users, + get_guild=lambda gid: guild, + user=user, + guilds=guilds if guilds is not None else ([guild] if guild else []), + process_commands=AsyncMock(), + ) + bot.get_channel = lambda cid: bot.channels.get(cid) + bot.get_user = lambda uid: bot.users_by_id.get(uid) + return bot + + +def make_ctx(author=None, channel=None, content="", reference=None): + """A fake ``commands.Context`` - text-command style invocation (as + opposed to a slash-command ``discord.Interaction``).""" + channel = channel if channel is not None else make_text_channel() + message = SimpleNamespace( + content=content, + channel=channel, + reference=reference, + delete=AsyncMock(), + ) + return SimpleNamespace( + author=author if author is not None else make_member(), + channel=channel, + message=message, + send=AsyncMock(), + defer=AsyncMock(), + ) + + +def make_interaction(user=None, channel=None): + """Uses ``spec=discord.Interaction`` for the ``isinstance()`` checks + Moderacion uses to tell slash-command interactions apart from regular + text-command invocations.""" + interaction = MagicMock(spec=discord.Interaction) + interaction.user = user if user is not None else make_member() + interaction.channel = channel if channel is not None else make_text_channel() + # `_is_valid_channel` compares `channel_mod.id == ctx.message.channel.id` + # even for interactions, so this needs to line up with `.channel` too. + interaction.message = SimpleNamespace(channel=interaction.channel) + interaction.response = MagicMock() + interaction.response.send_message = AsyncMock() + interaction.response.send_modal = AsyncMock() + return interaction + + +def encode_for_mod_row(text: str) -> str: + """Matches ``Moderacion.on_message``'s encoding of a message's content + into the ``data_mod``/log-file ``message`` column: a base64-encoded + ``bytes`` object rendered through an f-string (so later ``eval()``'d + back into a real ``bytes`` object by the accept/reject/list commands).""" + import base64 + + return f"{base64.b64encode(text.encode('utf-8'))}" + + +def make_png_bytes(color=(255, 0, 0), size=(8, 8)): + """A tiny, valid PNG - for tests that hash/sanitize real image bytes.""" + buf = BytesIO() + Image.new("RGB", size, color=color).save(buf, format="PNG") + return buf.getvalue() + + +def bind_commands(cog): + """Set ``command.cog`` for every ``@commands.command``/``hybrid_command`` + defined on this cog, the way ``bot.add_cog()`` would. + + Our tests instantiate cogs directly without registering them on a real + ``discord.ext.commands.Bot``, but discord.py's ``Command.__call__`` only + binds ``self`` (the cog instance) to the callback when ``command.cog`` + is set - otherwise calling e.g. ``cog.some_command(ctx)`` directly (or a + command calling a sibling command the same way) raises a confusing + "missing 1 required positional argument: 'ctx'". + """ + for command in cog.get_commands(): + command.cog = cog + return cog + + +def prime_cog(cog, message): + """Mirror the attribute setup ``FloodSpam.on_message`` does before + delegating to its individual ``*_check`` methods, so those methods can + be unit-tested directly without going through the full listener.""" + cog._msg_channel = message.channel + cog._msg_content = strip_message(message.content) + cog._msg_author = message.author + cog._msg_author_mention = message.author.mention diff --git a/tests/test_archivar.py b/tests/test_archivar.py new file mode 100644 index 0000000..bfaab1b --- /dev/null +++ b/tests/test_archivar.py @@ -0,0 +1,116 @@ +from unittest.mock import MagicMock + +import discord + +from comandos.archivar import Archivar +from tests.factories import ( + bind_commands, + make_bot, + make_category, + make_ctx, + make_dm_channel, + make_message, + make_text_channel, +) + + +class TestArchivarCanal: + def test_writes_header_and_rows(self, tmp_path): + cog = Archivar(make_bot()) + channel = make_text_channel(id=1, name="general") + messages = [ + make_message(id=10, content="hola\ncon salto de linea", channel=channel), + make_message(id=11, content="segundo mensaje", channel=channel), + ] + target = tmp_path / "archivo.csv" + + status = cog.archivar_canal(str(target), messages) + + assert status == (True, str(target)) + lines = target.read_text().splitlines() + assert lines[0].startswith("id;content;channel_id") + assert len(lines) == 3 + # Newlines inside content are escaped, not left as real line breaks. + assert "hola\\ncon salto de linea" in lines[1] + + def test_stops_and_returns_none_for_unsupported_channel_type(self, tmp_path): + cog = Archivar(make_bot()) + messages = [make_message(id=1, channel=make_dm_channel())] + target = tmp_path / "archivo.csv" + + assert cog.archivar_canal(str(target), messages) is None + + def test_write_failure_returns_false_none(self, tmp_path): + cog = Archivar(make_bot()) + messages = [make_message(id=1)] + # A directory can't be opened for writing as a file. + bad_target = tmp_path + + assert cog.archivar_canal(str(bad_target), messages) == (False, None) + + def test_failure_tuple_is_still_truthy(self, tmp_path): + """Documents a real footgun in archivar(): ``status = archivar_canal(...)`` + followed by ``if status:`` treats a write failure as success, because + ``(False, None)`` is a non-empty tuple and therefore truthy in Python. + """ + cog = Archivar(make_bot()) + status = cog.archivar_canal(str(tmp_path), [make_message(id=1)]) + + assert status == (False, None) + assert bool(status) is True # the actual footgun + + +class TestArchivarCommand: + async def test_sends_success_embed_with_the_file(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + mod_channel = make_text_channel(id=1, name="mod") + cog = bind_commands(Archivar(make_bot())) + cog.mod_channel = mod_channel + + channel = make_text_channel( + id=2, + name="general", + history_messages=[make_message(id=1), make_message(id=2)], + ) + ctx = make_ctx(channel=mod_channel) + + await cog.archivar(ctx, channel=channel) + + mod_channel.send.assert_awaited_once() + _, kwargs = mod_channel.send.call_args + assert "2 mensajes" in kwargs["embed"].description + + async def test_sends_error_embed_when_archiving_fails(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + mod_channel = make_text_channel(id=1, name="mod") + cog = bind_commands(Archivar(make_bot())) + cog.mod_channel = mod_channel + + channel = make_text_channel( + id=2, name="general", history_messages=[make_message(id=1, channel=make_dm_channel())] + ) + ctx = make_ctx(channel=mod_channel) + + await cog.archivar(ctx, channel=channel) + + mod_channel.send.assert_awaited_once() + (msg,), _ = mod_channel.send.call_args + assert "Error" in msg + + +class TestArchivarCategoria: + async def test_only_archives_text_channels(self, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + mod_channel = make_text_channel(id=1, name="mod") + cog = bind_commands(Archivar(make_bot())) + cog.mod_channel = mod_channel + + text_channel = make_text_channel(id=2, name="canal-texto", history_messages=[]) + voice_channel = MagicMock(spec=discord.VoiceChannel) + category = make_category(channels=[text_channel, voice_channel]) + ctx = make_ctx(channel=mod_channel) + + await cog.archivar_categoria(ctx, category=category) + + # Only one embed sent - for the TextChannel. The VoiceChannel was skipped. + mod_channel.send.assert_awaited_once() diff --git a/tests/test_ayuda.py b/tests/test_ayuda.py new file mode 100644 index 0000000..a81fedd --- /dev/null +++ b/tests/test_ayuda.py @@ -0,0 +1,61 @@ +from comandos.ayuda import Ayuda +from tests.factories import bind_commands, make_bot, make_ctx, make_member, make_text_channel + + +class TestGetModHelp: + def test_lists_moderation_commands(self): + cog = Ayuda(make_bot()) + + embed = cog.get_mod_help() + + names = [f.name for f in embed.fields] + assert "`%mod`" in names + assert "`%aceptar ID`" in names + assert "`%rechazar ID RAZON`" in names + + +class TestGetMainHelp: + def test_lists_encuesta_usage(self): + cog = Ayuda(make_bot()) + + embed = cog.get_main_help() + + assert any("encuesta" in f.name for f in embed.fields) + + +class TestMensajeAyuda: + async def test_ignores_the_bot_itself(self, config): + cog = bind_commands(Ayuda(make_bot())) + ctx = make_ctx(author=make_member(id=config.BOT_ID)) + + await cog.mensaje_ayuda(ctx) + + ctx.channel.send.assert_not_awaited() + + async def test_sends_mod_help_inside_a_moderation_channel(self): + mod_channel = make_text_channel(id=1, name="mod") + bot = make_bot(channels={mod_channel.id: mod_channel}) + cog = bind_commands(Ayuda(bot)) + ctx = make_ctx(channel=mod_channel) + + await cog.mensaje_ayuda(ctx) + + mod_channel.send.assert_awaited_once() + _, kwargs = mod_channel.send.call_args + assert kwargs["embed"].title == "Comandos Disponibles" + names = [f.name for f in kwargs["embed"].fields] + assert "`%mod`" in names + + async def test_sends_main_help_outside_a_moderation_channel(self): + # bot.get_channel(ctx.channel.id) returns None - not a known channel. + bot = make_bot(channels={}) + cog = bind_commands(Ayuda(bot)) + channel = make_text_channel(id=99, name="general") + ctx = make_ctx(channel=channel) + + await cog.mensaje_ayuda(ctx) + + channel.send.assert_awaited_once() + _, kwargs = channel.send.call_args + names = [f.name for f in kwargs["embed"].fields] + assert any("encuesta" in name for name in names) diff --git a/tests/test_configuration.py b/tests/test_configuration.py new file mode 100644 index 0000000..9390be8 --- /dev/null +++ b/tests/test_configuration.py @@ -0,0 +1,61 @@ +from configuration import Config + + +class TestSingleton: + def test_config_is_a_singleton(self, config): + assert Config() is config + + def test_reflects_sandboxed_toml(self, config): + assert config.MOD_ROLE == "Coordinacion" + assert config.MUTED_ROLE == "Muted" + assert config.BOT_ID == 111111111111111111 + assert "eventos" in config.CHANNELS + + def test_hardcoded_limits(self, config): + # Not read from config.toml - always set in __init__. + assert config.FLOOD_LIMIT == 3 + assert config.MENTIONS_LIMIT == 3 + assert config.IMAGE_ATTACHMENT_LIMIT == 2 + assert config.IMAGE_BURST_WINDOW == 60 * 5 + + +class TestCheckCreateFile: + def test_creates_missing_file_with_header(self, config, tmp_path): + target = tmp_path / "new_log.csv" + assert not target.exists() + + config.check_create_file(target, "a;b;c\n") + + assert target.read_text() == "a;b;c\n" + + def test_does_not_overwrite_existing_file(self, config, tmp_path): + target = tmp_path / "existing_log.csv" + target.write_text("already;here\nfoo;bar\n") + + config.check_create_file(target, "a;b;c\n") + + assert target.read_text() == "already;here\nfoo;bar\n" + + +class TestGetSpamMessages: + def test_reads_lines_as_a_set(self, config, isolated_logs): + isolated_logs.log_spam_file.write_text("mensaje uno\nmensaje dos\n") + + assert config.get_spam_messages() == {"mensaje uno", "mensaje dos"} + + def test_empty_file_returns_empty_set(self, config, isolated_logs): + isolated_logs.log_spam_file.write_text("") + + assert config.get_spam_messages() == set() + + +class TestGetSpamImageHashes: + def test_reads_hashes_as_a_set(self, config, isolated_logs): + isolated_logs.log_image_spam_file.write_text("abc123\ndef456\n") + + assert config.get_spam_image_hashes() == {"abc123", "def456"} + + def test_skips_blank_lines(self, config, isolated_logs): + isolated_logs.log_image_spam_file.write_text("abc123\n\n\ndef456\n") + + assert config.get_spam_image_hashes() == {"abc123", "def456"} diff --git a/tests/test_enviar.py b/tests/test_enviar.py new file mode 100644 index 0000000..4cceca8 --- /dev/null +++ b/tests/test_enviar.py @@ -0,0 +1,32 @@ +from unittest.mock import AsyncMock + +from comandos.enviar import Enviar +from tests.factories import bind_commands, make_bot, make_ctx, make_text_channel + + +class TestEnviar: + async def test_replies_and_sends_to_target_channel(self): + cog = bind_commands(Enviar(make_bot())) + target = make_text_channel(id=1, name="anuncios") + ctx = make_ctx() + ctx.reply = AsyncMock() + + await cog.enviar(ctx, channel=target, message="hola a todos") + + ctx.reply.assert_awaited_once() + target.send.assert_awaited_once() + _, kwargs = target.send.call_args + assert kwargs["embed"].description == "hola a todos" + + async def test_falls_back_to_channel_send_when_reply_unavailable(self): + cog = bind_commands(Enviar(make_bot())) + target = make_text_channel(id=1, name="anuncios") + # A plain SimpleNamespace ctx has no ``.reply`` attribute at all, + # which is exactly the AttributeError the production code falls + # back on (e.g. a hybrid command invoked in a context without it). + ctx = make_ctx() + + await cog.enviar(ctx, channel=target, message="hola a todos") + + ctx.channel.send.assert_awaited_once() + target.send.assert_awaited_once() diff --git a/tests/test_flood.py b/tests/test_flood.py new file mode 100644 index 0000000..e8ff73c --- /dev/null +++ b/tests/test_flood.py @@ -0,0 +1,469 @@ +from unittest.mock import AsyncMock + +import discord +import pytest + +from tests.factories import ( + make_attachment, + make_member, + make_message, + make_png_bytes, + make_text_channel, + prime_cog, +) + + +# --------------------------------------------------------------------------- +# spam_check +# --------------------------------------------------------------------------- +class TestSpamCheck: + async def test_ignores_non_member_author(self, flood_cog): + message = make_message(content="discord nitro free http://evil") + message.author = object() # not a discord.Member + + assert await flood_cog.spam_check(message) is None + + async def test_no_match_returns_false(self, flood_cog): + message = make_message(content="hola a todos, buen dia") + + assert await flood_cog.spam_check(message) is False + + @pytest.mark.parametrize( + "content", + [ + "gana discord nitro free aqui http://scam.example", + "everyone free steam gift http://scam.example", + ], + ) + async def test_match_mutes_and_notifies(self, flood_cog, content): + member = make_member(name="victima") + message = make_message(content=content, author=member) + prime_cog(flood_cog, message) + + result = await flood_cog.spam_check(message) + + assert result is True + member.add_roles.assert_awaited_once_with(flood_cog.muted_role) + message.channel.send.assert_awaited_once() + _, kwargs = message.channel.send.call_args + assert kwargs["embed"].title.endswith("Alerta de posible SCAM") + + +# --------------------------------------------------------------------------- +# flood_check +# --------------------------------------------------------------------------- +class TestFloodCheck: + async def test_empty_content_is_a_noop(self, flood_cog): + message = make_message(content="") + prime_cog(flood_cog, message) + + assert await flood_cog.flood_check(message) is False + assert flood_cog.messages.normal == {} + + async def test_below_flood_limit_does_not_mute(self, flood_cog, config): + member = make_member(name="repetidor") + message = make_message(content="hola hola hola", author=member) + prime_cog(flood_cog, message) + + for _ in range(config.FLOOD_LIMIT - 1): + await flood_cog.flood_check(message) + + member.add_roles.assert_not_awaited() + + async def test_reaching_flood_limit_mutes_and_caches(self, flood_cog, config): + member = make_member(name="repetidor") + message = make_message(content="hola hola hola", author=member) + prime_cog(flood_cog, message) + + for _ in range(config.FLOOD_LIMIT): + await flood_cog.flood_check(message) + + member.add_roles.assert_awaited_once_with(flood_cog.muted_role) + assert "hola hola hola" in flood_cog.messages.spam + # Counter resets after muting + assert flood_cog.messages.normal[member] == {} + + async def test_different_authors_counted_separately(self, flood_cog, config): + alice = make_member(name="alice", id=1) + bob = make_member(name="bob", id=2) + + for _ in range(config.FLOOD_LIMIT - 1): + msg = make_message(content="mismo mensaje", author=alice) + prime_cog(flood_cog, msg) + await flood_cog.flood_check(msg) + + msg = make_message(content="mismo mensaje", author=bob) + prime_cog(flood_cog, msg) + await flood_cog.flood_check(msg) + + alice.add_roles.assert_not_awaited() + bob.add_roles.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# mention_check +# --------------------------------------------------------------------------- +class TestMentionCheck: + async def test_below_limit_returns_false(self, flood_cog, config): + mentions = [make_member(id=i) for i in range(config.MENTIONS_LIMIT - 1)] + message = make_message(mentions=mentions) + prime_cog(flood_cog, message) + + assert await flood_cog.mention_check(message) is False + + async def test_at_limit_mutes_and_alerts(self, flood_cog, config): + member = make_member(name="mencionador") + mentions = [make_member(id=i) for i in range(config.MENTIONS_LIMIT)] + message = make_message(author=member, mentions=mentions) + prime_cog(flood_cog, message) + + assert await flood_cog.mention_check(message) is True + member.add_roles.assert_awaited_once_with(flood_cog.muted_role) + + async def test_mentions_and_role_mentions_add_up(self, flood_cog, config): + member = make_member(name="mencionador") + mentions = [make_member(id=1)] + role_mentions = [object() for _ in range(config.MENTIONS_LIMIT - 1)] + message = make_message(author=member, mentions=mentions, role_mentions=role_mentions) + prime_cog(flood_cog, message) + + assert await flood_cog.mention_check(message) is True + + +# --------------------------------------------------------------------------- +# attachment_check +# --------------------------------------------------------------------------- +class TestAttachmentCheckFastPath: + async def test_no_images_returns_false(self, flood_cog): + message = make_message(attachments=[make_attachment(content_type="text/plain")]) + prime_cog(flood_cog, message) + + assert await flood_cog.attachment_check(message) is False + + async def test_known_hash_mutes_and_deletes_regardless_of_channel_count( + self, flood_cog, patched_message_delete + ): + data = make_png_bytes() + digest = __import__("hashlib").sha256(data).hexdigest() + flood_cog.messages.image_spam.add(digest) + + member = make_member(name="reincidente") + message = make_message( + author=member, + attachments=[make_attachment(data=data)], + ) + prime_cog(flood_cog, message) + + assert await flood_cog.attachment_check(message) is True + member.add_roles.assert_awaited_once_with(flood_cog.muted_role) + patched_message_delete.assert_awaited_once_with(message) + + +class TestAttachmentCheckBurstPath: + async def test_single_channel_two_images_does_not_trigger(self, flood_cog): + member = make_member(name="autor") + message = make_message( + author=member, + attachments=[make_attachment(filename="a.png"), make_attachment(filename="b.png")], + ) + prime_cog(flood_cog, message) + + assert await flood_cog.attachment_check(message) is False + member.add_roles.assert_not_awaited() + + async def test_single_image_across_channels_does_not_trigger(self, flood_cog): + member = make_member(name="autor") + channel_a = make_text_channel(id=1) + channel_b = make_text_channel(id=2) + + for channel in (channel_a, channel_b): + message = make_message( + author=member, channel=channel, attachments=[make_attachment()] + ) + prime_cog(flood_cog, message) + assert await flood_cog.attachment_check(message) is False + + member.add_roles.assert_not_awaited() + + async def test_two_images_two_channels_triggers_on_the_second_message( + self, flood_cog, patched_message_delete + ): + member = make_member(name="comprometido") + channel_a = make_text_channel(id=1) + channel_b = make_text_channel(id=2) + + first = make_message( + author=member, + channel=channel_a, + attachments=[make_attachment(filename="a1.png"), make_attachment(filename="a2.png")], + ) + prime_cog(flood_cog, first) + first_result = await flood_cog.attachment_check(first) + + second = make_message( + author=member, + channel=channel_b, + attachments=[make_attachment(filename="b1.png"), make_attachment(filename="b2.png")], + ) + prime_cog(flood_cog, second) + second_result = await flood_cog.attachment_check(second) + + # The first channel's message is never retroactively touched - only + # the message that crosses the 2-channel threshold gets acted on. + assert first_result is False + assert second_result is True + member.add_roles.assert_awaited_once_with(flood_cog.muted_role) + patched_message_delete.assert_awaited_once_with(second) + + async def test_images_get_cached_for_the_fast_path(self, flood_cog): + member = make_member(name="comprometido") + data_a, data_b = make_png_bytes((255, 0, 0)), make_png_bytes((0, 255, 0)) + + first = make_message( + author=member, + channel=make_text_channel(id=1), + attachments=[make_attachment(data=data_a), make_attachment(data=data_b)], + ) + prime_cog(flood_cog, first) + await flood_cog.attachment_check(first) + + second = make_message( + author=member, + channel=make_text_channel(id=2), + attachments=[make_attachment(data=data_a), make_attachment(data=data_b)], + ) + prime_cog(flood_cog, second) + await flood_cog.attachment_check(second) + + import hashlib + + assert hashlib.sha256(data_a).hexdigest() in flood_cog.messages.image_spam + assert hashlib.sha256(data_b).hexdigest() in flood_cog.messages.image_spam + + async def test_outside_burst_window_does_not_trigger(self, flood_cog, config, monkeypatch): + import comandos.flood as flood_module + + member = make_member(name="lento") + clock = iter([1000.0, 1000.0 + config.IMAGE_BURST_WINDOW + 1]) + monkeypatch.setattr(flood_module.time, "time", lambda: next(clock)) + + first = make_message( + author=member, + channel=make_text_channel(id=1), + attachments=[make_attachment(filename="a1.png"), make_attachment(filename="a2.png")], + ) + prime_cog(flood_cog, first) + await flood_cog.attachment_check(first) + + second = make_message( + author=member, + channel=make_text_channel(id=2), + attachments=[make_attachment(filename="b1.png"), make_attachment(filename="b2.png")], + ) + prime_cog(flood_cog, second) + result = await flood_cog.attachment_check(second) + + assert result is False + member.add_roles.assert_not_awaited() + + async def test_same_channel_twice_is_not_two_distinct_channels(self, flood_cog): + member = make_member(name="autor") + channel = make_text_channel(id=1) + + for _ in range(3): + message = make_message( + author=member, + channel=channel, + attachments=[make_attachment(filename="a.png"), make_attachment(filename="b.png")], + ) + prime_cog(flood_cog, message) + result = await flood_cog.attachment_check(message) + + assert result is False + member.add_roles.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# _sanitize_attachment / _hash_attachment +# --------------------------------------------------------------------------- +class TestSanitizeAttachment: + async def test_valid_image_round_trips_as_spoiler_file(self, flood_cog): + attachment = make_attachment(data=make_png_bytes()) + + result = await flood_cog._sanitize_attachment(attachment) + + assert result is not None + assert isinstance(result, discord.File) + assert result.spoiler is True + assert result.filename.endswith("evidencia.png") or "SPOILER" in result.filename + + async def test_garbage_bytes_returns_none(self, flood_cog): + attachment = make_attachment(data=b"not an image, just garbage" * 10) + + assert await flood_cog._sanitize_attachment(attachment) is None + + async def test_truncated_image_returns_none(self, flood_cog): + attachment = make_attachment(data=make_png_bytes()[:15]) + + assert await flood_cog._sanitize_attachment(attachment) is None + + +class TestHashAttachment: + async def test_matches_sha256_of_bytes(self, flood_cog): + import hashlib + + data = b"some bytes" + attachment = make_attachment(data=data) + + assert await flood_cog._hash_attachment(attachment) == hashlib.sha256(data).hexdigest() + + +# --------------------------------------------------------------------------- +# add_spam_message / add_spam_image_hash +# --------------------------------------------------------------------------- +class TestAddSpamHelpers: + def test_add_spam_message_persists_and_caches(self, flood_cog, isolated_logs): + flood_cog.add_spam_message("mensaje malo") + + assert "mensaje malo" in flood_cog.messages.spam + assert "mensaje malo" in isolated_logs.log_spam_file.read_text() + + def test_add_spam_image_hash_persists_and_caches(self, flood_cog, isolated_logs): + flood_cog.add_spam_image_hash("deadbeef") + + assert "deadbeef" in flood_cog.messages.image_spam + assert "deadbeef" in isolated_logs.log_image_spam_file.read_text() + + +# --------------------------------------------------------------------------- +# alert_moderation +# --------------------------------------------------------------------------- +class TestAlertModeration: + async def test_creates_thread_and_sends_embed(self, flood_cog): + member = make_member(name="alguien") + message = make_message(author=member) + prime_cog(flood_cog, message) + + await flood_cog.alert_moderation("Alerta de prueba", "scam") + + flood_cog.main_mod_channel.create_thread.assert_awaited_once() + _, kwargs = flood_cog.main_mod_channel.create_thread.call_args + assert member.mention in kwargs["name"] + + thread = flood_cog.main_mod_channel.create_thread.return_value + thread.send.assert_awaited_once() + + async def test_unknown_reason_raises(self, flood_cog): + message = make_message() + prime_cog(flood_cog, message) + + with pytest.raises(KeyError): + await flood_cog.alert_moderation("Título", "no-existe") + + async def test_attachments_are_forwarded_sanitized_and_spoilered(self, flood_cog): + message = make_message() + prime_cog(flood_cog, message) + images = [make_attachment(data=make_png_bytes())] + + await flood_cog.alert_moderation("Alerta", "known_image", attachments=images) + + thread = flood_cog.main_mod_channel.create_thread.return_value + _, kwargs = thread.send.call_args + assert len(kwargs["files"]) == 1 + assert kwargs["files"][0].spoiler is True + + async def test_undecodable_attachment_is_skipped_not_forwarded(self, flood_cog): + message = make_message() + prime_cog(flood_cog, message) + images = [make_attachment(data=b"garbage" * 10)] + + await flood_cog.alert_moderation("Alerta", "known_image", attachments=images) + + thread = flood_cog.main_mod_channel.create_thread.return_value + _, kwargs = thread.send.call_args + assert kwargs["files"] == [] + + async def test_no_attachments_means_no_warning_field(self, flood_cog): + message = make_message() + prime_cog(flood_cog, message) + + await flood_cog.alert_moderation("Alerta", "scam") + + thread = flood_cog.main_mod_channel.create_thread.return_value + _, kwargs = thread.send.call_args + field_names = [f.name for f in kwargs["embed"].fields] + assert not any("Imágenes adjuntas" in name for name in field_names) + + +# --------------------------------------------------------------------------- +# on_message pipeline +# --------------------------------------------------------------------------- +class TestOnMessagePipeline: + async def test_ignores_messages_from_bots(self, flood_cog): + member = make_member(name="unbot", bot=True) + message = make_message(content="discord nitro free http://x", author=member) + + await flood_cog.on_message(message) + + member.add_roles.assert_not_awaited() + + async def test_ignores_the_configured_bot_id(self, flood_cog, config): + member = make_member(name="elbot", id=config.BOT_ID, bot=False) + message = make_message(content="discord nitro free http://x", author=member) + + await flood_cog.on_message(message) + + member.add_roles.assert_not_awaited() + + async def test_ignores_short_textless_messages_without_attachments(self, flood_cog): + member = make_member(name="alguien") + message = make_message(content="ok", author=member) + + await flood_cog.on_message(message) + + # Never even gets far enough to set up per-message state. + assert flood_cog._msg_author is None + + async def test_short_caption_with_attachments_is_still_processed(self, flood_cog): + member = make_member(name="alguien") + message = make_message( + content="ok", + author=member, + attachments=[make_attachment(filename="a.png"), make_attachment(filename="b.png")], + ) + + await flood_cog.on_message(message) + + # It went through the pipeline (attachment_check saw it), even though + # the caption alone would have been skipped. + assert flood_cog._msg_author is member + + async def test_skips_coordination_role_members(self, flood_cog): + message = make_message( + content="discord nitro free http://x", + author=make_member(name="mod", roles=[flood_cog.coord_role]), + ) + + result = None + try: + result = await flood_cog.on_message(message) + finally: + pass + + message.author.add_roles.assert_not_awaited() + + async def test_known_spam_text_is_deleted_and_author_muted( + self, flood_cog, patched_message_delete + ): + flood_cog.messages.spam.add("mensaje ya conocido como spam") + member = make_member(name="repetidor") + channel = make_text_channel(id=42) + flood_cog.bot.channels[channel.id] = channel + message = make_message( + content="Mensaje YA conocido como SPAM", author=member, channel=channel + ) + + await flood_cog.on_message(message) + + member.add_roles.assert_awaited_once_with(flood_cog.muted_role) + patched_message_delete.assert_awaited_once_with(message) diff --git a/tests/test_limpia.py b/tests/test_limpia.py new file mode 100644 index 0000000..e43ce7e --- /dev/null +++ b/tests/test_limpia.py @@ -0,0 +1,50 @@ +from types import SimpleNamespace + +from comandos.limpia import Limpia +from tests.factories import ( + bind_commands, + make_bot, + make_ctx, + make_dm_channel, + make_message, + make_text_channel, +) + + +class TestPurge: + async def test_ignores_unsupported_channel_types(self): + cog = bind_commands(Limpia(make_bot())) + ctx = make_ctx(channel=make_dm_channel()) + + await cog.purge(ctx, limit=5) + + # Nothing to assert on the channel itself (it's not a spec that + # exposes purge/delete_messages), just that we returned early + # without raising. + + async def test_no_reply_purges_the_channel_and_the_command_message(self): + channel = make_text_channel(id=1, name="general") + cog = bind_commands(Limpia(make_bot())) + ctx = make_ctx(channel=channel) + + await cog.purge(ctx, limit=3) + + ctx.defer.assert_awaited_once_with(ephemeral=True) + channel.purge.assert_any_await(limit=4) + ctx.message.delete.assert_awaited_once() + ctx.send.assert_awaited_once() + channel.purge.assert_any_await(limit=1) + + async def test_reply_deletes_messages_up_to_the_referenced_one(self): + target = make_message(id=42) + history = [make_message(id=1), make_message(id=2), target, make_message(id=3)] + channel = make_text_channel(id=1, name="general", history_messages=history) + cog = bind_commands(Limpia(make_bot())) + ctx = make_ctx(channel=channel, reference=SimpleNamespace(message_id=42)) + ctx.message.channel = channel + + await cog.purge(ctx) + + channel.delete_messages.assert_awaited_once() + (deleted,), _ = channel.delete_messages.call_args + assert [m.id for m in deleted] == [1, 2, 42] diff --git a/tests/test_moderacion.py b/tests/test_moderacion.py new file mode 100644 index 0000000..6fd066c --- /dev/null +++ b/tests/test_moderacion.py @@ -0,0 +1,320 @@ +from unittest.mock import AsyncMock + +import pandas as pd +import pytest + +from tests.factories import ( + encode_for_mod_row, + make_ctx, + make_interaction, + make_member, +) + + +def add_pending_row(cog, post_id, *, channel="envio-eventos", author_id=42, author_name="autor", + content="contenido de prueba"): + new_row = { + "date": "2026-01-01 00:00:00", + "message_id": str(post_id), + "channel": channel, + "author_id": str(author_id), + "author": author_name, + "message": encode_for_mod_row(content), + } + cog.bot.data_mod = pd.concat([cog.bot.data_mod, pd.DataFrame([new_row])], ignore_index=True) + return new_row + + +# --------------------------------------------------------------------------- +# small helpers +# --------------------------------------------------------------------------- +class TestResolveAuthor: + def test_regular_context_uses_author(self, moderacion_cog): + member = make_member(name="alguien") + ctx = make_ctx(author=member) + + assert moderacion_cog._resolve_author(ctx) is member + + def test_interaction_uses_user(self, moderacion_cog): + member = make_member(name="alguien") + interaction = make_interaction(user=member) + + assert moderacion_cog._resolve_author(interaction) is member + + +class TestIsBot: + def test_true_for_configured_bot_id(self, moderacion_cog, config): + ctx = make_ctx(author=make_member(id=config.BOT_ID)) + + assert moderacion_cog._is_bot(ctx) is True + + def test_false_for_regular_user(self, moderacion_cog): + ctx = make_ctx(author=make_member(id=12345)) + + assert moderacion_cog._is_bot(ctx) is False + + +class TestIsValidChannel: + def test_true_when_channel_is_registered_on_the_bot(self, moderacion_cog, moderacion_channels): + ctx = make_ctx(channel=moderacion_channels["mod"]) + + assert moderacion_cog._is_valid_channel(ctx) is True + + +class TestGetChannelsMainModSub: + def test_resolves_the_three_channels(self, moderacion_cog, moderacion_channels): + main, mod, sub = moderacion_cog.get_channels_main_mod_sub(moderacion_channels["sub"].id) + + assert main is moderacion_channels["main"] + assert mod is moderacion_channels["mod"] + assert sub is moderacion_channels["sub"] + + +# --------------------------------------------------------------------------- +# _parse_post_id +# --------------------------------------------------------------------------- +class TestParsePostId: + async def test_interaction_with_message_id_returns_it_directly( + self, moderacion_cog, moderacion_channels + ): + interaction = make_interaction(channel=moderacion_channels["mod"]) + + result = await moderacion_cog._parse_post_id(interaction, 555, "%aceptar") + + assert result == "555" + + async def test_valid_numeric_id_from_message_content(self, moderacion_cog, moderacion_channels): + ctx = make_ctx(channel=moderacion_channels["mod"], content="%aceptar 123") + + assert await moderacion_cog._parse_post_id(ctx, None, "%aceptar") == "123" + + async def test_non_numeric_id_reports_error_and_returns_none( + self, moderacion_cog, moderacion_channels + ): + ctx = make_ctx(channel=moderacion_channels["mod"], content="%aceptar abc") + + result = await moderacion_cog._parse_post_id(ctx, None, "%aceptar") + + assert result is None + moderacion_channels["mod"].send.assert_awaited_once() + (msg,), _ = moderacion_channels["mod"].send.call_args + assert "abc" in msg + + +# --------------------------------------------------------------------------- +# _get_validated_post +# --------------------------------------------------------------------------- +class TestGetValidatedPost: + async def test_happy_path_resolves_everything(self, moderacion_cog, moderacion_channels): + add_pending_row(moderacion_cog, post_id=1, author_id=99) + moderacion_cog.bot.users_by_id[99] = make_member(id=99, name="remitente") + ctx = make_ctx(channel=moderacion_channels["mod"], content="%aceptar 1") + + vp = await moderacion_cog._get_validated_post(ctx, None, "%aceptar") + + assert vp is not None + assert vp.post_id == "1" + assert vp.message_dec == "contenido de prueba" + assert vp.ch_main is moderacion_channels["main"] + assert vp.ch_mod is moderacion_channels["mod"] + assert vp.ch_sub is moderacion_channels["sub"] + assert vp.author.id == 99 + + async def test_unknown_post_id_reports_error_and_returns_none( + self, moderacion_cog, moderacion_channels + ): + ctx = make_ctx(channel=moderacion_channels["mod"], content="%aceptar 999") + + vp = await moderacion_cog._get_validated_post(ctx, None, "%aceptar") + + assert vp is None + moderacion_channels["mod"].send.assert_awaited_once() + (msg,), _ = moderacion_channels["mod"].send.call_args + assert "999" in msg + + async def test_bot_author_returns_none(self, moderacion_cog, moderacion_channels, config): + add_pending_row(moderacion_cog, post_id=1) + ctx = make_ctx( + author=make_member(id=config.BOT_ID), + channel=moderacion_channels["mod"], + content="%aceptar 1", + ) + + assert await moderacion_cog._get_validated_post(ctx, None, "%aceptar") is None + + +# --------------------------------------------------------------------------- +# _log_action / log_on_message +# --------------------------------------------------------------------------- +class TestLogAction: + def test_aceptar_writes_expected_line(self, moderacion_cog, isolated_logs): + row = pd.DataFrame([add_pending_row(moderacion_cog, post_id=1)]) + + moderacion_cog._log_action("aceptar", row, "1", "moderador#0") + + content = isolated_logs.log_accepted_file.read_text() + assert '"1"' in content + assert '"moderador#0"' in content + + def test_rechazar_includes_reason(self, moderacion_cog, isolated_logs): + row = pd.DataFrame([add_pending_row(moderacion_cog, post_id=2)]) + + moderacion_cog._log_action("rechazar", row, "2", "moderador#0", "le falta info") + + content = isolated_logs.log_rejected_file.read_text() + assert '"le falta info"' in content + + +class TestLogOnMessage: + def test_appends_row_and_writes_log_line(self, moderacion_cog, isolated_logs): + moderacion_cog._msg_id = 777 + moderacion_cog._msg_enc = encode_for_mod_row("hola mundo") + author = make_member(id=55, name="autor") + + before = len(moderacion_cog.bot.data_mod) + moderacion_cog.log_on_message("envio-eventos", author) + + assert len(moderacion_cog.bot.data_mod) == before + 1 + assert "777" in isolated_logs.log_mod_file.read_text() + + +# --------------------------------------------------------------------------- +# get_mod_pending +# --------------------------------------------------------------------------- +class TestGetModPending: + def test_empty_data_sets_footer(self, moderacion_cog): + embed = moderacion_cog.get_mod_pending(moderacion_cog.bot.data_mod) + + assert embed.footer.text == "No hay mensajes pendientes de moderación" + assert len(embed.fields) == 0 + + def test_lists_pending_posts_with_known_authors(self, moderacion_cog): + add_pending_row(moderacion_cog, post_id=1, author_id=99, content="hola" * 20) + moderacion_cog.bot.users_by_id[99] = make_member(id=99, name="remitente") + + embed = moderacion_cog.get_mod_pending(moderacion_cog.bot.data_mod) + + assert len(embed.fields) == 1 + assert "1" in embed.fields[0].name + + def test_skips_posts_from_users_no_longer_in_the_server(self, moderacion_cog): + add_pending_row(moderacion_cog, post_id=1, author_id=404) + # bot.get_user(404) resolves to None - author has left the server. + + embed = moderacion_cog.get_mod_pending(moderacion_cog.bot.data_mod) + + assert len(embed.fields) == 0 + assert embed.footer.text == "No hay mensajes pendientes de moderación" + + +# --------------------------------------------------------------------------- +# _aceptar_mensaje / _rechazar_mensaje +# --------------------------------------------------------------------------- +class TestAceptarMensaje: + async def test_removes_pending_row_and_notifies_channels( + self, moderacion_cog, moderacion_channels, config + ): + add_pending_row(moderacion_cog, post_id=1, author_id=99, content="contenido aprobado") + moderacion_cog.bot.users_by_id[99] = make_member(id=99, name="remitente") + moderacion_cog._msg_id = 12345 + + ctx = make_ctx( + author=make_member(name="moderador"), + channel=moderacion_channels["mod"], + content="%aceptar 1", + ) + + await moderacion_cog._aceptar_mensaje(ctx) + + assert moderacion_cog.bot.data_mod.empty + moderacion_channels["mod"].send.assert_awaited_once() + moderacion_channels["main"].send.assert_awaited_once() + (msg,), _ = moderacion_channels["main"].send.call_args + assert "contenido aprobado" in msg + + async def test_unknown_post_id_does_not_touch_channels(self, moderacion_cog, moderacion_channels): + ctx = make_ctx(channel=moderacion_channels["mod"], content="%aceptar 999") + + await moderacion_cog._aceptar_mensaje(ctx) + + moderacion_channels["main"].send.assert_not_awaited() + + +class TestRechazarMensaje: + async def test_removes_pending_row_and_notifies_with_reason( + self, moderacion_cog, moderacion_channels + ): + add_pending_row(moderacion_cog, post_id=2, author_id=99, content="contenido rechazado") + moderacion_cog.bot.users_by_id[99] = make_member(id=99, name="remitente") + + ctx = make_ctx( + author=make_member(name="moderador"), + channel=moderacion_channels["mod"], + content="%rechazar 2 le falta info aqui", + ) + + await moderacion_cog._rechazar_mensaje(ctx) + + assert moderacion_cog.bot.data_mod.empty + moderacion_channels["sub"].send.assert_awaited_once() + _, kwargs = moderacion_channels["sub"].send.call_args + assert "le falta info aqui" in kwargs["embed"].fields[0].value + + async def test_interaction_path_uses_the_provided_reason( + self, moderacion_cog, moderacion_channels + ): + add_pending_row(moderacion_cog, post_id=3, author_id=99, content="contenido") + moderacion_cog.bot.users_by_id[99] = make_member(id=99, name="remitente") + interaction = make_interaction(channel=moderacion_channels["mod"]) + + await moderacion_cog._rechazar_mensaje(interaction, message_id=3, reason="motivo modal") + + _, kwargs = moderacion_channels["sub"].send.call_args + assert "motivo modal" in kwargs["embed"].fields[0].value + + +# --------------------------------------------------------------------------- +# on_message listener +# --------------------------------------------------------------------------- +class TestOnMessage: + async def test_ignores_limpia_command(self, moderacion_cog, moderacion_channels): + message = make_ctx(channel=moderacion_channels["sub"], content="%limpia").message + message.author = make_member() + message.id = 1 + + before = len(moderacion_cog.bot.data_mod) + await moderacion_cog.on_message(message) + + assert len(moderacion_cog.bot.data_mod) == before + + async def test_ignores_channels_outside_the_submission_mapping( + self, moderacion_cog, moderacion_channels + ): + other_channel = moderacion_channels["mod"] # not a "submission" channel + message = make_ctx(channel=other_channel, content="hola").message + message.author = make_member() + message.id = 2 + + before = len(moderacion_cog.bot.data_mod) + await moderacion_cog.on_message(message) + + assert len(moderacion_cog.bot.data_mod) == before + + async def test_submission_gets_logged_and_forwarded_to_mod_channel( + self, moderacion_cog, moderacion_channels, monkeypatch + ): + import comandos.moderacion as moderacion_module + + monkeypatch.setattr(moderacion_module.asyncio, "sleep", AsyncMock()) + + member = make_member(name="remitente") + message = make_ctx(channel=moderacion_channels["sub"], content="mi propuesta de evento").message + message.author = member + message.id = 321 + + before = len(moderacion_cog.bot.data_mod) + await moderacion_cog.on_message(message) + + assert len(moderacion_cog.bot.data_mod) == before + 1 + moderacion_channels["sub"].send.assert_awaited_once() + moderacion_channels["mod"].send.assert_awaited_once() diff --git a/tests/test_ping.py b/tests/test_ping.py new file mode 100644 index 0000000..857ccba --- /dev/null +++ b/tests/test_ping.py @@ -0,0 +1,63 @@ +from types import SimpleNamespace + +from comandos.ping import Ping +from tests.factories import bind_commands, make_bot, make_ctx, make_message + + +def make_bot_user(mentioned=True): + return SimpleNamespace(mentioned_in=lambda message: mentioned) + + +class TestHasGlobalMention: + def test_true_for_everyone(self): + cog = Ping(make_bot()) + assert cog.has_global_mention(SimpleNamespace(content="hola @everyone")) is True + + def test_true_for_here(self): + cog = Ping(make_bot()) + assert cog.has_global_mention(SimpleNamespace(content="hola @here")) is True + + def test_false_for_a_regular_mention(self): + cog = Ping(make_bot()) + assert cog.has_global_mention(SimpleNamespace(content="hola <@123>")) is False + + +class TestPingPong: + async def test_replies_pong_ephemerally(self): + cog = bind_commands(Ping(make_bot())) + ctx = make_ctx() + + await cog.pingpong(ctx) + + ctx.send.assert_awaited_once_with("pong", ephemeral=True) + + +class TestOnMessage: + async def test_sends_the_llama_gif_when_mentioned(self): + bot = make_bot(user=make_bot_user(mentioned=True)) + cog = Ping(bot) + message = make_message(content="hola bot") + + await cog.on_message(message) + + message.channel.send.assert_awaited_once() + _, kwargs = message.channel.send.call_args + assert kwargs["file"].filename == "resources/llama.gif" + + async def test_ignores_global_mentions(self): + bot = make_bot(user=make_bot_user(mentioned=True)) + cog = Ping(bot) + message = make_message(content="@everyone hola bot") + + await cog.on_message(message) + + message.channel.send.assert_not_awaited() + + async def test_ignores_messages_that_do_not_mention_the_bot(self): + bot = make_bot(user=make_bot_user(mentioned=False)) + cog = Ping(bot) + message = make_message(content="hola a todos") + + await cog.on_message(message) + + message.channel.send.assert_not_awaited() diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..2ac0a48 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,51 @@ +from tests.factories import make_message +from utils import get_message_to_moderate, get_moderation_channel, strip_message + + +class TestStripMessage: + def test_lowercases(self): + assert strip_message("HOLA Mundo") == "hola mundo" + + def test_removes_newlines_and_tabs(self): + assert strip_message("hola\nmundo\tcon\rtabs") == "hola mundo con tabs" + + def test_collapses_multiple_whitespace(self): + assert strip_message("hola mundo") == "hola mundo" + + def test_removes_mentions(self): + assert strip_message("hola <@123456789> mundo") == "hola mundo" + + def test_removes_multiple_mentions(self): + assert strip_message("<@111> hola <@!222> mundo") == "hola mundo" + + def test_strips_surrounding_whitespace(self): + assert strip_message(" hola mundo ") == "hola mundo" + + def test_empty_string(self): + assert strip_message("") == "" + + +class TestGetModerationChannel: + def test_returns_bot_get_channel_result(self): + sentinel = object() + + class FakeBot: + def get_channel(self, channel_id): + assert channel_id == 42 + return sentinel + + assert get_moderation_channel(FakeBot(), 42) is sentinel + + +class TestGetMessageToModerate: + def test_embed_contains_message_and_commands(self): + message = make_message(content="hola, este es mi post") + message.id = 4242 + + embed = get_message_to_moderate(message) + + assert "hola, este es mi post" in embed.description + assert "%aceptar 4242" in embed.description + assert "%rechazar 4242" in embed.description + assert message.channel.mention in embed.description + assert message.author.mention in embed.description diff --git a/utils.py b/utils.py index e8b5e60..41cb7c1 100644 --- a/utils.py +++ b/utils.py @@ -1,6 +1,6 @@ import re import discord -from datetime import datetime +from datetime import datetime, timezone from configuration import Config @@ -17,7 +17,7 @@ def get_moderation_channel(bot, channel_id): def get_message_to_moderate(message): msg = ( - f"{datetime.utcnow()} UTC\n" + f"{datetime.now(timezone.utc).replace(tzinfo=None)} UTC\n" f"Mensaje enviado desde {message.channel.mention} por {message.author.mention}\n\n" f"```\n{message.content}\n```\n\n**¿Cumple con todos los requisitos?**\n\n" f"{aceptar_emoji} Para aceptarlo, envía el siguiente mensaje:\n\n"