Modifications whitlist_partners
This commit is contained in:
parent
c14e2fb7e3
commit
0d24e8dfca
7 changed files with 9 additions and 687 deletions
|
|
@ -1,6 +1,6 @@
|
||||||
# 🤖 Kuby - Bot Discord Multi-Fonctions
|
# 🤖 Kuby - Bot Discord Multi-Fonctions
|
||||||
|
|
||||||
Kuby est un bot Discord complet développé en Python avec `discord.py`. Il offre une suite complète d'outils de modération, de sécurité, de sauvegarde et de gestion communautaire.
|
Kuby est un bot Discord complet développé en Python avec `disnake.py`. Il offre une suite complète d'outils de modération, de sécurité, de sauvegarde et de gestion communautaire.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
|
|
||||||
15
TODO.md
15
TODO.md
|
|
@ -1,15 +0,0 @@
|
||||||
# TODO - Désactivation fonctionnalités premium (SKU) & correction accès /ticket
|
|
||||||
|
|
||||||
## Étape 1 — Compréhension
|
|
||||||
- [x] Identifier le check premium bloquant /ticket (utils/premium.py)
|
|
||||||
- [x] Vérifier où le check est appliqué (commandes/ticket/__init__.py)
|
|
||||||
|
|
||||||
## Étape 2 — Mise en conformité
|
|
||||||
- [x] Désactivation temporaire via env var : DISABLE_PREMIUM_SKU_CHECK=true
|
|
||||||
- [ ] Ajouter un switch clair pour ré-activer facilement (déjà géré par la même env var)
|
|
||||||
|
|
||||||
|
|
||||||
## Étape 3 — Validation
|
|
||||||
- [ ] Lancer un lint/tests rapides ou démarrer le bot (si possible)
|
|
||||||
- [ ] Vérifier que /ticket répond sur le serveur bloqué (Omega Kube)
|
|
||||||
|
|
||||||
259
TÂCHE_VICY.py
259
TÂCHE_VICY.py
|
|
@ -1,259 +0,0 @@
|
||||||
import discord
|
|
||||||
from discord import app_commands
|
|
||||||
from discord.ext import commands, tasks
|
|
||||||
from datetime import datetime
|
|
||||||
from zoneinfo import ZoneInfo
|
|
||||||
from utils.gestion_taches import (
|
|
||||||
ajouter_tache, lister_taches_utilisateur, taches_proches,
|
|
||||||
update_tache_status, charger_taches, supprimer_tache, sauvegarder_taches
|
|
||||||
)
|
|
||||||
|
|
||||||
PRIORITES = ["Haute", "Moyenne", "Basse", "Critique", "Faible"]
|
|
||||||
|
|
||||||
class RappelTacheView(discord.ui.View):
|
|
||||||
def __init__(self, tid: str):
|
|
||||||
super().__init__(timeout=None)
|
|
||||||
self.tid = tid
|
|
||||||
|
|
||||||
@discord.ui.button(label="✅ Marquer comme faite", style=discord.ButtonStyle.success, custom_id="btn_mark_done")
|
|
||||||
async def mark_done(self, interaction: discord.Interaction, button: discord.ui.Button):
|
|
||||||
success = await update_tache_status(self.tid, "Terminée")
|
|
||||||
if success:
|
|
||||||
await interaction.response.send_message(
|
|
||||||
f"✅ La tâche #{self.tid} a été marquée comme terminée !", ephemeral=True)
|
|
||||||
else:
|
|
||||||
await interaction.response.send_message(
|
|
||||||
"❌ Impossible de mettre à jour la tâche. Peut-être déjà terminée ?", ephemeral=True)
|
|
||||||
self.disable_all_items()
|
|
||||||
await interaction.message.edit(view=self)
|
|
||||||
|
|
||||||
class MesTachesPaginator(discord.ui.View):
|
|
||||||
def __init__(self, interaction: discord.Interaction, taches_list: list[tuple], per_page=5):
|
|
||||||
super().__init__(timeout=120)
|
|
||||||
self.interaction = interaction
|
|
||||||
self.taches = taches_list
|
|
||||||
self.per_page = per_page
|
|
||||||
self.page = 0
|
|
||||||
self.max_page = (len(taches_list) - 1) // per_page
|
|
||||||
|
|
||||||
def get_embed(self):
|
|
||||||
embed = discord.Embed(title="📝 Vos Tâches Assignées", color=discord.Color.green())
|
|
||||||
start = self.page * self.per_page
|
|
||||||
end = start + self.per_page
|
|
||||||
for tid, t in self.taches[start:end]:
|
|
||||||
embed.add_field(
|
|
||||||
name=f"#{tid} - {t['titre']}",
|
|
||||||
value=f"📅 Échéance : {t['échéance']}\n⚠️ Priorité : {t['priorité']}\n📌 Statut : {t['statut']}",
|
|
||||||
inline=False
|
|
||||||
)
|
|
||||||
embed.set_footer(text=f"Page {self.page + 1} / {self.max_page + 1}")
|
|
||||||
return embed
|
|
||||||
|
|
||||||
@discord.ui.button(label="⬅️", style=discord.ButtonStyle.secondary)
|
|
||||||
async def previous_page(self, interaction: discord.Interaction, button: discord.ui.Button):
|
|
||||||
if interaction.user.id != self.interaction.user.id:
|
|
||||||
await interaction.response.send_message("❌ Ce n'est pas votre pagination.", ephemeral=True)
|
|
||||||
return
|
|
||||||
if self.page > 0:
|
|
||||||
self.page -= 1
|
|
||||||
await interaction.response.edit_message(embed=self.get_embed(), view=self)
|
|
||||||
else:
|
|
||||||
await interaction.response.defer()
|
|
||||||
|
|
||||||
@discord.ui.button(label="➡️", style=discord.ButtonStyle.secondary)
|
|
||||||
async def next_page(self, interaction: discord.Interaction, button: discord.ui.Button):
|
|
||||||
if interaction.user.id != self.interaction.user.id:
|
|
||||||
await interaction.response.send_message("❌ Ce n'est pas votre pagination.", ephemeral=True)
|
|
||||||
return
|
|
||||||
if self.page < self.max_page:
|
|
||||||
self.page += 1
|
|
||||||
await interaction.response.edit_message(embed=self.get_embed(), view=self)
|
|
||||||
else:
|
|
||||||
await interaction.response.defer()
|
|
||||||
|
|
||||||
class JiraDiscord(commands.Cog):
|
|
||||||
def __init__(self, bot):
|
|
||||||
self.bot = bot
|
|
||||||
self.rappels.start()
|
|
||||||
|
|
||||||
def cog_unload(self):
|
|
||||||
self.rappels.cancel()
|
|
||||||
|
|
||||||
async def autocomplete_priorite(self, interaction: discord.Interaction, current: str):
|
|
||||||
return [
|
|
||||||
app_commands.Choice(name=prio, value=prio)
|
|
||||||
for prio in PRIORITES if current.lower() in prio.lower()
|
|
||||||
][:5]
|
|
||||||
|
|
||||||
@app_commands.command(name="nouvelle_tache", description="Créer une nouvelle tâche")
|
|
||||||
@app_commands.describe(
|
|
||||||
titre="Titre de la tâche",
|
|
||||||
description="Description de la tâche",
|
|
||||||
assigne="Utilisateur assigné",
|
|
||||||
priorite="Niveau de priorité (ex: Haute, Moyenne, Basse)",
|
|
||||||
echeance="Date d'échéance au format JJ/MM/AAAA"
|
|
||||||
)
|
|
||||||
async def nouvelle_tache(
|
|
||||||
self, interaction: discord.Interaction, titre: str, description: str,
|
|
||||||
assigne: discord.User, priorite: str, echeance: str
|
|
||||||
):
|
|
||||||
data = await charger_taches()
|
|
||||||
for tid, tache in data.get("taches", {}).items():
|
|
||||||
if tache["assigné"] == assigne.id and tache["titre"].lower() == titre.lower():
|
|
||||||
return await interaction.response.send_message(
|
|
||||||
f"❌ {assigne.mention} a déjà une tâche avec ce titre.", ephemeral=True)
|
|
||||||
|
|
||||||
try:
|
|
||||||
date_obj = datetime.strptime(echeance, "%d/%m/%Y")
|
|
||||||
date_str = date_obj.strftime("%Y-%m-%d")
|
|
||||||
except ValueError:
|
|
||||||
return await interaction.response.send_message(
|
|
||||||
"❌ Format de date invalide (JJ/MM/AAAA).", ephemeral=True)
|
|
||||||
|
|
||||||
id_tache = await ajouter_tache(titre, description, assigne.id, priorite, date_str, interaction.user.id)
|
|
||||||
|
|
||||||
embed = discord.Embed(
|
|
||||||
title=f"📌 Nouvelle Tâche #{id_tache}",
|
|
||||||
description=description,
|
|
||||||
color=discord.Color.blurple()
|
|
||||||
)
|
|
||||||
embed.set_author(name=f"Assignée à {assigne.display_name}", icon_url=assigne.display_avatar.url)
|
|
||||||
embed.add_field(name="🗂 Titre", value=titre, inline=False)
|
|
||||||
embed.add_field(name="⚠️ Priorité", value=priorite, inline=True)
|
|
||||||
embed.add_field(name="📅 Échéance", value=echeance, inline=True)
|
|
||||||
embed.set_footer(text=f"Créée par {interaction.user.display_name}", icon_url=interaction.user.display_avatar.url)
|
|
||||||
|
|
||||||
await interaction.response.send_message(embed=embed)
|
|
||||||
|
|
||||||
try:
|
|
||||||
await assigne.send(
|
|
||||||
f"👋 Bonjour {assigne.display_name},\n\nUne nouvelle tâche t'a été assignée par {interaction.user.mention} :",
|
|
||||||
embed=embed
|
|
||||||
)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
@nouvelle_tache.autocomplete("priorite")
|
|
||||||
async def autocomplete_priorite_handler(self, interaction: discord.Interaction, current: str):
|
|
||||||
return await self.autocomplete_priorite(interaction, current)
|
|
||||||
|
|
||||||
@app_commands.command(name="mes_taches", description="Lister mes tâches assignées")
|
|
||||||
async def mes_taches(self, interaction: discord.Interaction):
|
|
||||||
taches = await lister_taches_utilisateur(interaction.user.id)
|
|
||||||
if not taches:
|
|
||||||
await interaction.response.send_message("🚫 Aucune tâche ne vous est assignée.", ephemeral=True)
|
|
||||||
return
|
|
||||||
|
|
||||||
taches_list = list(taches.items())
|
|
||||||
if len(taches_list) <= 5:
|
|
||||||
embed = discord.Embed(title="📝 Vos Tâches Assignées", color=discord.Color.green())
|
|
||||||
for tid, t in taches_list:
|
|
||||||
embed.add_field(
|
|
||||||
name=f"#{tid} - {t['titre']}",
|
|
||||||
value=f"📅 Échéance : {t['échéance']}\n⚠️ Priorité : {t['priorité']}\n📌 Statut : {t['statut']}",
|
|
||||||
inline=False
|
|
||||||
)
|
|
||||||
await interaction.response.send_message(embed=embed, ephemeral=True)
|
|
||||||
else:
|
|
||||||
view = MesTachesPaginator(interaction, taches_list)
|
|
||||||
await interaction.response.send_message(embed=view.get_embed(), view=view, ephemeral=True)
|
|
||||||
|
|
||||||
@app_commands.command(name="update_tache", description="Mettre à jour le statut et la priorité d'une tâche")
|
|
||||||
@app_commands.describe(
|
|
||||||
id_tache="ID de la tâche à mettre à jour",
|
|
||||||
statut="Nouveau statut de la tâche",
|
|
||||||
priorite="Nouvelle priorité (optionnel)"
|
|
||||||
)
|
|
||||||
async def update_tache(
|
|
||||||
self, interaction: discord.Interaction, id_tache: str, statut: str, priorite: str = None
|
|
||||||
):
|
|
||||||
data = await charger_taches()
|
|
||||||
if id_tache not in data.get("taches", {}):
|
|
||||||
return await interaction.response.send_message("❌ Tâche non trouvée.", ephemeral=True)
|
|
||||||
|
|
||||||
tache = data["taches"][id_tache]
|
|
||||||
if tache["assigné"] != interaction.user.id:
|
|
||||||
return await interaction.response.send_message("❌ Vous ne pouvez pas mettre à jour le statut de cette tâche.", ephemeral=True)
|
|
||||||
|
|
||||||
statut_update = await update_tache_status(id_tache, statut)
|
|
||||||
priorite_update = True
|
|
||||||
if priorite:
|
|
||||||
tache["priorité"] = priorite
|
|
||||||
await sauvegarder_taches(data)
|
|
||||||
|
|
||||||
if statut_update and priorite_update:
|
|
||||||
await interaction.response.send_message(
|
|
||||||
f"✅ Le statut de la tâche #{id_tache} a été mis à jour à **{statut}**." +
|
|
||||||
(f"\nLa priorité est maintenant **{priorite}**." if priorite else ""),
|
|
||||||
ephemeral=True
|
|
||||||
)
|
|
||||||
if tache["créé_par"] != interaction.user.id:
|
|
||||||
try:
|
|
||||||
creator = await self.bot.fetch_user(tache["créé_par"])
|
|
||||||
await creator.send(
|
|
||||||
f"🔔 La tâche #{id_tache} que vous avez créée a été mise à jour par {interaction.user.mention} :\n"
|
|
||||||
f"**Nouveau statut** : {statut}" +
|
|
||||||
(f"\n**Nouvelle priorité** : {priorite}" if priorite else "")
|
|
||||||
)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
else:
|
|
||||||
await interaction.response.send_message("❌ Erreur lors de la mise à jour de la tâche.", ephemeral=True)
|
|
||||||
|
|
||||||
@update_tache.autocomplete("priorite")
|
|
||||||
async def autocomplete_priorite_update(self, interaction: discord.Interaction, current: str):
|
|
||||||
return await self.autocomplete_priorite(interaction, current)
|
|
||||||
|
|
||||||
@app_commands.command(name="supprimer_tache", description="Supprimer une tâche")
|
|
||||||
@app_commands.describe(
|
|
||||||
id_tache="ID de la tâche à supprimer"
|
|
||||||
)
|
|
||||||
async def supprimer_tache_cmd(self, interaction: discord.Interaction, id_tache: str):
|
|
||||||
data = await charger_taches()
|
|
||||||
if id_tache not in data.get("taches", {}):
|
|
||||||
return await interaction.response.send_message("❌ Tâche non trouvée.", ephemeral=True)
|
|
||||||
|
|
||||||
tache = data["taches"][id_tache]
|
|
||||||
perms = interaction.user.guild_permissions
|
|
||||||
can_delete = (
|
|
||||||
tache["créé_par"] == interaction.user.id or
|
|
||||||
tache["assigné"] == interaction.user.id or
|
|
||||||
perms.administrator
|
|
||||||
)
|
|
||||||
if not can_delete:
|
|
||||||
return await interaction.response.send_message("❌ Vous ne pouvez pas supprimer cette tâche.", ephemeral=True)
|
|
||||||
|
|
||||||
success = await supprimer_tache(id_tache)
|
|
||||||
if success:
|
|
||||||
await interaction.response.send_message(f"🗑️ La tâche #{id_tache} a été supprimée.", ephemeral=True)
|
|
||||||
else:
|
|
||||||
await interaction.response.send_message("❌ Erreur lors de la suppression de la tâche.", ephemeral=True)
|
|
||||||
|
|
||||||
@tasks.loop(seconds=30)
|
|
||||||
async def rappels(self):
|
|
||||||
now = datetime.now(ZoneInfo("Europe/Paris"))
|
|
||||||
if now.hour == 9 and now.minute == 0:
|
|
||||||
taches = await taches_proches()
|
|
||||||
for tid, t in taches.items():
|
|
||||||
try:
|
|
||||||
user = await self.bot.fetch_user(t["assigné"])
|
|
||||||
embed = discord.Embed(
|
|
||||||
title=f"🔔 Rappel Tâche #{tid}",
|
|
||||||
description=t["description"],
|
|
||||||
color=discord.Color.orange()
|
|
||||||
)
|
|
||||||
embed.add_field(name="🗂 Titre", value=t["titre"], inline=False)
|
|
||||||
embed.add_field(name="⚠️ Priorité", value=t["priorité"], inline=True)
|
|
||||||
embed.add_field(name="📅 Échéance", value=t["échéance"], inline=True)
|
|
||||||
embed.set_footer(text="Clique sur le bouton si tu as fini !")
|
|
||||||
view = RappelTacheView(tid)
|
|
||||||
await user.send(embed=embed, view=view)
|
|
||||||
except Exception:
|
|
||||||
continue
|
|
||||||
|
|
||||||
@rappels.before_loop
|
|
||||||
async def avant_rappel(self):
|
|
||||||
await self.bot.wait_until_ready()
|
|
||||||
|
|
||||||
async def setup(bot):
|
|
||||||
await bot.add_cog(JiraDiscord(bot))
|
|
||||||
|
|
@ -8,6 +8,7 @@
|
||||||
"rules_channel_id": 1369669999677145101,
|
"rules_channel_id": 1369669999677145101,
|
||||||
"rules_accept_role_id": 1369769836028231730,
|
"rules_accept_role_id": 1369769836028231730,
|
||||||
"rules_message_type": "embed",
|
"rules_message_type": "embed",
|
||||||
"role_restore_enabled": true
|
"role_restore_enabled": true,
|
||||||
|
"server_logs": true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -1,210 +0,0 @@
|
||||||
import unittest
|
|
||||||
from unittest.mock import AsyncMock, MagicMock, patch
|
|
||||||
import os
|
|
||||||
import json
|
|
||||||
import asyncio
|
|
||||||
|
|
||||||
# On injecte les vrais IDs à 18 chiffres de ton .env pour le test
|
|
||||||
os.environ["STARTER_SKU_ID"] = "111111111111111111"
|
|
||||||
os.environ["MEDIUM_SKU_ID"] = "222222222222222222"
|
|
||||||
os.environ["LARGE_SKU_ID"] = "333333333333333333"
|
|
||||||
|
|
||||||
from utils.premium import load_partners_whitelist, get_sku_ids, check_premium_tier, WHITELIST_PARTNERS_FILE
|
|
||||||
|
|
||||||
class TestPremiumSystem(unittest.IsolatedAsyncioTestCase):
|
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
# Nettoyage du fichier JSON avant chaque test
|
|
||||||
if os.path.exists(WHITELIST_PARTNERS_FILE):
|
|
||||||
try:
|
|
||||||
os.remove(WHITELIST_PARTNERS_FILE)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def tearDown(self):
|
|
||||||
# Nettoyage du fichier JSON après chaque test
|
|
||||||
if os.path.exists(WHITELIST_PARTNERS_FILE):
|
|
||||||
try:
|
|
||||||
os.remove(WHITELIST_PARTNERS_FILE)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
async def test_get_sku_ids(self):
|
|
||||||
"""Vérifie que la lecture du .env convertit bien les chaînes en nombres entiers (int)"""
|
|
||||||
starter, medium, large = get_sku_ids()
|
|
||||||
self.assertEqual(starter, 111111111111111111)
|
|
||||||
self.assertEqual(medium, 222222222222222222)
|
|
||||||
self.assertEqual(large, 333333333333333333)
|
|
||||||
|
|
||||||
async def test_load_partners_whitelist_empty(self):
|
|
||||||
"""Vérifie que le système gère proprement un fichier whitelist vide sans crasher"""
|
|
||||||
partners = await load_partners_whitelist()
|
|
||||||
self.assertEqual(partners, [])
|
|
||||||
self.assertTrue(os.path.exists(WHITELIST_PARTNERS_FILE))
|
|
||||||
|
|
||||||
async def test_load_partners_whitelist_with_data(self):
|
|
||||||
"""Vérifie le chargement des partenaires depuis le fichier JSON"""
|
|
||||||
with open(WHITELIST_PARTNERS_FILE, "w") as f:
|
|
||||||
json.dump({"partners": [123, 456]}, f)
|
|
||||||
partners = await load_partners_whitelist()
|
|
||||||
self.assertEqual(partners, [123, 456])
|
|
||||||
|
|
||||||
async def test_check_premium_tier_dm_fails(self):
|
|
||||||
"""Vérifie que le bot bloque proprement les commandes exécutées en DM (hors serveur)"""
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = None
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertFalse(result)
|
|
||||||
interaction.response.send_message.assert_called_once_with(
|
|
||||||
"❌ Cette commande ne peut être exécutée que sur un serveur Discord.",
|
|
||||||
ephemeral=True
|
|
||||||
)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_whitelist_passes(self):
|
|
||||||
"""Vérifie qu'un serveur dans la Whitelist passe sans condition, même sans SKU"""
|
|
||||||
with open(WHITELIST_PARTNERS_FILE, "w") as f:
|
|
||||||
json.dump({"partners": [12345]}, f)
|
|
||||||
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 12345
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertTrue(result)
|
|
||||||
interaction.response.send_message.assert_not_called()
|
|
||||||
|
|
||||||
async def test_check_premium_tier_invalid_member_count_fails(self):
|
|
||||||
"""Vérifie la sécurité (Fail-Safe) si le nombre de membres Discord renvoie une erreur"""
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = None
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertFalse(result)
|
|
||||||
interaction.response.send_message.assert_called_once_with(
|
|
||||||
"❌ Impossible de vérifier la taille du serveur par sécurité. Veuillez réessayer plus tard.",
|
|
||||||
ephemeral=True
|
|
||||||
)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_less_than_500_with_starter(self):
|
|
||||||
"""Vérifie qu'un petit serveur (<500 membres) avec le SKU Starter est bien AUTORISÉ"""
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 100
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 111111111111111111
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertTrue(result)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_500_to_2500_with_starter_only_fails(self):
|
|
||||||
"""Vérifie qu'un serveur moyen qui a grandi (>500 membres) est BLOQUÉ s'il n'a qu'un SKU Starter"""
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 1000
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 111111111111111111
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertFalse(result)
|
|
||||||
interaction.response.send_message.assert_called_once()
|
|
||||||
self.assertIn("Palier Medium", interaction.response.send_message.call_args[0][0])
|
|
||||||
|
|
||||||
async def test_check_premium_tier_500_to_2500_with_medium_passes(self):
|
|
||||||
"""Vérifie qu'un serveur moyen (1000 membres) avec le SKU Medium est bien AUTORISÉ"""
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 1000
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 222222222222222222
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertTrue(result)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_above_2500_with_large_passes(self):
|
|
||||||
"""Vérifie qu'un très gros serveur (>2500 membres) avec le SKU Large est bien AUTORISÉ"""
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 3000
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 333333333333333333
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertTrue(result)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_above_2500_with_medium_fails(self):
|
|
||||||
"""Vérifie qu'un très gros serveur (>2500 membres) est BLOQUÉ s'il n'a qu'un SKU Medium"""
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 3000
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 222222222222222222
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertFalse(result)
|
|
||||||
self.assertIn("Palier Large", interaction.response.send_message.call_args[0][0])
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
200
test_premium.py
200
test_premium.py
|
|
@ -1,200 +0,0 @@
|
||||||
import unittest
|
|
||||||
from unittest.mock import AsyncMock, MagicMock, patch
|
|
||||||
import os
|
|
||||||
import json
|
|
||||||
import asyncio
|
|
||||||
|
|
||||||
# Setup env variables before importing premium module
|
|
||||||
os.environ["STARTER_SKU_ID"] = "111111"
|
|
||||||
os.environ["MEDIUM_SKU_ID"] = "222222"
|
|
||||||
os.environ["LARGE_SKU_ID"] = "333333"
|
|
||||||
|
|
||||||
from utils.premium import load_partners_whitelist, get_sku_ids, check_premium_tier, WHITELIST_PARTNERS_FILE
|
|
||||||
|
|
||||||
class TestPremiumSystem(unittest.IsolatedAsyncioTestCase):
|
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
# Reset whitelist file before each test
|
|
||||||
if os.path.exists(WHITELIST_PARTNERS_FILE):
|
|
||||||
try:
|
|
||||||
os.remove(WHITELIST_PARTNERS_FILE)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def tearDown(self):
|
|
||||||
# Clean up whitelist file
|
|
||||||
if os.path.exists(WHITELIST_PARTNERS_FILE):
|
|
||||||
try:
|
|
||||||
os.remove(WHITELIST_PARTNERS_FILE)
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
async def test_get_sku_ids(self):
|
|
||||||
starter, medium, large = get_sku_ids()
|
|
||||||
self.assertEqual(starter, 111111)
|
|
||||||
self.assertEqual(medium, 222222)
|
|
||||||
self.assertEqual(large, 333333)
|
|
||||||
|
|
||||||
async def test_load_partners_whitelist_empty(self):
|
|
||||||
partners = await load_partners_whitelist()
|
|
||||||
self.assertEqual(partners, [])
|
|
||||||
self.assertTrue(os.path.exists(WHITELIST_PARTNERS_FILE))
|
|
||||||
|
|
||||||
async def test_load_partners_whitelist_with_data(self):
|
|
||||||
with open(WHITELIST_PARTNERS_FILE, "w") as f:
|
|
||||||
json.dump({"partners": [123, 456]}, f)
|
|
||||||
partners = await load_partners_whitelist()
|
|
||||||
self.assertEqual(partners, [123, 456])
|
|
||||||
|
|
||||||
async def test_check_premium_tier_dm_fails(self):
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = None
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertFalse(result)
|
|
||||||
interaction.response.send_message.assert_called_once_with(
|
|
||||||
"❌ Cette commande ne peut être exécutée que sur un serveur Discord.",
|
|
||||||
ephemeral=True
|
|
||||||
)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_whitelist_passes(self):
|
|
||||||
with open(WHITELIST_PARTNERS_FILE, "w") as f:
|
|
||||||
json.dump({"partners": [12345]}, f)
|
|
||||||
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 12345
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertTrue(result)
|
|
||||||
interaction.response.send_message.assert_not_called()
|
|
||||||
|
|
||||||
async def test_check_premium_tier_invalid_member_count_fails(self):
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = None
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertFalse(result)
|
|
||||||
interaction.response.send_message.assert_called_once_with(
|
|
||||||
"❌ Impossible de vérifier la taille du serveur par sécurité. Veuillez réessayer plus tard.",
|
|
||||||
ephemeral=True
|
|
||||||
)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_less_than_500_with_starter(self):
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 100
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
# Mock entitlements
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 111111
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertTrue(result)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_500_to_2500_with_starter_only_fails(self):
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 1000
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 111111
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertFalse(result)
|
|
||||||
interaction.response.send_message.assert_called_once()
|
|
||||||
self.assertIn("Palier Medium", interaction.response.send_message.call_args[0][0])
|
|
||||||
|
|
||||||
async def test_check_premium_tier_500_to_2500_with_medium_passes(self):
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 1000
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 222222
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertTrue(result)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_above_2500_with_large_passes(self):
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 3000
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 333333
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertTrue(result)
|
|
||||||
|
|
||||||
async def test_check_premium_tier_above_2500_with_medium_fails(self):
|
|
||||||
interaction = MagicMock()
|
|
||||||
interaction.guild = MagicMock()
|
|
||||||
interaction.guild.id = 999
|
|
||||||
interaction.guild.member_count = 3000
|
|
||||||
interaction.response = MagicMock()
|
|
||||||
interaction.response.send_message = AsyncMock()
|
|
||||||
|
|
||||||
entitlement = MagicMock()
|
|
||||||
entitlement.sku_id = 222222
|
|
||||||
entitlement.is_active.return_value = True
|
|
||||||
interaction.entitlements = [entitlement]
|
|
||||||
|
|
||||||
decorator = check_premium_tier()
|
|
||||||
predicate = decorator.predicate
|
|
||||||
|
|
||||||
result = await predicate(interaction)
|
|
||||||
self.assertFalse(result)
|
|
||||||
self.assertIn("Palier Large", interaction.response.send_message.call_args[0][0])
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
|
|
@ -1,5 +1,10 @@
|
||||||
{
|
{
|
||||||
"partners": [
|
"partners": [
|
||||||
1369669999345537145
|
1369669999345537145,
|
||||||
|
1277303870657400882,
|
||||||
|
1366461904976871565,
|
||||||
|
1313868393513881622,
|
||||||
|
1376435153731326084,
|
||||||
|
1370362952938688573
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
Loading…
Add table
Add a link
Reference in a new issue