from future import annotations

import threading
import unittest
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Dict, Hashable, Optional

@dataclass(frozen=True)
class TenantConfig:
"""Limites de débit pour un locataire.

client_limit s'applique indépendamment à chaque client, tandis que tenant_limit s'applique
à toutes les requêtes admises du locataire. Les deux limites utilisent des fenêtres glissantes exactes.
active_timeout_ms limite la durée pendant laquelle un attenteur d'équité absent peut conserver son tour.
eviction_ms contrôle la rétention de l'état du client.
"""

client_limit: int
client_window_ms: int
tenant_limit: int
tenant_window_ms: int
active_timeout_ms: int = 60_000
eviction_ms: int = 120_000

def validate(self) -> None:
    if self.client_limit <= 0 or self.tenant_limit <= 0:
        raise ValueError("les limites doivent être positives")
    if self.client_window_ms <= 0 or self.tenant_window_ms <= 0:
        raise ValueError("les longueurs de fenêtre doivent être positives")
    if self.active_timeout_ms <= 0:
        raise ValueError("active_timeout_ms doit être positif")
    if self.eviction_ms < max(self.client_window_ms, self.tenant_window_ms):
        raise ValueError("eviction_ms doit être au moins aussi grand que les deux fenêtres")
    if self.eviction_ms < self.active_timeout_ms:
        raise ValueError("eviction_ms doit être au moins active_timeout_ms")

@dataclass(frozen=True)
class AllowResult:
allowed: bool
retry_after_ms: int = 0

@dataclass
class _ClientState:
admitted: Deque[int] = field(default_factory=deque)
last_seen_ms: int = 0

@dataclass
class _TenantState:
admitted: Deque[int] = field(default_factory=deque)
clients: Dict[Hashable, _ClientState] = field(default_factory=dict)
waiters: Deque[Hashable] = field(default_factory=deque)
waiter_set: set[Hashable] = field(default_factory=set)
last_now_ms: Optional[int] = None
last_seen_ms: int = 0

class SlidingWindowRateLimiter:
"""Limiteur de débit à fenêtre glissante, thread-safe, mono-processus.

Algorithme de fenêtre glissante :
  Chaque horodatage de requête admise est stocké dans une deque. Avant d'évaluer une
  requête, les horodatages t satisfaisant t <= now - window sont supprimés. Ainsi une
  requête vieille d'une fenêtre complète ne consomme plus de capacité. Il s'agit d'une
  véritable fenêtre glissante basée sur les journaux, elle n'a donc pas de rafale de limite de compartiment calendaire.

  Le résultat est exact, avec des opérations de deque amorties en O(1) et une mémoire
  proportionnelle aux requêtes admises encore dans les fenêtres configurées. C'est
  plus précis que les compartiments pondérés mais utilise plus de mémoire. Les limites fournissent une
  limite stricte aux entrées d'horodatage actives, et les objets clients obsolètes sont évincés.

Partage équitable :
  La capacité de locataire non contestée est économe en travail et peut être utilisée par n'importe quel
  client. Une fois le plafond du locataire atteint, les clients refusés entrent dans une file FIFO
  avec au plus une entrée par client. Au fur et à mesure que la capacité expire, la tête obtient
  l'admission suivante, puis quitte la file. Un client continuellement occupé doit
  se réinscrire derrière d'autres prétendants, implémentant une redistribution round-robin
  sans réserver de manière permanente des parts inutilisées par client. Les entrées de file inactives
  expirent après active_timeout_ms afin qu'un tour abandonné ne puisse pas bloquer
  le locataire indéfiniment.

Comportement de l'horloge :
  now_ms doit provenir d'une horloge monotone. Si un appelant fournit un horodatage dupliqué
  ou décroissant, le temps est limité à l'horodatage le plus élevé observé par le locataire.
  Cela empêche l'historique expiré de redevenir actif.

Concurrence et hypothèses de déploiement :
  Un seul verrou réentrant rend l'autorisation, la configuration et le nettoyage atomiques
  entre les threads et les tâches asynchrones partageant cet objet. Pour une utilisation distribuée,
  la même transition d'état peut être implémentée atomiquement dans Redis avec un
  script Lua et des ensembles triés, ou dans un magasin fortement cohérent transactionnel.
  Les métadonnées de l'attenteur FIFO doivent être mises à jour dans la même transaction que les
  journaux de requêtes ; les verrous locaux seuls ne coordonnent pas plusieurs processus.
"""

