286 lines
8.4 KiB
Python
Executable File
286 lines
8.4 KiB
Python
Executable File
import asyncio
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, "/home/marco/MonicaAgentIA/app")
|
|
|
|
os.environ["TELEGRAM_BOT_TOKEN"] = "test"
|
|
os.environ["TELEGRAM_CHAT_ID"] = "123"
|
|
|
|
from monica_agent import (
|
|
useragenda_command,
|
|
load_user_profile,
|
|
save_user_profile,
|
|
clear_all_data,
|
|
is_valid_time,
|
|
format_user_agenda,
|
|
get_user_agenda,
|
|
get_current_activity,
|
|
message_handler,
|
|
)
|
|
|
|
class FakeMessage:
|
|
def __init__(self, text):
|
|
self.text = text
|
|
self.chat_id = 7889499475
|
|
|
|
async def reply_text(self, text, **kwargs):
|
|
print(f"BOT RESPONSE: {text[:500]}")
|
|
|
|
class FakeUpdate:
|
|
def __init__(self, text):
|
|
self.message = FakeMessage(text)
|
|
self.effective_chat = self
|
|
self.effective_user = self
|
|
self.effective_message = self.message
|
|
|
|
@property
|
|
def id(self):
|
|
return self.message.chat_id
|
|
|
|
@property
|
|
def username(self):
|
|
return "test_user"
|
|
|
|
@property
|
|
def first_name(self):
|
|
return "Marco"
|
|
|
|
class FakeContext:
|
|
def __init__(self):
|
|
self.bot = self
|
|
|
|
async def send_message(self, **kwargs):
|
|
print(f"BOT SEND: {kwargs.get('text', '')[:500]}")
|
|
|
|
async def test_useragenda_command():
|
|
print("=== Testando /useragenda ===")
|
|
update = FakeUpdate("/useragenda")
|
|
context = FakeContext()
|
|
try:
|
|
await useragenda_command(update, context)
|
|
print("✅ useragenda_command executou sem erros")
|
|
except Exception as e:
|
|
print(f"❌ Erro: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
async def test_useragenda_flow():
|
|
print("\n=== Testando fluxo /useragenda ===")
|
|
|
|
profile = load_user_profile()
|
|
profile["user_agenda"] = {}
|
|
save_user_profile(profile)
|
|
|
|
update = FakeUpdate("/useragenda")
|
|
context = FakeContext()
|
|
await useragenda_command(update, context)
|
|
|
|
update = FakeUpdate("1")
|
|
context = FakeContext()
|
|
handled = await message_handler(update, context)
|
|
print(f" Após selecionar atividade 1: handled={handled}")
|
|
|
|
update = FakeUpdate("07:00")
|
|
context = FakeContext()
|
|
handled = await message_handler(update, context)
|
|
print(f" Após enviar horário: handled={handled}")
|
|
|
|
update = FakeUpdate("2")
|
|
context = FakeContext()
|
|
handled = await message_handler(update, context)
|
|
print(f" Após selecionar atividade 2: handled={handled}")
|
|
|
|
update = FakeUpdate("08:00")
|
|
context = FakeContext()
|
|
handled = await message_handler(update, context)
|
|
print(f" Após enviar horário: handled={handled}")
|
|
|
|
update = FakeUpdate("fim")
|
|
context = FakeContext()
|
|
handled = await message_handler(update, context)
|
|
print(f" Após finalizar: handled={handled}")
|
|
|
|
profile = load_user_profile()
|
|
agenda = profile.get("user_agenda", {})
|
|
print(f" Agenda salva: {agenda}")
|
|
if agenda.get("acordar") == "07:00" and agenda.get("cafe_da_manha") == "08:00":
|
|
print("✅ Agenda salva corretamente")
|
|
else:
|
|
print("❌ Agenda não foi salva como esperado")
|
|
|
|
async def test_useragenda_all_activities():
|
|
print("\n=== Testando fluxo completo /useragenda ===")
|
|
|
|
profile = load_user_profile()
|
|
profile["user_agenda"] = {}
|
|
save_user_profile(profile)
|
|
|
|
update = FakeUpdate("/useragenda")
|
|
context = FakeContext()
|
|
await useragenda_command(update, context)
|
|
|
|
activities = [
|
|
("1", "06:30"),
|
|
("2", "07:00"),
|
|
("3", "08:00"),
|
|
("4", "09:00"),
|
|
("5", "12:00"),
|
|
("6", "15:00"),
|
|
("7", "19:00"),
|
|
("8", "20:00"),
|
|
("9", "21:00"),
|
|
("10", "22:00"),
|
|
("11", "23:00"),
|
|
("fim", ""),
|
|
]
|
|
for activity, time in activities:
|
|
update = FakeUpdate(activity)
|
|
context = FakeContext()
|
|
handled = await message_handler(update, context)
|
|
print(f" Atividade {activity}: handled={handled}")
|
|
if time:
|
|
update = FakeUpdate(time)
|
|
context = FakeContext()
|
|
handled = await message_handler(update, context)
|
|
print(f" Horário {time}: handled={handled}")
|
|
|
|
profile = load_user_profile()
|
|
agenda = profile.get("user_agenda", {})
|
|
print(f" Agenda salva: {agenda}")
|
|
if len(agenda) == 11:
|
|
print("✅ Todas as 11 atividades foram salvas")
|
|
else:
|
|
print(f"❌ Esperado 11 atividades, salvas {len(agenda)}")
|
|
|
|
async def test_useragenda_invalid_option():
|
|
print("\n=== Testando opção inválida ===")
|
|
profile = load_user_profile()
|
|
profile["user_agenda"] = {}
|
|
save_user_profile(profile)
|
|
|
|
update = FakeUpdate("/useragenda")
|
|
context = FakeContext()
|
|
await useragenda_command(update, context)
|
|
|
|
update = FakeUpdate("15")
|
|
context = FakeContext()
|
|
handled = await message_handler(update, context)
|
|
print(f" Após opção inválida: handled={handled}")
|
|
|
|
profile = load_user_profile()
|
|
agenda = profile.get("user_agenda", {})
|
|
if not agenda:
|
|
print("✅ Opção inválida foi rejeitada corretamente")
|
|
else:
|
|
print("❌ Opção inválida foi aceita (erro)")
|
|
|
|
async def test_useragenda_invalid_time():
|
|
print("\n=== Testando horário inválido ===")
|
|
profile = load_user_profile()
|
|
profile["user_agenda"] = {}
|
|
save_user_profile(profile)
|
|
|
|
update = FakeUpdate("/useragenda")
|
|
context = FakeContext()
|
|
await useragenda_command(update, context)
|
|
|
|
update = FakeUpdate("1")
|
|
context = FakeContext()
|
|
await message_handler(update, context)
|
|
|
|
update = FakeUpdate("25:00")
|
|
context = FakeContext()
|
|
handled = await message_handler(update, context)
|
|
print(f" Após horário inválido: handled={handled}")
|
|
|
|
profile = load_user_profile()
|
|
agenda = profile.get("user_agenda", {})
|
|
if not agenda:
|
|
print("✅ Horário inválido foi rejeitado corretamente")
|
|
else:
|
|
print("❌ Horário inválido foi aceito (erro)")
|
|
|
|
async def test_clear_all_data():
|
|
print("\n=== Testando /limpar ===")
|
|
profile = load_user_profile()
|
|
profile["name"] = "Marco"
|
|
profile["user_agenda"] = {
|
|
"acordar": "06:30",
|
|
"cafe_da_manha": "07:00",
|
|
"trabalho": "08:00",
|
|
}
|
|
save_user_profile(profile)
|
|
clear_all_data()
|
|
profile = load_user_profile()
|
|
if not profile.get("user_agenda"):
|
|
print("✅ /limpar removeu a agenda do usuário")
|
|
else:
|
|
print("❌ /limpar não removeu a agenda do usuário")
|
|
|
|
def test_format_user_agenda():
|
|
print("\n=== Testando format_user_agenda ===")
|
|
agenda = {
|
|
"acordar": "06:30",
|
|
"cafe_da_manha": "07:00",
|
|
"trabalho": "08:00",
|
|
"almoco": "12:00",
|
|
}
|
|
formatted = format_user_agenda(agenda)
|
|
print(f" Agenda formatada:\n{formatted}")
|
|
if "Acordar: 06:30" in formatted and "Trabalho: 08:00" in formatted:
|
|
print("✅ format_user_agenda funcionou corretamente")
|
|
else:
|
|
print("❌ format_user_agenda não funcionou como esperado")
|
|
|
|
def test_get_current_activity():
|
|
print("\n=== Testando get_current_activity ===")
|
|
from datetime import datetime
|
|
from unittest.mock import patch
|
|
|
|
agenda = {
|
|
"acordar": "06:00",
|
|
"cafe_da_manha": "07:00",
|
|
"trabalho": "08:00",
|
|
"almoco": "12:00",
|
|
}
|
|
|
|
test_cases = [
|
|
(agenda, "2026-08-24 06:30", "acordar"),
|
|
(agenda, "2026-08-24 07:30", "cafe_da_manha"),
|
|
(agenda, "2026-08-24 10:00", "trabalho"),
|
|
(agenda, "2026-08-24 12:30", "almoco"),
|
|
({}, "2026-08-24 12:30", None),
|
|
]
|
|
|
|
all_ok = True
|
|
for agenda, dt_str, expected in test_cases:
|
|
test_dt = datetime.strptime(dt_str, "%Y-%m-%d %H:%M")
|
|
with patch("monica_agent.datetime") as mock_dt:
|
|
mock_dt.now.return_value = test_dt
|
|
mock_dt.side_effect = datetime
|
|
result = get_current_activity(agenda)
|
|
status = "✅" if result == expected else "❌"
|
|
print(f" {status} {dt_str}: {result} (esperado {expected})")
|
|
if result != expected:
|
|
all_ok = False
|
|
return all_ok
|
|
|
|
async def main():
|
|
await test_useragenda_command()
|
|
await test_useragenda_flow()
|
|
await test_useragenda_all_activities()
|
|
await test_useragenda_invalid_option()
|
|
await test_useragenda_invalid_time()
|
|
await test_clear_all_data()
|
|
test_format_user_agenda()
|
|
time_ok = test_get_current_activity()
|
|
|
|
print("\n=== Todos os testes concluídos ===")
|
|
if time_ok:
|
|
print("✅ Todos os testes de useragenda passaram!")
|
|
else:
|
|
print("❌ Alguns testes falharam")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|