opsxsql/tests/test_redis.py

"""Tests for Redis management API endpoints — all methods mocked."""
import pytest
pytestmark = pytest.mark.skip(reason="Go backend endpoint not implemented")

from unittest.mock import AsyncMock, patch

from tests.conftest import auth_header


# ── Helpers ──────────────────────────────────────────────────────

REDIS_INSTANCE_DATA = {
    "name": "Test Redis",
    "db_type": "redis",
    "host": "127.0.0.1",
    "port": 6379,
    "username": "default",
    "password": "test123",
    "database": "0",
    "env": "development",
    "description": "Test Redis instance",
}

MOCK_INFO = {
    "redis_version": "7.2.3",
    "uptime_in_days": 30,
    "connected_clients": 42,
    "db0": {"keys": 1000, "expires": 500},
    "used_memory_human": "256M",
    "keyspace_hits": 10000,
    "keyspace_misses": 100,
}


async def _create_redis_instance(client, admin_token):
    """Create a Redis instance and return its ID."""


    resp = await client.post(
        "/api/v1/instances",
        json=REDIS_INSTANCE_DATA,
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 201
    return resp.json()["id"]


# ── Info endpoint ────────────────────────────────────────────────

@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_get_redis_info(mock_cls, client, admin_token):
    mock_cls.get_info = AsyncMock(return_value=MOCK_INFO)
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.get(
        "/api/v1/redis/" + str(iid) + "/info",
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["redis_version"] == "7.2.3"


# ── DBSIZE endpoint ──────────────────────────────────────────────

@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_get_redis_dbsize(mock_cls, client, admin_token):
    mock_cls.get_dbsize = AsyncMock(return_value=1000)
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.get(
        "/api/v1/redis/" + str(iid) + "/dbsize?db=0",
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["size"] == 1000


# ── Keys list ────────────────────────────────────────────────────

@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_get_redis_keys(mock_cls, client, admin_token):
    mock_keys = [
        {"key": "user:1", "type": "string", "ttl": 3600},
        {"key": "session:abc", "type": "hash", "ttl": -1},
    ]
    mock_cls.scan_keys = AsyncMock(return_value=(0, mock_keys))
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.get(
        "/api/v1/redis/" + str(iid) + "/keys?db=0&pattern=*&count=50",
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200
    data = resp.json()
    assert len(data["keys"]) == 2
    assert data["keys"][0]["key"] == "user:1"


# ── Key detail ───────────────────────────────────────────────────

@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_get_redis_key_detail(mock_cls, client, admin_token):
    mock_cls.get_key_value = AsyncMock(return_value={
        "key": "user:1", "type": "string", "value": "hello", "ttl": 3600,
    })
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.get(
        "/api/v1/redis/" + str(iid) + "/keys/user:1?db=0",
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["value"] == "hello"


# ── Key CRUD ─────────────────────────────────────────────────────

@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_create_redis_key(mock_cls, client, admin_token):
    mock_cls.set_key_value = AsyncMock(return_value="OK")
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.post(
        "/api/v1/redis/" + str(iid) + "/keys",
        json={"db": 0, "key": "test_key", "value": "test_value", "type": "string", "ttl": 60},
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200
    assert resp.json()["message"] == "操作成功"


@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_delete_redis_key(mock_cls, client, admin_token):
    mock_cls.del_key = AsyncMock(return_value=1)
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.delete(
        "/api/v1/redis/" + str(iid) + "/keys/test_key?db=0",
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200
    assert resp.json()["deleted"] == 1


@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_set_expire(mock_cls, client, admin_token):
    mock_cls.set_ttl = AsyncMock(return_value=True)
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.put(
        "/api/v1/redis/" + str(iid) + "/keys/test_key/expire",
        json={"db": 0, "seconds": 300},
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200


# ── Command execution ────────────────────────────────────────────

@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_execute_command(mock_cls, client, admin_token):
    mock_cls.execute_command = AsyncMock(return_value="PONG")
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.post(
        "/api/v1/redis/" + str(iid) + "/command",
        json={"db": 0, "command": "PING"},
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200
    assert resp.json()["output"] == "PONG"


# ── Slowlog & Clients ────────────────────────────────────────────

@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_get_slowlog(mock_cls, client, admin_token):
    mock_cls.get_slowlog = AsyncMock(return_value=[
        {"id": 1, "timestamp": 1700000000, "duration_us": 5000, "command": "KEYS *"},
    ])
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.get(
        "/api/v1/redis/" + str(iid) + "/slowlog?count=10",
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200
    assert len(resp.json()) == 1


@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_get_clients(mock_cls, client, admin_token):
    mock_cls.get_clients = AsyncMock(return_value=[
        {"id": "1", "addr": "127.0.0.1:54321", "cmd": "ping"},
    ])
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.get(
        "/api/v1/redis/" + str(iid) + "/clients",
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200
    assert len(resp.json()) == 1


# ── Permission checks ────────────────────────────────────────────

@pytest.mark.asyncio
async def test_create_key_requires_admin_or_dba(client, developer_token, admin_token):
    """Non-admin/dba cannot create keys."""
    # Need a redis instance first
    iid_data = dict(REDIS_INSTANCE_DATA)
    resp = await client.post(
        "/api/v1/instances",
        json=iid_data,
        headers=auth_header(admin_token),
    )
    iid = resp.json()["id"]

    resp = await client.post(
        "/api/v1/redis/" + str(iid) + "/keys",
        json={"db": 0, "key": "k", "value": "v", "type": "string"},
        headers=auth_header(developer_token),
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_flushdb_requires_admin(client, dba_token, admin_token):
    """Non-admin cannot flush DB."""
    iid_data = dict(REDIS_INSTANCE_DATA)
    resp = await client.post(
        "/api/v1/instances",
        json=iid_data,
        headers=auth_header(admin_token),
    )
    iid = resp.json()["id"]

    resp = await client.delete(
        "/api/v1/redis/" + str(iid) + "/flushdb?db=0",
        headers=auth_header(dba_token),
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
@patch("app.api.v1.endpoints.redis.redis_connector.RedisConnector")
async def test_flushdb_admin_can(mock_cls, client, admin_token):
    """Admin can flush DB."""
    mock_cls.flush_db = AsyncMock(return_value="OK")
    iid = await _create_redis_instance(client, admin_token)

    resp = await client.delete(
        "/api/v1/redis/" + str(iid) + "/flushdb?db=0",
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 200


# ── Non-Redis instance rejection ──────────────────────────────────

@pytest.mark.asyncio
async def test_non_redis_instance_rejected(client, admin_token):
    """A MySQL instance should be rejected on Redis endpoints."""
    mysql_data = {
        "name": "Test MySQL",
        "db_type": "mysql",
        "host": "127.0.0.1",
        "port": 3306,
        "username": "root",
        "password": "test123",
        "database": "test_db",
        "env": "development",
    }
    resp = await client.post(
        "/api/v1/instances",
        json=mysql_data,
        headers=auth_header(admin_token),
    )
    iid = resp.json()["id"]

    resp = await client.get(
        "/api/v1/redis/" + str(iid) + "/info",
        headers=auth_header(admin_token),
    )
    assert resp.status_code == 400
    assert "Redis" in resp.json()["detail"]