def __init__(self, default_config: Optional[TenantConfig] = None) -> None:
    if default_config is not None:
        default_config.validate()
    self._default_config = default_config
    self._configs: Dict[Hashable, TenantConfig] = {}
    self._tenants: Dict[Hashable, _TenantState] = {}
    self._lock = threading.RLock()

def configure_tenant(self, tenant_id: Hashable, config: TenantConfig) -> None:
    """Crée ou remplace la configuration d'un locataire de manière atomique."""
    config.validate()
    with self._lock:
        self._configs[tenant_id] = config

def allow(self, tenant_id: Hashable, client_id: Hashable, now_ms: int) -> AllowResult:
    """Décide de manière atomique s'il faut admettre une requête.

    retry_after_ms est exact pour un refus direct client/fenêtre ou locataire/fenêtre.
    Pour un refus d'équité FIFO, il s'agit du premier délai de nouvelle tentative temporel ;
    l'admission réelle dépend également des attenteurs actifs précédents qui prennent ou
    abandonnent leurs tours.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms doit être un entier")

    with self._lock:
        cfg = self._config_for(tenant_id)
        tenant = self._tenants.get(tenant_id)
        if tenant is None:
            tenant = _TenantState(last_now_ms=now_ms, last_seen_ms=now_ms)
            self._tenants[tenant_id] = tenant

        now = self._normalize_time(tenant, now_ms)
        tenant.last_seen_ms = now
        self._prune(tenant.admitted, now, cfg.tenant_window_ms)

        client = tenant.clients.get(client_id)
        if client is None:
            client = _ClientState(last_seen_ms=now)
            tenant.clients[client_id] = client
        client.last_seen_ms = now
        self._prune(client.admitted, now, cfg.client_window_ms)

        self._purge_waiters(tenant, cfg, now)
        self._evict_clients(tenant, cfg, now, preserve=client_id)

        if len(client.admitted) >= cfg.client_limit:
            self._remove_waiter(tenant, client_id)
            retry = client.admitted[0] + cfg.client_window_ms - now
            return AllowResult(False, max(0, retry))

        if len(tenant.admitted) >= cfg.tenant_limit:
            self._enqueue_waiter(tenant, client_id)
            retry = tenant.admitted[0] + cfg.tenant_window_ms - now
            return AllowResult(False, max(0, retry))

        # La capacité existe. En cas de contention, seule la tête FIFO peut l'utiliser.
        if tenant.waiters and tenant.waiters[0] != client_id:
            self._enqueue_waiter(tenant, client_id)
            return AllowResult(False, 0)

        if tenant.waiters and tenant.waiters[0] == client_id:
            tenant.waiters.popleft()
            tenant.waiter_set.remove(client_id)

        client.admitted.append(now)
        tenant.admitted.append(now)
        return AllowResult(True, 0)

def sweep(self, now_ms: int) -> int:
    """Évince les clients obsolètes et nettoie les états d'exécution des locataires.

    Les applications peuvent appeler cela périodiquement lorsque le trafic est faible. Les appels
    normaux à allow nettoient également le locataire accédé. Retourne le nombre d'objets clients
    et locataires d'exécution supprimés. Les configurations sont conservées.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms doit être un entier")

    removed = 0
    with self._lock:
        for tenant_id in list(self._tenants):
            cfg = self._config_for(tenant_id)
            tenant = self._tenants[tenant_id]
            now = self._normalize_time(tenant, now_ms)
            self._prune(tenant.admitted, now, cfg.tenant_window_ms)
            self._purge_waiters(tenant, cfg, now)
            before = len(tenant.clients)
            self._evict_clients(tenant, cfg, now, preserve=None)
            removed += before - len(tenant.clients)

            if (not tenant.clients and not tenant.admitted and not tenant.waiters
                    and now - tenant.last_seen_ms >= cfg.eviction_ms):
                del self._tenants[tenant_id]
                removed += 1
    return removed

