pip install pymemcache testcontainers   # Python
npm install --save-dev memjs testcontainers   # Node (binary protocol)

from testcontainers.memcached import MemcachedContainer

with MemcachedContainer("memcached:1.6-alpine") as mc:
    host, port = mc.get_host_and_port()

import pytest
from pymemcache.client.base import Client
from testcontainers.memcached import MemcachedContainer

@pytest.fixture(scope="session")
def mc_addr():
    with MemcachedContainer("memcached:1.6-alpine") as mc:
        yield mc.get_host_and_port()   # (host, port)

@pytest.fixture
def mc(mc_addr):
    host, port = mc_addr
    client = Client((host, port), default_value=None)
    yield client
    client.flush_all()   # Reset between tests

def test_set_and_get(mc):
    mc.set("k", b"hello")
    assert mc.get("k") == b"hello"

def test_add_only_when_absent(mc):
    assert mc.add("k", b"first") is True
    assert mc.add("k", b"second") is False   # NOT_STORED: key exists
    assert mc.get("k") == b"first"

def test_get_absent_returns_none(mc):
    assert mc.get("no-such-key") is None

import time

def test_ttl_zero_never_expires(mc):
    mc.set("k", b"v", expire=0)
    time.sleep(0.1)
    assert mc.get("k") == b"v"

def test_key_expires_after_ttl(mc):
    mc.set("k", b"v", expire=1)
    assert mc.get("k") == b"v"
    time.sleep(1.5)
    assert mc.get("k") is None

def test_short_ttl_via_pexpire_pattern(mc):
    # pymemcache does not expose millisecond TTLs; use 1-second minimum
    mc.set("k", b"val", expire=1)
    time.sleep(1.5)
    assert mc.get("k") is None, "Key must expire after 1 s TTL"

def test_cas_succeeds_when_token_matches(mc):
    mc.set("k", b"v1")
    value, cas_token = mc.gets("k")
    result = mc.cas("k", b"v2", cas_token)
    assert result is True
    assert mc.get("k") == b"v2"

def test_cas_fails_after_concurrent_write(mc):
    mc.set("k", b"original")
    _, old_token = mc.gets("k")
    mc.set("k", b"concurrent-update")   # token now stale
    result = mc.cas("k", b"late-writer", old_token)
    assert result is False   # EXISTS: token mismatch
    assert mc.get("k") == b"concurrent-update"

def test_incr_increments_existing_counter(mc):
    mc.set("counter", b"10")
    result = mc.incr("counter", 5)
    assert result == 15

def test_incr_absent_key_returns_none(mc):
    assert mc.incr("no-such-counter", 1) is None

def test_incr_uses_add_to_initialise(mc):
    # Per github.com/memcached/memcached/wiki/Programming:
    # add is the correct initialiser for counters
    mc.add("hits", b"0")
    mc.incr("hits", 1)
    assert mc.get("hits") == b"1"

def test_decr_does_not_go_below_zero(mc):
    mc.set("counter", b"3")
    mc.decr("counter", 10)
    assert mc.get("counter") == b"0"   # unsigned floor at 0

def test_lru_evicts_cold_keys_under_pressure():
    """
    Launch a small-memory container to verify LRU eviction.
    The -m flag caps Memcached's RAM (MB).
    """
    from testcontainers.memcached import MemcachedContainer
    from pymemcache.client.base import Client

    with MemcachedContainer("memcached:1.6-alpine") as mc:
        mc.get_wrapped_container().exec_run   # introspect if needed
        host, port = mc.get_host_and_port()
        # Restart with low memory cap via Docker command override
    
    # Use a separate docker run with -m 8m for a tighter eviction test;
    # or accept that testcontainers default image evicts eventually.
    # The key assertion: after filling cache, a cold key may be absent.

def test_no_data_survives_restart(mc_addr):
    """Memcached has no persistence: data is gone after any restart."""
    host, port = mc_addr
    c = Client((host, port))
    c.set("persistent", b"should-not-survive")
    # Simulate application expectation: always handle cache miss gracefully
    # after a node restart or replacement (e.g., ElastiCache node failure).
    assert c.get("persistent") is not None  # warm path
    # After restart (modelled here as flush_all), data is gone:
    c.flush_all()
    assert c.get("persistent") is None, "Memcached is not persistent"

def test_hash_client_distributes_keys():
    from testcontainers.memcached import MemcachedContainer
    from pymemcache.client.hash import HashClient

    with MemcachedContainer("memcached:1.6-alpine") as mc1, \
         MemcachedContainer("memcached:1.6-alpine") as mc2:

        h1, p1 = mc1.get_host_and_port()
        h2, p2 = mc2.get_host_and_port()
        cluster = HashClient([(h1, p1), (h2, p2)])

        keys = [f"key:{i}" for i in range(100)]
        for k in keys:
            cluster.set(k, b"v")

        # Verify distribution: each node should hold some keys
        direct1 = sum(
            1 for k in keys if Client((h1, p1)).get(k) is not None
        )
        direct2 = sum(
            1 for k in keys if Client((h2, p2)).get(k) is not None
        )
        assert direct1 > 0, "Node 1 should hold some keys"
        assert direct2 > 0, "Node 2 should hold some keys"
        assert direct1 + direct2 == 100, "Every key must be on exactly one node"

def test_hash_client_handles_node_removal():
    """After removing a node, the remaining node serves all keys."""
    from testcontainers.memcached import MemcachedContainer
    from pymemcache.client.hash import HashClient

    with MemcachedContainer("memcached:1.6-alpine") as mc1, \
         MemcachedContainer("memcached:1.6-alpine") as mc2:

        h1, p1 = mc1.get_host_and_port()
        h2, p2 = mc2.get_host_and_port()
        full_cluster = HashClient([(h1, p1), (h2, p2)])

        for i in range(20):
            full_cluster.set(f"k{i}", b"val")

        # Simulate node removal: re-create client with one node
        degraded = HashClient([(h1, p1)])
        # Keys that were on node 2 are now misses - application must
        # handle gracefully (cache miss -> read-through from source of truth)
        miss_count = sum(
            1 for i in range(20) if degraded.get(f"k{i}") is None
        )
        assert miss_count >= 0  # Some keys lost; app must tolerate it

def test_elasticache_auto_discovery_endpoint(monkeypatch):
    """
    Integration smoke test: verify the app resolves a configuration
    endpoint and discovers cluster nodes.
    Runs only when ELASTICACHE_CONFIG_ENDPOINT is set.
    """
    import os
    endpoint = os.getenv("ELASTICACHE_CONFIG_ENDPOINT")
    if not endpoint:
        pytest.skip("ELASTICACHE_CONFIG_ENDPOINT not set (ElastiCache env only)")

    from pymemcache.client.hash import HashClient
    # The ElastiCache Cluster Client for Python resolves the cfg endpoint
    # and populates the server list automatically via the config get cluster
    # Memcached command.
    client = HashClient([endpoint])
    client.set("smoke-test", b"ok")
    assert client.get("smoke-test") == b"ok"

pytest tests/memcached/ -v

jobs:
  memcached-tests:
    runs-on: ubuntu-latest
    services:
      memcached:
        image: memcached:1.6-alpine
        ports:
          - 11211:11211
    steps:
      - uses: actions/checkout@v5
      - uses: actions/setup-python@v5
      - run: pip install -e ".[test]"
      - run: pytest tests/memcached/ --tb=short
        env:
          MEMCACHED_HOST: localhost
          MEMCACHED_PORT: 11211