AI-Powered Development Tools

# You type:
# Function to validate email addresses using regex

# Copilot suggests:
import re

def validate_email(email):
    """
    Validate email address format.

    Args:
        email: Email address to validate

    Returns:
        bool: True if valid email format, False otherwise
    """
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))


# Result: Complete, working function from a single comment!

# You type:
class User:
    def __init__(self, name, email):

# Copilot suggests:
        self.name = name
        self.email = email
        self.created_at = datetime.now()
        self.id = str(uuid.uuid4())

    def __repr__(self):
        return f"User(name={self.name}, email={self.email})"

    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "email": self.email,
            "created_at": self.created_at.isoformat()
        }


# Copilot often suggests standard methods and patterns

# You have this function:
def calculate_discount(price, discount_percent):
    """Calculate final price after discount."""
    if discount_percent < 0 or discount_percent > 100:
        raise ValueError("Discount must be between 0 and 100")
    return price * (1 - discount_percent / 100)


# You type:
def test_calculate_discount():

# Copilot suggests complete test:
    # Test normal discount
    assert calculate_discount(100, 20) == 80
    assert calculate_discount(50, 10) == 45

    # Test edge cases
    assert calculate_discount(100, 0) == 100
    assert calculate_discount(100, 100) == 0

    # Test invalid input
    with pytest.raises(ValueError):
        calculate_discount(100, -10)

    with pytest.raises(ValueError):
        calculate_discount(100, 150)


# Copilot often generates comprehensive tests from function signature

# Start Redis on the default port
docker run -d --name ratelimit-redis -p 6379:6379 redis:8-alpine

# Python dependencies
pip install redis flask pytest

# ... work through the steps below ...

# Tear it down when you're done
docker rm -f ratelimit-redis

# Rate limiter using Redis
# - Track requests per user in Redis
# - Sliding window: count requests in the last 60 seconds
# - Return True if allowed, False if rate limited
# - Handle Redis connection errors gracefully

import time
from typing import Tuple

import redis


def check_rate_limit(user_id: str, redis_client: redis.Redis) -> Tuple[bool, int]:
    """Check if user has exceeded rate limit.

    Returns:
        Tuple of (allowed: bool, remaining: int)
    """
    key = f"rate_limit:{user_id}"
    current_time = time.time()
    window = 60  # seconds
    limit = 100

    try:
        # Remove old entries outside the window
        redis_client.zremrangebyscore(key, 0, current_time - window)

        # Count current requests in window
        current_count = redis_client.zcard(key)

        if current_count < limit:
            redis_client.zadd(key, {str(current_time): current_time})
            redis_client.expire(key, window)
            return True, limit - current_count - 1
        else:
            return False, 0

    except redis.RedisError as e:
        # Log error, but allow the request through (fail open)
        print(f"Redis error: {e}")
        return True, -1

def check_rate_limit_improved(user_id: str, redis_client: redis.Redis) -> Tuple[bool, int]:
    """Rate limit with atomic operations."""
    key = f"rate_limit:{user_id}"
    current_time = time.time()
    window = 60
    limit = 100

    try:
        # redis-py pipelines default to transaction=True, which wraps the
        # commands in MULTI/EXEC, so no other client interleaves with them.
        pipe = redis_client.pipeline()
        pipe.zremrangebyscore(key, 0, current_time - window)
        pipe.zadd(key, {str(current_time): current_time})
        pipe.expire(key, window)
        pipe.zcard(key)

        results = pipe.execute()
        current_count = results[-1]   # last result is the count

        if current_count <= limit:
            return True, limit - current_count
        else:
            # Over the limit: take back the request we just added
            redis_client.zrem(key, str(current_time))
            return False, 0

    except redis.RedisError as e:
        print(f"Redis error: {e}")
        return True, -1  # fail open

from unittest.mock import Mock

import redis