def debug_state(self, tenant_id: Hashable) -> dict[str, int]:
    """Retourne les tailles d'état agrégées, utiles pour la surveillance et les tests."""
    with self._lock:
        tenant = self._tenants.get(tenant_id)
        if tenant is None:
            return {"clients": 0, "tenant_events": 0, "waiters": 0,
                    "client_events": 0}
        return {
            "clients": len(tenant.clients),
            "tenant_events": len(tenant.admitted),
            "waiters": len(tenant.waiters),
            "client_events": sum(len(c.admitted) for c in tenant.clients.values()),
        }

def _config_for(self, tenant_id: Hashable) -> TenantConfig:
    cfg = self._configs.get(tenant_id, self._default_config)
    if cfg is None:
        raise KeyError(f"le locataire {tenant_id!r} n'a pas de configuration")
    return cfg

@staticmethod
def _normalize_time(tenant: _TenantState, supplied: int) -> int:
    if tenant.last_now_ms is None or supplied >= tenant.last_now_ms:
        tenant.last_now_ms = supplied
        return supplied
    return tenant.last_now_ms

@staticmethod
def _prune(events: Deque[int], now: int, window_ms: int) -> None:
    boundary = now - window_ms
    while events and events[0] <= boundary:
        events.popleft()

@staticmethod
def _enqueue_waiter(tenant: _TenantState, client_id: Hashable) -> None:
    if client_id not in tenant.waiter_set:
        tenant.waiters.append(client_id)
        tenant.waiter_set.add(client_id)

@staticmethod
def _remove_waiter(tenant: _TenantState, client_id: Hashable) -> None:
    if client_id not in tenant.waiter_set:
        return
    tenant.waiters = deque(x for x in tenant.waiters if x != client_id)
    tenant.waiter_set.remove(client_id)

def _purge_waiters(self, tenant: _TenantState, cfg: TenantConfig, now: int) -> None:
    kept: Deque[Hashable] = deque()
    kept_set: set[Hashable] = set()
    for client_id in tenant.waiters:
        client = tenant.clients.get(client_id)
        if client is None or now - client.last_seen_ms >= cfg.active_timeout_ms:
            continue
        self._prune(client.admitted, now, cfg.client_window_ms)
        # Un client bloqué par son propre quota ne réserve pas un tour de locataire.
        if len(client.admitted) >= cfg.client_limit:
            continue
        kept.append(client_id)
        kept_set.add(client_id)
    tenant.waiters = kept
    tenant.waiter_set = kept_set

@staticmethod
def _evict_clients(tenant: _TenantState, cfg: TenantConfig, now: int,
                   preserve: Optional[Hashable]) -> None:
    for client_id in list(tenant.clients):
        if preserve is not None and client_id == preserve:
            continue
        client = tenant.clients[client_id]
        boundary = now - cfg.client_window_ms
        while client.admitted and client.admitted[0] <= boundary:
            client.admitted.popleft()
        if (not client.admitted and client_id not in tenant.waiter_set
                and now - client.last_seen_ms >= cfg.eviction_ms):
            del tenant.clients[client_id]

class SlidingWindowRateLimiterTests(unittest.TestCase):
def make_limiter(self, **overrides: int) -> SlidingWindowRateLimiter:
values = dict(client_limit=3, client_window_ms=1000,
tenant_limit=20, tenant_window_ms=1000,
active_timeout_ms=1000, eviction_ms=2000)
values.update(overrides)
return SlidingWindowRateLimiter(TenantConfig(**values))

