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:
"""Rate limits for one tenant.

client_limit applies independently to each client, while tenant_limit applies
to all admitted requests from the tenant. Both limits use exact rolling
windows. active_timeout_ms bounds how long an absent fairness waiter can
retain its turn. eviction_ms controls client-state retention.
"""

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("limits must be positive")
    if self.client_window_ms <= 0 or self.tenant_window_ms <= 0:
        raise ValueError("window lengths must be positive")
    if self.active_timeout_ms <= 0:
        raise ValueError("active_timeout_ms must be positive")
    if self.eviction_ms < max(self.client_window_ms, self.tenant_window_ms):
        raise ValueError("eviction_ms must be at least as large as both windows")
    if self.eviction_ms < self.active_timeout_ms:
        raise ValueError("eviction_ms must be at least 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:
"""Thread-safe, single-process, exact sliding-window rate limiter.

Sliding-window algorithm:
  Every admitted request timestamp is stored in a deque. Before evaluating a
  request, timestamps t satisfying t <= now - window are removed. Thus a
  request exactly one full window old no longer consumes capacity. This is a
  true log-based sliding window, so it has no calendar-bucket boundary burst.

  The result is exact, with O(1) amortized deque operations and memory
  proportional to admitted requests still inside configured windows. This is
  more accurate than weighted buckets but uses more memory. Limits provide a
  hard bound on live timestamp entries, and stale client objects are evicted.

Fair sharing:
  Uncontended tenant capacity is work-conserving and can be used by any
  client. Once the tenant ceiling is reached, denied clients enter a FIFO
  queue with at most one entry per client. As capacity expires, the head gets
  the next admission, then leaves the queue. A continuously busy client must
  rejoin behind other contenders, implementing round-robin redistribution
  without permanently reserving unused per-client shares. Inactive queue
  entries expire after active_timeout_ms so an abandoned turn cannot block
  the tenant forever.

Clock behavior:
  now_ms should come from a monotonic clock. If a caller supplies a duplicate
  or decreasing timestamp, time is clamped to the tenant's greatest observed
  timestamp. This prevents expired history from becoming live again.

Concurrency and deployment assumptions:
  A single re-entrant lock makes allow, configuration, and sweeping atomic
  across threads and async tasks sharing this object. For distributed use,
  the same state transition can be implemented atomically in Redis with a
  Lua script and sorted sets, or in a transactional strongly consistent
  store. FIFO waiter metadata must be updated in the same transaction as the
  request logs; local locks alone do not coordinate multiple processes.
"""

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:
    """Create or replace a tenant's configuration atomically."""
    config.validate()
    with self._lock:
        self._configs[tenant_id] = config

def allow(self, tenant_id: Hashable, client_id: Hashable, now_ms: int) -> AllowResult:
    """Atomically decide whether to admit one request.

    retry_after_ms is exact for a direct client/window or tenant/window
    denial. For a FIFO fairness denial it is the earliest temporal retry;
    actual admission also depends on earlier active waiters taking or
    abandoning their turns.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms must be an integer")

    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))

        # Capacity exists. During contention, only the FIFO head may take it.
        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:
    """Evict stale clients and empty tenant runtime states.

    Applications can call this periodically when traffic is low. Normal
    allow calls also clean the tenant being accessed. Returns the number of
    client and tenant runtime objects removed. Configurations are retained.
    """
    if not isinstance(now_ms, int):
        raise TypeError("now_ms must be an integer")

    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]:
    """Return aggregate state sizes, useful for monitoring and 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"tenant {tenant_id!r} has no 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)
        # A client blocked by its own quota does not reserve a tenant turn.
        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)

    # Both old slots expire, but a cannot monopolize them: b and c queued.
    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)
    # At 1001 the waiter is inactive, so another client may use capacity.
    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)

    # The active request is not dropped before its own window expires.
    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