def test_rate_limit_allows_under_limit():
    """Requests are allowed when under the limit."""
    mock_redis = Mock()
    mock_redis.pipeline.return_value.execute.return_value = [None, None, None, 50]

    allowed, remaining = check_rate_limit_improved("user123", mock_redis)

    assert allowed is True
    assert remaining == 50


def test_rate_limit_blocks_over_limit():
    """Requests are blocked once the count passes the limit."""
    mock_redis = Mock()
    mock_redis.pipeline.return_value.execute.return_value = [None, None, None, 101]

    allowed, remaining = check_rate_limit_improved("user123", mock_redis)

    assert allowed is False
    assert remaining == 0


def test_rate_limit_handles_redis_error():
    """A Redis outage fails open rather than locking everyone out."""
    mock_redis = Mock()
    mock_redis.pipeline.side_effect = redis.RedisError("Connection failed")

    allowed, remaining = check_rate_limit_improved("user123", mock_redis)

    assert allowed is True
    assert remaining == -1

def test_sliding_window_expires_old_requests(redis_client):
    """After the window passes, earlier requests stop counting."""
    # needs real Redis, and a clock you control

def test_two_requests_in_the_same_instant_both_count(redis_client):
    """Regression test for the timestamp-collision bug found above."""

def test_limit_is_per_user(redis_client):
    """One user hitting the limit must not affect anyone else."""

from flask import Flask, g, jsonify, request

app = Flask(__name__)
redis_client = redis.Redis(host="localhost", port=6379)


@app.before_request
def rate_limit_middleware():
    user_id = request.headers.get("X-User-ID")
    if not user_id:
        return jsonify({"error": "Missing user ID"}), 400

    allowed, remaining = check_rate_limit_improved(user_id, redis_client)

    if not allowed:
        return jsonify({"error": "Rate limit exceeded", "retry_after": 60}), 429

    g.rate_limit_remaining = remaining   # stash for the response hook


@app.after_request
def add_rate_limit_headers(response):
    remaining = getattr(g, "rate_limit_remaining", None)
    if remaining is not None:
        response.headers["X-RateLimit-Limit"] = "100"
        response.headers["X-RateLimit-Remaining"] = str(remaining)
    return response

# app.py
import time
from typing import Tuple

import redis
from flask import Flask, g, jsonify, request

app = Flask(__name__)
redis_client = redis.Redis(host="localhost", port=6379)


def check_rate_limit_improved(user_id: str, redis_client: redis.Redis) -> Tuple[bool, int]:
    """Rate limit with atomic operations (Step 5)."""
    key = f"rate_limit:{user_id}"
    current_time = time.time()
    window = 60
    limit = 100

    try:
        pipe = redis_client.pipeline()
        pipe.zremrangebyscore(key, 0, current_time - window)
        pipe.zadd(key, {str(current_time): current_time})
        pipe.expire(key, window)
        pipe.zcard(key)

        results = pipe.execute()
        current_count = results[-1]

        if current_count <= limit:
            return True, limit - current_count
        else:
            redis_client.zrem(key, str(current_time))
            return False, 0

    except redis.RedisError as e:
        print(f"Redis error: {e}")
        return True, -1


@app.before_request
def rate_limit_middleware():
    user_id = request.headers.get("X-User-ID")
    if not user_id:
        return jsonify({"error": "Missing user ID"}), 400

    allowed, remaining = check_rate_limit_improved(user_id, redis_client)

    if not allowed:
        return jsonify({"error": "Rate limit exceeded", "retry_after": 60}), 429

    g.rate_limit_remaining = remaining


@app.after_request
def add_rate_limit_headers(response):
    remaining = getattr(g, "rate_limit_remaining", None)
    if remaining is not None:
        response.headers["X-RateLimit-Limit"] = "100"
        response.headers["X-RateLimit-Remaining"] = str(remaining)
    return response


@app.get("/hello")
def hello():
    return jsonify({"message": "within your limit"})