verify_redis.py

"""Test reference Redis primitives in an isolated disposable server, not the API."""
from pathlib import Path
import shutil
import socket
import subprocess
import tempfile
import time
import unittest

HERE = Path(__file__).parent
PUT = (HERE / "redis/put_immutable.lua").read_bytes()
PROMOTE = (HERE / "redis/promote_record.lua").read_bytes()


class RedisError(Exception):
    pass


class Connection:
    def __init__(self, path):
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.sock.settimeout(3)
        try:
            self.sock.connect(path)
        except OSError:
            self.sock.close()
            raise
        self.reader = self.sock.makefile("rb")

    def command(self, *args):
        values = [a if isinstance(a, bytes) else str(a).encode() for a in args]
        payload = b"*%d\r\n" % len(values)
        for value in values:
            payload += b"$%d\r\n" % len(value) + value + b"\r\n"
        self.sock.sendall(payload)
        return self.read()

    def read(self):
        line = self.reader.readline()
        if not line:
            raise ConnectionError("Redis closed connection")
        kind, body = line[:1], line[1:-2]
        if kind == b"+":
            return body
        if kind == b"-":
            raise RedisError(body.decode())
        if kind == b":":
            return int(body)
        if kind == b"$":
            length = int(body)
            if length == -1:
                return None
            value = self.reader.read(length)
            if self.reader.read(2) != b"\r\n":
                raise ConnectionError("Invalid RESP bulk string")
            return value
        if kind == b"*":
            count = int(body)
            return None if count == -1 else [self.read() for _ in range(count)]
        raise ConnectionError("Unexpected RESP type")

    def close(self):
        self.reader.close()
        self.sock.close()


class RedisContract(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        executable = shutil.which("redis-server")
        if not executable:
            raise unittest.SkipTest("redis-server 7.2+ is required")
        cls.temp = tempfile.TemporaryDirectory(prefix="ms-redis-contract-", dir="/tmp")
        cls.path = str(Path(cls.temp.name) / "redis.sock")
        cls.log = open(Path(cls.temp.name) / "server.log", "wb")
        cls.server = subprocess.Popen([
            executable, "--port", "0", "--unixsocket", cls.path,
            "--unixsocketperm", "700", "--dir", cls.temp.name,
            "--save", "", "--appendonly", "yes", "--appendfsync", "everysec",
            "--maxmemory-policy", "noeviction", "--daemonize", "no",
        ], stdout=cls.log, stderr=subprocess.STDOUT)
        for _ in range(100):
            if cls.server.poll() is not None:
                cls.log.flush()
                diagnostic = (Path(cls.temp.name) / "server.log").read_text()
                cls.cleanup()
                raise RuntimeError("Disposable Redis failed to start: " + diagnostic)
            try:
                probe = Connection(cls.path)
                probe.command("PING")
                probe.close()
                return
            except (FileNotFoundError, ConnectionRefusedError):
                time.sleep(0.02)
        cls.cleanup()
        raise RuntimeError("Disposable Redis startup timeout")

    @classmethod
    def cleanup(cls):
        if cls.server.poll() is None:
            cls.server.terminate()
            try:
                cls.server.wait(timeout=5)
            except subprocess.TimeoutExpired:
                cls.server.kill()
                cls.server.wait(timeout=5)
        cls.log.close()
        cls.temp.cleanup()

    @classmethod
    def tearDownClass(cls):
        cls.cleanup()

    def setUp(self):
        self.r = Connection(self.path)
        self.key = "minesweeper:dev:v1:test:" + self._testMethodName
        secs, micros = self.r.command("TIME")
        self.now = int(secs) * 1000 + int(micros) // 1000

    def tearDown(self):
        self.r.close()

    def put(self, value=b"\x1f\x8b\x00binary", deadline=None):
        return self.r.command("EVAL", PUT, 1, self.key, value, deadline or self.now + 10000)

    def test_binary_round_trip_with_atomic_absolute_expiry(self):
        data = bytes(range(256))
        self.assertEqual(self.put(data)[0], b"created")
        self.assertEqual(self.r.command("GET", self.key), data)
        self.assertEqual(self.r.command("PEXPIRETIME", self.key), self.now + 10000)

    def test_retry_does_not_slide_expiry(self):
        self.put()
        self.assertEqual(self.put(deadline=self.now + 30000)[0], b"existing")
        self.assertEqual(self.r.command("PEXPIRETIME", self.key), self.now + 10000)

    def test_different_payload_cannot_replace_existing_record(self):
        self.put(b"first")
        with self.assertRaisesRegex(RedisError, "CONTENT_MISMATCH"):
            self.put(b"second")
        self.assertEqual(self.r.command("GET", self.key), b"first")

    def test_past_deadline_rejected_and_expiry_needs_no_job(self):
        with self.assertRaisesRegex(RedisError, "DEADLINE_EXPIRED"):
            self.put(deadline=self.now - 1)
        self.put(deadline=self.now + 100)
        time.sleep(0.15)
        self.assertIsNone(self.r.command("GET", self.key))
        self.assertEqual(self.r.command("PTTL", self.key), -2)

    def test_promotion_has_fixed_expiry_and_does_not_rewrite_body(self):
        self.put(b"verified")
        for _ in range(2):
            self.r.command("EVAL", PROMOTE, 1, self.key, self.now + 10000, self.now + 60000)
        self.assertEqual(self.r.command("PEXPIRETIME", self.key), self.now + 60000)
        self.assertEqual(self.r.command("GET", self.key), b"verified")

    def test_promotion_cannot_recreate_missing_record_or_cross_date(self):
        with self.assertRaisesRegex(RedisError, "REPLAY_MISSING"):
            self.r.command("EVAL", PROMOTE, 1, self.key, self.now + 10000, self.now + 60000)
        self.put()
        with self.assertRaisesRegex(RedisError, "BOARD_FROZEN"):
            self.r.command("EVAL", PROMOTE, 1, self.key, self.now - 1, self.now + 60000)

    def test_existing_key_without_ttl_is_rejected(self):
        self.r.command("SET", self.key, b"bad")
        with self.assertRaisesRegex(RedisError, "INVALID_EXISTING_TTL"):
            self.put(b"bad")
        with self.assertRaisesRegex(RedisError, "INVALID_EXISTING_TTL"):
            self.r.command("EVAL", PROMOTE, 1, self.key, self.now + 10000, self.now + 60000)

    def test_local_aof_barrier_and_noeviction_configuration(self):
        self.put()
        local, replicas = self.r.command("WAITAOF", 1, 0, 2000)
        self.assertEqual(local, 1)
        self.assertEqual(replicas, 0)
        self.assertEqual(self.r.command("CONFIG", "GET", "maxmemory-policy"),
                         [b"maxmemory-policy", b"noeviction"])


if __name__ == "__main__":
    unittest.main(verbosity=2)