Hilfsdatenklasse für die Mandantenkonfiguration

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:
"""
Ein threadsicherer, mandantenfähiger Rate Limiter, der einen Sliding-Window-Log-Algorithmus verwendet.

Diese Implementierung bietet eine Ratenbegrenzung pro Client unter einer globalen Obergrenze pro Mandant.
Sie verwendet eine Fair-Sharing-Richtlinie, bei der kein einzelner Client das Kontingent des Mandanten
aufgrund seines eigenen individuellen Limits ausschöpfen kann.

Algorithmus: Sliding Window Log
- Für jeden Client und Mandanten verwalten wir eine deque (einen listenähnlichen Container mit schnellem Anhängen
  und Entfernen von beiden Enden) von Zeitstempeln der innerhalb des Fensters empfangenen Anfragen.
- Wenn eine neue Anfrage eintrifft, verwerfen wir zuerst alle Zeitstempel, die älter sind als die aktuelle
  Zeit minus die Fenstergröße.
- Dann prüfen wir, ob die Anzahl der verbleibenden Zeitstempel kleiner ist als das konfigurierte Ratenlimit.
- Wenn die Anfrage erlaubt ist, wird der aktuelle Zeitstempel dem Log hinzugefügt.

Abwägungen:
- Genauigkeit: Dieser Algorithmus ist vollkommen genau. Er verfolgt die Anzahl der Anfragen
  innerhalb des präzisen gleitenden Fensters korrekt und vermeidet Probleme mit Fixed-Window-Zählern, bei denen
  Bursts an der Grenze von Fenstern die Rate überschreiten können.
- Speicherverbrauch: Der Speicherbedarf ist proportional zum Ratenlimit, da ein
  Zeitstempel pro erlaubter Anfrage innerhalb des Fensters gespeichert wird. Für einen Client mit einer Rate von 1000
  werden bis zu 1000 Zeitstempel gespeichert (ca. 8 KB pro Client). Dies kann für
  Systeme mit einer sehr großen Anzahl aktiver Clients oder sehr hohen Ratenlimits erheblich sein.

Annahmen:
- Single-Process-Bereitstellung: Der Zustand wird im Speicher innerhalb der Klasseninstanz gespeichert. Er ist
  threadsicher für die Verwendung innerhalb eines einzelnen Anwendungsprozesses.
- Monoton steigende Uhr: Es wird erwartet, dass der Parameter `now_ms` ein monoton
  steigender Zeitstempel in Millisekunden ist. Der Limiter schützt sich vor rückwärts laufender Zeit,
  indem er Anfragen mit Zeitstempeln ignoriert, die älter sind als der zuletzt verarbeitete.

Verteilte Erweiterung:
Um eine verteilte Umgebung zu unterstützen (z. B. mehrere API-Server), muss der Zustand in
einen zentralen, latenzarmen Speicher wie Redis ausgelagert werden.
- Zustandsspeicherung: Ein Redis Sorted Set könnte für jedes Client-/Mandanten-Log verwendet werden, wobei sowohl der
  Score als auch der Wert der Anfragezeitstempel sind. Dies ermöglicht effiziente Bereichsabfragen.
- Atomarität: Redis-Transaktionen (MULTI/EXEC) oder Lua-Skripte wären unerlässlich, um
  die Check-and-Update-Operation (alte Zeitstempel bereinigen, Anzahl prüfen, neuen Zeitstempel hinzufügen)
  atomar auszuführen und Race Conditions zwischen verschiedenen Servern zu verhindern.
- Entfernung: Redis' eingebaute Schlüsselablauffunktion (`EXPIRE`) würde die Entfernung
  veralteter Client-Daten auf natürliche Weise übernehmen.
"""

def __init__(self):
    self._tenants: Dict[str, Dict] = {}
    self._lock = threading.Lock()  # Globale Sperre zum Hinzufügen/Entfernen von Mandanten

def add_tenant(self, tenant_id: str, config: TenantConfig):
    """Fügt die Konfiguration eines Mandanten hinzu oder aktualisiert sie."""
    with self._lock:
        if tenant_id not in self._tenants:
            self._tenants[tenant_id] = {
                "config": config,
                "clients": {},
                "tenant_log": deque(),
                "lock": threading.Lock() # Sperre pro Mandant für Thread-Sicherheit
            }
        else:
            # Konfiguration im laufenden Betrieb aktualisieren
            self._tenants[tenant_id]['config'] = config

def allow(self, tenant_id: str, client_id: str, now_ms: Optional[int] = None) -> Tuple[bool, int]:
    """
    Bestimmt, ob eine Anfrage für einen gegebenen Mandanten und Client erlaubt ist.

    Argumente:
        tenant_id: Der Bezeichner für den Mandanten.
        client_id: Der Bezeichner für den Client.
        now_ms: Der aktuelle Zeitstempel in Millisekunden. Wenn None, wird die Systemzeit verwendet.

    Rückgabewerte:
        Ein Tupel (is_allowed, retry_after_ms).
        - is_allowed (bool): True, wenn die Anfrage zugelassen ist, andernfalls False.
        - retry_after_ms (int): Wenn abgelehnt, die Anzahl der Millisekunden, die gewartet werden soll, bevor die
          nächste Anfrage möglicherweise erlaubt wird. 0, wenn die Anfrage erlaubt ist.
    """
    if now_ms is None:
        now_ms = int(time.time() * 1000)

    with self._lock:
        if tenant_id not in self._tenants:
            raise ValueError(f"Mandant '{tenant_id}' ist nicht konfiguriert.")
        tenant_state = self._tenants[tenant_id]

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

        # 1. Client-Zustand abrufen oder erstellen
        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]

        # Gegen rückwärts laufende Uhr verteidigen
        if now_ms < client_state['last_seen_ms']:
            return (False, 1) # Ablehnen und eine kurze Wiederholungsverzögerung vorschlagen

        # 2. Limit pro Client bereinigen und prüfen
        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. Limit pro Mandant bereinigen und prüfen
        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. Wenn beide erlaubt sind, die Anfrage aufzeichnen und sie zulassen
        client_state['log'].append(now_ms)
        client_state['last_seen_ms'] = now_ms
        tenant_state['tenant_log'].append(now_ms)

        # 5. Opportunistische Bereinigung veralteter Clients für diesen Mandanten
        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]:
    """Hilfsmethode, um ein Zeitstempel-Log zu bereinigen und gegen ein Ratenlimit zu prüfen."""
    # Alte Zeitstempel bereinigen
    window_start_ms = now_ms - window_ms
    while log and log[0] <= window_start_ms:
        log.popleft()

    # Prüfen, ob das Limit überschritten ist
    if len(log) >= rate:
        oldest_timestamp = log[0]
        retry_after_ms = (oldest_timestamp + window_ms) - now_ms
        return (False, max(1, retry_after_ms)) # Sicherstellen, dass die Wiederholung mindestens 1 ms beträgt

    return (True, 0)