def test_exact_window_boundary_is_expired(self) -> None:
    limiter = self.make_limiter(client_limit=1)
    self.assertTrue(limiter.allow("t", "c", 0).allowed)
    denied = limiter.allow("t", "c", 999)
    self.assertFalse(denied.allowed)
    self.assertEqual(denied.retry_after_ms, 1)
    self.assertTrue(limiter.allow("t", "c", 1000).allowed)

def test_idle_client_returns_after_full_window(self) -> None:
    limiter = self.make_limiter(client_limit=2)
    self.assertTrue(limiter.allow("t", "c", 10).allowed)
    self.assertTrue(limiter.allow("t", "c", 11).allowed)
    self.assertFalse(limiter.allow("t", "c", 12).allowed)
    self.assertTrue(limiter.allow("t", "c", 1011).allowed)

def test_duplicate_and_backwards_clock_are_clamped(self) -> None:
    limiter = self.make_limiter(client_limit=2)
    self.assertTrue(limiter.allow("t", "c", 100).allowed)
    self.assertTrue(limiter.allow("t", "c", 100).allowed)
    denied = limiter.allow("t", "c", 50)
    self.assertFalse(denied.allowed)
    self.assertEqual(denied.retry_after_ms, 1000)
    self.assertTrue(limiter.allow("t", "c", 1100).allowed)

