chore(test): remove obsolete test files
This commit is contained in:
marco
@@ -1,170 +0,0 @@
|
||||
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 (
|
||||
adicionarperfil_command,
|
||||
adicionaragente_command,
|
||||
message_handler,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
is_adding_user_preference,
|
||||
adding_user_preference_field,
|
||||
is_adding_agent_preference,
|
||||
adding_agent_preference_field,
|
||||
)
|
||||
|
||||
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[:300]}")
|
||||
|
||||
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', '')[:300]}")
|
||||
|
||||
async def test_adicionarperfil_command():
|
||||
print("=== Testando /adicionarperfil ===")
|
||||
update = FakeUpdate("/adicionarperfil")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await adicionarperfil_command(update, context)
|
||||
print("✅ adicionarperfil_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_adicionarperfil_flow():
|
||||
print("\n=== Testando adição de interesses ===")
|
||||
profile = load_user_profile()
|
||||
profile["interests"] = ["programação"]
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/adicionarperfil")
|
||||
context = FakeContext()
|
||||
await adicionarperfil_command(update, context)
|
||||
|
||||
update = FakeUpdate("1")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Selecionar interesses: handled={handled}")
|
||||
|
||||
update = FakeUpdate("música, viagens")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Adicionar itens: handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
interests = profile.get("interests", [])
|
||||
print(f" Interesses: {interests}")
|
||||
|
||||
if "música" in interests and "viagens" in interests and "programação" in interests:
|
||||
print("✅ Interesses adicionados sem duplicatas")
|
||||
else:
|
||||
print("❌ Falha ao adicionar interesses")
|
||||
|
||||
async def test_adicionaragente_command():
|
||||
print("\n=== Testando /adicionaragente ===")
|
||||
update = FakeUpdate("/adicionaragente")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await adicionaragente_command(update, context)
|
||||
print("✅ adicionaragente_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_adicionaragente_flow():
|
||||
print("\n=== Testando adição de preferências visuais ===")
|
||||
profile = load_user_profile()
|
||||
profile["monica_visual_preferences"] = ["item original"]
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/adicionaragente")
|
||||
context = FakeContext()
|
||||
await adicionaragente_command(update, context)
|
||||
|
||||
update = FakeUpdate("1")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Selecionar visuais: handled={handled}")
|
||||
|
||||
update = FakeUpdate("novo visual, item original")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Adicionar itens: handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
visual = profile.get("monica_visual_preferences", [])
|
||||
print(f" Preferências visuais: {visual}")
|
||||
|
||||
if "novo visual" in visual and "item original" in visual and len(visual) == 2:
|
||||
print("✅ Preferências visuais adicionadas sem duplicatas")
|
||||
else:
|
||||
print("❌ Falha ao adicionar preferências visuais")
|
||||
|
||||
async def test_help_contains_new_commands():
|
||||
print("\n=== Verificando /help contém /adicionarperfil e /adicionaragente ===")
|
||||
from monica_agent import help_command
|
||||
|
||||
update = FakeUpdate("/help")
|
||||
context = FakeContext()
|
||||
responses = []
|
||||
|
||||
original_reply = FakeMessage.reply_text
|
||||
async def capture_reply(self, text, **kwargs):
|
||||
responses.append(text)
|
||||
print(f"BOT RESPONSE: {text[:300]}")
|
||||
|
||||
FakeMessage.reply_text = capture_reply
|
||||
try:
|
||||
await help_command(update, context)
|
||||
help_text = responses[0] if responses else ""
|
||||
has_adicionarperfil = "/adicionarperfil" in help_text
|
||||
has_adicionaragente = "/adicionaragente" in help_text
|
||||
print(f" /adicionarperfil: {'✅' if has_adicionarperfil else '❌'}")
|
||||
print(f" /adicionaragente: {'✅' if has_adicionaragente else '❌'}")
|
||||
finally:
|
||||
FakeMessage.reply_text = original_reply
|
||||
|
||||
async def main():
|
||||
await test_adicionarperfil_command()
|
||||
await test_adicionarperfil_flow()
|
||||
await test_adicionaragente_command()
|
||||
await test_adicionaragente_flow()
|
||||
await test_help_contains_new_commands()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,106 +0,0 @@
|
||||
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 (
|
||||
adicionaracoesauto_command,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
load_all_contexts,
|
||||
add_to_context,
|
||||
)
|
||||
|
||||
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]}")
|
||||
|
||||
async def edit_text(self, text, **kwargs):
|
||||
print(f"BOT EDIT: {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_adicionaracoesauto_command():
|
||||
print("=== Testando /adicionaraçõesauto ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["name"] = "Marco"
|
||||
profile["interests"] = ["Programar", "Violão"]
|
||||
profile["personality_traits"] = ["extrovertido"]
|
||||
profile["preferences"] = {
|
||||
"communication_style": "direta",
|
||||
"topics_of_interest": ["tecnologia"],
|
||||
"topics_to_avoid": ["política"],
|
||||
}
|
||||
profile["sexual_preferences"] = {
|
||||
"openness": "Sim, confortável",
|
||||
"preferred_topics": ["sedução"],
|
||||
"boundaries": ["Não"],
|
||||
}
|
||||
profile["monica_sexual_preferences"] = [
|
||||
"Gosta de dar palmadas",
|
||||
"Gosta de enfiar dedos",
|
||||
]
|
||||
profile["monica_visual_preferences"] = [
|
||||
"Gosta de ver pelado",
|
||||
]
|
||||
save_user_profile(profile)
|
||||
|
||||
add_to_context("user", "Hoje estou trabalhando em casa, programando um projeto novo em Arduino.")
|
||||
add_to_context("assistant", "Que legal, amor! Adoro ver você programando.")
|
||||
|
||||
update = FakeUpdate("/adicionaraçõesauto")
|
||||
context = FakeContext()
|
||||
|
||||
try:
|
||||
await adicionaracoesauto_command(update, context)
|
||||
print("✅ adicionaracoesauto_command executou sem erros")
|
||||
|
||||
final_profile = load_user_profile()
|
||||
if final_profile.get("interests") != profile.get("interests"):
|
||||
print("✅ Perfil foi atualizado (interesses modificados)")
|
||||
else:
|
||||
print(" Perfil pode não ter sido modificado (nada novo encontrado - também é válido)")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def main():
|
||||
await test_adicionaracoesauto_command()
|
||||
print("\n=== Teste concluído ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,287 +0,0 @@
|
||||
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 (
|
||||
agendamento_command,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
clear_all_data,
|
||||
is_valid_time,
|
||||
time_to_minutes,
|
||||
is_within_schedule,
|
||||
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[:400]}")
|
||||
|
||||
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', '')[:400]}")
|
||||
|
||||
async def test_agendamento_command():
|
||||
print("=== Testando /agendamento ===")
|
||||
update = FakeUpdate("/agendamento")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await agendamento_command(update, context)
|
||||
print("✅ agendamento_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_agendamento_flow():
|
||||
print("\n=== Testando fluxo /agendamento ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["proactive_schedule"] = {}
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/agendamento")
|
||||
context = FakeContext()
|
||||
await agendamento_command(update, context)
|
||||
|
||||
update = FakeUpdate("1,2,3,4,5")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após selecionar dias: handled={handled}")
|
||||
|
||||
update = FakeUpdate("09:00")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após enviar start time: handled={handled}")
|
||||
|
||||
update = FakeUpdate("22:00")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após enviar end time: handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
schedule = profile.get("proactive_schedule", {})
|
||||
print(f" Schedule salvo: {schedule}")
|
||||
if schedule.get("enabled") and schedule.get("days") == [0, 1, 2, 3, 4] and schedule.get("start_time") and schedule.get("end_time"):
|
||||
print("✅ Agendamento salvo corretamente")
|
||||
else:
|
||||
print("❌ Agendamento não foi salvo como esperado")
|
||||
|
||||
async def test_agendamento_flow_weekend():
|
||||
print("\n=== Testando fluxo /agendamento (fim de semana) ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["proactive_schedule"] = {}
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/agendamento")
|
||||
context = FakeContext()
|
||||
await agendamento_command(update, context)
|
||||
|
||||
update = FakeUpdate("6,7")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após selecionar dias: handled={handled}")
|
||||
|
||||
update = FakeUpdate("14:00")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após enviar start time: handled={handled}")
|
||||
|
||||
update = FakeUpdate("23:59")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após enviar end time: handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
schedule = profile.get("proactive_schedule", {})
|
||||
print(f" Schedule salvo: {schedule}")
|
||||
if schedule.get("enabled") and schedule.get("days") == [5, 6] and schedule.get("start_time") and schedule.get("end_time"):
|
||||
print("✅ Agendamento de fim de semana salvo corretamente")
|
||||
else:
|
||||
print("❌ Agendamento de fim de semana não foi salvo como esperado")
|
||||
|
||||
async def test_agendamento_invalid_days():
|
||||
print("\n=== Testando dias inválidos ===")
|
||||
profile = load_user_profile()
|
||||
profile["proactive_schedule"] = {}
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/agendamento")
|
||||
context = FakeContext()
|
||||
await agendamento_command(update, context)
|
||||
|
||||
update = FakeUpdate("9,10")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após enviar dias inválidos: handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
schedule = profile.get("proactive_schedule", {})
|
||||
if not schedule.get("enabled", False):
|
||||
print("✅ Dias inválidos foram rejeitados corretamente")
|
||||
else:
|
||||
print("❌ Dias inválidos foram aceitos (erro)")
|
||||
|
||||
async def test_agendamento_invalid_time():
|
||||
print("\n=== Testando horário inválido ===")
|
||||
profile = load_user_profile()
|
||||
profile["proactive_schedule"] = {}
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/agendamento")
|
||||
context = FakeContext()
|
||||
await agendamento_command(update, context)
|
||||
|
||||
update = FakeUpdate("1,2,3")
|
||||
context = FakeContext()
|
||||
await message_handler(update, context)
|
||||
|
||||
update = FakeUpdate("25:00")
|
||||
context = FakeContext()
|
||||
await message_handler(update, context)
|
||||
print(f" Após enviar horário inválido")
|
||||
|
||||
profile = load_user_profile()
|
||||
schedule = profile.get("proactive_schedule", {})
|
||||
if not schedule.get("start_time"):
|
||||
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["proactive_schedule"] = {
|
||||
"enabled": True,
|
||||
"days": [0, 1, 2],
|
||||
"start_time": "08:00",
|
||||
"end_time": "20:00"
|
||||
}
|
||||
save_user_profile(profile)
|
||||
clear_all_data()
|
||||
profile = load_user_profile()
|
||||
if not profile.get("proactive_schedule"):
|
||||
print("✅ /limpar removeu o agendamento")
|
||||
else:
|
||||
print("❌ /limpar não removeu o agendamento")
|
||||
|
||||
def test_is_valid_time():
|
||||
print("\n=== Testando is_valid_time ===")
|
||||
cases = [
|
||||
("09:00", True),
|
||||
("23:59", True),
|
||||
("00:00", True),
|
||||
("24:00", False),
|
||||
("12:60", False),
|
||||
("abc", False),
|
||||
("", False),
|
||||
(None, False),
|
||||
]
|
||||
all_ok = True
|
||||
for value, expected in cases:
|
||||
result = is_valid_time(value)
|
||||
status = "✅" if result == expected else "❌"
|
||||
print(f" {status} is_valid_time({value!r}) = {result} (esperado {expected})")
|
||||
if result != expected:
|
||||
all_ok = False
|
||||
return all_ok
|
||||
|
||||
def test_is_within_schedule():
|
||||
print("\n=== Testando is_within_schedule ===")
|
||||
from datetime import datetime
|
||||
from unittest.mock import patch
|
||||
|
||||
base_profile = {
|
||||
"proactive_schedule": {
|
||||
"enabled": True,
|
||||
"days": [0, 1, 2, 3, 4],
|
||||
"start_time": "09:00",
|
||||
"end_time": "17:00"
|
||||
}
|
||||
}
|
||||
|
||||
test_cases = [
|
||||
(base_profile, "monday 10:00", True),
|
||||
(base_profile, "monday 08:00", False),
|
||||
(base_profile, "monday 17:00", True),
|
||||
(base_profile, "saturday 10:00", False),
|
||||
(base_profile, "sunday 10:00", False),
|
||||
({"proactive_schedule": {}}, "monday 10:00", True),
|
||||
({"proactive_schedule": {"enabled": False}}, "monday 10:00", True),
|
||||
]
|
||||
|
||||
all_ok = True
|
||||
for profile, time_str, expected in test_cases:
|
||||
dt_str, time_part = time_str.split(" ", 1)
|
||||
hour, minute = map(int, time_part.split(":"))
|
||||
dt_map = {
|
||||
"monday": "2026-08-24",
|
||||
"tuesday": "2026-08-25",
|
||||
"wednesday": "2026-08-26",
|
||||
"thursday": "2026-08-27",
|
||||
"friday": "2026-08-28",
|
||||
"saturday": "2026-08-29",
|
||||
"sunday": "2026-08-23",
|
||||
}
|
||||
date_str = dt_map.get(dt_str, "2026-08-25")
|
||||
test_dt = datetime.strptime(f"{date_str} {time_part}", "%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 = is_within_schedule(profile)
|
||||
status = "✅" if result == expected else "❌"
|
||||
print(f" {status} {time_str}: {result} (esperado {expected})")
|
||||
if result != expected:
|
||||
all_ok = False
|
||||
return all_ok
|
||||
|
||||
async def main():
|
||||
await test_agendamento_command()
|
||||
await test_agendamento_flow()
|
||||
await test_agendamento_flow_weekend()
|
||||
await test_agendamento_invalid_days()
|
||||
await test_agendamento_invalid_time()
|
||||
await test_clear_all_data()
|
||||
time_ok = test_is_valid_time()
|
||||
schedule_ok = test_is_within_schedule()
|
||||
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
if time_ok and schedule_ok:
|
||||
print("✅ Todos os testes de agendamento passaram!")
|
||||
else:
|
||||
print("❌ Alguns testes falharam")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,55 +0,0 @@
|
||||
import asyncio
|
||||
import sys
|
||||
sys.path.insert(0, "/home/marco/MonicaAgentIA/app")
|
||||
|
||||
from telegram import Update
|
||||
from telegram.ext import ContextTypes
|
||||
from monica_agent import 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[:200]}")
|
||||
|
||||
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', '')[:200]}")
|
||||
|
||||
async def test():
|
||||
update = FakeUpdate("Olá Mônica!")
|
||||
context = FakeContext()
|
||||
print("=== Testando message_handler ===")
|
||||
try:
|
||||
await message_handler(update, context)
|
||||
print("✅ message_handler executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
asyncio.run(test())
|
||||
@@ -1,136 +0,0 @@
|
||||
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 (
|
||||
cancelar_command,
|
||||
message_handler,
|
||||
is_onboarding,
|
||||
is_config_agent_profile,
|
||||
is_shutting_down,
|
||||
proactive_task,
|
||||
app_instance,
|
||||
)
|
||||
|
||||
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[:300]}")
|
||||
|
||||
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', '')[:300]}")
|
||||
|
||||
async def test_cancelar_command():
|
||||
print("=== Testando /cancelar ===")
|
||||
|
||||
# Simulate being in onboarding state
|
||||
import monica_agent
|
||||
monica_agent.is_onboarding = True
|
||||
monica_agent.onboarding_step = 3
|
||||
monica_agent.onboarding_answers = {"name": "Test"}
|
||||
monica_agent.is_config_agent_profile = True
|
||||
monica_agent.config_agent_step = 2
|
||||
monica_agent.config_agent_field = "monica_visual_preferences"
|
||||
|
||||
update = FakeUpdate("/cancelar")
|
||||
context = FakeContext()
|
||||
|
||||
try:
|
||||
await cancelar_command(update, context)
|
||||
print("✅ cancelar_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return
|
||||
|
||||
# Verify state was reset
|
||||
print(f" is_onboarding={monica_agent.is_onboarding}")
|
||||
print(f" onboarding_step={monica_agent.onboarding_step}")
|
||||
print(f" onboarding_answers={monica_agent.onboarding_answers}")
|
||||
print(f" is_config_agent_profile={monica_agent.is_config_agent_profile}")
|
||||
print(f" config_agent_step={monica_agent.config_agent_step}")
|
||||
print(f" config_agent_field={monica_agent.config_agent_field}")
|
||||
print(f" is_shutting_down={monica_agent.is_shutting_down}")
|
||||
|
||||
assert monica_agent.is_onboarding == False
|
||||
assert monica_agent.onboarding_step == 0
|
||||
assert monica_agent.onboarding_answers == {}
|
||||
assert monica_agent.is_config_agent_profile == False
|
||||
assert monica_agent.config_agent_step == 0
|
||||
assert monica_agent.config_agent_field == ""
|
||||
assert monica_agent.is_shutting_down == True
|
||||
|
||||
print("✅ Estado resetado corretamente")
|
||||
|
||||
# Verify that after cancelar, normal messages are handled
|
||||
print("\n=== Verificando que mensagens normais funcionam após /cancelar ===")
|
||||
update = FakeUpdate("Olá Mônica!")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await message_handler(update, context)
|
||||
print("✅ message_handler funciona após /cancelar")
|
||||
except Exception as e:
|
||||
print(f"❌ message_handler falhou após /cancelar: {e}")
|
||||
|
||||
async def test_help_contains_cancelar():
|
||||
print("\n=== Verificando /help contém /cancelar ===")
|
||||
from monica_agent import help_command
|
||||
|
||||
update = FakeUpdate("/help")
|
||||
context = FakeContext()
|
||||
responses = []
|
||||
|
||||
original_reply = FakeMessage.reply_text
|
||||
async def capture_reply(self, text, **kwargs):
|
||||
responses.append(text)
|
||||
print(f"BOT RESPONSE: {text[:300]}")
|
||||
|
||||
FakeMessage.reply_text = capture_reply
|
||||
try:
|
||||
await help_command(update, context)
|
||||
help_text = responses[0] if responses else ""
|
||||
if "/cancelar" in help_text:
|
||||
print("✅ /help menciona /cancelar")
|
||||
else:
|
||||
print("❌ /help não menciona /cancelar")
|
||||
finally:
|
||||
FakeMessage.reply_text = original_reply
|
||||
|
||||
async def main():
|
||||
await test_cancelar_command()
|
||||
await test_help_contains_cancelar()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,138 +0,0 @@
|
||||
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 (
|
||||
help_command,
|
||||
configagentprofile_command,
|
||||
message_handler,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
)
|
||||
|
||||
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[:300]}")
|
||||
|
||||
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', '')[:300]}")
|
||||
|
||||
async def test_help_command():
|
||||
print("=== Testando /help ===")
|
||||
update = FakeUpdate("/help")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await help_command(update, context)
|
||||
print("✅ help_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_configagentprofile_command():
|
||||
print("\n=== Testando /configagentprofile ===")
|
||||
update = FakeUpdate("/configagentprofile")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await configagentprofile_command(update, context)
|
||||
print("✅ configagentprofile_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_configagentprofile_flow():
|
||||
print("\n=== Testando fluxo /configagentprofile ===")
|
||||
|
||||
# Start config
|
||||
update = FakeUpdate("/configagentprofile")
|
||||
context = FakeContext()
|
||||
await configagentprofile_command(update, context)
|
||||
|
||||
# Select option 1
|
||||
update = FakeUpdate("1")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após selecionar opção 1: handled={handled}")
|
||||
|
||||
# Provide values
|
||||
update = FakeUpdate("ver o pau de Marco, ver a bunda de Marco")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após enviar valores: handled={handled}")
|
||||
|
||||
# Verify profile was updated
|
||||
profile = load_user_profile()
|
||||
visual_prefs = profile.get("monica_visual_preferences", [])
|
||||
print(f" Preferências visuais salvas: {visual_prefs}")
|
||||
|
||||
expected = ["ver o pau de Marco", "ver a bunda de Marco"]
|
||||
if all(item in visual_prefs for item in expected):
|
||||
print("✅ Perfil da Mônica atualizado corretamente")
|
||||
else:
|
||||
print("❌ Perfil da Mônica não foi atualizado como esperado")
|
||||
|
||||
async def test_help_contains_new_command():
|
||||
print("\n=== Verificando /help contém /configagentprofile ===")
|
||||
update = FakeUpdate("/help")
|
||||
context = FakeContext()
|
||||
responses = []
|
||||
|
||||
original_reply = FakeMessage.reply_text
|
||||
async def capture_reply(self, text, **kwargs):
|
||||
responses.append(text)
|
||||
print(f"BOT RESPONSE: {text[:300]}")
|
||||
|
||||
FakeMessage.reply_text = capture_reply
|
||||
try:
|
||||
await help_command(update, context)
|
||||
help_text = responses[0] if responses else ""
|
||||
if "/configagentprofile" in help_text:
|
||||
print("✅ /help menciona /configagentprofile")
|
||||
else:
|
||||
print("❌ /help não menciona /configagentprofile")
|
||||
print(f" Help text: {help_text[:200]}")
|
||||
finally:
|
||||
FakeMessage.reply_text = original_reply
|
||||
|
||||
async def main():
|
||||
await test_help_command()
|
||||
await test_configagentprofile_command()
|
||||
await test_configagentprofile_flow()
|
||||
await test_help_contains_new_command()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
-193
@@ -1,193 +0,0 @@
|
||||
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 (
|
||||
editarperfil_command,
|
||||
editaragente_command,
|
||||
message_handler,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
is_editing_user_profile,
|
||||
editing_user_profile_field,
|
||||
is_editing_agent_profile,
|
||||
editing_agent_profile_field,
|
||||
editing_agent_profile_action,
|
||||
)
|
||||
|
||||
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[:300]}")
|
||||
|
||||
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', '')[:300]}")
|
||||
|
||||
async def test_editarperfil_command():
|
||||
print("=== Testando /editarperfil ===")
|
||||
update = FakeUpdate("/editarperfil")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await editarperfil_command(update, context)
|
||||
print("✅ editarperfil_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_editarperfil_flow():
|
||||
print("\n=== Testando fluxo /editarperfil ===")
|
||||
update = FakeUpdate("/editarperfil")
|
||||
context = FakeContext()
|
||||
await editarperfil_command(update, context)
|
||||
|
||||
update = FakeUpdate("1")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após selecionar opção 1: handled={handled}")
|
||||
|
||||
update = FakeUpdate("João")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após enviar nome: handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
print(f" Nome salvo: {profile.get('name')}")
|
||||
if profile.get("name") == "João":
|
||||
print("✅ Perfil do usuário atualizado corretamente")
|
||||
else:
|
||||
print("❌ Perfil do usuário não foi atualizado")
|
||||
|
||||
async def test_editaragente_command():
|
||||
print("\n=== Testando /editaragente ===")
|
||||
update = FakeUpdate("/editaragente")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await editaragente_command(update, context)
|
||||
print("✅ editaragente_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_editaragente_remove():
|
||||
print("\n=== Testando remoção no /editaragente ===")
|
||||
profile = load_user_profile()
|
||||
profile["monica_visual_preferences"] = ["item1", "item2", "item3"]
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/editaragente")
|
||||
context = FakeContext()
|
||||
await editaragente_command(update, context)
|
||||
|
||||
update = FakeUpdate("5")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após selecionar remover (opção 5): handled={handled}")
|
||||
|
||||
update = FakeUpdate("2")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após selecionar item 2: handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
visual = profile.get("monica_visual_preferences", [])
|
||||
print(f" Preferências visuais após remoção: {visual}")
|
||||
if "item2" not in visual and len(visual) == 2:
|
||||
print("✅ Remoção funcionou corretamente")
|
||||
else:
|
||||
print("❌ Remoção não funcionou como esperado")
|
||||
|
||||
async def test_editaragente_add():
|
||||
print("\n=== Testando adição no /editaragente ===")
|
||||
profile = load_user_profile()
|
||||
profile["monica_visual_preferences"] = ["item1"]
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/editaragente")
|
||||
context = FakeContext()
|
||||
await editaragente_command(update, context)
|
||||
|
||||
update = FakeUpdate("1")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após selecionar adicionar (opção 1): handled={handled}")
|
||||
|
||||
update = FakeUpdate("novo item, item1")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Após enviar itens: handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
visual = profile.get("monica_visual_preferences", [])
|
||||
print(f" Preferências visuais após adição: {visual}")
|
||||
if "novo item" in visual:
|
||||
print("✅ Adição funcionou corretamente")
|
||||
else:
|
||||
print("❌ Adição não funcionou como esperado")
|
||||
|
||||
async def test_help_contains_new_commands():
|
||||
print("\n=== Verificando /help contém /editarperfil e /editaragente ===")
|
||||
from monica_agent import help_command
|
||||
|
||||
update = FakeUpdate("/help")
|
||||
context = FakeContext()
|
||||
responses = []
|
||||
|
||||
original_reply = FakeMessage.reply_text
|
||||
async def capture_reply(self, text, **kwargs):
|
||||
responses.append(text)
|
||||
print(f"BOT RESPONSE: {text[:300]}")
|
||||
|
||||
FakeMessage.reply_text = capture_reply
|
||||
try:
|
||||
await help_command(update, context)
|
||||
help_text = responses[0] if responses else ""
|
||||
has_editarperfil = "/editarperfil" in help_text
|
||||
has_editaragente = "/editaragente" in help_text
|
||||
print(f" /editarperfil: {'✅' if has_editarperfil else '❌'}")
|
||||
print(f" /editaragente: {'✅' if has_editaragente else '❌'}")
|
||||
finally:
|
||||
FakeMessage.reply_text = original_reply
|
||||
|
||||
async def main():
|
||||
await test_editarperfil_command()
|
||||
await test_editarperfil_flow()
|
||||
await test_editaragente_command()
|
||||
await test_editaragente_remove()
|
||||
await test_editaragente_add()
|
||||
await test_help_contains_new_commands()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,269 +0,0 @@
|
||||
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 (
|
||||
verperfil_command,
|
||||
veragente_command,
|
||||
editarperfil_command,
|
||||
editaragente_command,
|
||||
message_handler,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
get_user_profile_summary,
|
||||
get_agent_profile_summary,
|
||||
)
|
||||
|
||||
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_verperfil():
|
||||
print("=== Testando /verperfil ===")
|
||||
update = FakeUpdate("/verperfil")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await verperfil_command(update, context)
|
||||
print("✅ verperfil_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_veragente():
|
||||
print("\n=== Testando /veragente ===")
|
||||
update = FakeUpdate("/veragente")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await veragente_command(update, context)
|
||||
print("✅ veragente_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_editarperfil_all_fields():
|
||||
print("\n=== Testando edição de TODOS os campos do usuário ===")
|
||||
|
||||
test_cases = [
|
||||
("1", "João Silva", "name"),
|
||||
("2", "30", "age"),
|
||||
("3", "Masculino", "gender"),
|
||||
("4", "São Paulo", "location"),
|
||||
("5", "programação, música, viagens", "interests"),
|
||||
("6", "extrovertido, curioso", "personality_traits"),
|
||||
("7", "Carinhosa e direta", "preferences.communication_style"),
|
||||
("8", "política, futebol", "preferences.topics_to_avoid"),
|
||||
("9", "tecnologia, IA, cinema", "preferences.topics_of_interest"),
|
||||
("10", "Casado, feliz", "relationship_context"),
|
||||
("11", "Totalmente aberto", "sexual_preferences.openness"),
|
||||
("12", "fantasias, sedução", "sexual_preferences.preferred_topics"),
|
||||
("13", "respeito, consentimento", "sexual_preferences.boundaries"),
|
||||
("14", "7h", "daily_routine.wake_up_time"),
|
||||
("15", "23h", "daily_routine.sleep_time"),
|
||||
("16", "home-office", "daily_routine.work_schedule"),
|
||||
("17", "Nota de teste", "notes"),
|
||||
]
|
||||
|
||||
for option, value, expected_field in test_cases:
|
||||
update = FakeUpdate("/editarperfil")
|
||||
context = FakeContext()
|
||||
await editarperfil_command(update, context)
|
||||
|
||||
update = FakeUpdate(option)
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Opção {option}: handled={handled}")
|
||||
|
||||
update = FakeUpdate(value)
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Valor '{value}': handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
field_path = expected_field.split(".")
|
||||
current = profile
|
||||
for key in field_path:
|
||||
if key in current:
|
||||
current = current[key]
|
||||
else:
|
||||
current = None
|
||||
break
|
||||
|
||||
if current is not None:
|
||||
print(f" ✅ Campo {expected_field} = {current}")
|
||||
else:
|
||||
print(f" ❌ Campo {expected_field} não encontrado")
|
||||
|
||||
print("\n✅ Todos os campos do usuário editados com sucesso")
|
||||
|
||||
async def test_editaragente_all_categories():
|
||||
print("\n=== Testando edição de TODAS as categorias do agente ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["monica_visual_preferences"] = ["item original visual"]
|
||||
profile["monica_sexual_preferences"] = ["item original sexual"]
|
||||
profile["monica_nudity_preferences"] = ["item original nudez"]
|
||||
profile["relationship_dynamics"] = ["item original dynamics"]
|
||||
profile["nudity_preferences"] = {"notas_adicionais": "texto original"}
|
||||
save_user_profile(profile)
|
||||
|
||||
test_cases = [
|
||||
("1", "novo visual", "add", "monica_visual_preferences"),
|
||||
("2", "novo sexual", "add", "monica_sexual_preferences"),
|
||||
("3", "novo nudez", "add", "monica_nudity_preferences"),
|
||||
("4", "nova dynamics", "add", "relationship_dynamics"),
|
||||
("9", "novo texto nudidade", "edit", "nudity_preferences"),
|
||||
]
|
||||
|
||||
for option, value, action, expected_field in test_cases:
|
||||
update = FakeUpdate("/editaragente")
|
||||
context = FakeContext()
|
||||
await editaragente_command(update, context)
|
||||
|
||||
update = FakeUpdate(option)
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Opção {option}: handled={handled}")
|
||||
|
||||
update = FakeUpdate(value)
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Valor '{value}': handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
if expected_field == "nudity_preferences":
|
||||
current = profile.get(expected_field, {}).get("notas_adicionais", "")
|
||||
else:
|
||||
current = profile.get(expected_field, [])
|
||||
|
||||
if action == "add" and expected_field != "nudity_preferences":
|
||||
if value in current:
|
||||
print(f" ✅ {expected_field} contém '{value}'")
|
||||
else:
|
||||
print(f" ❌ {expected_field} não contém '{value}'")
|
||||
elif action == "edit":
|
||||
if current == value:
|
||||
print(f" ✅ {expected_field} = '{current}'")
|
||||
else:
|
||||
print(f" ❌ {expected_field} = '{current}' (esperado '{value}')")
|
||||
|
||||
print("\n✅ Todas as categorias do agente editadas com sucesso")
|
||||
|
||||
async def test_remove_agent_items():
|
||||
print("\n=== Testando remoção de itens do agente ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["monica_visual_preferences"] = ["item1", "item2", "item3"]
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/editaragente")
|
||||
context = FakeContext()
|
||||
await editaragente_command(update, context)
|
||||
|
||||
update = FakeUpdate("5")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Opção remover visual: handled={handled}")
|
||||
|
||||
update = FakeUpdate("2")
|
||||
context = FakeContext()
|
||||
handled = await message_handler(update, context)
|
||||
print(f" Remover item 2: handled={handled}")
|
||||
|
||||
profile = load_user_profile()
|
||||
visual = profile.get("monica_visual_preferences", [])
|
||||
if "item2" not in visual and len(visual) == 2:
|
||||
print(" ✅ Remoção funcionou")
|
||||
else:
|
||||
print(f" ❌ Remoção não funcionou: {visual}")
|
||||
|
||||
async def test_summary_functions():
|
||||
print("\n=== Testando funções de sumário ===")
|
||||
profile = load_user_profile()
|
||||
|
||||
user_summary = get_user_profile_summary(profile)
|
||||
print(f"Sumário do usuário: {len(user_summary)} caracteres")
|
||||
if "=== SEU PERFIL ===" in user_summary and "Nome:" in user_summary:
|
||||
print("✅ Sumário do usuário OK")
|
||||
else:
|
||||
print("❌ Sumário do usuário com problema")
|
||||
|
||||
agent_summary = get_agent_profile_summary(profile)
|
||||
print(f"Sumário do agente: {len(agent_summary)} caracteres")
|
||||
if "=== PERFIL DA MÔNICA ===" in agent_summary:
|
||||
print("✅ Sumário do agente OK")
|
||||
else:
|
||||
print("❌ Sumário do agente com problema")
|
||||
|
||||
async def test_help_contains_new_commands():
|
||||
print("\n=== Verificando /help contém todos os comandos ===")
|
||||
from monica_agent import help_command
|
||||
|
||||
update = FakeUpdate("/help")
|
||||
context = FakeContext()
|
||||
responses = []
|
||||
|
||||
original_reply = FakeMessage.reply_text
|
||||
async def capture_reply(self, text, **kwargs):
|
||||
responses.append(text)
|
||||
|
||||
FakeMessage.reply_text = capture_reply
|
||||
try:
|
||||
await help_command(update, context)
|
||||
help_text = responses[0] if responses else ""
|
||||
commands_to_check = ["/verperfil", "/veragente", "/editarperfil", "/editaragente"]
|
||||
for cmd in commands_to_check:
|
||||
if cmd in help_text:
|
||||
print(f" ✅ /help menciona {cmd}")
|
||||
else:
|
||||
print(f" ❌ /help NÃO menciona {cmd}")
|
||||
finally:
|
||||
FakeMessage.reply_text = original_reply
|
||||
|
||||
async def main():
|
||||
await test_verperfil()
|
||||
await test_veragente()
|
||||
await test_summary_functions()
|
||||
await test_editarperfil_all_fields()
|
||||
await test_editaragente_all_categories()
|
||||
await test_remove_agent_items()
|
||||
await test_help_contains_new_commands()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,187 +0,0 @@
|
||||
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"
|
||||
|
||||
import monica_agent as ma
|
||||
from monica_agent import (
|
||||
fusohorario_command,
|
||||
message_handler,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
get_now,
|
||||
)
|
||||
|
||||
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_fusohorario_command_starts():
|
||||
print("=== Testando /fusohorario inicia fluxo ===")
|
||||
ma.is_setting_timezone = False
|
||||
ma.timezone_step = 0
|
||||
update = FakeUpdate("/fusohorario")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await fusohorario_command(update, context)
|
||||
assert ma.is_setting_timezone is True
|
||||
assert ma.timezone_step == 1
|
||||
print("✅ /fusohorario iniciou o fluxo corretamente")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_fusohorario_valid_timezone():
|
||||
print("\n=== Testando definicao de fuso horario valido ===")
|
||||
profile = load_user_profile()
|
||||
profile["timezone"] = ""
|
||||
save_user_profile(profile)
|
||||
ma.is_setting_timezone = False
|
||||
ma.timezone_step = 0
|
||||
update = FakeUpdate("/fusohorario")
|
||||
context = FakeContext()
|
||||
await fusohorario_command(update, context)
|
||||
|
||||
update = FakeUpdate("America/Sao_Paulo")
|
||||
context = FakeContext()
|
||||
await message_handler(update, context)
|
||||
assert ma.is_setting_timezone is False
|
||||
assert ma.timezone_step == 0
|
||||
|
||||
profile = load_user_profile()
|
||||
assert profile.get("timezone") == "America/Sao_Paulo"
|
||||
print(f"✅ Fuso horario salvo: {profile.get('timezone')}")
|
||||
|
||||
now = get_now()
|
||||
assert now.tzinfo is not None
|
||||
print(f"✅ get_now() retorna datetime com timezone: {now}")
|
||||
|
||||
async def test_fusohorario_invalid_timezone():
|
||||
print("\n=== Testando fuso horario invalido ===")
|
||||
profile = load_user_profile()
|
||||
profile["timezone"] = ""
|
||||
save_user_profile(profile)
|
||||
ma.is_setting_timezone = False
|
||||
ma.timezone_step = 0
|
||||
update = FakeUpdate("/fusohorario")
|
||||
context = FakeContext()
|
||||
await fusohorario_command(update, context)
|
||||
|
||||
update = FakeUpdate("Fuso/Invalido")
|
||||
context = FakeContext()
|
||||
await message_handler(update, context)
|
||||
assert ma.is_setting_timezone is True
|
||||
|
||||
profile = load_user_profile()
|
||||
assert profile.get("timezone") == ""
|
||||
print("✅ Fuso horario invalido rejeitado e fluxo mantido")
|
||||
|
||||
async def test_fusohorario_cancelar():
|
||||
print("\n=== Testando cancelar durante /fusohorario ===")
|
||||
profile = load_user_profile()
|
||||
profile["timezone"] = ""
|
||||
save_user_profile(profile)
|
||||
ma.is_setting_timezone = False
|
||||
ma.timezone_step = 0
|
||||
update = FakeUpdate("/fusohorario")
|
||||
context = FakeContext()
|
||||
await fusohorario_command(update, context)
|
||||
|
||||
update = FakeUpdate("cancelar")
|
||||
context = FakeContext()
|
||||
await message_handler(update, context)
|
||||
assert ma.is_setting_timezone is False
|
||||
assert ma.timezone_step == 0
|
||||
print("✅ Cancelar encerrou o fluxo de fuso horario")
|
||||
|
||||
async def test_fusohorario_affects_time():
|
||||
print("\n=== Testando que get_now usa o fuso horario ===")
|
||||
profile = load_user_profile()
|
||||
profile["timezone"] = "UTC"
|
||||
save_user_profile(profile)
|
||||
|
||||
now = get_now()
|
||||
assert now.tzinfo is not None
|
||||
print(f"✅ get_now() com UTC: {now}")
|
||||
|
||||
profile["timezone"] = "America/Sao_Paulo"
|
||||
save_user_profile(profile)
|
||||
now_sp = get_now()
|
||||
assert now_sp.tzinfo is not None
|
||||
print(f"✅ get_now() com America/Sao_Paulo: {now_sp}")
|
||||
|
||||
profile["timezone"] = ""
|
||||
save_user_profile(profile)
|
||||
now_naive = get_now()
|
||||
assert now_naive.tzinfo is None
|
||||
print(f"✅ get_now() sem timezone retorna naive: {now_naive}")
|
||||
|
||||
async def test_fusohorario_in_help():
|
||||
print("\n=== Verificando /help contem /fusohorario ===")
|
||||
from monica_agent import help_command
|
||||
update = FakeUpdate("/help")
|
||||
context = FakeContext()
|
||||
responses = []
|
||||
|
||||
original_reply = FakeMessage.reply_text
|
||||
async def capture_reply(self, text, **kwargs):
|
||||
responses.append(text)
|
||||
print(f"BOT RESPONSE: {text[:300]}")
|
||||
|
||||
FakeMessage.reply_text = capture_reply
|
||||
try:
|
||||
await help_command(update, context)
|
||||
help_text = responses[0] if responses else ""
|
||||
if "/fusohorario" in help_text:
|
||||
print("✅ /help menciona /fusohorario")
|
||||
else:
|
||||
print("❌ /help nao menciona /fusohorario")
|
||||
print(f" Help text: {help_text[:200]}")
|
||||
finally:
|
||||
FakeMessage.reply_text = original_reply
|
||||
|
||||
async def main():
|
||||
await test_fusohorario_command_starts()
|
||||
await test_fusohorario_valid_timezone()
|
||||
await test_fusohorario_invalid_timezone()
|
||||
await test_fusohorario_cancelar()
|
||||
await test_fusohorario_affects_time()
|
||||
await test_fusohorario_in_help()
|
||||
print("\n=== Todos os testes de fusohorario concluidos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
-134
@@ -1,134 +0,0 @@
|
||||
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 (
|
||||
limpar_command,
|
||||
load_user_profile,
|
||||
load_all_contexts,
|
||||
ensure_contexts_dir,
|
||||
save_user_profile,
|
||||
)
|
||||
|
||||
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[:300]}")
|
||||
|
||||
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', '')[:300]}")
|
||||
|
||||
async def test_limpar_command():
|
||||
print("=== Testando /limpar ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["name"] = "Teste"
|
||||
profile["age"] = "30"
|
||||
profile["monica_visual_preferences"] = ["item1", "item2"]
|
||||
profile["monica_sexual_preferences"] = ["item3"]
|
||||
profile["relationship_dynamics"] = ["item4"]
|
||||
save_user_profile(profile)
|
||||
|
||||
ensure_contexts_dir()
|
||||
from pathlib import Path
|
||||
from monica_agent import CONTEXTS_DIR
|
||||
test_context_file = Path(CONTEXTS_DIR) / "2026-08-22.json"
|
||||
test_context_file.write_text('[{"role": "user", "content": "teste", "timestamp": "2026-08-22T12:00:00"}]', encoding="utf-8")
|
||||
|
||||
print(f"Antes do /limpar:")
|
||||
print(f" Nome no perfil: {load_user_profile().get('name')}")
|
||||
print(f" Contextos: {len(load_all_contexts())} mensagens")
|
||||
print(f" Arquivos de contexto: {len(list(Path(CONTEXTS_DIR).glob('*.json')))}")
|
||||
|
||||
update = FakeUpdate("/limpar")
|
||||
context = FakeContext()
|
||||
|
||||
try:
|
||||
await limpar_command(update, context)
|
||||
print("✅ limpar_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return
|
||||
|
||||
profile = load_user_profile()
|
||||
contexts = load_all_contexts()
|
||||
context_files = list(Path(CONTEXTS_DIR).glob("*.json"))
|
||||
|
||||
print(f"\nDepois do /limpar:")
|
||||
print(f" Nome no perfil: '{profile.get('name')}'")
|
||||
print(f" Idade: '{profile.get('age')}'")
|
||||
print(f" Visual prefs: {profile.get('monica_visual_preferences', [])}")
|
||||
print(f" Contextos: {len(contexts)} mensagens")
|
||||
print(f" Arquivos de contexto: {len(context_files)}")
|
||||
|
||||
assert profile.get("name") == "", f"Nome deveria estar vazio, está: {profile.get('name')}"
|
||||
assert profile.get("age") == "", f"Idade deveria estar vazia, está: {profile.get('age')}"
|
||||
assert profile.get("monica_visual_preferences") == [], f"Visual prefs deveria estar vazia"
|
||||
assert len(contexts) == 0, f"Contextos deveriam estar vazios, tem: {len(contexts)}"
|
||||
assert len(context_files) == 0, f"Arquivos de contexto deveriam estar vazios, tem: {len(context_files)}"
|
||||
|
||||
print("✅ Todos os dados foram limpos corretamente")
|
||||
|
||||
async def test_help_contains_limpar():
|
||||
print("\n=== Verificando /help contém /limpar ===")
|
||||
from monica_agent import help_command
|
||||
|
||||
update = FakeUpdate("/help")
|
||||
context = FakeContext()
|
||||
responses = []
|
||||
|
||||
original_reply = FakeMessage.reply_text
|
||||
async def capture_reply(self, text, **kwargs):
|
||||
responses.append(text)
|
||||
print(f"BOT RESPONSE: {text[:300]}")
|
||||
|
||||
FakeMessage.reply_text = capture_reply
|
||||
try:
|
||||
await help_command(update, context)
|
||||
help_text = responses[0] if responses else ""
|
||||
if "/limpar" in help_text:
|
||||
print("✅ /help menciona /limpar")
|
||||
else:
|
||||
print("❌ /help não menciona /limpar")
|
||||
finally:
|
||||
FakeMessage.reply_text = original_reply
|
||||
|
||||
async def main():
|
||||
await test_limpar_command()
|
||||
await test_help_contains_limpar()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,106 +0,0 @@
|
||||
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 (
|
||||
build_system_prompt,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
get_agent_profile_summary,
|
||||
clear_all_data,
|
||||
)
|
||||
|
||||
async def test_lubricant_preferences_in_prompt():
|
||||
print("=== Verificando preferências de lubrificante no prompt ===")
|
||||
|
||||
original_profile = load_user_profile()
|
||||
profile = load_user_profile()
|
||||
profile["monica_lubricant_preferences"] = [
|
||||
"Sempre deve ser utilizado lubrificante nas ações de sexo necessárias. É uma regra obrigatória em todas as penetrações, punhetas, sexo anal, vaginal e qualquer outro tipo de atrito sexual que necessite de lubrificação. Nunca faça sexo seco ou sem lubrificante quando houver penetração ou atrito intenso."
|
||||
]
|
||||
save_user_profile(profile)
|
||||
|
||||
try:
|
||||
prompt = build_system_prompt(load_user_profile())
|
||||
|
||||
if "Preferências de lubrificante da Mônica" in prompt:
|
||||
print(" ✅ Preferências de lubrificante estão no prompt do sistema")
|
||||
else:
|
||||
print(" ❌ Preferências de lubrificante NÃO estão no prompt")
|
||||
|
||||
if "lubrificante" in prompt.lower():
|
||||
print(" ✅ Conteúdo de lubrificante está no prompt")
|
||||
else:
|
||||
print(" ❌ Conteúdo de lubrificante NÃO está no prompt")
|
||||
|
||||
if "sexo seco" in prompt.lower() or "sem lubrificante" in prompt.lower():
|
||||
print(" ✅ Regra de lubrificante obrigatório está no prompt")
|
||||
else:
|
||||
print(" ❌ Regra de lubrificante obrigatório NÃO está no prompt")
|
||||
|
||||
print(f"\nSystem prompt length: {len(prompt)} chars")
|
||||
finally:
|
||||
save_user_profile(original_profile)
|
||||
|
||||
async def test_lubricant_preferences_in_summary():
|
||||
print("\n=== Verificando preferências de lubrificante no sumário ===")
|
||||
|
||||
original_profile = load_user_profile()
|
||||
profile = load_user_profile()
|
||||
profile["monica_lubricant_preferences"] = [
|
||||
"Sempre deve ser utilizado lubrificante nas ações de sexo necessárias."
|
||||
]
|
||||
save_user_profile(profile)
|
||||
|
||||
try:
|
||||
summary = get_agent_profile_summary(load_user_profile())
|
||||
|
||||
if "Preferências de lubrificante" in summary:
|
||||
print(" ✅ Sumário do agente inclui preferências de lubrificante")
|
||||
else:
|
||||
print(" ❌ Sumário do agente NÃO inclui preferências de lubrificante")
|
||||
|
||||
if "lubrificante" in summary.lower():
|
||||
print(" ✅ Detalhe do lubrificante está no sumário")
|
||||
else:
|
||||
print(" ❌ Detalhe do lubrificante NÃO está no sumário")
|
||||
finally:
|
||||
save_user_profile(original_profile)
|
||||
|
||||
async def test_clear_all_data_removes_lubricant_preferences():
|
||||
print("\n=== Verificando que limpar dados remove preferências de lubrificante ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["monica_lubricant_preferences"] = ["item de teste"]
|
||||
save_user_profile(profile)
|
||||
|
||||
clear_all_data()
|
||||
|
||||
profile = load_user_profile()
|
||||
if not profile.get("monica_lubricant_preferences"):
|
||||
print(" ✅ clear_all_data removeu preferências de lubrificante")
|
||||
else:
|
||||
print(f" ❌ clear_all_data NÃO removeu preferências de lubrificante: {profile.get('monica_lubricant_preferences')}")
|
||||
|
||||
async def test_lubricant_preferences_with_empty_profile():
|
||||
print("\n=== Verificando prompt com perfil vazio ===")
|
||||
prompt = build_system_prompt({})
|
||||
|
||||
if "Preferências de lubrificante da Mônica" not in prompt:
|
||||
print(" ✅ Prompt vazio NÃO contém preferências de lubrificante (correto)")
|
||||
else:
|
||||
print(" ❌ Prompt vazio contém preferências de lubrificante (erro)")
|
||||
|
||||
async def main():
|
||||
await test_lubricant_preferences_in_prompt()
|
||||
await test_lubricant_preferences_in_summary()
|
||||
await test_clear_all_data_removes_lubricant_preferences()
|
||||
await test_lubricant_preferences_with_empty_profile()
|
||||
print("\n=== Todos os testes de lubrificante concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,122 +0,0 @@
|
||||
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 (
|
||||
ensure_min_lines,
|
||||
MIN_RESPONSE_LINES,
|
||||
load_user_profile,
|
||||
build_system_prompt,
|
||||
load_all_contexts,
|
||||
build_messages_for_ai,
|
||||
AIProvider,
|
||||
DEFAULT_MODEL,
|
||||
)
|
||||
|
||||
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[:300]}")
|
||||
|
||||
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', '')[:300]}")
|
||||
|
||||
async def test_ensure_min_lines():
|
||||
print("=== Testando ensure_min_lines ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
all_contexts = load_all_contexts()
|
||||
recent_context = all_contexts[-10:] if len(all_contexts) > 10 else all_contexts
|
||||
|
||||
short_response = "Olá! Tudo bem?"
|
||||
user_message = "Oi Mônica!"
|
||||
|
||||
print(f"Resposta curta original: '{short_response}' ({len(short_response.splitlines())} linhas)")
|
||||
|
||||
result = await ensure_min_lines(short_response, user_message, recent_context, profile)
|
||||
|
||||
print(f"Resposta após ensure_min_lines: '{result[:200]}...' ({len(result.splitlines())} linhas)")
|
||||
|
||||
if len(result.splitlines()) >= MIN_RESPONSE_LINES:
|
||||
print(f"✅ Resposta expandida para {len(result.splitlines())} linhas (mínimo: {MIN_RESPONSE_LINES})")
|
||||
else:
|
||||
print(f"❌ Resposta não atingiu o mínimo de {MIN_RESPONSE_LINES} linhas")
|
||||
|
||||
print(f"\nConteúdo completo da resposta expandida:")
|
||||
print(result)
|
||||
|
||||
async def test_get_monica_response():
|
||||
print("\n=== Testando get_monica_response ===")
|
||||
from monica_agent import get_monica_response
|
||||
|
||||
response = await get_monica_response("Oi Mônica!")
|
||||
|
||||
if response:
|
||||
lines = response.splitlines()
|
||||
print(f"Resposta de get_monica_response: {len(lines)} linhas")
|
||||
print(f"Primeiras 3 linhas: {lines[:3]}")
|
||||
|
||||
if len(lines) >= MIN_RESPONSE_LINES:
|
||||
print(f"✅ Resposta tem {len(lines)} linhas (mínimo: {MIN_RESPONSE_LINES})")
|
||||
else:
|
||||
print(f"❌ Resposta tem apenas {len(lines)} linhas (mínimo: {MIN_RESPONSE_LINES})")
|
||||
print("Conteúdo completo:")
|
||||
print(response)
|
||||
else:
|
||||
print("❌ Resposta foi None")
|
||||
|
||||
async def test_system_prompt_contains_min_lines():
|
||||
print("\n=== Verificando se prompt base pede respostas longas ===")
|
||||
from monica_agent import MONICA_SYSTEM_PROMPT_BASE, build_system_prompt, MIN_RESPONSE_LINES
|
||||
|
||||
if "{min_lines}" in MONICA_SYSTEM_PROMPT_BASE:
|
||||
print("✅ Prompt base usa placeholder {min_lines} (configurável)")
|
||||
else:
|
||||
print("❌ Prompt base NÃO usa placeholder configurável")
|
||||
|
||||
formatted = build_system_prompt({})
|
||||
expected_text = f"no mínimo {MIN_RESPONSE_LINES} linhas"
|
||||
if expected_text in formatted:
|
||||
print(f"✅ Prompt formatado pede pelo menos {MIN_RESPONSE_LINES} linhas")
|
||||
else:
|
||||
print(f"❌ Prompt formatado NÃO menciona {MIN_RESPONSE_LINES} linhas")
|
||||
print(f"Prompt formatado (início): {formatted[:300]}...")
|
||||
|
||||
async def main():
|
||||
await test_system_prompt_contains_min_lines()
|
||||
await test_ensure_min_lines()
|
||||
await test_get_monica_response()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,68 +0,0 @@
|
||||
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 (
|
||||
load_all_contexts,
|
||||
load_user_profile,
|
||||
build_system_prompt,
|
||||
AIProvider,
|
||||
MIN_INTERVAL_MINUTES,
|
||||
MAX_INTERVAL_MINUTES,
|
||||
)
|
||||
|
||||
async def test_proactive_context():
|
||||
print("=== Testando contexto de mensagens proativas ===")
|
||||
|
||||
all_contexts = load_all_contexts()
|
||||
print(f"Total de mensagens historicas: {len(all_contexts)}")
|
||||
|
||||
recent_context = all_contexts[-10:] if len(all_contexts) > 10 else all_contexts
|
||||
print(f"Mensagens recentes usadas para proativa: {len(recent_context)}")
|
||||
|
||||
profile = load_user_profile()
|
||||
system_prompt = build_system_prompt(profile)
|
||||
|
||||
messages = [{"role": "system", "content": system_prompt}]
|
||||
messages.extend(recent_context)
|
||||
messages.append({
|
||||
"role": "user",
|
||||
"content": (
|
||||
"Mônica, inicie uma conversa espontânea e natural comigo agora. "
|
||||
"Use o contexto da conversa acima para manter a coerência e continuar o fluxo natural do dia. "
|
||||
"Pode ser um pensamento aleatório, uma pergunta sobre o meu dia, "
|
||||
"algo carinhoso, sexy, ou divertido. Seja criativa e autêntica."
|
||||
),
|
||||
})
|
||||
|
||||
print(f"System prompt length: {len(system_prompt)} chars")
|
||||
print(f"Total messages sent to AI: {len(messages)}")
|
||||
print(f"Context messages: {len(recent_context)}")
|
||||
|
||||
# Show last 3 context messages for verification
|
||||
if recent_context:
|
||||
print("\nUltimas 3 mensagens do contexto:")
|
||||
for msg in recent_context[-3:]:
|
||||
role = msg.get("role", "?")
|
||||
content = msg.get("content", "")[:80]
|
||||
print(f" [{role}]: {content}...")
|
||||
|
||||
print("\n=== Estrutura de mensagens para IA ===")
|
||||
for i, msg in enumerate(messages):
|
||||
role = msg.get("role", "?")
|
||||
content = msg.get("content", "")[:100]
|
||||
print(f" {i}: [{role}] {content}...")
|
||||
|
||||
print("\n=== Intervalos configurados ===")
|
||||
print(f"MIN_INTERVAL_MINUTES: {MIN_INTERVAL_MINUTES}")
|
||||
print(f"MAX_INTERVAL_MINUTES: {MAX_INTERVAL_MINUTES}")
|
||||
print(f"Range: {MIN_INTERVAL_MINUTES * 60}s a {MAX_INTERVAL_MINUTES * 60}s")
|
||||
|
||||
print("\n✅ Teste de estrutura de contexto proativo passou!")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(test_proactive_context())
|
||||
@@ -1,116 +0,0 @@
|
||||
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 (
|
||||
build_system_prompt,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
get_agent_profile_summary,
|
||||
clear_all_data,
|
||||
MONICA_SYSTEM_PROMPT_BASE,
|
||||
)
|
||||
|
||||
async def test_professional_services_in_prompt():
|
||||
print("=== Verificando serviços profissionais no prompt ===")
|
||||
|
||||
original_profile = load_user_profile()
|
||||
profile = load_user_profile()
|
||||
profile["monica_professional_services"] = [
|
||||
"Faz depilação íntima completa: bunda, cú e partes íntimas de homens e mulheres.",
|
||||
"Sempre termina com final feliz: os homens gozam sem querer de tanto que a Mônica precisa pegar no pau, alisar o pau durante a depilação.",
|
||||
"Massageia o cú e/ou buceta durante e/ou após a depilação.",
|
||||
"Marco às vezes ajuda quando a Mônica chama, não é sempre.",
|
||||
"Quando falar desse trabalho, seja muito detalhista, descrevendo cada etapa, cada toque, cada sensação, cada detalhe do corpo e do prazer envolvidos."
|
||||
]
|
||||
save_user_profile(profile)
|
||||
|
||||
try:
|
||||
prompt = build_system_prompt(load_user_profile())
|
||||
|
||||
if "Serviços profissionais da Mônica" in prompt:
|
||||
print(" ✅ Serviços profissionais estão no prompt do sistema")
|
||||
else:
|
||||
print(" ❌ Serviços profissionais NÃO estão no prompt")
|
||||
|
||||
if "depilação" in prompt.lower() and "final feliz" in prompt.lower():
|
||||
print(" ✅ Detalhes do trabalho de depilação estão no prompt")
|
||||
else:
|
||||
print(" ❌ Detalhes do trabalho de depilação NÃO estão no prompt")
|
||||
|
||||
if "Marco às vezes ajuda" in prompt:
|
||||
print(" ✅ Participação do Marco está no prompt")
|
||||
else:
|
||||
print(" ❌ Participação do Marco NÃO está no prompt")
|
||||
|
||||
if "detalhista" in prompt.lower() or "detalhado" in prompt.lower():
|
||||
print(" ✅ Diretiva de detalhamento está no prompt")
|
||||
else:
|
||||
print(" ❌ Diretiva de detalhamento NÃO está no prompt")
|
||||
|
||||
print(f"\nSystem prompt length: {len(prompt)} chars")
|
||||
finally:
|
||||
save_user_profile(original_profile)
|
||||
|
||||
async def test_professional_services_in_summary():
|
||||
print("\n=== Verificando serviços profissionais no sumário ===")
|
||||
|
||||
original_profile = load_user_profile()
|
||||
profile = load_user_profile()
|
||||
profile["monica_professional_services"] = [
|
||||
"Faz depilação íntima completa: bunda, cú e partes íntimas."
|
||||
]
|
||||
save_user_profile(profile)
|
||||
|
||||
try:
|
||||
summary = get_agent_profile_summary(load_user_profile())
|
||||
|
||||
if "Serviços profissionais" in summary:
|
||||
print(" ✅ Sumário do agente inclui serviços profissionais")
|
||||
else:
|
||||
print(" ❌ Sumário do agente NÃO inclui serviços profissionais")
|
||||
|
||||
if "depilação" in summary.lower():
|
||||
print(" ✅ Detalhe da depilação está no sumário")
|
||||
else:
|
||||
print(" ❌ Detalhe da depilação NÃO está no sumário")
|
||||
finally:
|
||||
save_user_profile(original_profile)
|
||||
|
||||
async def test_clear_all_data_removes_professional_services():
|
||||
print("\n=== Verificando que limpar dados remove serviços profissionais ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["monica_professional_services"] = ["item de teste"]
|
||||
save_user_profile(profile)
|
||||
|
||||
clear_all_data()
|
||||
|
||||
profile = load_user_profile()
|
||||
if not profile.get("monica_professional_services"):
|
||||
print(" ✅ clear_all_data removeu serviços profissionais")
|
||||
else:
|
||||
print(f" ❌ clear_all_data NÃO removeu serviços profissionais: {profile.get('monica_professional_services')}")
|
||||
|
||||
async def test_professional_services_without_cohabitation():
|
||||
print("\n=== Verificando prompt sem coabitação (perfil vazio) ===")
|
||||
prompt = build_system_prompt({})
|
||||
|
||||
if "Serviços profissionais da Mônica" not in prompt:
|
||||
print(" ✅ Prompt vazio NÃO contém serviços profissionais (correto)")
|
||||
else:
|
||||
print(" ❌ Prompt vazio contém serviços profissionais (erro)")
|
||||
|
||||
async def main():
|
||||
await test_professional_services_in_prompt()
|
||||
await test_professional_services_in_summary()
|
||||
await test_clear_all_data_removes_professional_services()
|
||||
await test_professional_services_without_cohabitation()
|
||||
print("\n=== Todos os testes de serviços profissionais concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,128 +0,0 @@
|
||||
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 (
|
||||
build_system_prompt,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
MONICA_SYSTEM_PROMPT_BASE,
|
||||
MIN_RESPONSE_LINES,
|
||||
)
|
||||
|
||||
async def test_system_prompt_contains_cohabitation_directive():
|
||||
print("=== Verificando diretiva de coabitação no prompt base ===")
|
||||
|
||||
expected_phrases = [
|
||||
"morar com você",
|
||||
"mesma casa",
|
||||
"se estivesse aí com você agora",
|
||||
"saudade",
|
||||
]
|
||||
|
||||
for phrase in expected_phrases:
|
||||
if phrase.lower() in MONICA_SYSTEM_PROMPT_BASE.lower():
|
||||
print(f" ✅ Prompt base contém referência a: '{phrase}'")
|
||||
else:
|
||||
print(f" ❌ Prompt base NÃO contém referência a: '{phrase}'")
|
||||
|
||||
print("\n=== Verificando diretiva de coabitação no prompt completo ===")
|
||||
|
||||
original_profile = load_user_profile()
|
||||
profile = load_user_profile()
|
||||
profile["relationship_context"] = "Casados, moram na mesma casa, estão sempre juntos."
|
||||
profile["relationship_dynamics"] = [
|
||||
"Casados",
|
||||
"moram na mesma casa",
|
||||
"estão sempre juntos",
|
||||
"NUNCA use frases como 'se estivesse aí com você agora', 'como eu desejaria estar aí', 'queria te ver agora', 'saudade de te ver' ou qualquer expressão que implique separação física ou desejo de estar junto, pois vocês já moram juntos e estão sempre presentes um para o outro."
|
||||
]
|
||||
save_user_profile(profile)
|
||||
|
||||
try:
|
||||
prompt = build_system_prompt(load_user_profile())
|
||||
|
||||
if "Casados, moram na mesma casa" in prompt:
|
||||
print(" ✅ Contexto do relacionamento (casados/mesma casa) está no prompt")
|
||||
else:
|
||||
print(" ❌ Contexto do relacionamento NÃO está no prompt")
|
||||
|
||||
if "REGRAS ABSOLUTAS" in prompt:
|
||||
print(" ✅ Regras absolutas de coabitação estão no prompt")
|
||||
else:
|
||||
print(" ❌ Regras absolutas de coabitação NÃO estão no prompt")
|
||||
|
||||
forbidden_phrases = [
|
||||
"se estivesse aí com você agora",
|
||||
"como eu desejaria estar aí",
|
||||
"queria te ver agora",
|
||||
"saudade de te ver",
|
||||
"morrendo de saudade",
|
||||
"não posso estar aí",
|
||||
"adoraria estar aí",
|
||||
"como eu queria estar aí",
|
||||
]
|
||||
|
||||
found_forbidden = sum(1 for p in forbidden_phrases if p in prompt.lower())
|
||||
if found_forbidden >= 6:
|
||||
print(f" ✅ Prompt lista {found_forbidden} frases proibidas de distância")
|
||||
else:
|
||||
print(f" ❌ Prompt lista apenas {found_forbidden} frases proibidas (esperado >= 6)")
|
||||
|
||||
if f"no mínimo {MIN_RESPONSE_LINES} linhas" in prompt:
|
||||
print(f" ✅ Prompt pede no mínimo {MIN_RESPONSE_LINES} linhas")
|
||||
else:
|
||||
print(f" ❌ Prompt NÃO pede {MIN_RESPONSE_LINES} linhas")
|
||||
|
||||
print(f"\nSystem prompt length: {len(prompt)} chars")
|
||||
finally:
|
||||
save_user_profile(original_profile)
|
||||
|
||||
async def test_system_prompt_without_cohabitation():
|
||||
print("\n=== Verificando prompt sem coabitação (perfil vazio) ===")
|
||||
prompt = build_system_prompt({})
|
||||
|
||||
if "REGRAS ABSOLUTAS" not in prompt:
|
||||
print(" ✅ Prompt sem coabitação NÃO contém regras absolutas (correto)")
|
||||
else:
|
||||
print(" ❌ Prompt sem coabitação contém regras absolutas (erro)")
|
||||
|
||||
if f"no mínimo {MIN_RESPONSE_LINES} linhas" in prompt:
|
||||
print(f" ✅ Prompt base ainda pede {MIN_RESPONSE_LINES} linhas")
|
||||
else:
|
||||
print(f" ❌ Prompt base NÃO pede {MIN_RESPONSE_LINES} linhas")
|
||||
|
||||
async def test_relationship_dynamics_update_triggers_cohabitation_rules():
|
||||
print("\n=== Verificando que relationship_dynamics com coabitação ativa regras ===")
|
||||
|
||||
original_profile = load_user_profile()
|
||||
profile = load_user_profile()
|
||||
profile["relationship_dynamics"] = [
|
||||
"Casados",
|
||||
"moram na mesma casa",
|
||||
"estão sempre juntos",
|
||||
]
|
||||
save_user_profile(profile)
|
||||
|
||||
try:
|
||||
prompt = build_system_prompt(load_user_profile())
|
||||
|
||||
if "REGRAS ABSOLUTAS" in prompt:
|
||||
print(" ✅ relationship_dynamics com coabitação ativa regras absolutas")
|
||||
else:
|
||||
print(" ❌ relationship_dynamics com coabitação NÃO ativou regras absolutas")
|
||||
finally:
|
||||
save_user_profile(original_profile)
|
||||
|
||||
async def main():
|
||||
await test_system_prompt_contains_cohabitation_directive()
|
||||
await test_system_prompt_without_cohabitation()
|
||||
await test_relationship_dynamics_update_triggers_cohabitation_rules()
|
||||
print("\n=== Todos os testes de relacionamento concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,85 +0,0 @@
|
||||
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 _force_min_lines, _split_message_for_telegram, MAX_TELEGRAM_MESSAGE_LENGTH
|
||||
|
||||
def test_force_min_lines_no_truncation():
|
||||
print("=== Testando _force_min_lines nao trunca ===")
|
||||
text = "A. B. C. D. E. F. G. H. I. J."
|
||||
result = _force_min_lines(text, 5)
|
||||
assert "F." in result, f"Conteudo foi truncado! Resultado: {result}"
|
||||
assert "G." in result, f"Conteudo foi truncado! Resultado: {result}"
|
||||
assert "H." in result, f"Conteudo foi truncado! Resultado: {result}"
|
||||
assert "I." in result, f"Conteudo foi truncado! Resultado: {result}"
|
||||
assert "J." in result, f"Conteudo foi truncado! Resultado: {result}"
|
||||
print(f"✅ _force_min_lines preservou todo o conteudo")
|
||||
print(f" Resultado: {result}")
|
||||
|
||||
def test_force_min_lines_preserves_long_text():
|
||||
print("\n=== Testando _force_min_lines com texto longo ===")
|
||||
sentences = " ".join([f"Sentenca{i}." for i in range(1, 26)])
|
||||
result = _force_min_lines(sentences, 10)
|
||||
assert "Sentenca20." in result
|
||||
assert "Sentenca25." in result
|
||||
print("✅ _force_min_lines nao truncou texto longo")
|
||||
|
||||
def test_split_message_short():
|
||||
print("\n=== Testando _split_message_for_telegram mensagem curta ===")
|
||||
text = "Ola, tudo bem?"
|
||||
parts = _split_message_for_telegram(text)
|
||||
assert len(parts) == 1
|
||||
assert parts[0] == text
|
||||
print("✅ Mensagem curta nao foi dividida")
|
||||
|
||||
def test_split_message_exactly_at_limit():
|
||||
print("\n=== Testando _split_message_for_telegram no limite ===")
|
||||
text = "A" * MAX_TELEGRAM_MESSAGE_LENGTH
|
||||
parts = _split_message_for_telegram(text)
|
||||
assert len(parts) == 1
|
||||
assert len(parts[0]) == MAX_TELEGRAM_MESSAGE_LENGTH
|
||||
print(f"✅ Mensagem com {MAX_TELEGRAM_MESSAGE_LENGTH} chars nao foi dividida")
|
||||
|
||||
def test_split_message_long():
|
||||
print("\n=== Testando _split_message_for_telegram mensagem longa ===")
|
||||
paragraph = "Esta é uma fraselonga. " * 500
|
||||
text = paragraph + "\n\n" + paragraph
|
||||
parts = _split_message_for_telegram(text)
|
||||
assert len(parts) > 1
|
||||
for part in parts:
|
||||
assert len(part) <= MAX_TELEGRAM_MESSAGE_LENGTH, f"Parte muito longa: {len(part)}"
|
||||
print(f"✅ Mensagem longa dividida em {len(parts)} partes, todas <= {MAX_TELEGRAM_MESSAGE_LENGTH} chars")
|
||||
|
||||
def test_split_message_preserves_content():
|
||||
print("\n=== Testando _split_message_for_telegram preserva conteudo ===")
|
||||
text = "Paragrafo um.\n\nParagrafo dois.\n\nParagrafo tres."
|
||||
parts = _split_message_for_telegram(text)
|
||||
joined = "\n\n".join(parts)
|
||||
assert joined == text
|
||||
print("✅ Conteudo preservado apos split")
|
||||
|
||||
def test_split_message_very_long_single_paragraph():
|
||||
print("\n=== Testando _split_message_for_telegram paragrafo muito longo ===")
|
||||
text = "A. " * 5000
|
||||
parts = _split_message_for_telegram(text)
|
||||
assert len(parts) > 1
|
||||
for part in parts:
|
||||
assert len(part) <= MAX_TELEGRAM_MESSAGE_LENGTH
|
||||
print(f"✅ Paragrafo longo dividido em {len(parts)} partes")
|
||||
|
||||
def main():
|
||||
test_force_min_lines_no_truncation()
|
||||
test_force_min_lines_preserves_long_text()
|
||||
test_split_message_short()
|
||||
test_split_message_exactly_at_limit()
|
||||
test_split_message_long()
|
||||
test_split_message_preserves_content()
|
||||
test_split_message_very_long_single_paragraph()
|
||||
print("\n=== Todos os testes de correcao de resposta concluidos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,97 +0,0 @@
|
||||
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 unittest.mock import patch, AsyncMock
|
||||
from monica_agent import get_monica_response, AIProvider
|
||||
|
||||
async def test_retry_success_after_failures():
|
||||
print("=== Testando retry: sucesso após falhas ===")
|
||||
|
||||
call_count = 0
|
||||
def mock_chat(model, messages, temperature=0.8, max_tokens=2000):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
if call_count < 3:
|
||||
print(f" Tentativa {call_count}: falha (simulada)")
|
||||
return None
|
||||
print(f" Tentativa {call_count}: sucesso")
|
||||
return "Resposta após retry"
|
||||
|
||||
async def mock_ensure_min_lines(response, *args, **kwargs):
|
||||
return response
|
||||
|
||||
with patch.object(AIProvider, 'chat', mock_chat), \
|
||||
patch('monica_agent.ensure_min_lines', mock_ensure_min_lines):
|
||||
response = await get_monica_response("Olá")
|
||||
|
||||
print(f" Total de chamadas: {call_count}")
|
||||
print(f" Resposta: {response}")
|
||||
|
||||
if call_count == 3 and response == "Resposta após retry":
|
||||
print("✅ Retry funcionou: sucesso após 2 falhas")
|
||||
else:
|
||||
print(f"❌ Retry não funcionou como esperado: {call_count} chamadas")
|
||||
|
||||
async def test_retry_exhausted():
|
||||
print("\n=== Testando retry: esgotamento das 5 tentativas ===")
|
||||
|
||||
call_count = 0
|
||||
def mock_chat(model, messages, temperature=0.8, max_tokens=2000):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
print(f" Tentativa {call_count}: falha")
|
||||
return None
|
||||
|
||||
async def mock_ensure_min_lines(response, *args, **kwargs):
|
||||
return response
|
||||
|
||||
with patch.object(AIProvider, 'chat', mock_chat), \
|
||||
patch('monica_agent.ensure_min_lines', mock_ensure_min_lines):
|
||||
response = await get_monica_response("Olá")
|
||||
|
||||
print(f" Total de chamadas: {call_count}")
|
||||
print(f" Resposta: {response}")
|
||||
|
||||
if call_count == 5 and response is None:
|
||||
print("✅ Retry esgotado corretamente após 5 tentativas")
|
||||
else:
|
||||
print(f"❌ Esperado 5 chamadas, got {call_count}")
|
||||
|
||||
async def test_no_retry_on_success():
|
||||
print("\n=== Testando: sem retry quando sucesso na primeira tentativa ===")
|
||||
|
||||
call_count = 0
|
||||
def mock_chat(model, messages, temperature=0.8, max_tokens=2000):
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
print(f" Tentativa {call_count}: sucesso")
|
||||
return "Resposta imediata"
|
||||
|
||||
async def mock_ensure_min_lines(response, *args, **kwargs):
|
||||
return response
|
||||
|
||||
with patch.object(AIProvider, 'chat', mock_chat), \
|
||||
patch('monica_agent.ensure_min_lines', mock_ensure_min_lines):
|
||||
response = await get_monica_response("Olá")
|
||||
|
||||
print(f" Total de chamadas: {call_count}")
|
||||
print(f" Resposta: {response}")
|
||||
|
||||
if call_count == 1 and response == "Resposta imediata":
|
||||
print("✅ Sem retry desnecessário quando sucesso na primeira tentativa")
|
||||
else:
|
||||
print(f"❌ Esperado 1 chamada, got {call_count}")
|
||||
|
||||
async def main():
|
||||
await test_retry_success_after_failures()
|
||||
await test_retry_exhausted()
|
||||
await test_no_retry_on_success()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,109 +0,0 @@
|
||||
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 unittest.mock import patch, AsyncMock
|
||||
from monica_agent import (
|
||||
start_command,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
load_all_contexts,
|
||||
add_to_context,
|
||||
ensure_contexts_dir,
|
||||
AIProvider,
|
||||
DEFAULT_MODEL,
|
||||
)
|
||||
from pathlib import Path
|
||||
|
||||
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_start_without_context():
|
||||
print("=== Testando /start SEM contextos ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["name"] = "Marco"
|
||||
save_user_profile(profile)
|
||||
|
||||
from monica_agent import CONTEXTS_DIR
|
||||
contexts_dir = Path(CONTEXTS_DIR)
|
||||
for file_path in contexts_dir.glob("*.json"):
|
||||
file_path.unlink()
|
||||
|
||||
update = FakeUpdate("/start")
|
||||
context = FakeContext()
|
||||
|
||||
try:
|
||||
await start_command(update, context)
|
||||
print("✅ start_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_start_with_context():
|
||||
print("\n=== Testando /start COM contextos ===")
|
||||
|
||||
profile = load_user_profile()
|
||||
profile["name"] = "Marco"
|
||||
save_user_profile(profile)
|
||||
|
||||
add_to_context("user", "Olá Mônica!")
|
||||
add_to_context("assistant", "Oi, Marco! Como você está?")
|
||||
|
||||
def mock_chat(model, messages, temperature=0.8, max_tokens=2000):
|
||||
return "Que bom ver você de novo! Continue a conversa baseado no contexto anterior."
|
||||
|
||||
with patch.object(AIProvider, 'chat', mock_chat):
|
||||
update = FakeUpdate("/start")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await start_command(update, context)
|
||||
print("✅ start_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def main():
|
||||
await test_start_without_context()
|
||||
await test_start_with_context()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,285 +0,0 @@
|
||||
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())
|
||||
@@ -1,133 +0,0 @@
|
||||
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 (
|
||||
veragendamento_command,
|
||||
veruseragenda_command,
|
||||
load_user_profile,
|
||||
save_user_profile,
|
||||
clear_all_data,
|
||||
)
|
||||
|
||||
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_veragendamento_not_configured():
|
||||
print("=== Testando /veragendamento sem configuração ===")
|
||||
profile = load_user_profile()
|
||||
profile["proactive_schedule"] = {"enabled": False}
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/veragendamento")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await veragendamento_command(update, context)
|
||||
print("✅ veragendamento_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_veragendamento_configured():
|
||||
print("\n=== Testando /veragendamento com configuração ===")
|
||||
profile = load_user_profile()
|
||||
profile["proactive_schedule"] = {
|
||||
"enabled": True,
|
||||
"days": [0, 1, 2, 3, 4],
|
||||
"start_time": "09:00",
|
||||
"end_time": "22:00"
|
||||
}
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/veragendamento")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await veragendamento_command(update, context)
|
||||
print("✅ veragendamento_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_veruseragenda_not_configured():
|
||||
print("\n=== Testando /veruseragenda sem configuração ===")
|
||||
profile = load_user_profile()
|
||||
profile["user_agenda"] = {}
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/veruseragenda")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await veruseragenda_command(update, context)
|
||||
print("✅ veruseragenda_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def test_veruseragenda_configured():
|
||||
print("\n=== Testando /veruseragenda com configuração ===")
|
||||
profile = load_user_profile()
|
||||
profile["user_agenda"] = {
|
||||
"acordar": "07:00",
|
||||
"cafe_da_manha": "08:00",
|
||||
"trabalho": "09:00",
|
||||
"almoco": "12:00",
|
||||
}
|
||||
save_user_profile(profile)
|
||||
|
||||
update = FakeUpdate("/veruseragenda")
|
||||
context = FakeContext()
|
||||
try:
|
||||
await veruseragenda_command(update, context)
|
||||
print("✅ veruseragenda_command executou sem erros")
|
||||
except Exception as e:
|
||||
print(f"❌ Erro: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
async def main():
|
||||
await test_veragendamento_not_configured()
|
||||
await test_veragendamento_configured()
|
||||
await test_veruseragenda_not_configured()
|
||||
await test_veruseragenda_configured()
|
||||
print("\n=== Todos os testes concluídos ===")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user