def _cleanup_stale_clients(self, tenant_state: Dict, now_ms: int):
    """
    Entfernt Clients, die lange Zeit inaktiv waren.
    Ein inaktiver Client ist einer, dessen letzte Anfrage älter ist als das Doppelte der Fenstergröße.
    Dies ist eine opportunistische Bereinigung, die inline mit einer Anfrage ausgeführt wird.
    """
    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]

--- Testsuite ---

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):
    """Ein einzelner Client sollte gedrosselt werden, nachdem er seine Rate überschritten hat."""
    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) # Sollte etwa 1000 ms - 5 ms betragen

def test_window_sliding(self):
    """Anfragen sollten wieder erlaubt werden, nachdem das Fenster vergangen ist."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)

    # Diese Anfrage sollte abgelehnt werden
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 6)
    self.assertFalse(allowed)

    # Nachdem das Fenster für die erste Anfrage vergangen ist, ist eine neue erlaubt
    # Die erste Anfrage war bei self.now, das Fenster ist 1000 ms.
    # Bei self.now + 1001 läuft die erste Anfrage ab.
    allowed, _ = self.limiter.allow('test_tenant', 'client1', self.now + 1001)
    self.assertTrue(allowed, "Die Anfrage sollte erlaubt sein, nachdem das Fenster weitergleitet")

def test_idle_client_returns(self):
    """Das Kontingent eines inaktiven Clients sollte frisch sein, wenn er zurückkehrt."""
    for i in range(5):
        self.limiter.allow('test_tenant', 'client1', self.now + i)
    
    # Client bleibt länger als ein Fenster inaktiv
    new_time = self.now + 2000
    for i in range(5):
        allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + i)
        self.assertTrue(allowed, "Anfragen sollten für einen zurückkehrenden Client erlaubt sein")
    
    allowed, _ = self.limiter.allow('test_tenant', 'client1', new_time + 5)
    self.assertFalse(allowed, "Client sollte nach Nutzung des neuen Kontingents erneut gedrosselt werden")

def test_tenant_limit_and_fairness(self):
    """Mehrere Clients sollten gemeinsam das Mandantenlimit erreichen und damit Fairness demonstrieren."""
    # 3 Clients, jeder stellt 5 Anfragen. Insgesamt 15 Anfragen, was dem Mandantenlimit entspricht.
    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])

    # An diesem Punkt hat das Mandanten-Log 15 Anfragen. Die nächste Anfrage von einem beliebigen Client sollte fehlschlagen.
    self.assertFalse(self.limiter.allow('test_tenant', 'c4', self.now + 6)[0])
    self.assertFalse(self.limiter.allow('test_tenant', 'c1', self.now + 6)[0])

    # Fairness testen: Ein Client erreicht sein Limit, aber ein anderer kann fortfahren.
    limiter = SlidingWindowRateLimiter()
    limiter.add_tenant('fair_tenant', TenantConfig(client_rate=2, client_window_ms=1000, tenant_rate=10, tenant_window_ms=1000))
    # Hog-Client verbraucht sein Kontingent
    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]) # Hog ist blockiert

    # Ein anderer Client sollte weiterhin Anfragen stellen können
    self.assertTrue(limiter.allow('fair_tenant', 'fair_client', self.now + 3)[0])

def test_stale_client_eviction(self):
    """Veraltete Clients sollten aus dem Speicher entfernt werden."""
    self.limiter.allow('test_tenant', 'stale_client', self.now)
    self.assertIn('stale_client', self.limiter._tenants['test_tenant']['clients'])

    # Eine neue Anfrage weit in der Zukunft stellen, wodurch die Bereinigung ausgelöst wird
    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):
    """Sollte gleichzeitige Anfragen an denselben Client korrekt verarbeiten."""
    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) # kurze Pause, um die Wahrscheinlichkeit von Konkurrenz zu erhöhen

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

    # Die Anzahl der erlaubten Anfragen sollte genau dem Client-Ratenlimit entsprechen.
    self.assertEqual(allowed_count[0], 100)

def test_clock_regression(self):
    """Sollte Anfragen ablehnen, wenn die Uhr rückwärts läuft."""
    self.assertTrue(self.limiter.allow('test_tenant', 'client1', self.now + 100)[0])
    # Eine Anfrage mit einem früheren Zeitstempel sollte abgelehnt werden
    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()