diff --git a/test/conftest.py b/test/conftest.py index b57cf62..dc799b0 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,17 +1,32 @@ import os - +import asyncio from datetime import datetime from functools import wraps +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + from peewee import SqliteDatabase -from telegram import Bot -from pycamp_bot.models import Pycampista, Slot, Pycamp, WizardAtPycamp, PycampistaAtPycamp +from pycamp_bot.models import ( + Pycampista, Slot, Pycamp, WizardAtPycamp, PycampistaAtPycamp, Project, Vote +) +# ----------------------------------------------------------------------------- +# Causa raíz de fallos en tests de handlers (PTB v21): +# En python-telegram-bot v21 los objetos Telegram (Message, CallbackQuery, etc.) +# son inmutables ("frozen"): no se puede asignar msg.reply_text = AsyncMock(). +# Doc oficial (v20.0+): "Objects of this class (or subclasses) are now immutable. +# This means that you can't set or delete attributes anymore." +# https://docs.python-telegram-bot.org/en/v21.10/telegram.telegramobject.html +# Por eso los builders de updates usan dobles mutables (SimpleNamespace) que +# exponen la misma interfaz que los handlers (message.text, message.reply_text, +# callback_query.answer, etc.) sin tocar objetos reales de la librería. +# ----------------------------------------------------------------------------- # use an in-memory SQLite for tests. test_db = SqliteDatabase(':memory:') -MODELS = [Pycampista, Slot, Pycamp, WizardAtPycamp, PycampistaAtPycamp] +MODELS = [Pycampista, Slot, Pycamp, WizardAtPycamp, PycampistaAtPycamp, Project, Vote] def use_test_database(fn): @@ -25,3 +40,91 @@ def inner(self): finally: test_db.drop_tables(MODELS) return inner + + +def use_test_database_async(fn): + """Bind the given models to the db for the duration of an async test.""" + @wraps(fn) + def inner(self): + with test_db.bind_ctx(MODELS): + test_db.create_tables(MODELS) + try: + asyncio.get_event_loop().run_until_complete(fn(self)) + finally: + test_db.drop_tables(MODELS) + return inner + + +def make_user(username="testuser", user_id=12345, first_name="Test"): + """Doble mutable de telegram.User para tests (PTB v21 usa objetos congelados).""" + return SimpleNamespace( + id=user_id, + first_name=first_name, + is_bot=False, + username=username, + ) + + +def make_chat(chat_id=67890, chat_type="private"): + """Doble mutable de telegram.Chat para tests.""" + return SimpleNamespace(id=chat_id, type=chat_type) + + +def make_message(text="/start", username="testuser", chat_id=67890, + user_id=12345, message_id=1): + """Doble mutable de telegram.Message: misma interfaz que usan los handlers.""" + user = make_user(username=username, user_id=user_id) + chat = make_chat(chat_id=chat_id) + msg = SimpleNamespace( + message_id=message_id, + date=datetime.now(), + chat=chat, + from_user=user, + text=text, + chat_id=chat_id, + reply_text=AsyncMock(), + ) + return msg + + +def make_update(text="/start", username="testuser", chat_id=67890, + user_id=12345, update_id=1): + """Doble mutable de telegram.Update con message; interfaz usada por handlers.""" + message = make_message( + text=text, username=username, chat_id=chat_id, + user_id=user_id, + ) + return SimpleNamespace(update_id=update_id, message=message) + + +def make_callback_update(data="vote:si", username="testuser", chat_id=67890, + user_id=12345, message_text="Proyecto1", update_id=1): + """Doble mutable de Update con callback_query para botones inline.""" + user = make_user(username=username, user_id=user_id) + chat = make_chat(chat_id=chat_id) + message = SimpleNamespace( + message_id=1, + date=datetime.now(), + chat=chat, + text=message_text, + chat_id=chat_id, + reply_text=AsyncMock(), + ) + callback_query = SimpleNamespace( + id="test_callback_1", + from_user=user, + chat_instance="test_instance", + data=data, + message=message, + answer=AsyncMock(), + ) + return SimpleNamespace(update_id=update_id, callback_query=callback_query) + + +def make_context(): + """Crea un mock de CallbackContext con bot mockeado según la API v21.""" + context = MagicMock() + context.bot = AsyncMock() + context.bot.send_message = AsyncMock() + context.bot.edit_message_text = AsyncMock() + return context diff --git a/test/test_auth.py b/test/test_auth.py new file mode 100644 index 0000000..b4613b1 --- /dev/null +++ b/test/test_auth.py @@ -0,0 +1,56 @@ +from pycamp_bot.models import Pycampista +from pycamp_bot.commands.auth import get_admins_username +from test.conftest import use_test_database, test_db, MODELS + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestGetAdminsUsername: + + @use_test_database + def test_returns_empty_when_no_admins(self): + Pycampista.create(username="pepe", admin=False) + assert get_admins_username() == [] + + @use_test_database + def test_returns_admin_usernames(self): + Pycampista.create(username="admin1", admin=True) + result = get_admins_username() + assert result == ["admin1"] + + @use_test_database + def test_excludes_non_admin_users(self): + Pycampista.create(username="admin1", admin=True) + Pycampista.create(username="user1", admin=False) + result = get_admins_username() + assert "admin1" in result + assert "user1" not in result + + @use_test_database + def test_multiple_admins(self): + Pycampista.create(username="admin1", admin=True) + Pycampista.create(username="admin2", admin=True) + Pycampista.create(username="user1", admin=False) + result = get_admins_username() + assert len(result) == 2 + assert "admin1" in result + assert "admin2" in result + + @use_test_database + def test_admin_with_null_flag(self): + Pycampista.create(username="pepe", admin=None) + result = get_admins_username() + assert result == [] + + @use_test_database + def test_no_users_returns_empty(self): + result = get_admins_username() + assert result == [] diff --git a/test/test_conftest_builders.py b/test/test_conftest_builders.py new file mode 100644 index 0000000..9a00418 --- /dev/null +++ b/test/test_conftest_builders.py @@ -0,0 +1,92 @@ +"""Tests de los builders de conftest: garantizan la interfaz que usan los handlers. + +Si PTB o los handlers cambian de atributos, estos tests fallan y evitan +regresiones silenciosas en los tests de handlers. +""" +import pytest + +from test.conftest import ( + make_user, + make_chat, + make_message, + make_update, + make_callback_update, + make_context, +) + + +class TestMakeUser: + def test_exposes_username_id_first_name(self): + u = make_user(username="pepe", user_id=999, first_name="Pepe") + assert u.username == "pepe" + assert u.id == 999 + assert u.first_name == "Pepe" + assert u.is_bot is False + + def test_accepts_username_none(self): + u = make_user(username=None) + assert u.username is None + + +class TestMakeChat: + def test_exposes_id_and_type(self): + c = make_chat(chat_id=111, chat_type="private") + assert c.id == 111 + assert c.type == "private" + + +class TestMakeMessage: + def test_exposes_text_chat_id_from_user_reply_text(self): + msg = make_message(text="/start", username="u1", chat_id=222) + assert msg.text == "/start" + assert msg.chat_id == 222 + assert msg.from_user.username == "u1" + assert msg.chat.id == 222 + assert callable(msg.reply_text) + + def test_reply_text_is_awaitable(self): + msg = make_message() + import asyncio + asyncio.get_event_loop().run_until_complete(msg.reply_text("hi")) + + +class TestMakeUpdate: + def test_exposes_message_with_same_interface(self): + up = make_update(text="/cmd", username="alice", chat_id=333) + assert up.message.text == "/cmd" + assert up.message.chat_id == 333 + assert up.message.from_user.username == "alice" + assert callable(up.message.reply_text) + + def test_update_id_set(self): + up = make_update(update_id=42) + assert up.update_id == 42 + + +class TestMakeCallbackUpdate: + def test_exposes_callback_query_data_and_message(self): + up = make_callback_update( + data="vote:si", + username="voter", + message_text="ProyectoX", + chat_id=444, + ) + assert up.callback_query.data == "vote:si" + assert up.callback_query.from_user.username == "voter" + assert up.callback_query.message.text == "ProyectoX" + assert up.callback_query.message.chat_id == 444 + assert up.callback_query.message.chat.id == 444 + assert callable(up.callback_query.answer) + + def test_answer_is_awaitable(self): + up = make_callback_update() + import asyncio + asyncio.get_event_loop().run_until_complete(up.callback_query.answer()) + + +class TestMakeContext: + def test_exposes_bot_send_message_and_edit_message_text(self): + ctx = make_context() + assert ctx.bot is not None + assert callable(ctx.bot.send_message) + assert callable(ctx.bot.edit_message_text) diff --git a/test/test_db_to_json.py b/test/test_db_to_json.py new file mode 100644 index 0000000..daf4f6d --- /dev/null +++ b/test/test_db_to_json.py @@ -0,0 +1,88 @@ +from datetime import datetime +from pycamp_bot.models import Pycampista, Project, Slot, Vote +from pycamp_bot.scheduler.db_to_json import export_db_2_json +from test.conftest import use_test_database, test_db, MODELS + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestExportDb2Json: + + @use_test_database + def test_empty_projects_returns_empty_structure(self): + result = export_db_2_json() + assert result["projects"] == {} + assert result["available_slots"] == [] + assert result["responsable_available_slots"] == {} + + @use_test_database + def test_exports_project_with_votes(self): + owner = Pycampista.create(username="pepe") + voter = Pycampista.create(username="juan") + slot = Slot.create(code="A1", start=datetime(2024, 6, 21, 10, 0), current_wizard=owner) + project = Project.create(name="MiProyecto", owner=owner, topic="django", difficult_level=2) + Vote.create( + project=project, pycampista=voter, interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + + result = export_db_2_json() + assert "MiProyecto" in result["projects"] + proj_data = result["projects"]["MiProyecto"] + assert proj_data["responsables"] == ["pepe"] + assert "juan" in proj_data["votes"] + assert proj_data["difficult_level"] == 2 + assert proj_data["theme"] == "django" + + @use_test_database + def test_exports_available_slots(self): + owner = Pycampista.create(username="pepe") + Slot.create(code="A1", start=datetime(2024, 6, 21, 10, 0), current_wizard=owner) + Slot.create(code="A2", start=datetime(2024, 6, 21, 11, 0), current_wizard=owner) + Slot.create(code="B1", start=datetime(2024, 6, 22, 10, 0), current_wizard=owner) + + result = export_db_2_json() + assert "A1" in result["available_slots"] + assert "A2" in result["available_slots"] + assert "B1" in result["available_slots"] + assert len(result["available_slots"]) == 3 + + @use_test_database + def test_responsable_available_slots_includes_all(self): + owner = Pycampista.create(username="pepe") + Slot.create(code="A1", start=datetime(2024, 6, 21, 10, 0), current_wizard=owner) + Slot.create(code="A2", start=datetime(2024, 6, 21, 11, 0), current_wizard=owner) + Project.create(name="MiProyecto", owner=owner) + + result = export_db_2_json() + assert "pepe" in result["responsable_available_slots"] + assert result["responsable_available_slots"]["pepe"] == ["A1", "A2"] + + @use_test_database + def test_vote_interest_filter(self): + owner = Pycampista.create(username="pepe") + voter1 = Pycampista.create(username="juan") + voter2 = Pycampista.create(username="maria") + project = Project.create(name="MiProyecto", owner=owner) + + Vote.create( + project=project, pycampista=voter1, interest=True, + _project_pycampista_id=f"{project.id}-{voter1.id}", + ) + Vote.create( + project=project, pycampista=voter2, interest=False, + _project_pycampista_id=f"{project.id}-{voter2.id}", + ) + + result = export_db_2_json() + votes = result["projects"]["MiProyecto"]["votes"] + assert "juan" in votes + assert "maria" not in votes diff --git a/test/test_handler_announcements.py b/test/test_handler_announcements.py new file mode 100644 index 0000000..ab41495 --- /dev/null +++ b/test/test_handler_announcements.py @@ -0,0 +1,173 @@ +from telegram.ext import ConversationHandler +from pycamp_bot.models import Pycampista, Pycamp, Project, Vote +from pycamp_bot.commands.announcements import ( + announce, get_project, meeting_place, message_project, cancel, + user_is_admin, should_be_able_to_announce, + AnnouncementState, state, ERROR_MESSAGES, + PROYECTO, LUGAR, MENSAJE, +) +from test.conftest import ( + use_test_database_async, test_db, MODELS, + make_update, make_context, +) + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestUserIsAdmin: + + @use_test_database_async + async def test_returns_true_for_admin(self): + Pycampista.create(username="admin1", admin=True) + assert await user_is_admin("admin1") is True + + @use_test_database_async + async def test_returns_false_for_non_admin(self): + Pycampista.create(username="regular", admin=False) + assert await user_is_admin("regular") is False + + +class TestShouldBeAbleToAnnounce: + + @use_test_database_async + async def test_owner_can_announce(self): + owner = Pycampista.create(username="pepe") + project = Project.create(name="MiProj", owner=owner, topic="test") + assert await should_be_able_to_announce("pepe", project) is True + + @use_test_database_async + async def test_admin_can_announce_any_project(self): + owner = Pycampista.create(username="pepe") + Pycampista.create(username="admin1", admin=True) + project = Project.create(name="MiProj", owner=owner, topic="test") + assert await should_be_able_to_announce("admin1", project) is True + + @use_test_database_async + async def test_non_owner_non_admin_cannot_announce(self): + owner = Pycampista.create(username="pepe") + Pycampista.create(username="intruso", admin=False) + project = Project.create(name="MiProj", owner=owner, topic="test") + assert await should_be_able_to_announce("intruso", project) is False + + +class TestAnnouncementState: + + def test_initial_state(self): + s = AnnouncementState() + assert s.username is None + assert s.p_name == '' + assert s.current_project is False + assert s.projects == [] + assert s.owner == '' + assert s.lugar == '' + assert s.mensaje == '' + + +class TestAnnounce: + + @use_test_database_async + async def test_owner_with_projects_returns_proyecto_state(self): + Pycamp.create(headquarters="Narnia", active=True) + owner = Pycampista.create(username="pepe") + Project.create(name="MiProj", owner=owner, topic="test") + update = make_update(text="/anunciar", username="pepe") + context = make_context() + result = await announce(update, context) + assert result == PROYECTO + + @use_test_database_async + async def test_non_admin_no_projects_is_rejected(self): + Pycamp.create(headquarters="Narnia", active=True) + Pycampista.create(username="nadie", admin=False) + update = make_update(text="/anunciar", username="nadie") + context = make_context() + result = await announce(update, context) + assert result == ConversationHandler.END + + +class TestGetProject: + + @use_test_database_async + async def test_valid_project_returns_lugar(self): + owner = Pycampista.create(username="pepe") + Project.create(name="MiProj", owner=owner, topic="test") + state.username = "pepe" + update = make_update(text="MiProj", username="pepe") + context = make_context() + result = await get_project(update, context) + assert result == LUGAR + + @use_test_database_async + async def test_nonexistent_project_returns_proyecto(self): + Pycampista.create(username="pepe") + state.username = "pepe" + update = make_update(text="Fantasma", username="pepe") + context = make_context() + result = await get_project(update, context) + assert result == PROYECTO + + +class TestMeetingPlace: + + @use_test_database_async + async def test_sets_lugar_returns_mensaje(self): + update = make_update(text="sala principal") + context = make_context() + result = await meeting_place(update, context) + assert result == MENSAJE + assert state.lugar == "Sala principal" + + +class TestMessageProject: + + @use_test_database_async + async def test_sends_notifications_to_voters(self): + owner = Pycampista.create(username="pepe") + voter = Pycampista.create(username="juan", chat_id="111") + project = Project.create(name="MiProj", owner=owner, topic="test") + Vote.create( + project=project, pycampista=voter, interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + state.username = "pepe" + state.current_project = project + state.p_name = "MiProj" + state.owner = "pepe" + state.lugar = "Sala 1" + update = make_update(text="Arrancamos!", username="pepe") + context = make_context() + result = await message_project(update, context) + assert result == ConversationHandler.END + # Debe haber enviado mensajes al voter + assert context.bot.send_message.call_count >= 1 + + +class TestCancelAnnouncement: + + @use_test_database_async + async def test_cancel_returns_end(self): + update = make_update(text="/cancel") + context = make_context() + result = await cancel(update, context) + assert result == ConversationHandler.END + + +class TestErrorMessages: + + def test_error_messages_dict_has_expected_keys(self): + assert "format_error" in ERROR_MESSAGES + assert "not_admin" in ERROR_MESSAGES + assert "not_found" in ERROR_MESSAGES + assert "no_admin" in ERROR_MESSAGES + + def test_not_found_formats_with_project_name(self): + msg = ERROR_MESSAGES["not_found"].format(project_name="TestProj") + assert "TestProj" in msg diff --git a/test/test_handler_auth.py b/test/test_handler_auth.py new file mode 100644 index 0000000..3efcc6e --- /dev/null +++ b/test/test_handler_auth.py @@ -0,0 +1,155 @@ +import os +from unittest.mock import patch +from pycamp_bot.models import Pycampista +from pycamp_bot.commands.auth import ( + grant_admin, revoke_admin, list_admins, is_admin, admin_needed, +) +from test.conftest import ( + use_test_database_async, test_db, MODELS, + make_update, make_context, +) + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestIsAdmin: + + @use_test_database_async + async def test_returns_true_for_admin_user(self): + Pycampista.create(username="admin1", admin=True) + update = make_update(username="admin1") + context = make_context() + assert is_admin(update, context) is True + + @use_test_database_async + async def test_returns_false_for_non_admin_user(self): + Pycampista.create(username="regular", admin=False) + update = make_update(username="regular") + context = make_context() + assert is_admin(update, context) is False + + @use_test_database_async + async def test_returns_false_for_unknown_user(self): + update = make_update(username="unknown") + context = make_context() + assert is_admin(update, context) is False + + +class TestAdminNeeded: + + @use_test_database_async + async def test_allows_admin_to_proceed(self): + Pycampista.create(username="admin1", admin=True) + update = make_update(username="admin1") + context = make_context() + + called = False + async def handler(update, context): + nonlocal called + called = True + + wrapped = admin_needed(handler) + await wrapped(update, context) + assert called is True + + @use_test_database_async + async def test_blocks_non_admin(self): + Pycampista.create(username="regular", admin=False) + update = make_update(username="regular") + context = make_context() + + called = False + async def handler(update, context): + nonlocal called + called = True + + wrapped = admin_needed(handler) + await wrapped(update, context) + assert called is False + context.bot.send_message.assert_called_once() + call_kwargs = context.bot.send_message.call_args[1] + assert "No estas Autorizadx" in call_kwargs["text"] + + +class TestGrantAdmin: + + @use_test_database_async + @patch.dict(os.environ, {"PYCAMP_BOT_MASTER_KEY": "secreto123"}) + async def test_grants_admin_with_correct_password(self): + update = make_update(text="/su secreto123", username="pepe") + context = make_context() + await grant_admin(update, context) + user = Pycampista.get(Pycampista.username == "pepe") + assert user.admin is True + context.bot.send_message.assert_called_once() + assert "poder" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + @patch.dict(os.environ, {"PYCAMP_BOT_MASTER_KEY": "secreto123"}) + async def test_rejects_wrong_password(self): + update = make_update(text="/su wrongpass", username="pepe") + context = make_context() + await grant_admin(update, context) + user = Pycampista.get(Pycampista.username == "pepe") + assert user.admin is not True + assert "magic word" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + async def test_rejects_missing_parameter(self): + update = make_update(text="/su", username="pepe") + context = make_context() + await grant_admin(update, context) + assert "Parametros incorrectos" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + @patch.dict(os.environ, {}, clear=True) + async def test_error_when_env_not_set(self): + # Limpiar PYCAMP_BOT_MASTER_KEY si existe + os.environ.pop("PYCAMP_BOT_MASTER_KEY", None) + update = make_update(text="/su algo", username="pepe") + context = make_context() + await grant_admin(update, context) + assert "problema en el servidor" in context.bot.send_message.call_args[1]["text"] + + +class TestRevokeAdmin: + + @use_test_database_async + async def test_revokes_admin_privileges(self): + Pycampista.create(username="admin1", admin=True) + Pycampista.create(username="fallen", admin=True) + update = make_update(text="/degradar fallen", username="admin1") + context = make_context() + await revoke_admin(update, context) + fallen = Pycampista.get(Pycampista.username == "fallen") + assert fallen.admin is False + + @use_test_database_async + async def test_revoke_rejects_missing_parameter(self): + Pycampista.create(username="admin1", admin=True) + update = make_update(text="/degradar", username="admin1") + context = make_context() + await revoke_admin(update, context) + assert "Parametros incorrectos" in context.bot.send_message.call_args[1]["text"] + + +class TestListAdmins: + + @use_test_database_async + async def test_lists_all_admins(self): + Pycampista.create(username="admin1", admin=True) + Pycampista.create(username="admin2", admin=True) + update = make_update(username="admin1") + context = make_context() + await list_admins(update, context) + text = context.bot.send_message.call_args[1]["text"] + assert "@admin1" in text + assert "@admin2" in text diff --git a/test/test_handler_base.py b/test/test_handler_base.py new file mode 100644 index 0000000..6a226b1 --- /dev/null +++ b/test/test_handler_base.py @@ -0,0 +1,76 @@ +import os +from unittest.mock import patch, AsyncMock +from pycamp_bot.commands.base import start, msg_to_active_pycamp_chat +from pycamp_bot.commands.help_msg import get_help, HELP_MESSAGE, HELP_MESSAGE_ADMIN +from pycamp_bot.models import Pycampista +from test.conftest import ( + use_test_database_async, test_db, MODELS, + make_update, make_context, +) + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestStart: + + @use_test_database_async + async def test_welcomes_user_with_username(self): + update = make_update(text="/start", username="pepe") + context = make_context() + await start(update, context) + text = context.bot.send_message.call_args[1]["text"] + assert "pepe" in text + assert "Bienvenidx" in text + + @use_test_database_async + async def test_asks_for_username_when_missing(self): + update = make_update(text="/start", username=None) + context = make_context() + await start(update, context) + text = context.bot.send_message.call_args[1]["text"] + assert "username" in text.lower() + + +class TestMsgToActivePycampChat: + + @use_test_database_async + @patch.dict(os.environ, {"TEST_CHAT_ID": "12345"}) + async def test_sends_message_when_env_set(self): + bot = AsyncMock() + await msg_to_active_pycamp_chat(bot, "Test message") + bot.send_message.assert_called_once() + assert bot.send_message.call_args[1]["text"] == "Test message" + + @use_test_database_async + async def test_does_nothing_when_env_not_set(self): + os.environ.pop("TEST_CHAT_ID", None) + bot = AsyncMock() + await msg_to_active_pycamp_chat(bot, "Test message") + bot.send_message.assert_not_called() + + +class TestGetHelp: + + @use_test_database_async + async def test_returns_admin_help_for_admin(self): + Pycampista.create(username="admin1", admin=True) + update = make_update(username="admin1") + context = make_context() + result = get_help(update, context) + assert result == HELP_MESSAGE_ADMIN + + @use_test_database_async + async def test_returns_normal_help_for_user(self): + Pycampista.create(username="regular", admin=False) + update = make_update(username="regular") + context = make_context() + result = get_help(update, context) + assert result == HELP_MESSAGE diff --git a/test/test_handler_devtools.py b/test/test_handler_devtools.py new file mode 100644 index 0000000..dacf423 --- /dev/null +++ b/test/test_handler_devtools.py @@ -0,0 +1,76 @@ +"""Tests para handlers de devtools.py: /mostrar_version.""" +from unittest.mock import patch, MagicMock + +from pycamp_bot.commands.devtools import show_version +from test.conftest import ( + use_test_database_async, test_db, MODELS, + make_update, make_context, +) + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestShowVersion: + + @use_test_database_async + @patch("pycamp_bot.commands.devtools.subprocess.run") + async def test_shows_version_info(self, mock_run): + """Verifica que show_version envía info de commit, Python y deps.""" + # Simular los 4 subprocess.run: rev-parse, log, diff, pip freeze + mock_run.side_effect = [ + MagicMock(stdout=b"abc1234\n", returncode=0), # git rev-parse + MagicMock(stdout=b"2024-06-20 10:00:00 -0300\n", returncode=0), # git log + MagicMock(returncode=0), # git diff (clean) + MagicMock(stdout=b"python-telegram-bot==21.10\npeewee==3.17.0\n", returncode=0), # pip freeze + ] + update = make_update(text="/mostrar_version") + context = make_context() + await show_version(update, context) + update.message.reply_text.assert_called_once() + text = update.message.reply_text.call_args[0][0] + assert "abc1234" in text + # escape_markdown escapa guiones y puntos; comprobar nombre y versión + assert "python" in text and "21" in text and "telegram" in text + + @use_test_database_async + @patch("pycamp_bot.commands.devtools.subprocess.run") + async def test_dirty_worktree_shows_red(self, mock_run): + """Verifica que worktree sucio muestra indicador rojo.""" + mock_run.side_effect = [ + MagicMock(stdout=b"abc1234\n", returncode=0), + MagicMock(stdout=b"2024-06-20 10:00:00 -0300\n", returncode=0), + MagicMock(returncode=1), # git diff: dirty + MagicMock(stdout=b"peewee==3.17.0\n", returncode=0), + ] + update = make_update(text="/mostrar_version") + context = make_context() + await show_version(update, context) + text = update.message.reply_text.call_args[0][0] + # Indicador rojo para worktree sucio + assert "\U0001f534" in text # 🔴 + + @use_test_database_async + @patch("pycamp_bot.commands.devtools.subprocess.run") + @patch.dict("os.environ", {"SENTRY_DATA_SOURCE_NAME": "https://sentry.io/123"}) + async def test_sentry_env_set_shows_green(self, mock_run): + """Verifica que con Sentry configurado muestra indicador verde.""" + mock_run.side_effect = [ + MagicMock(stdout=b"abc1234\n", returncode=0), + MagicMock(stdout=b"2024-06-20 10:00:00 -0300\n", returncode=0), + MagicMock(returncode=0), + MagicMock(stdout=b"peewee==3.17.0\n", returncode=0), + ] + update = make_update(text="/mostrar_version") + context = make_context() + await show_version(update, context) + text = update.message.reply_text.call_args[0][0] + # Último indicador debe ser verde (Sentry definida) + assert text.count("\U0001f7e2") >= 2 # 🟢 para clean worktree + sentry diff --git a/test/test_handler_manage_pycamp.py b/test/test_handler_manage_pycamp.py new file mode 100644 index 0000000..3a45fbf --- /dev/null +++ b/test/test_handler_manage_pycamp.py @@ -0,0 +1,212 @@ +import datetime as dt +from telegram.ext import ConversationHandler +from pycamp_bot.models import Pycamp, Pycampista, PycampistaAtPycamp +from pycamp_bot.commands.manage_pycamp import ( + add_pycamp, define_start_date, define_duration, end_pycamp, + set_active_pycamp, add_pycampista_to_pycamp, list_pycamps, + list_pycampistas, cancel, SET_DATE_STATE, SET_DURATION_STATE, +) +from test.conftest import ( + use_test_database_async, test_db, MODELS, + make_update, make_context, +) + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestAddPycamp: + + @use_test_database_async + async def test_creates_pycamp_and_returns_set_date_state(self): + Pycampista.create(username="admin1", admin=True) + update = make_update(text="/empezar_pycamp Narnia", username="admin1") + context = make_context() + result = await add_pycamp(update, context) + assert result == SET_DATE_STATE + pycamp = Pycamp.get(Pycamp.headquarters == "Narnia") + assert pycamp.active is True + + @use_test_database_async + async def test_rejects_missing_name(self): + Pycampista.create(username="admin1", admin=True) + update = make_update(text="/empezar_pycamp", username="admin1") + context = make_context() + result = await add_pycamp(update, context) + assert result is None + assert "necesita un parametro" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + async def test_rejects_empty_name(self): + Pycampista.create(username="admin1", admin=True) + update = make_update(text="/empezar_pycamp ", username="admin1") + context = make_context() + result = await add_pycamp(update, context) + assert result is None + assert "vacío" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + async def test_deactivates_previous_pycamp(self): + Pycampista.create(username="admin1", admin=True) + Pycamp.create(headquarters="Viejo", active=True) + update = make_update(text="/empezar_pycamp Nuevo", username="admin1") + context = make_context() + result = await add_pycamp(update, context) + assert result == SET_DATE_STATE + viejo = Pycamp.get(Pycamp.headquarters == "Viejo") + assert viejo.active is False + nuevo = Pycamp.get(Pycamp.headquarters == "Nuevo") + assert nuevo.active is True + + @use_test_database_async + async def test_non_admin_is_blocked(self): + Pycampista.create(username="user1", admin=False) + update = make_update(text="/empezar_pycamp Narnia", username="user1") + context = make_context() + result = await add_pycamp(update, context) + assert "No estas Autorizadx" in context.bot.send_message.call_args[1]["text"] + + +class TestDefineStartDate: + + @use_test_database_async + async def test_valid_date_returns_duration_state(self): + Pycamp.create(headquarters="Narnia", active=True) + update = make_update(text="2024-06-20") + context = make_context() + result = await define_start_date(update, context) + assert result == SET_DURATION_STATE + pycamp = Pycamp.get(Pycamp.active == True) + assert pycamp.init == dt.datetime(2024, 6, 20) + + @use_test_database_async + async def test_invalid_date_returns_same_state(self): + Pycamp.create(headquarters="Narnia", active=True) + update = make_update(text="no-es-fecha") + context = make_context() + result = await define_start_date(update, context) + assert result == SET_DATE_STATE + + +class TestDefineDuration: + + @use_test_database_async + async def test_valid_duration_sets_end_and_finishes(self): + Pycamp.create( + headquarters="Narnia", active=True, + init=dt.datetime(2024, 6, 20), + ) + update = make_update(text="4") + context = make_context() + result = await define_duration(update, context) + assert result == ConversationHandler.END + pycamp = Pycamp.get(Pycamp.active == True) + assert pycamp.end.day == 23 + + @use_test_database_async + async def test_invalid_duration_returns_same_state(self): + Pycamp.create( + headquarters="Narnia", active=True, + init=dt.datetime(2024, 6, 20), + ) + update = make_update(text="abc") + context = make_context() + result = await define_duration(update, context) + assert result == SET_DURATION_STATE + + +class TestEndPycamp: + + @use_test_database_async + async def test_deactivates_pycamp(self): + Pycampista.create(username="admin1", admin=True) + Pycamp.create(headquarters="Narnia", active=True) + update = make_update(text="/terminar_pycamp", username="admin1") + context = make_context() + await end_pycamp(update, context) + pycamp = Pycamp.get(Pycamp.headquarters == "Narnia") + assert pycamp.active is False + + +class TestSetActivePycamp: + + @use_test_database_async + async def test_activates_named_pycamp(self): + Pycampista.create(username="admin1", admin=True) + Pycamp.create(headquarters="Narnia", active=False) + update = make_update(text="/activar_pycamp Narnia", username="admin1") + context = make_context() + await set_active_pycamp(update, context) + pycamp = Pycamp.get(Pycamp.headquarters == "Narnia") + assert pycamp.active is True + + @use_test_database_async + async def test_rejects_nonexistent_pycamp(self): + Pycampista.create(username="admin1", admin=True) + update = make_update(text="/activar_pycamp Mordor", username="admin1") + context = make_context() + await set_active_pycamp(update, context) + assert "no existe" in context.bot.send_message.call_args[1]["text"] + + +class TestAddPycampistaToP: + + @use_test_database_async + async def test_adds_user_to_active_pycamp(self): + Pycamp.create(headquarters="Narnia", active=True) + update = make_update(text="/voy_al_pycamp", username="pepe") + context = make_context() + await add_pycampista_to_pycamp(update, context) + assert PycampistaAtPycamp.select().count() == 1 + + +class TestListPycamps: + + @use_test_database_async + async def test_lists_all_pycamps(self): + Pycamp.create(headquarters="Narnia") + Pycamp.create(headquarters="Mordor") + update = make_update(text="/pycamps") + context = make_context() + await list_pycamps(update, context) + update.message.reply_text.assert_called_once() + text = update.message.reply_text.call_args[0][0] + assert "Narnia" in text + assert "Mordor" in text + + +class TestListPycampistas: + + @use_test_database_async + async def test_lists_pycampistas_in_active_pycamp(self): + p = Pycamp.create(headquarters="Narnia", active=True) + user1 = Pycampista.create(username="pepe", chat_id="111") + user2 = Pycampista.create(username="juan", chat_id="222") + PycampistaAtPycamp.create(pycamp=p, pycampista=user1) + PycampistaAtPycamp.create(pycamp=p, pycampista=user2) + update = make_update(text="/pycampistas", username="pepe") + context = make_context() + # Nota: list_pycampistas tiene un bug en la línea final + # (concatena str + int), así que este test documentará el fallo. + try: + await list_pycampistas(update, context) + except TypeError: + # Bug conocido: `text + len(pycampistas_at_pycamp)` falla + pass + + +class TestCancel: + + @use_test_database_async + async def test_cancel_returns_end(self): + update = make_update(text="/cancel") + context = make_context() + result = await cancel(update, context) + assert result == ConversationHandler.END diff --git a/test/test_handler_projects.py b/test/test_handler_projects.py new file mode 100644 index 0000000..190ca3a --- /dev/null +++ b/test/test_handler_projects.py @@ -0,0 +1,461 @@ +import peewee +from peewee import JOIN +from telegram.ext import ConversationHandler +from pycamp_bot.models import Pycampista, Pycamp, Project, Vote, Slot +from pycamp_bot.commands.projects import ( + load_project, naming_project, project_level, project_topic, + save_project, ask_if_repository_exists, ask_if_group_exists, + project_repository, project_group, cancel, + ask_project_name, ask_repository_url, ask_group_url, + add_repository, add_group, + delete_project, show_projects, show_participants, show_my_projects, + start_project_load, end_project_load, + current_projects, + NOMBRE, DIFICULTAD, TOPIC, CHECK_REPOSITORIO, REPOSITORIO, CHECK_GRUPO, GRUPO, +) +from test.conftest import ( + use_test_database_async, test_db, MODELS, + make_update, make_callback_update, make_context, +) + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestLoadProject: + + @use_test_database_async + async def test_starts_dialog_returns_nombre(self): + Pycampista.create(username="admin1", admin=True) + p = Pycamp.create(headquarters="Narnia", active=True, project_load_authorized=True) + update = make_update(text="/cargar_proyecto", username="admin1") + context = make_context() + result = await load_project(update, context) + assert result == NOMBRE + + @use_test_database_async + async def test_blocked_when_not_authorized(self): + Pycampista.create(username="user1") + Pycamp.create(headquarters="Narnia", active=True, project_load_authorized=False) + update = make_update(text="/cargar_proyecto", username="user1") + context = make_context() + result = await load_project(update, context) + assert result is None + assert "no está autorizada" in context.bot.send_message.call_args[1]["text"] + + +class TestNamingProject: + + @use_test_database_async + async def test_sets_project_name_returns_dificultad(self): + # chat_id debe coincidir con make_update() para que get_or_create encuentre al usuario + Pycampista.create(username="pepe", chat_id=str(67890)) + update = make_update(text="Mi Proyecto Genial", username="pepe") + context = make_context() + result = await naming_project(update, context) + assert result == DIFICULTAD + assert "pepe" in current_projects + assert current_projects["pepe"].name == "Mi Proyecto Genial" + + @use_test_database_async + async def test_handles_cargar_proyecto_reentry(self): + update = make_update(text="/cargar_proyecto", username="pepe") + context = make_context() + result = await naming_project(update, context) + assert result == NOMBRE + + +class TestProjectLevel: + + @use_test_database_async + async def test_valid_level_returns_topic(self): + owner = Pycampista.create(username="pepe") + proj = Project(name="Test", owner=owner) + current_projects["pepe"] = proj + update = make_update(text="2", username="pepe") + context = make_context() + result = await project_level(update, context) + assert result == TOPIC + assert current_projects["pepe"].difficult_level == "2" + + @use_test_database_async + async def test_invalid_level_returns_dificultad(self): + update = make_update(text="5", username="pepe") + context = make_context() + result = await project_level(update, context) + assert result == DIFICULTAD + + +class TestProjectTopic: + + @use_test_database_async + async def test_sets_topic_returns_check_repositorio(self): + owner = Pycampista.create(username="pepe") + proj = Project(name="Test", owner=owner) + current_projects["pepe"] = proj + update = make_update(text="django", username="pepe") + context = make_context() + result = await project_topic(update, context) + assert result == CHECK_REPOSITORIO + assert current_projects["pepe"].topic == "django" + + +class TestAskIfRepositoryExists: + + @use_test_database_async + async def test_yes_returns_repositorio(self): + update = make_callback_update(data="repoexists:si") + context = make_context() + result = await ask_if_repository_exists(update, context) + assert result == REPOSITORIO + + @use_test_database_async + async def test_no_returns_check_grupo(self): + owner = Pycampista.create(username="testuser", chat_id=str(67890)) + proj = Project(name="ProjTmp", owner=owner, topic="test") + current_projects["testuser"] = proj + update = make_callback_update(data="groupexists:no", username="testuser") + context = make_context() + result = await ask_if_group_exists(update, context) + assert result == ConversationHandler.END + assert Project.select().where(Project.name == "ProjTmp").exists() + + +class TestAskIfGroupExists: + + @use_test_database_async + async def test_yes_returns_grupo(self): + update = make_callback_update(data="groupexists:si") + context = make_context() + result = await ask_if_group_exists(update, context) + assert result == GRUPO + + @use_test_database_async + async def test_no_saves_project_and_ends(self): + owner = Pycampista.create(username="testuser") + proj = Project(name="TestProj", owner=owner, topic="django") + current_projects["testuser"] = proj + update = make_callback_update(data="groupexists:no", username="testuser") + context = make_context() + result = await ask_if_group_exists(update, context) + assert result == ConversationHandler.END + assert Project.select().where(Project.name == "TestProj").exists() + + +class TestSaveProject: + + @use_test_database_async + async def test_saves_project_successfully(self): + owner = Pycampista.create(username="pepe") + proj = Project(name="NuevoProj", owner=owner, topic="flask") + current_projects["pepe"] = proj + context = make_context() + await save_project("pepe", 67890, context) + assert Project.select().where(Project.name == "NuevoProj").exists() + assert "cargado" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + async def test_handles_duplicate_name(self): + owner = Pycampista.create(username="pepe") + Project.create(name="Existente", owner=owner, topic="flask") + proj = Project(name="Existente", owner=owner, topic="django") + current_projects["pepe"] = proj + context = make_context() + await save_project("pepe", 67890, context) + assert "ya fue cargado" in context.bot.send_message.call_args[1]["text"] + + +class TestProjectRepository: + + @use_test_database_async + async def test_sets_repo_url_returns_check_grupo(self): + owner = Pycampista.create(username="pepe") + proj = Project(name="Test", owner=owner) + current_projects["pepe"] = proj + update = make_update(text="https://github.com/test", username="pepe") + context = make_context() + result = await project_repository(update, context) + assert result == CHECK_GRUPO + assert current_projects["pepe"].repository_url == "https://github.com/test" + + +class TestProjectGroup: + + @use_test_database_async + async def test_sets_group_url_and_saves(self): + owner = Pycampista.create(username="pepe") + proj = Project(name="TestGrp", owner=owner, topic="flask") + current_projects["pepe"] = proj + update = make_update(text="https://t.me/grupo", username="pepe") + context = make_context() + result = await project_group(update, context) + assert result == ConversationHandler.END + assert Project.select().where(Project.name == "TestGrp").exists() + + +class TestDeleteProject: + + @use_test_database_async + async def test_owner_can_delete_project(self): + owner = Pycampista.create(username="pepe") + Project.create(name="borrame", owner=owner, topic="test") + update = make_update(text="/borrar_proyecto borrame", username="pepe") + context = make_context() + await delete_project(update, context) + assert not Project.select().where(Project.name == "borrame").exists() + assert "eliminado" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + async def test_admin_can_delete_project(self): + owner = Pycampista.create(username="pepe") + Pycampista.create(username="admin1", admin=True) + Project.create(name="borrame", owner=owner, topic="test") + update = make_update(text="/borrar_proyecto borrame", username="admin1") + context = make_context() + await delete_project(update, context) + assert not Project.select().where(Project.name == "borrame").exists() + + @use_test_database_async + async def test_non_owner_non_admin_cannot_delete(self): + owner = Pycampista.create(username="pepe") + Pycampista.create(username="intruso", admin=False) + Project.create(name="borrame", owner=owner, topic="test") + update = make_update(text="/borrar_proyecto borrame", username="intruso") + context = make_context() + await delete_project(update, context) + assert Project.select().where(Project.name == "borrame").exists() + assert "Careta" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + async def test_missing_project_name_shows_help(self): + update = make_update(text="/borrar_proyecto", username="pepe") + context = make_context() + await delete_project(update, context) + assert "nombre de proyecto" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + async def test_nonexistent_project_shows_error(self): + update = make_update(text="/borrar_proyecto fantasma", username="pepe") + context = make_context() + await delete_project(update, context) + assert "No se encontró" in context.bot.send_message.call_args[1]["text"] + + +class TestShowProjects: + + @use_test_database_async + async def test_shows_projects_list(self): + owner = Pycampista.create(username="pepe") + Project.create(name="Proyecto1", owner=owner, topic="django", difficult_level=1) + Project.create(name="Proyecto2", owner=owner, topic="flask", difficult_level=2) + update = make_update(text="/proyectos") + context = make_context() + await show_projects(update, context) + text = update.message.reply_text.call_args[1].get("text", + update.message.reply_text.call_args[0][0] if update.message.reply_text.call_args[0] else "") + assert "Proyecto1" in text or "Proyecto2" in text + + @use_test_database_async + async def test_shows_no_projects_message(self): + update = make_update(text="/proyectos") + context = make_context() + await show_projects(update, context) + text = update.message.reply_text.call_args[0][0] + assert "no hay" in text.lower() + + +class TestShowParticipants: + + @use_test_database_async + async def test_shows_participants_for_project(self): + owner = Pycampista.create(username="pepe") + voter = Pycampista.create(username="juan", chat_id="111") + project = Project.create(name="MiProyecto", owner=owner, topic="test") + Vote.create( + project=project, pycampista=voter, interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + update = make_update(text="/participantes MiProyecto") + context = make_context() + await show_participants(update, context) + text = update.message.reply_text.call_args[0][0] + assert "@juan" in text + + @use_test_database_async + async def test_missing_project_name(self): + update = make_update(text="/participantes") + context = make_context() + await show_participants(update, context) + assert "nombre del proyecto" in context.bot.send_message.call_args[1]["text"] + + +class TestStartEndProjectLoad: + + @use_test_database_async + async def test_start_project_load_opens_loading(self): + Pycampista.create(username="admin1", admin=True) + Pycamp.create(headquarters="Narnia", active=True, project_load_authorized=False) + update = make_update(text="/empezar_carga_proyectos", username="admin1") + context = make_context() + await start_project_load(update, context) + pycamp = Pycamp.get(Pycamp.active == True) + assert pycamp.project_load_authorized is True + + @use_test_database_async + async def test_end_project_load_closes_loading(self): + Pycampista.create(username="admin1", admin=True) + Pycamp.create(headquarters="Narnia", active=True, project_load_authorized=True) + update = make_update(text="/terminar_carga_proyectos", username="admin1") + context = make_context() + await end_project_load(update, context) + pycamp = Pycamp.get(Pycamp.active == True) + assert pycamp.project_load_authorized is False + + +class TestShowMyProjects: + + @use_test_database_async + async def test_shows_voted_projects(self): + owner = Pycampista.create(username="pepe") + voter = Pycampista.create(username="juan", chat_id="111") + project = Project.create(name="MiProj", owner=owner, topic="test") + Vote.create( + project=project, pycampista=voter, interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + update = make_update(text="/mis_proyectos", username="juan") + context = make_context() + await show_my_projects(update, context) + text = update.message.reply_text.call_args[0][0] + assert "MiProj" in text + + @use_test_database_async + async def test_no_votes_shows_message(self): + Pycampista.create(username="pepe") + update = make_update(text="/mis_proyectos", username="pepe") + context = make_context() + await show_my_projects(update, context) + text = update.message.reply_text.call_args[0][0] + assert "No votaste" in text + + @use_test_database_async + async def test_shows_projects_with_assigned_slots(self): + import datetime as dt + Pycamp.create( + headquarters="Narnia", active=True, + init=dt.datetime(2024, 6, 20), + ) + owner = Pycampista.create(username="pepe") + voter = Pycampista.create(username="juan", chat_id="111") + slot_a1 = Slot.create(code="A1", start=9) + project = Project.create( + name="MiProj", owner=owner, topic="test", slot=slot_a1, + ) + Vote.create( + project=project, pycampista=voter, interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + update = make_update(text="/mis_proyectos", username="juan") + context = make_context() + await show_my_projects(update, context) + text = update.message.reply_text.call_args[0][0] + assert "MiProj" in text + assert "9:00" in text + + +class TestAskProjectName: + + @use_test_database_async + async def test_shows_user_projects_for_selection(self): + Pycamp.create(headquarters="Narnia", active=True) + owner = Pycampista.create(username="pepe") + Project.create(name="Proj1", owner=owner, topic="test") + Project.create(name="Proj2", owner=owner, topic="test") + update = make_update(text="/agregar_repositorio", username="pepe") + context = make_context() + result = await ask_project_name(update, context) + assert result == 1 + assert "modificar" in context.bot.send_message.call_args[1]["text"] + + @use_test_database_async + async def test_no_projects_shows_message(self): + Pycamp.create(headquarters="Narnia", active=True) + Pycampista.create(username="pepe") + update = make_update(text="/agregar_repositorio", username="pepe") + context = make_context() + result = await ask_project_name(update, context) + assert result == ConversationHandler.END + assert "No cargaste" in context.bot.send_message.call_args[1]["text"] + + +class TestAskRepositoryUrl: + + @use_test_database_async + async def test_stores_project_id_and_asks_url(self): + update = make_callback_update(data="projectname:42", username="pepe") + context = make_context() + result = await ask_repository_url(update, context) + assert result == 2 + assert current_projects["pepe"] == "42" + assert "URL" in context.bot.send_message.call_args[1]["text"] + + +class TestAskGroupUrl: + + @use_test_database_async + async def test_stores_project_id_and_asks_url(self): + update = make_callback_update(data="projectname:42", username="pepe") + context = make_context() + result = await ask_group_url(update, context) + assert result == 2 + assert current_projects["pepe"] == "42" + assert "URL" in context.bot.send_message.call_args[1]["text"] + + +class TestAddRepository: + + @use_test_database_async + async def test_adds_repository_url(self): + owner = Pycampista.create(username="pepe") + project = Project.create(name="Proj1", owner=owner, topic="test") + current_projects["pepe"] = str(project.id) + update = make_update(text="https://github.com/mi-repo", username="pepe") + context = make_context() + result = await add_repository(update, context) + assert result == ConversationHandler.END + proj = Project.get(Project.id == project.id) + assert proj.repository_url == "https://github.com/mi-repo" + assert "Repositorio agregado" in context.bot.send_message.call_args[1]["text"] + + +class TestAddGroup: + + @use_test_database_async + async def test_adds_group_url(self): + owner = Pycampista.create(username="pepe") + project = Project.create(name="Proj1", owner=owner, topic="test") + current_projects["pepe"] = str(project.id) + update = make_update(text="https://t.me/grupo", username="pepe") + context = make_context() + result = await add_group(update, context) + assert result == ConversationHandler.END + proj = Project.get(Project.id == project.id) + assert proj.group_url == "https://t.me/grupo" + assert "Grupo agregado" in context.bot.send_message.call_args[1]["text"] + + +class TestCancelProject: + + @use_test_database_async + async def test_cancel_returns_end(self): + update = make_update(text="/cancel") + context = make_context() + result = await cancel(update, context) + assert result == ConversationHandler.END diff --git a/test/test_handler_raffle.py b/test/test_handler_raffle.py new file mode 100644 index 0000000..c3365e8 --- /dev/null +++ b/test/test_handler_raffle.py @@ -0,0 +1,45 @@ +from unittest.mock import patch +from pycamp_bot.models import Pycampista +from pycamp_bot.commands.raffle import get_random_user +from test.conftest import ( + use_test_database_async, test_db, MODELS, + make_update, make_context, +) + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestGetRandomUser: + + @use_test_database_async + async def test_returns_a_username(self): + Pycampista.create(username="admin1", admin=True) + Pycampista.create(username="pepe") + Pycampista.create(username="juan") + update = make_update(text="/rifar", username="admin1") + context = make_context() + await get_random_user(update, context) + update.message.reply_text.assert_called_once() + username = update.message.reply_text.call_args[0][0] + assert username in ["admin1", "pepe", "juan"] + + @use_test_database_async + @patch("pycamp_bot.commands.raffle.random.randint", return_value=1) + async def test_returns_specific_user_with_mocked_random(self, mock_randint): + Pycampista.create(username="admin1", admin=True) + Pycampista.create(username="pepe") + update = make_update(text="/rifar", username="admin1") + context = make_context() + await get_random_user(update, context) + update.message.reply_text.assert_called_once() + # Con randint=1, retorna el primer Pycampista creado + username = update.message.reply_text.call_args[0][0] + assert username == "admin1" diff --git a/test/test_handler_voting.py b/test/test_handler_voting.py new file mode 100644 index 0000000..f2e3935 --- /dev/null +++ b/test/test_handler_voting.py @@ -0,0 +1,170 @@ +import peewee +from pycamp_bot.models import Pycampista, Pycamp, Project, Vote +from pycamp_bot.commands.voting import ( + start_voting, end_voting, vote, button, vote_count, +) +from test.conftest import ( + use_test_database_async, test_db, MODELS, + make_update, make_callback_update, make_context, +) + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestStartVoting: + + @use_test_database_async + async def test_opens_voting(self): + Pycampista.create(username="admin1", admin=True) + Pycamp.create(headquarters="Narnia", active=True, vote_authorized=False) + update = make_update(text="/empezar_votacion_proyectos", username="admin1") + context = make_context() + await start_voting(update, context) + pycamp = Pycamp.get(Pycamp.active == True) + assert pycamp.vote_authorized is True + + @use_test_database_async + async def test_already_open_voting(self): + Pycampista.create(username="admin1", admin=True) + Pycamp.create(headquarters="Narnia", active=True, vote_authorized=True) + update = make_update(text="/empezar_votacion_proyectos", username="admin1") + context = make_context() + await start_voting(update, context) + update.message.reply_text.assert_called_once() + text = update.message.reply_text.call_args[0][0] + assert "ya estaba abierta" in text + + +class TestEndVoting: + + @use_test_database_async + async def test_closes_voting(self): + Pycampista.create(username="admin1", admin=True) + Pycamp.create(headquarters="Narnia", active=True, vote_authorized=True) + update = make_update(text="/terminar_votacion_proyectos", username="admin1") + context = make_context() + await end_voting(update, context) + pycamp = Pycamp.get(Pycamp.active == True) + assert pycamp.vote_authorized is False + + +class TestVote: + + @use_test_database_async + async def test_vote_sends_project_list(self): + Pycamp.create(headquarters="Narnia", active=True, vote_authorized=True) + owner = Pycampista.create(username="pepe") + Project.create(name="Proyecto1", owner=owner, topic="django") + Project.create(name="Proyecto2", owner=owner, topic="flask") + update = make_update(text="/votar", username="pepe") + context = make_context() + await vote(update, context) + # Debe enviar reply_text por cada proyecto + el mensaje inicial + assert update.message.reply_text.call_count >= 2 + + @use_test_database_async + async def test_vote_creates_test_project_if_empty(self): + Pycamp.create(headquarters="Narnia", active=True, vote_authorized=True) + update = make_update(text="/votar", username="pepe") + context = make_context() + await vote(update, context) + assert Project.select().where(Project.name == "PROYECTO DE PRUEBA").exists() + + +class TestButton: + + @use_test_database_async + async def test_vote_si_saves_interest_true(self): + owner = Pycampista.create(username="owner1", chat_id="11111") + voter = Pycampista.create(username="voter1", chat_id="67890") + project = Project.create(name="Proyecto1", owner=owner, topic="test") + update = make_callback_update( + data="vote:si", username="voter1", + message_text="Proyecto1", + ) + context = make_context() + await button(update, context) + vote_obj = Vote.get(Vote.pycampista == voter) + assert vote_obj.interest is True + text = context.bot.edit_message_text.call_args[1]["text"] + assert "Sumade" in text + + @use_test_database_async + async def test_vote_no_saves_interest_false(self): + owner = Pycampista.create(username="owner1", chat_id="11111") + voter = Pycampista.create(username="voter1", chat_id="67890") + project = Project.create(name="Proyecto1", owner=owner, topic="test") + update = make_callback_update( + data="vote:no", username="voter1", + message_text="Proyecto1", + ) + context = make_context() + await button(update, context) + vote_obj = Vote.get(Vote.pycampista == voter) + assert vote_obj.interest is False + + @use_test_database_async + async def test_duplicate_vote_shows_warning(self): + owner = Pycampista.create(username="owner1", chat_id="11111") + voter = Pycampista.create(username="voter1", chat_id="67890") + project = Project.create(name="Proyecto1", owner=owner, topic="test") + Vote.create( + project=project, pycampista=voter, interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + update = make_callback_update( + data="vote:si", username="voter1", + message_text="Proyecto1", + ) + context = make_context() + await button(update, context) + text = context.bot.edit_message_text.call_args[1]["text"] + assert "Ya te habías sumado" in text + + +class TestVoteCount: + + @use_test_database_async + async def test_counts_unique_voters(self): + Pycampista.create(username="admin1", admin=True) + owner = Pycampista.create(username="owner1") + v1 = Pycampista.create(username="voter1") + v2 = Pycampista.create(username="voter2") + p1 = Project.create(name="P1", owner=owner, topic="test") + p2 = Project.create(name="P2", owner=owner, topic="test") + Vote.create(project=p1, pycampista=v1, interest=True, _project_pycampista_id=f"{p1.id}-{v1.id}") + Vote.create(project=p2, pycampista=v1, interest=True, _project_pycampista_id=f"{p2.id}-{v1.id}") + Vote.create(project=p1, pycampista=v2, interest=True, _project_pycampista_id=f"{p1.id}-{v2.id}") + + update = make_update(text="/contar_votos", username="admin1") + context = make_context() + await vote_count(update, context) + text = context.bot.send_message.call_args[1]["text"] + assert "2" in text + + @use_test_database_async + async def test_zero_votes(self): + Pycampista.create(username="admin1", admin=True) + update = make_update(text="/contar_votos", username="admin1") + context = make_context() + await vote_count(update, context) + text = context.bot.send_message.call_args[1]["text"] + assert "0" in text + + @use_test_database_async + async def test_contar_votos_rejects_non_admin(self): + Pycampista.create(username="user1", admin=False) + update = make_update(text="/contar_votos", username="user1") + context = make_context() + await vote_count(update, context) + context.bot.send_message.assert_called_once() + text = context.bot.send_message.call_args[1]["text"] + assert "No estas Autorizadx" in text diff --git a/test/test_handler_wizard.py b/test/test_handler_wizard.py new file mode 100644 index 0000000..d53df76 --- /dev/null +++ b/test/test_handler_wizard.py @@ -0,0 +1,294 @@ +"""Tests para handlers de wizard.py: /ser_magx, /ver_magx, /evocar_magx, /agendar_magx, /ver_agenda_magx.""" +from datetime import datetime +from freezegun import freeze_time +from pycamp_bot.models import Pycampista, Pycamp, PycampistaAtPycamp, WizardAtPycamp +from pycamp_bot.commands.wizard import ( + become_wizard, list_wizards, summon_wizard, schedule_wizards, + show_wizards_schedule, format_wizards_schedule, + persist_wizards_schedule_in_db, aux_resolve_show_all, +) +from test.conftest import ( + use_test_database, use_test_database_async, test_db, MODELS, + make_update, make_message, make_context, +) + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestBecomeWizard: + + @use_test_database_async + async def test_registers_user_as_wizard(self): + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + update = make_update(text="/ser_magx", username="gandalf") + context = make_context() + await become_wizard(update, context, pycamp=p) + wizard = Pycampista.get(Pycampista.username == "gandalf") + assert wizard.wizard is True + assert "registrado como magx" in context.bot.send_message.call_args[1]["text"] + + +class TestListWizards: + + @use_test_database_async + async def test_lists_registered_wizards(self): + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + p.add_wizard("gandalf", "111") + p.add_wizard("merlin", "222") + update = make_update(text="/ver_magx", username="admin1") + context = make_context() + await list_wizards(update, context, pycamp=p) + text = context.bot.send_message.call_args[1]["text"] + assert "@gandalf" in text + assert "@merlin" in text + + @use_test_database_async + async def test_empty_list(self): + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + update = make_update(text="/ver_magx", username="admin1") + context = make_context() + # list_wizards con msg vacío: BadRequest se ignora internamente + await list_wizards(update, context, pycamp=p) + + +class TestSummonWizard: + + @use_test_database_async + @freeze_time("2024-06-21 15:30:00") + async def test_summons_current_wizard(self): + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + w = p.add_wizard("gandalf", "111") + persist_wizards_schedule_in_db(p) + + update = make_update(text="/evocar_magx", username="pepe") + context = make_context() + await summon_wizard(update, context, pycamp=p) + # Debe haber enviado un PING al wizard + sent_texts = [call[1]["text"] for call in context.bot.send_message.call_args_list] + assert any("PING" in t or "magx" in t.lower() for t in sent_texts) + + @use_test_database_async + async def test_no_wizard_scheduled(self): + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + update = make_update(text="/evocar_magx", username="pepe") + context = make_context() + await summon_wizard(update, context, pycamp=p) + text = context.bot.send_message.call_args[1]["text"] + assert "No hay" in text + + @use_test_database_async + @freeze_time("2024-06-21 15:30:00") + async def test_wizard_summons_self(self): + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + w = p.add_wizard("gandalf", "111") + persist_wizards_schedule_in_db(p) + + update = make_update(text="/evocar_magx", username="gandalf") + context = make_context() + await summon_wizard(update, context, pycamp=p) + sent_texts = [call[1]["text"] for call in context.bot.send_message.call_args_list] + assert any("sombrero" in t for t in sent_texts) + + +class TestScheduleWizards: + + @use_test_database_async + async def test_schedules_and_persists(self): + Pycampista.create(username="admin1", admin=True) + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + p.add_wizard("gandalf", "111") + update = make_update(text="/agendar_magx", username="admin1") + context = make_context() + await schedule_wizards(update, context, pycamp=p) + assert WizardAtPycamp.select().where(WizardAtPycamp.pycamp == p).count() > 0 + + @use_test_database_async + async def test_clears_previous_schedule(self): + """Agendar de nuevo borra la agenda anterior.""" + Pycampista.create(username="admin1", admin=True) + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + w = p.add_wizard("gandalf", "111") + # Primera agenda + persist_wizards_schedule_in_db(p) + count1 = WizardAtPycamp.select().where(WizardAtPycamp.pycamp == p).count() + # Segunda agenda via handler + update = make_update(text="/agendar_magx", username="admin1") + context = make_context() + await schedule_wizards(update, context, pycamp=p) + count2 = WizardAtPycamp.select().where(WizardAtPycamp.pycamp == p).count() + assert count2 == count1 # misma cantidad, no acumulada + + +class TestShowWizardsSchedule: + + @use_test_database_async + @freeze_time("2024-06-21 10:00:00") + async def test_shows_remaining_schedule(self): + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + w = p.add_wizard("gandalf", "111") + persist_wizards_schedule_in_db(p) + + update = make_update(text="/ver_agenda_magx", username="pepe") + context = make_context() + await show_wizards_schedule(update, context, pycamp=p) + text = context.bot.send_message.call_args[1]["text"] + assert "Agenda" in text or "magx" in text.lower() + + @use_test_database_async + @freeze_time("2024-06-21 10:00:00") + async def test_shows_complete_schedule_with_flag(self): + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + w = p.add_wizard("gandalf", "111") + persist_wizards_schedule_in_db(p) + + update = make_update(text="/ver_agenda_magx completa", username="pepe") + context = make_context() + await show_wizards_schedule(update, context, pycamp=p) + text = context.bot.send_message.call_args[1]["text"] + assert "Agenda" in text or "magx" in text.lower() + + @use_test_database_async + async def test_wrong_parameter_shows_error(self): + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + update = make_update(text="/ver_agenda_magx basura", username="pepe") + context = make_context() + context.args = ["basura"] + await show_wizards_schedule(update, context, pycamp=p) + text = context.bot.send_message.call_args[1]["text"] + assert "parámetro" in text.lower() or "completa" in text.lower() + + @use_test_database_async + async def test_empty_agenda_sends_hint_without_parse_mode(self): + """Sin turnos se envía hint en texto plano (sin MarkdownV2) para evitar BadRequest por '.'.""" + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + # Sin magxs ni /agendar_magx no hay WizardAtPycamp + update = make_update(text="/ver_agenda_magx", username="pepe") + context = make_context() + await show_wizards_schedule(update, context, pycamp=p) + kwargs = context.bot.send_message.call_args[1] + assert "No hay turnos cargados" in kwargs["text"] + assert kwargs.get("parse_mode") is None + + @use_test_database_async + @freeze_time("2024-06-21 10:00:00") + async def test_empty_futuros_sends_hint_without_parse_mode(self): + """Con 'futuros' y sin turnos futuros, hint en texto plano.""" + p = Pycamp.create( + headquarters="Narnia", active=True, + init=datetime(2020, 1, 1), end=datetime(2020, 1, 2), + ) + update = make_update(text="/ver_agenda_magx futuros", username="pepe") + context = make_context() + context.args = ["futuros"] + await show_wizards_schedule(update, context, pycamp=p) + kwargs = context.bot.send_message.call_args[1] + assert "No hay turnos futuros" in kwargs["text"] + assert kwargs.get("parse_mode") is None + + +class TestFormatWizardsSchedule: + + @use_test_database + def test_formats_agenda(self): + p = Pycamp.create( + headquarters="Narnia", + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + w = Pycampista.create(username="gandalf", wizard=True) + WizardAtPycamp.create( + pycamp=p, wizard=w, + init=datetime(2024, 6, 21, 9, 0), + end=datetime(2024, 6, 21, 10, 0), + ) + agenda = WizardAtPycamp.select().where(WizardAtPycamp.pycamp == p) + msg = format_wizards_schedule(agenda) + assert "Agenda de magxs" in msg + assert "@gandalf" in msg + assert "09:00" in msg + + @use_test_database + def test_empty_agenda(self): + p = Pycamp.create( + headquarters="Narnia", + init=datetime(2024, 6, 20), end=datetime(2024, 6, 23), + ) + agenda = WizardAtPycamp.select().where(WizardAtPycamp.pycamp == p) + msg = format_wizards_schedule(agenda) + assert "Agenda de magxs" in msg + + +class TestAuxResolveShowAll: + """aux_resolve_show_all recibe context (con context.args).""" + + def test_no_parameter_returns_true(self): + context = make_context() + context.args = [] + assert aux_resolve_show_all(context) is True # por defecto se muestra agenda completa + + def test_completa_returns_true(self): + context = make_context() + context.args = ["completa"] + assert aux_resolve_show_all(context) is True + + def test_futuros_returns_false(self): + context = make_context() + context.args = ["futuros"] + assert aux_resolve_show_all(context) is False + + def test_wrong_parameter_raises(self): + import pytest + context = make_context() + context.args = ["basura"] + with pytest.raises(ValueError): + aux_resolve_show_all(context) + + def test_too_many_parameters_raises(self): + import pytest + context = make_context() + context.args = ["completa", "extra"] + with pytest.raises(ValueError): + aux_resolve_show_all(context) diff --git a/test/test_manage_pycamp.py b/test/test_manage_pycamp.py new file mode 100644 index 0000000..fa494e2 --- /dev/null +++ b/test/test_manage_pycamp.py @@ -0,0 +1,109 @@ +import datetime as dt +from pycamp_bot.models import Pycamp +from pycamp_bot.commands.manage_pycamp import get_pycamp_by_name, get_active_pycamp +from test.conftest import use_test_database, test_db, MODELS + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestGetPycampByName: + + @use_test_database + def test_returns_pycamp_when_exists(self): + Pycamp.create(headquarters="Narnia") + result = get_pycamp_by_name("Narnia") + assert result is not None + assert result.headquarters == "Narnia" + + @use_test_database + def test_returns_none_when_not_exists(self): + result = get_pycamp_by_name("Inexistente") + assert result is None + + @use_test_database + def test_finds_correct_pycamp_among_many(self): + Pycamp.create(headquarters="Narnia") + Pycamp.create(headquarters="Mordor") + result = get_pycamp_by_name("Mordor") + assert result.headquarters == "Mordor" + + +class TestGetActivePycamp: + + @use_test_database + def test_returns_false_none_when_no_active(self): + Pycamp.create(headquarters="Narnia", active=False) + is_active, pycamp = get_active_pycamp() + assert is_active is False + assert pycamp is None + + @use_test_database + def test_returns_true_pycamp_when_active(self): + Pycamp.create(headquarters="Narnia", active=True) + is_active, pycamp = get_active_pycamp() + assert is_active is True + assert pycamp.headquarters == "Narnia" + + @use_test_database + def test_returns_false_none_when_no_pycamps(self): + is_active, pycamp = get_active_pycamp() + assert is_active is False + assert pycamp is None + + @use_test_database + def test_inactive_pycamp_not_returned(self): + Pycamp.create(headquarters="Narnia", active=False) + Pycamp.create(headquarters="Mordor", active=False) + is_active, pycamp = get_active_pycamp() + assert is_active is False + + +class TestPycampDurationCalculation: + + @use_test_database + def test_duration_one_day(self): + p = Pycamp.create( + headquarters="Test", + init=dt.datetime(2024, 6, 20), + active=True, + ) + duration = 1 + p.end = p.init + dt.timedelta( + days=duration - 1, + hours=23, + minutes=59, + seconds=59, + milliseconds=99, + ) + p.save() + # Con duración 1 día, init y end deben ser el mismo día + assert p.end.date() == p.init.date() + + @use_test_database + def test_duration_four_days(self): + p = Pycamp.create( + headquarters="Test", + init=dt.datetime(2024, 6, 20), + active=True, + ) + duration = 4 + p.end = p.init + dt.timedelta( + days=duration - 1, + hours=23, + minutes=59, + seconds=59, + milliseconds=99, + ) + p.save() + # 4 días desde el 20: 20, 21, 22, 23 -> end el 23 + assert p.end.day == 23 + assert p.end.hour == 23 + assert p.end.minute == 59 diff --git a/test/test_models_extended.py b/test/test_models_extended.py new file mode 100644 index 0000000..3cd5878 --- /dev/null +++ b/test/test_models_extended.py @@ -0,0 +1,235 @@ +from datetime import datetime, timedelta +import peewee +from pycamp_bot.models import ( + Pycamp, Pycampista, PycampistaAtPycamp, WizardAtPycamp, + Slot, Project, Vote, DEFAULT_SLOT_PERIOD, +) +from test.conftest import use_test_database, test_db, MODELS + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestPycampSetAsOnlyActive: + + @use_test_database + def test_activates_pycamp(self): + p = Pycamp.create(headquarters="Narnia") + assert not p.active + p.set_as_only_active() + p = Pycamp.get_by_id(p.id) + assert p.active + + @use_test_database + def test_deactivates_other_pycamps(self): + p1 = Pycamp.create(headquarters="Narnia", active=True) + p2 = Pycamp.create(headquarters="Mordor") + p2.set_as_only_active() + p1 = Pycamp.get_by_id(p1.id) + assert not p1.active + p2 = Pycamp.get_by_id(p2.id) + assert p2.active + + @use_test_database + def test_single_pycamp_active(self): + p = Pycamp.create(headquarters="Narnia") + p.set_as_only_active() + active_count = Pycamp.select().where(Pycamp.active).count() + assert active_count == 1 + + @use_test_database + def test_multiple_active_pycamps_resolved(self): + p1 = Pycamp.create(headquarters="Narnia", active=True) + p2 = Pycamp.create(headquarters="Mordor", active=True) + p3 = Pycamp.create(headquarters="Rivendel") + p3.set_as_only_active() + active_count = Pycamp.select().where(Pycamp.active).count() + assert active_count == 1 + p3 = Pycamp.get_by_id(p3.id) + assert p3.active + + +class TestPycampClearWizardsSchedule: + + @use_test_database + def test_clears_all_wizard_assignments(self): + p = Pycamp.create( + headquarters="Narnia", + init=datetime(2024, 6, 20), + end=datetime(2024, 6, 23), + ) + w = Pycampista.create(username="gandalf", wizard=True) + WizardAtPycamp.create( + pycamp=p, wizard=w, + init=datetime(2024, 6, 21, 9, 0), + end=datetime(2024, 6, 21, 10, 0), + ) + WizardAtPycamp.create( + pycamp=p, wizard=w, + init=datetime(2024, 6, 21, 10, 0), + end=datetime(2024, 6, 21, 11, 0), + ) + assert WizardAtPycamp.select().where(WizardAtPycamp.pycamp == p).count() == 2 + p.clear_wizards_schedule() + assert WizardAtPycamp.select().where(WizardAtPycamp.pycamp == p).count() == 0 + + @use_test_database + def test_returns_count_of_deleted(self): + p = Pycamp.create(headquarters="Narnia") + w = Pycampista.create(username="gandalf", wizard=True) + WizardAtPycamp.create( + pycamp=p, wizard=w, + init=datetime(2024, 6, 21, 9, 0), + end=datetime(2024, 6, 21, 10, 0), + ) + deleted = p.clear_wizards_schedule() + assert deleted == 1 + + @use_test_database + def test_no_wizards_returns_zero(self): + p = Pycamp.create(headquarters="Narnia") + deleted = p.clear_wizards_schedule() + assert deleted == 0 + + +class TestSlotGetEndTime: + + @use_test_database + def test_returns_start_plus_60_minutes(self): + pycamper = Pycampista.create(username="pepe") + slot = Slot.create(code="A1", start=datetime(2024, 6, 21, 10, 0), current_wizard=pycamper) + expected = datetime(2024, 6, 21, 11, 0) + assert slot.get_end_time() == expected + + @use_test_database + def test_end_time_crosses_hour_boundary(self): + pycamper = Pycampista.create(username="pepe") + slot = Slot.create(code="A1", start=datetime(2024, 6, 21, 10, 30), current_wizard=pycamper) + expected = datetime(2024, 6, 21, 11, 30) + assert slot.get_end_time() == expected + + @use_test_database + def test_default_slot_period_is_60(self): + assert DEFAULT_SLOT_PERIOD == 60 + + +class TestProjectModel: + + @use_test_database + def test_create_project(self): + owner = Pycampista.create(username="pepe") + project = Project.create(name="Mi Proyecto", owner=owner) + assert project.name == "Mi Proyecto" + assert project.owner.username == "pepe" + + @use_test_database + def test_unique_name_constraint(self): + owner = Pycampista.create(username="pepe") + Project.create(name="Mi Proyecto", owner=owner) + with self._raises_integrity_error(): + Project.create(name="Mi Proyecto", owner=owner) + + @use_test_database + def test_project_with_slot_assignment(self): + owner = Pycampista.create(username="pepe") + slot = Slot.create(code="A1", start=datetime(2024, 6, 21, 10, 0), current_wizard=owner) + project = Project.create(name="Mi Proyecto", owner=owner, slot=slot) + assert project.slot.code == "A1" + + @use_test_database + def test_project_optional_fields_null(self): + owner = Pycampista.create(username="pepe") + project = Project.create(name="Mi Proyecto", owner=owner) + assert project.topic is None + assert project.repository_url is None + assert project.group_url is None + assert project.slot_id is None + + @use_test_database + def test_project_with_all_fields(self): + owner = Pycampista.create(username="pepe") + project = Project.create( + name="Mi Proyecto", + owner=owner, + difficult_level=2, + topic="django", + repository_url="https://github.com/test", + group_url="https://t.me/test", + ) + assert project.difficult_level == 2 + assert project.topic == "django" + assert project.repository_url == "https://github.com/test" + + @staticmethod + def _raises_integrity_error(): + import contextlib + return contextlib.suppress(peewee.IntegrityError) if False else __import__('pytest').raises(peewee.IntegrityError) + + +class TestVoteModel: + + @use_test_database + def test_create_vote_with_interest(self): + owner = Pycampista.create(username="pepe") + voter = Pycampista.create(username="juan") + project = Project.create(name="Proyecto1", owner=owner) + vote = Vote.create( + project=project, + pycampista=voter, + interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + assert vote.interest is True + + @use_test_database + def test_unique_constraint_prevents_duplicate(self): + owner = Pycampista.create(username="pepe") + voter = Pycampista.create(username="juan") + project = Project.create(name="Proyecto1", owner=owner) + Vote.create( + project=project, + pycampista=voter, + interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + import pytest + with pytest.raises(peewee.IntegrityError): + Vote.create( + project=project, + pycampista=voter, + interest=False, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + + @use_test_database + def test_vote_interest_false(self): + owner = Pycampista.create(username="pepe") + voter = Pycampista.create(username="juan") + project = Project.create(name="Proyecto1", owner=owner) + vote = Vote.create( + project=project, + pycampista=voter, + interest=False, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + assert vote.interest is False + + @use_test_database + def test_vote_interest_null(self): + owner = Pycampista.create(username="pepe") + voter = Pycampista.create(username="juan") + project = Project.create(name="Proyecto1", owner=owner) + vote = Vote.create( + project=project, + pycampista=voter, + interest=None, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + assert vote.interest is None diff --git a/test/test_scheduler.py b/test/test_scheduler.py new file mode 100644 index 0000000..ea687d6 --- /dev/null +++ b/test/test_scheduler.py @@ -0,0 +1,261 @@ +from pycamp_bot.scheduler.schedule_calculator import ( + PyCampScheduleProblem, + hill_climbing, + IMPOSIBLE_COST, +) + + +def _make_problem_data(projects=None, slots=None): + """Helper para crear datos de problema de scheduling.""" + if slots is None: + slots = ["A1", "A2", "B1", "B2"] + if projects is None: + projects = { + "proyecto1": { + "responsables": ["pepe"], + "votes": ["juan", "maria"], + "difficult_level": 1, + "theme": "django", + "priority_slots": [], + }, + "proyecto2": { + "responsables": ["ana"], + "votes": ["juan", "carlos"], + "difficult_level": 2, + "theme": "flask", + "priority_slots": [], + }, + } + responsable_available_slots = {} + for proj_data in projects.values(): + for resp in proj_data["responsables"]: + responsable_available_slots[resp] = slots + + return { + "projects": projects, + "available_slots": slots, + "responsable_available_slots": responsable_available_slots, + } + + +class TestPyCampScheduleProblemInit: + + def test_creates_problem_from_data(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + assert problem.data is not None + + def test_responsables_added_to_votes(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + assert "pepe" in problem.data.projects.proyecto1.votes + assert "ana" in problem.data.projects.proyecto2.votes + + def test_project_list_extracted(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + assert "proyecto1" in problem.project_list + assert "proyecto2" in problem.project_list + assert len(problem.project_list) == 2 + + def test_total_participants_calculated(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + # Votantes únicos: juan, maria, carlos, pepe, ana (responsables se agregan a votos) + all_voters = set() + for proj in data["projects"].values(): + all_voters.update(proj["votes"]) + all_voters.update(proj["responsables"]) + assert problem.total_participants == len(all_voters) + + +class TestGenerateRandomState: + + def test_returns_all_projects(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + state = problem.generate_random_state() + project_names = [proj for proj, _ in state] + assert "proyecto1" in project_names + assert "proyecto2" in project_names + + def test_each_project_has_valid_slot(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + state = problem.generate_random_state() + for _, slot in state: + assert slot in data["available_slots"] + + def test_random_state_length(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + state = problem.generate_random_state() + assert len(state) == len(data["projects"]) + + +class TestNeighboors: + + def test_includes_single_reassignment(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + state = [("proyecto1", "A1"), ("proyecto2", "A2")] + neighbors = problem.neighboors(state) + # Debe incluir reasignaciones de proyecto1 a A2, B1, B2 + reassigned = [n for n in neighbors if dict(n)["proyecto1"] != "A1" and dict(n)["proyecto2"] == "A2"] + assert len(reassigned) == 3 # A2, B1, B2 + + def test_includes_swaps(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + state = [("proyecto1", "A1"), ("proyecto2", "A2")] + neighbors = problem.neighboors(state) + # Debe incluir swap: proyecto1->A2, proyecto2->A1 + swapped = [n for n in neighbors if dict(n)["proyecto1"] == "A2" and dict(n)["proyecto2"] == "A1"] + assert len(swapped) == 1 + + def test_neighboors_count(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + state = [("proyecto1", "A1"), ("proyecto2", "A2")] + neighbors = problem.neighboors(state) + # Reasignaciones: 2 proyectos * 3 slots alternativos = 6 + # Swaps: C(2,2) = 1 (solo si slots diferentes) + assert len(neighbors) == 7 + + +class TestValue: + + def test_no_collisions_returns_negative_value(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + # Proyectos en slots distintos: sin colisiones + state = [("proyecto1", "A1"), ("proyecto2", "B1")] + value = problem.value(state) + assert value < 0 # Siempre negativo por slot_population_cost y most_voted_cost + + def test_responsable_collision_impossible_cost(self): + projects = { + "proyecto1": { + "responsables": ["pepe"], + "votes": ["juan"], + "difficult_level": 1, + "theme": "django", + "priority_slots": [], + }, + "proyecto2": { + "responsables": ["pepe"], # Mismo responsable + "votes": ["maria"], + "difficult_level": 2, + "theme": "flask", + "priority_slots": [], + }, + } + data = _make_problem_data(projects=projects) + problem = PyCampScheduleProblem(data) + # Mismos responsables en el mismo slot + state_collision = [("proyecto1", "A1"), ("proyecto2", "A1")] + state_no_collision = [("proyecto1", "A1"), ("proyecto2", "B1")] + val_collision = problem.value(state_collision) + val_no_collision = problem.value(state_no_collision) + # La colisión de responsables debe hacer mucho peor el valor + assert val_collision < val_no_collision + assert (val_no_collision - val_collision) >= IMPOSIBLE_COST + + def test_voter_collision_increases_cost(self): + projects = { + "proyecto1": { + "responsables": ["pepe"], + "votes": ["juan", "maria"], + "difficult_level": 1, + "theme": "django", + "priority_slots": [], + }, + "proyecto2": { + "responsables": ["ana"], + "votes": ["juan", "maria"], # Mismos votantes + "difficult_level": 2, + "theme": "flask", + "priority_slots": [], + }, + } + data = _make_problem_data(projects=projects) + problem = PyCampScheduleProblem(data) + state_collision = [("proyecto1", "A1"), ("proyecto2", "A1")] + state_no_collision = [("proyecto1", "A1"), ("proyecto2", "B1")] + assert problem.value(state_collision) < problem.value(state_no_collision) + + def test_same_difficulty_penalized(self): + projects = { + "proyecto1": { + "responsables": ["pepe"], + "votes": [], + "difficult_level": 1, + "theme": "django", + "priority_slots": [], + }, + "proyecto2": { + "responsables": ["ana"], + "votes": [], + "difficult_level": 1, # Mismo nivel + "theme": "flask", + "priority_slots": [], + }, + } + data = _make_problem_data(projects=projects) + problem = PyCampScheduleProblem(data) + state_same_slot = [("proyecto1", "A1"), ("proyecto2", "A1")] + state_diff_slot = [("proyecto1", "A1"), ("proyecto2", "B1")] + assert problem.value(state_same_slot) <= problem.value(state_diff_slot) + + def test_same_theme_penalized(self): + projects = { + "proyecto1": { + "responsables": ["pepe"], + "votes": [], + "difficult_level": 1, + "theme": "django", + "priority_slots": [], + }, + "proyecto2": { + "responsables": ["ana"], + "votes": [], + "difficult_level": 2, + "theme": "django", # Mismo tema + "priority_slots": [], + }, + } + data = _make_problem_data(projects=projects) + problem = PyCampScheduleProblem(data) + state_same_slot = [("proyecto1", "A1"), ("proyecto2", "A1")] + state_diff_slot = [("proyecto1", "A1"), ("proyecto2", "B1")] + assert problem.value(state_same_slot) <= problem.value(state_diff_slot) + + +class TestHillClimbing: + + def test_returns_valid_state(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + initial = problem.generate_random_state() + result = hill_climbing(problem, initial) + project_names = [proj for proj, _ in result] + assert "proyecto1" in project_names + assert "proyecto2" in project_names + + def test_result_is_local_optimum(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + initial = problem.generate_random_state() + result = hill_climbing(problem, initial) + result_value = problem.value(result) + # Ningún vecino debe ser mejor + for neighbor in problem.neighboors(result): + assert problem.value(neighbor) <= result_value + + def test_improves_or_maintains_initial_value(self): + data = _make_problem_data() + problem = PyCampScheduleProblem(data) + initial = problem.generate_random_state() + initial_value = problem.value(initial) + result = hill_climbing(problem, initial) + assert problem.value(result) >= initial_value diff --git a/test/test_utils.py b/test/test_utils.py new file mode 100644 index 0000000..e1277d1 --- /dev/null +++ b/test/test_utils.py @@ -0,0 +1,76 @@ +from datetime import datetime +from pycamp_bot.utils import escape_markdown, get_slot_weekday_name +from pycamp_bot.models import Pycamp +from test.conftest import use_test_database, test_db, MODELS + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestEscapeMarkdown: + + def test_escapes_asterisks(self): + assert escape_markdown("*bold*") == "\\*bold\\*" + + def test_escapes_underscores(self): + assert escape_markdown("_italic_") == "\\_italic\\_" + + def test_escapes_all_special_chars(self): + for char in "_*[]()~`>#+-=|{}.!": + result = escape_markdown(char) + assert result == f"\\{char}", f"Failed for char: {char}" + + def test_no_change_on_plain_text(self): + assert escape_markdown("hello world") == "hello world" + + def test_empty_string(self): + assert escape_markdown("") == "" + + def test_mixed_text_and_special_chars(self): + result = escape_markdown("hello *world* (test)") + assert result == "hello \\*world\\* \\(test\\)" + + +class TestGetSlotWeekdayName: + + @use_test_database + def test_first_day_returns_correct_weekday(self): + # 2024-06-20 es jueves (weekday=3) + Pycamp.create( + headquarters="Test", + init=datetime(2024, 6, 20), + end=datetime(2024, 6, 23), + active=True, + ) + assert get_slot_weekday_name("A") == "Jueves" + + @use_test_database + def test_second_day_returns_next_weekday(self): + # 2024-06-20 es jueves, B = viernes + Pycamp.create( + headquarters="Test", + init=datetime(2024, 6, 20), + end=datetime(2024, 6, 23), + active=True, + ) + assert get_slot_weekday_name("B") == "Viernes" + + @use_test_database + def test_monday_start_offset(self): + # 2024-06-17 es lunes (weekday=0) + Pycamp.create( + headquarters="Test", + init=datetime(2024, 6, 17), + end=datetime(2024, 6, 21), + active=True, + ) + assert get_slot_weekday_name("A") == "Lunes" + assert get_slot_weekday_name("B") == "Martes" + assert get_slot_weekday_name("C") == "Miércoles" diff --git a/test/test_voting_logic.py b/test/test_voting_logic.py new file mode 100644 index 0000000..354a7e1 --- /dev/null +++ b/test/test_voting_logic.py @@ -0,0 +1,117 @@ +import peewee +import pytest +from pycamp_bot.models import Pycampista, Project, Vote +from test.conftest import use_test_database, test_db, MODELS + + +def setup_module(module): + test_db.bind(MODELS, bind_refs=False, bind_backrefs=False) + test_db.connect() + + +def teardown_module(module): + test_db.drop_tables(MODELS) + test_db.close() + + +class TestVoteCreation: + + @use_test_database + def test_create_vote_with_interest_true(self): + owner = Pycampista.create(username="owner1") + voter = Pycampista.create(username="voter1") + project = Project.create(name="Proyecto1", owner=owner) + vote = Vote.create( + project=project, + pycampista=voter, + interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + assert vote.interest is True + assert vote.project.name == "Proyecto1" + + @use_test_database + def test_create_vote_with_interest_false(self): + owner = Pycampista.create(username="owner1") + voter = Pycampista.create(username="voter1") + project = Project.create(name="Proyecto1", owner=owner) + vote = Vote.create( + project=project, + pycampista=voter, + interest=False, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + assert vote.interest is False + + @use_test_database + def test_duplicate_vote_raises_integrity_error(self): + owner = Pycampista.create(username="owner1") + voter = Pycampista.create(username="voter1") + project = Project.create(name="Proyecto1", owner=owner) + Vote.create( + project=project, + pycampista=voter, + interest=True, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + with pytest.raises(peewee.IntegrityError): + Vote.create( + project=project, + pycampista=voter, + interest=False, + _project_pycampista_id=f"{project.id}-{voter.id}", + ) + + @use_test_database + def test_project_pycampista_id_format(self): + owner = Pycampista.create(username="owner1") + voter = Pycampista.create(username="voter1") + project = Project.create(name="Proyecto1", owner=owner) + expected_id = f"{project.id}-{voter.id}" + vote = Vote.create( + project=project, + pycampista=voter, + interest=True, + _project_pycampista_id=expected_id, + ) + assert vote._project_pycampista_id == expected_id + + +class TestVoteCount: + + @use_test_database + def test_count_unique_voters(self): + owner = Pycampista.create(username="owner1") + voter1 = Pycampista.create(username="voter1") + voter2 = Pycampista.create(username="voter2") + project = Project.create(name="Proyecto1", owner=owner) + + Vote.create(project=project, pycampista=voter1, interest=True, + _project_pycampista_id=f"{project.id}-{voter1.id}") + Vote.create(project=project, pycampista=voter2, interest=True, + _project_pycampista_id=f"{project.id}-{voter2.id}") + + votes = [vote.pycampista_id for vote in Vote.select()] + assert len(set(votes)) == 2 + + @use_test_database + def test_count_zero_when_no_votes(self): + votes = [vote.pycampista_id for vote in Vote.select()] + assert len(set(votes)) == 0 + + @use_test_database + def test_count_deduplicates_same_user_multiple_projects(self): + owner = Pycampista.create(username="owner1") + voter = Pycampista.create(username="voter1") + p1 = Project.create(name="Proyecto1", owner=owner) + p2 = Project.create(name="Proyecto2", owner=owner) + + Vote.create(project=p1, pycampista=voter, interest=True, + _project_pycampista_id=f"{p1.id}-{voter.id}") + Vote.create(project=p2, pycampista=voter, interest=True, + _project_pycampista_id=f"{p2.id}-{voter.id}") + + votes = [vote.pycampista_id for vote in Vote.select()] + # Mismo usuario votó 2 veces, pero unique count es 1 + assert len(votes) == 2 + assert len(set(votes)) == 1