def test_concurrent_requests_are_atomic(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=100)
    barrier = threading.Barrier(40)
    results: list[bool] = []
    results_lock = threading.Lock()

    def worker() -> None:
        barrier.wait()
        value = limiter.allow("t", "same-client", 500).allowed
        with results_lock:
            results.append(value)

    threads = [threading.Thread(target=worker) for _ in range(40)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    self.assertEqual(sum(results), 10)
    self.assertEqual(len(results), 40)

def test_tenant_cap_combines_all_clients(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=2)
    self.assertTrue(limiter.allow("t", "a", 0).allowed)
    self.assertTrue(limiter.allow("t", "b", 1).allowed)
    denied = limiter.allow("t", "c", 2)
    self.assertFalse(denied.allowed)
    self.assertEqual(denied.retry_after_ms, 998)

def test_saturated_tenant_uses_fifo_fair_redistribution(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=2)
    self.assertTrue(limiter.allow("t", "a", 0).allowed)
    self.assertTrue(limiter.allow("t", "a", 1).allowed)

    self.assertFalse(limiter.allow("t", "b", 2).allowed)
    self.assertFalse(limiter.allow("t", "c", 3).allowed)

    # Les deux anciens slots expirent, mais a ne peut pas les monopoliser : b et c sont en file d'attente.
    self.assertFalse(limiter.allow("t", "a", 1001).allowed)
    self.assertTrue(limiter.allow("t", "b", 1001).allowed)
    self.assertFalse(limiter.allow("t", "b", 1001).allowed)
    self.assertTrue(limiter.allow("t", "c", 1001).allowed)

def test_abandoned_fairness_turn_eventually_expires(self) -> None:
    limiter = self.make_limiter(client_limit=10, tenant_limit=1,
                                active_timeout_ms=1000, eviction_ms=2000)
    self.assertTrue(limiter.allow("t", "a", 0).allowed)
    self.assertFalse(limiter.allow("t", "gone", 1).allowed)
    # À 1001, l'attenteur est inactif, donc un autre client peut utiliser la capacité.
    self.assertTrue(limiter.allow("t", "b", 1001).allowed)

def test_eviction_removes_stale_but_not_active_clients(self) -> None:
    limiter = self.make_limiter(client_window_ms=100, tenant_window_ms=100,
                                active_timeout_ms=100, eviction_ms=200)
    self.assertTrue(limiter.allow("t", "stale", 0).allowed)
    self.assertTrue(limiter.allow("t", "active", 150).allowed)

    limiter.sweep(201)
    state = limiter.debug_state("t")
    self.assertEqual(state["clients"], 1)
    self.assertEqual(state["client_events"], 1)

    # La requête active n'est pas abandonnée avant l'expiration de sa propre fenêtre.
    self.assertFalse(limiter.allow("t", "active", 201).allowed
                     if limiter._default_config.client_limit == 1 else False)
    self.assertEqual(limiter.debug_state("t")["clients"], 1)
    limiter.sweep(351)
    self.assertEqual(limiter.debug_state("t")["clients"], 0)

def test_configuration_is_per_tenant(self) -> None:
    limiter = SlidingWindowRateLimiter()
    limiter.configure_tenant("small", TenantConfig(1, 100, 1, 100, 100, 100))
    limiter.configure_tenant("large", TenantConfig(2, 100, 3, 100, 100, 100))
    self.assertTrue(limiter.allow("small", "c", 0).allowed)
    self.assertFalse(limiter.allow("small", "c", 1).allowed)
    self.assertTrue(limiter.allow("large", "c", 0).allowed)
    self.assertTrue(limiter.allow("large", "c", 1).allowed)

def test_memory_is_bounded_by_live_limits_and_sweep(self) -> None:
    limiter = self.make_limiter(client_limit=5, tenant_limit=7,
                                client_window_ms=100, tenant_window_ms=100,
                                active_timeout_ms=100, eviction_ms=100)
    for index in range(100):
        limiter.allow("t", f"c{index}", 0)
    state = limiter.debug_state("t")
    self.assertLessEqual(state["tenant_events"], 7)
    self.assertLessEqual(state["client_events"], 7)
    limiter.sweep(100)
    self.assertEqual(limiter.debug_state("t")["clients"], 0)

if name == "main":
unittest.main()

import time
import threading
from collections import deque
import unittest
from typing import Dict, Tuple, Optional

Classe de données auxiliaire pour la configuration d’un locataire

class TenantConfig:
def init(self, client_rate: int, client_window_ms: int, tenant_rate: int, tenant_window_ms: int):
self.client_rate = client_rate
self.client_window_ms = client_window_ms
self.tenant_rate = tenant_rate
self.tenant_window_ms = tenant_window_ms

class SlidingWindowRateLimiter:
"""
Un limiteur de débit multi-locataire, sûr pour les threads, utilisant un algorithme de journal à fenêtre glissante.

Cette implémentation fournit une limitation de débit par client sous un plafond global par locataire.
Elle utilise une politique de partage équitable dans laquelle aucun client unique ne peut épuiser le quota du locataire
grâce à sa propre limite individuelle.

Algorithme : journal à fenêtre glissante
- Pour chaque client et locataire, nous maintenons une deque (un conteneur de type liste avec des ajouts
  et suppressions rapides aux deux extrémités) des horodatages des requêtes reçues dans la fenêtre.
- Lorsqu’une nouvelle requête arrive, nous supprimons d’abord tous les horodatages plus anciens que l’heure actuelle
  moins la taille de la fenêtre.
- Ensuite, nous vérifions si le nombre d’horodatages restants est inférieur à la limite de débit configurée.
- Si la requête est autorisée, l’horodatage actuel est ajouté au journal.

Compromis :
- Précision : cet algorithme est parfaitement précis. Il suit correctement le nombre de requêtes
  dans la fenêtre glissante exacte, évitant les problèmes des compteurs à fenêtre fixe où
  des rafales à la limite des fenêtres peuvent dépasser le débit.
- Utilisation mémoire : l’empreinte mémoire est proportionnelle à la limite de débit, car elle stocke un
  horodatage par requête autorisée dans la fenêtre. Pour un client avec un débit de 1000, elle
  stockera jusqu’à 1000 horodatages (environ 8 Ko par client). Cela peut être significatif pour
  les systèmes ayant un très grand nombre de clients actifs ou des limites de débit très élevées.

Hypothèses :
- Déploiement dans un seul processus : l’état est stocké en mémoire dans l’instance de la classe. Il est
  sûr pour les threads dans un seul processus applicatif.
- Horloge monotone croissante : le paramètre `now_ms` est supposé être un horodatage monotone
  croissant en millisecondes. Le limiteur se protège contre le retour en arrière du temps
  en ignorant les requêtes dont les horodatages sont antérieurs au dernier horodatage traité.

Extension distribuée :
Pour prendre en charge un environnement distribué (par ex., plusieurs serveurs d’API), l’état doit être
externalisé vers un magasin central à faible latence comme Redis.
- Stockage de l’état : un Sorted Set Redis pourrait être utilisé pour chaque journal client/locataire, avec à la fois
  le score et la valeur correspondant à l’horodatage de la requête. Cela permet des requêtes de plage efficaces.
- Atomicité : les transactions Redis (MULTI/EXEC) ou les scripts Lua seraient essentiels pour effectuer
  l’opération de vérification et mise à jour (élaguer les anciens horodatages, vérifier le nombre, ajouter un nouvel horodatage)
  de manière atomique, empêchant les conditions de concurrence entre différents serveurs.
- Éviction : l’expiration de clés intégrée de Redis (`EXPIRE`) gérerait naturellement l’éviction
  des données client obsolètes.
"""

def __init__(self):
    self._tenants: Dict[str, Dict] = {}
    self._lock = threading.Lock()  # Verrou global pour ajouter/supprimer des locataires

def add_tenant(self, tenant_id: str, config: TenantConfig):
    """Ajoute ou met à jour la configuration d’un locataire."""
    with self._lock:
        if tenant_id not in self._tenants:
            self._tenants[tenant_id] = {
                "config": config,
                "clients": {},
                "tenant_log": deque(),
                "lock": threading.Lock() # Verrou par locataire pour la sûreté des threads
            }
        else:
            # Met à jour la configuration à la volée
            self._tenants[tenant_id]['config'] = config

def allow(self, tenant_id: str, client_id: str, now_ms: Optional[int] = None) -> Tuple[bool, int]:
    """
    Détermine si une requête est autorisée pour un locataire et un client donnés.

    Args:
        tenant_id : l’identifiant du locataire.
        client_id : l’identifiant du client.
        now_ms : l’horodatage actuel en millisecondes. Si None, l’heure système est utilisée.

    Returns:
        Un tuple (is_allowed, retry_after_ms).
        - is_allowed (bool) : True si la requête est autorisée, False sinon.
        - retry_after_ms (int) : si elle est refusée, le nombre de millisecondes à attendre avant que
          la prochaine requête puisse être autorisée. 0 si la requête est autorisée.
    """
    if now_ms is None:
        now_ms = int(time.time() * 1000)

    with self._lock:
        if tenant_id not in self._tenants:
            raise ValueError(f"Locataire '{tenant_id}' non configuré.")
        tenant_state = self._tenants[tenant_id]

    with tenant_state['lock']:
        config = tenant_state['config']

        # 1. Obtenir ou créer l’état du client
        if client_id not in tenant_state['clients']:
            tenant_state['clients'][client_id] = {"log": deque(), "last_seen_ms": 0}
        client_state = tenant_state['clients'][client_id]

        # Se défendre contre le retour en arrière de l’horloge
        if now_ms < client_state['last_seen_ms']:
            return (False, 1) # Refuser et suggérer un petit délai avant réessai

        # 2. Élaguer et vérifier la limite par client
        client_allowed, client_retry_ms = self._check_limit(
            client_state['log'], config.client_rate, config.client_window_ms, now_ms
        )
        if not client_allowed:
            return (False, client_retry_ms)

        # 3. Élaguer et vérifier la limite par locataire
        tenant_allowed, tenant_retry_ms = self._check_limit(
            tenant_state['tenant_log'], config.tenant_rate, config.tenant_window_ms, now_ms
        )
        if not tenant_allowed:
            return (False, tenant_retry_ms)

        # 4. Si les deux sont autorisés, enregistrer la requête et l’autoriser
        client_state['log'].append(now_ms)
        client_state['last_seen_ms'] = now_ms
        tenant_state['tenant_log'].append(now_ms)

        # 5. Nettoyage opportuniste des clients obsolètes pour ce locataire
        self._cleanup_stale_clients(tenant_state, now_ms)

        return (True, 0)

def _check_limit(self, log: deque, rate: int, window_ms: int, now_ms: int) -> Tuple[bool, int]:
    """Auxiliaire pour élaguer un journal d’horodatages et le vérifier par rapport à une limite de débit."""
    # Élaguer les anciens horodatages
    window_start_ms = now_ms - window_ms
    while log and log[0] <= window_start_ms:
        log.popleft()

    # Vérifier si la limite est dépassée
    if len(log) >= rate:
        oldest_timestamp = log[0]
        retry_after_ms = (oldest_timestamp + window_ms) - now_ms
        return (False, max(1, retry_after_ms)) # Garantir que le réessai est d’au moins 1 ms

    return (True, 0)

def _cleanup_stale_clients(self, tenant_state: Dict, now_ms: int):
    """
    Supprime les clients qui sont inactifs depuis longtemps.
    Un client inactif est un client dont la dernière requête est plus ancienne que deux fois la taille de la fenêtre.
    Il s’agit d’un nettoyage opportuniste, exécuté en ligne avec une requête.
    """
    config = tenant_state['config']
    stale_threshold_ms = config.client_window_ms * 2
    stale_clients = [
        cid for cid, cstate in tenant_state['clients'].items()
        if now_ms - cstate['last_seen_ms'] > stale_threshold_ms
    ]
    for client_id in stale_clients:
        del tenant_state['clients'][client_id]

--- Suite de tests ---

class TestSlidingWindowRateLimiter(unittest.TestCase):

def setUp(self):
    self.limiter = SlidingWindowRateLimiter()
    self.tenant_config = TenantConfig(
        client_rate=5,
        client_window_ms=1000,
        tenant_rate=15,
        tenant_window_ms=1000
    )
    self.limiter.add_tenant('test_tenant', self.tenant_config)
    self.now = int(time.time() * 1000)

def test_basic_client_limit(self):
    """Un client unique devrait être limité après avoir dépassé son débit."""
    for i in range(5):
        allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + i)
        self.assertTrue(allowed)

    allowed, retry_after = self.limiter.allow('test_tenant', 'client1', self.now + 5)
    self.assertFalse(allowed)
    self.assertGreater(retry_after, 990) # Devrait être autour de 1000 ms - 5 ms

def test_window_sliding(self):
    """Les requêtes devraient être à nouveau autorisées une fois la fenêtre écoulée."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)

    # Cette requête devrait être refusée
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 6)
    self.assertFalse(allowed)

    # Une fois la fenêtre de la première requête passée, une nouvelle est autorisée
    # La première requête était à self.now, la fenêtre est de 1000 ms.
    # À self.now + 1001, la première requête expire.
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 1001)
    self.assertTrue(allowed, "La requête devrait être autorisée après le glissement de la fenêtre")

def test_idle_client_returns(self):
    """Le quota d’un client inactif devrait être réinitialisé lorsqu’il revient."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)
    
    # Le client reste inactif pendant plus d’une fenêtre
    new_time = self.now + 2000
    for i in range(5):
        allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + i)
        self.assertTrue(allowed, "Les requêtes devraient être autorisées pour un client qui revient")
    
    allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + 5)
    self.assertFalse(allowed, "Le client devrait être à nouveau limité après avoir utilisé son nouveau quota")

def test_tenant_limit_and_fairness(self):
    """Plusieurs clients devraient atteindre collectivement la limite du locataire, démontrant l’équité."""
    # 3 clients, chacun effectue 5 requêtes. Total 15 requêtes, ce qui est la limite du locataire.
    for i in range(5):
        # Client 1
        self.assertTrue(self.limiter.allow('test_tenant', 'c1', self.now + i)[0])
        # Client 2
        self.assertTrue(self.limiter.allow('test_tenant', 'c2', self.now + i)[0])
        # Client 3
        self.assertTrue(self.limiter.allow('test_tenant', 'c3', self.now + i)[0])

    # À ce stade, le journal du locataire contient 15 requêtes. La prochaine requête de n’importe quel client devrait échouer.
    self.assertFalse(self.limiter.allow('test_tenant', 'c4', self.now + 6)[0])
    self.assertFalse(self.limiter.allow('test_tenant', 'c1', self.now + 6)[0])

    # Tester l’équité : un client atteint sa limite, mais un autre peut continuer.
    limiter = SlidingWindowRateLimiter()
    limiter.add_tenant('fair_tenant', TenantConfig(client_rate=2, client_window_ms=1000, tenant_rate=10, tenant_window_ms=1000))
    # Le client accapareur utilise tout son quota
    self.assertTrue(limiter.allow('fair_tenant', 'hog', self.now)[0])
    self.assertTrue(limiter.allow('fair_tenant', 'hog', self.now + 1)[0])
    self.assertFalse(limiter.allow('fair_tenant', 'hog', self.now + 2)[0]) # L’accapareur est bloqué

    # Un autre client devrait toujours pouvoir effectuer des requêtes
    self.assertTrue(limiter.allow('fair_tenant', 'fair_client', self.now + 3)[0])

def test_stale_client_eviction(self):
    """Les clients obsolètes devraient être supprimés de la mémoire."""
    self.limiter.allow('test_tenant', 'stale_client', self.now)
    self.assertIn('stale_client', self.limiter._tenants['test_tenant']['clients'])

    # Effectuer une nouvelle requête loin dans le futur, déclenchant le nettoyage
    future_time = self.now + self.tenant_config.client_window_ms * 3
    self.limiter.allow('test_tenant', 'active_client', future_time)

    self.assertNotIn('stale_client', self.limiter._tenants['test_tenant']['clients'])

def test_concurrency(self):
    """Devrait gérer correctement les requêtes concurrentes vers le même client."""
    limiter = SlidingWindowRateLimiter()
    limiter.add_tenant('concurrent_tenant', TenantConfig(client_rate=100, client_window_ms=1000, tenant_rate=1000, tenant_window_ms=1000))
    start_time = int(time.time() * 1000)
    num_threads = 10
    reqs_per_thread = 15
    allowed_count = [0]
    lock = threading.Lock()

    def worker():
        for _ in range(reqs_per_thread):
            now = int(time.time() * 1000)
            allowed, _ = limiter.allow('concurrent_tenant', 'shared_client', now)
            if allowed:
                with lock:
                    allowed_count[0] += 1
            time.sleep(0.001) # petite pause pour augmenter les chances de contention

    threads = [threading.Thread(target=worker) for _ in range(num_threads)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    # Le nombre de requêtes autorisées devrait être exactement la limite de débit du client.
    self.assertEqual(allowed_count[0], 100)

def test_clock_regression(self):
    """Devrait refuser les requêtes si l’horloge recule."""
    self.assertTrue(self.limiter.allow('test_tenant', 'client1', self.now + 100)[0])
    # Une requête avec un horodatage antérieur devrait être refusée
    allowed, retry_after = self.limiter.allow('test_tenant', 'client1', self.now + 99)
    self.assertFalse(allowed)
    self.assertEqual(retry_after, 1)

if name == 'main':
unittest.main()