<?php

/*
 * This file is part of the Predis package.
 *
 * (c) 2009-2020 Daniele Alessandri
 * (c) 2021-2026 Till Krüss
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Predis\Transaction\Strategy;

use PHPUnit\Framework\MockObject\MockObject;
use PHPUnit\Framework\TestCase;
use Predis\ClientException;
use Predis\Command\CommandInterface;
use Predis\Command\Redis\DISCARD;
use Predis\Command\Redis\EXEC;
use Predis\Command\Redis\GET;
use Predis\Command\Redis\MULTI;
use Predis\Command\Redis\SET;
use Predis\Command\Redis\UNWATCH;
use Predis\Command\Redis\WATCH;
use Predis\Connection\Cluster\ClusterInterface;
use Predis\Connection\Cluster\RedisCluster;
use Predis\Connection\ConnectionException;
use Predis\Connection\NodeConnectionInterface;
use Predis\Connection\Resource\Exception\StreamInitException;
use Predis\Response\Error;
use Predis\Response\ErrorInterface;
use Predis\Response\ServerException;
use Predis\Response\Status;
use Predis\TimeoutException;
use Predis\Transaction\Exception\TransactionException;
use Predis\Transaction\MultiExecState;
use Predis\Transaction\Response\BypassTransactionResponse;
use RuntimeException;
use Throwable;

class ClusterConnectionStrategyTest extends TestCase
{
    /**
     * @var ClusterInterface
     */
    private $mockConnection;

    /**
     * @var \Predis\Cluster\StrategyInterface
     */
    private $mockStrategy;

    /**
     * @var NodeConnectionInterface
     */
    private $mockNodeConnection;

    /**
     * @var CommandInterface
     */
    private $mockCommand;

    /**
     * IDs of the commands executed through the cluster instead of the node.
     *
     * @var string[]
     */
    private $clusterCommands = [];

    /**
     * @return void
     */
    protected function setUp(): void
    {
        $this->mockConnection = $this->getMockBuilder(ClusterInterface::class)->getMock();
        $this->mockStrategy = $this->getMockBuilder(\Predis\Cluster\StrategyInterface::class)->getMock();
        $this->mockNodeConnection = $this->getMockBuilder(NodeConnectionInterface::class)->getMock();
        $this->mockConnection->method('getConnectionByCommand')->willReturn($this->mockNodeConnection);
        $this->mockConnection->method('executeCommand')->willReturnCallback($this->routeToMockNode());
        $this->mockCommand = $this->getMockBuilder(CommandInterface::class)->getMock();

        $this->mockConnection
            ->method('getClusterStrategy')
            ->willReturn($this->mockStrategy);
    }

    /**
     * @group disconnected
     * @return void
     */
    public function testExecuteCommandThrowsExceptionOnNonInitializedTransactionContext(): void
    {
        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());

        $this->expectException(TransactionException::class);
        $this->expectExceptionMessage('Transaction context should be initialized first');

        $strategy->executeCommand($this->mockCommand);
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testExecuteCommandAssignsHashSlotToSlotlessCommand(): void
    {
        $anotherCommand = new SET();

        $this->mockStrategy
            ->expects($this->exactly(2))
            ->method('getSlot')
            ->withConsecutive([$this->mockCommand], [$anotherCommand])
            ->willReturnOnConsecutiveCalls(10, null);

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();

        $this->assertEquals('QUEUED', $strategy->executeCommand($this->mockCommand));
        $this->assertEquals('QUEUED', $strategy->executeCommand($anotherCommand));
        $this->assertEquals(10, $anotherCommand->getSlot());
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testExecuteCommandReturnsErrorOnMissMatchingSlots(): void
    {
        $this->mockStrategy
            ->expects($this->exactly(2))
            ->method('getSlot')
            ->with($this->mockCommand)
            ->willReturnOnConsecutiveCalls(123, 124);

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();

        $strategy->executeCommand($this->mockCommand);
        $response = $strategy->executeCommand($this->mockCommand);

        $this->assertInstanceOf(ErrorInterface::class, $response);
        $this->assertSame(
            'To be able to execute a transaction against cluster, all commands should operate on the same hash slot',
            $response->getMessage()
        );
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testExecuteCommandQueueCommandsMappedToTheSameSlot(): void
    {
        $anotherMockCommand = $this->getMockBuilder(CommandInterface::class)->getMock();

        $this->mockStrategy
            ->expects($this->exactly(2))
            ->method('getSlot')
            ->withConsecutive([$this->mockCommand], [$anotherMockCommand])
            ->willReturnOnConsecutiveCalls([123], [123]);

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();

        $resp1 = $strategy->executeCommand($this->mockCommand);
        $resp2 = $strategy->executeCommand($this->mockCommand);

        $this->assertEquals('QUEUED', $resp1);
        $this->assertEquals('QUEUED', $resp2);
    }

    /**
     * @group disconnected
     * @return void
     */
    public function testInitializeTransactionContext(): void
    {
        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $this->assertTrue($strategy->initializeTransaction());
        $this->assertTrue($strategy->initializeTransaction());
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testExecuteTransactionReturnsNullOnInitializeError(): void
    {
        $this->mockNodeConnection
            ->expects($this->exactly(2))
            ->method('executeCommand')
            ->withConsecutive([new MULTI()], [new UNWATCH()])
            ->willReturnOnConsecutiveCalls('ERR', 'OK');
        $this->mockNodeConnection->expects($this->never())->method('disconnect');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        $this->assertNull($strategy->executeTransaction());
    }

    /**
     * @group disconnected
     */
    public function testRejectedNestedMultiDisconnectsTheNode(): void
    {
        $this->mockNodeConnection
            ->expects($this->exactly(2))
            ->method('executeCommand')
            ->withConsecutive([new MULTI()], [new UNWATCH()])
            ->willReturnOnConsecutiveCalls(new Error('ERR MULTI calls can not be nested'), new Status('QUEUED'));
        $this->mockNodeConnection->expects($this->once())->method('disconnect');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        $this->assertNull($strategy->executeTransaction());
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testExecuteTransactionReturnsNullOnQueueingError(): void
    {
        $this->mockNodeConnection
            ->expects($this->exactly(4))
            ->method('executeCommand')
            ->withConsecutive([new MULTI()], [new SET()], [new SET()], [new DISCARD()])
            ->willReturnOnConsecutiveCalls('OK', 'QUEUED', 'ERR', 'OK');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());
        $strategy->executeCommand(new SET());
        $strategy->executeCommand(new SET());

        $this->assertNull($strategy->executeTransaction());
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testExecuteTransactionReturnsResultOnSuccessfulTransaction(): void
    {
        $command1 = new SET();
        $command2 = new SET();
        $command3 = new SET();
        $command1->setArguments(['{foo}bar', 'value']);
        $command2->setArguments(['{foo}baz', 'value']);
        $command3->setArguments(['{foo}foo', 'value']);

        $this->mockNodeConnection
            ->expects($this->exactly(5))
            ->method('executeCommand')
            ->withConsecutive(
                [new MULTI()],
                [$command1],
                [$command2],
                [$command3],
                [new EXEC()]
            )
            ->willReturnOnConsecutiveCalls(
                'OK',
                'QUEUED',
                'QUEUED',
                'QUEUED',
                [new Status('OK'), new Status('OK'), new Status('OK')]
            );

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand($command1);
        $strategy->executeCommand($command2);
        $strategy->executeCommand($command3);

        $this->assertEquals(
            [new Status('OK'), new Status('OK'), new Status('OK')],
            $strategy->executeTransaction()
        );
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testMultiInitializesTransactionWithoutSendingMulti(): void
    {
        $this->mockNodeConnection->expects($this->never())->method('executeCommand');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $this->assertEquals('OK', $strategy->multi());
        $this->assertEquals('QUEUED', $strategy->executeCommand(new SET()));
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testWatchThrowsExceptionOnKeysPointingToDifferentSlots(): void
    {
        $this->mockStrategy
            ->expects($this->once())
            ->method('checkSameSlotForKeys')
            ->with(['key1', 'key2', 'key3'])
            ->willReturn(false);

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());

        $this->expectException(TransactionException::class);
        $this->expectExceptionMessage('WATCHed keys should point to the same hash slot');

        $strategy->watch(['key1', 'key2', 'key3']);
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testWatchReturnCorrectResponse(): void
    {
        $this->mockStrategy
            ->expects($this->once())
            ->method('checkSameSlotForKeys')
            ->with(['key1', 'key2', 'key3'])
            ->willReturn(true);

        $this->mockStrategy
            ->expects($this->once())
            ->method('getSlotByKey')
            ->with('key1')
            ->willReturn(10);

        $this->mockNodeConnection
            ->expects($this->once())
            ->method('executeCommand')
            ->withAnyParameters()
            ->willReturn(new Status('OK'));

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $this->assertTrue($strategy->watch(['key1', 'key2', 'key3']));
    }

    /**
     * @group disconnected
     * @return void
     * @throws TransactionException
     */
    public function testWatchAllowsImmediateReadsInCASTransaction(): void
    {
        $this->mockStrategy
            ->expects($this->once())
            ->method('checkSameSlotForKeys')
            ->with(['key1', 'key2', 'key3'])
            ->willReturn(true);

        $this->mockStrategy
            ->expects($this->once())
            ->method('getSlotByKey')
            ->with('key1')
            ->willReturn(10);

        $this->mockNodeConnection
            ->expects($this->exactly(2))
            ->method('executeCommand')
            ->willReturnOnConsecutiveCalls(new Status('OK'), 'value');

        $state = new MultiExecState();
        $state->set(MultiExecState::CAS);

        $strategy = new ClusterConnectionStrategy($this->mockConnection, $state);
        $this->assertTrue($strategy->watch(['key1', 'key2', 'key3']));
        $response = $strategy->executeCommand(new GET());
        $this->assertInstanceOf(BypassTransactionResponse::class, $response);
        $this->assertSame('value', $response->getResponse());
    }

    /**
     * @group disconnected
     * @return void
     */
    public function testDiscard(): void
    {
        $this->mockNodeConnection->expects($this->never())->method('executeCommand');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $this->assertEquals('OK', $strategy->discard());
    }

    /**
     * @group disconnected
     * @return void
     */
    public function testUnwatch(): void
    {
        $this->mockNodeConnection->expects($this->never())->method('executeCommand');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $this->assertEquals('OK', $strategy->unwatch());
    }

    /**
     * @group disconnected
     * @dataProvider clusterQueueErrors
     */
    public function testDiscardUsesTheNodeAndSlotThatReceivedMulti(string $error): void
    {
        $this->mockStrategy->method('getSlot')->willReturn(123);
        $this->mockConnection->expects($this->once())->method('getConnectionByCommand');
        $this->mockNodeConnection->expects($this->exactly(3))->method('executeCommand')
            ->withConsecutive([$this->callback(static function ($command) {
                return $command instanceof MULTI && $command->getSlot() === 123;
            })], [$this->callback(static function ($command) {
                return $command instanceof SET && $command->getSlot() === 123;
            })], [$this->callback(static function ($command) {
                return $command instanceof DISCARD && $command->getSlot() === 123;
            })])
            ->willReturnOnConsecutiveCalls('OK', new Error($error), 'OK');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        $this->assertNull($strategy->executeTransaction());
    }

    public function clusterQueueErrors(): array
    {
        return [
            ['OOM command not allowed'],
            ['MOVED 123 127.0.0.1:6380'],
            ['ASK 123 127.0.0.1:6380'],
            ['READONLY You cannot write against a read only replica'],
        ];
    }

    /**
     * @group disconnected
     */
    public function testFailedDiscardDisconnectsTheNode(): void
    {
        $this->mockNodeConnection->expects($this->exactly(3))->method('executeCommand')
            ->withConsecutive([new MULTI()], [new SET()], [new DISCARD()])
            ->willReturnOnConsecutiveCalls('OK', new Error('OOM'), new Error('NOPERM'));
        $this->mockNodeConnection->expects($this->once())->method('disconnect');
        $this->mockConnection->expects($this->never())->method('remove');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        $this->assertNull($strategy->executeTransaction());
    }

    /**
     * @group disconnected
     * @dataProvider transactionFailureStages
     */
    public function testExceptionDisconnectsTheNodeWithoutReplayingCommands(string $stage): void
    {
        $exception = new RuntimeException('Connection failed');
        $this->mockConnection->expects($this->once())->method('getConnectionByCommand');
        $this->mockNodeConnection->method('executeCommand')->willReturnCallback(
            static function ($command) use ($stage, $exception) {
                if ($command->getId() === $stage) {
                    throw $exception;
                }

                if ($command->getId() === 'MULTI') {
                    return 'OK';
                }

                return $stage === 'DISCARD' ? new Error('OOM') : 'QUEUED';
            }
        );
        $this->mockNodeConnection->expects($this->once())->method('disconnect');
        $this->mockConnection->expects($this->never())->method('remove');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        try {
            $strategy->executeTransaction();
            $this->fail('Expected the connection failure to be rethrown.');
        } catch (RuntimeException $caught) {
            $this->assertSame($exception, $caught);
        }

        $this->assertSame(['MULTI'], $this->clusterCommands);
    }

    public function transactionFailureStages(): array
    {
        return [['SET'], ['EXEC'], ['DISCARD']];
    }

    /**
     * @group disconnected
     */
    public function testFirstCommandIsLeftToTheClusterAndLaterOnesStayOnTheNode(): void
    {
        $this->mockStrategy->method('checkSameSlotForKeys')->willReturn(true);
        $this->mockNodeConnection->method('executeCommand')->willReturnCallback(
            static function ($command) {
                switch ($command->getId()) {
                    case 'GET':
                        return 'value';

                    case 'SET':
                        return 'QUEUED';

                    case 'EXEC':
                        return ['OK'];

                    default:
                        return 'OK';
                }
            }
        );

        $state = new MultiExecState();
        $state->flag(MultiExecState::CAS);
        $strategy = new ClusterConnectionStrategy($this->mockConnection, $state);

        $strategy->watch(['key1']);
        $strategy->executeCommand(new GET());
        $state->unflag(MultiExecState::CAS);
        $strategy->multi();
        $strategy->executeCommand(new SET());

        $this->assertSame(['OK'], $strategy->executeTransaction());
        $this->assertSame(['WATCH'], $this->clusterCommands);

        // The next transaction starts over, without a node holding any state.
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        $this->assertSame(['OK'], $strategy->executeTransaction());
        $this->assertSame(['WATCH', 'MULTI'], $this->clusterCommands);
    }

    /**
     * @group disconnected
     */
    public function testFailureOfTheFirstCommandIsRethrownWithoutTouchingAnyNode(): void
    {
        $exception = new ConnectionException($this->mockNodeConnection, 'Connection refused');
        $cluster = $this->getMockBuilder(RedisCluster::class)->disableOriginalConstructor()->getMock();
        $cluster->method('getClusterStrategy')->willReturn($this->mockStrategy);
        $cluster->expects($this->once())->method('executeCommand')->with(new MULTI())->willThrowException($exception);
        $cluster->expects($this->never())->method('getConnectionByCommand');
        $cluster->expects($this->never())->method('applyNodeFailure');
        $this->mockNodeConnection->expects($this->never())->method('executeCommand');
        $this->mockNodeConnection->expects($this->never())->method('disconnect');

        $strategy = new ClusterConnectionStrategy($cluster, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        try {
            $strategy->executeTransaction();
            $this->fail('Expected the connection failure to be rethrown.');
        } catch (ConnectionException $caught) {
            $this->assertSame($exception, $caught);
        }

        $this->expectException(TransactionException::class);

        $strategy->executeTransaction();
    }

    /**
     * @group disconnected
     * @dataProvider nodeFailures
     */
    public function testUnreachableNodeIsReportedToTheClusterWithoutReplayingCommands(string $stage, callable $failure): void
    {
        $exception = $failure($this->mockNodeConnection);
        $events = [];
        $cluster = $this->getMockRedisCluster();
        $cluster->expects($this->once())->method('applyNodeFailure')->with($this->mockNodeConnection)
            ->willReturnCallback(static function () use (&$events) {
                $events[] = 'applyNodeFailure';
            });
        $this->mockNodeConnection->method('executeCommand')->willReturnCallback(
            static function ($command) use ($stage, $exception, &$events) {
                $events[] = $id = $command->getId();

                if ($id === $stage) {
                    throw $exception;
                }

                return $id === 'MULTI' ? 'OK' : 'QUEUED';
            }
        );
        $this->mockNodeConnection->expects($this->once())->method('disconnect');

        $strategy = new ClusterConnectionStrategy($cluster, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        try {
            $strategy->executeTransaction();
            $this->fail('Expected the connection failure to be rethrown.');
        } catch (Throwable $caught) {
            $this->assertSame($exception, $caught);
        }

        $this->assertSame(array_merge($stage === 'SET' ? ['MULTI', 'SET'] : ['MULTI', 'SET', 'EXEC'], ['applyNodeFailure']), $events);
    }

    public function nodeFailures(): array
    {
        $connectionLost = static function ($connection) {
            return new ConnectionException($connection, 'Connection lost');
        };
        $connectionRefused = static function () {
            return new StreamInitException('Connection refused [tcp://127.0.0.1:6379]');
        };

        return [
            'connection lost while queueing' => ['SET', $connectionLost],
            'connection lost on EXEC' => ['EXEC', $connectionLost],
            'connection refused while queueing' => ['SET', $connectionRefused],
        ];
    }

    /**
     * @group disconnected
     * @dataProvider failuresKeepingTheTopology
     */
    public function testOtherFailuresLeaveTheTopologyUntouched(callable $failure): void
    {
        $exception = $failure($this->mockNodeConnection);
        $cluster = $this->getMockRedisCluster();
        $cluster->expects($this->never())->method('applyNodeFailure');
        $this->mockNodeConnection->method('executeCommand')->willReturnCallback(
            static function ($command) use ($exception) {
                if ($command->getId() === 'SET') {
                    throw $exception;
                }

                return 'OK';
            }
        );
        $this->mockNodeConnection->expects($this->once())->method('disconnect');

        $strategy = new ClusterConnectionStrategy($cluster, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        try {
            $strategy->executeTransaction();
            $this->fail('Expected the failure to be rethrown.');
        } catch (Throwable $caught) {
            $this->assertSame($exception, $caught);
        }
    }

    public function failuresKeepingTheTopology(): array
    {
        return [
            'timeout' => [static function ($connection) {
                return new TimeoutException($connection);
            }],
            'PHP error' => [static function () {
                return new RuntimeException('Unexpected failure');
            }],
        ];
    }

    /**
     * @group disconnected
     */
    public function testFailureToRefreshTheTopologyDoesNotHideTheOriginalException(): void
    {
        $exception = new ConnectionException($this->mockNodeConnection, 'Connection lost');
        $cluster = $this->getMockRedisCluster();
        $cluster->expects($this->once())->method('applyNodeFailure')
            ->willThrowException(new ClientException('No connections left in the pool for `CLUSTER SLOTS`'));
        $this->mockNodeConnection->method('executeCommand')->willReturnCallback(
            static function ($command) use ($exception) {
                if ($command->getId() === 'SET') {
                    throw $exception;
                }

                return 'OK';
            }
        );

        $strategy = new ClusterConnectionStrategy($cluster, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        try {
            $strategy->executeTransaction();
            $this->fail('Expected the connection failure to be rethrown.');
        } catch (ConnectionException $caught) {
            $this->assertSame($exception, $caught);
        }
    }

    /**
     * @group disconnected
     */
    public function testAbortClearsPendingCommandsBeforeReinitializing(): void
    {
        $this->mockNodeConnection->expects($this->exactly(6))->method('executeCommand')
            ->withConsecutive([new MULTI()], [new SET()], [new DISCARD()], [new MULTI()], [new GET()], [new EXEC()])
            ->willReturnOnConsecutiveCalls('OK', new Error('OOM'), 'OK', 'OK', 'QUEUED', ['value']);

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());
        $strategy->executeCommand(new SET());
        $this->assertNull($strategy->executeTransaction());

        $strategy->initializeTransaction();
        $strategy->executeCommand(new GET());
        $this->assertSame(['value'], $strategy->executeTransaction());
    }

    /**
     * @group disconnected
     */
    public function testCasWithoutWatchReadsImmediatelyAndSendsMultiOnlyOnce(): void
    {
        $this->mockNodeConnection->expects($this->exactly(4))->method('executeCommand')
            ->withConsecutive([new GET()], [new MULTI()], [new SET()], [new EXEC()])
            ->willReturnOnConsecutiveCalls('7', 'OK', 'QUEUED', [new Status('OK')]);
        $state = new MultiExecState();
        $state->flag(MultiExecState::CAS);
        $strategy = new ClusterConnectionStrategy($this->mockConnection, $state);

        $this->assertSame('7', $strategy->executeCommand(new GET())->getResponse());
        $state->unflag(MultiExecState::CAS);
        $this->assertEquals('OK', $strategy->multi());
        $strategy->executeCommand(new SET());
        $this->assertEquals([new Status('OK')], $strategy->executeTransaction());
    }

    /**
     * @group disconnected
     */
    public function testExecuteTransactionThrowsExceptionOnNonInitializedTransactionContext(): void
    {
        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());

        $this->expectException(TransactionException::class);
        $this->expectExceptionMessage('Transaction context should be initialized first');

        $strategy->executeTransaction();
    }

    /**
     * @group disconnected
     */
    public function testExecuteTransactionReturnsNullOnExecError(): void
    {
        $this->mockNodeConnection->expects($this->exactly(4))->method('executeCommand')
            ->withConsecutive([new MULTI()], [new SET()], [new EXEC()], [new UNWATCH()])
            ->willReturnOnConsecutiveCalls(
                'OK',
                'QUEUED',
                new Error('EXECABORT Transaction discarded because of previous errors.'),
                'OK'
            );
        $this->mockNodeConnection->expects($this->never())->method('disconnect');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        $this->assertNull($strategy->executeTransaction());
    }

    /**
     * @group disconnected
     */
    public function testWatchErrorUnwatchesTheNodeAndThrows(): void
    {
        $watch = new WATCH();
        $watch->setArguments(['key1', 'key2']);

        $this->mockStrategy->method('checkSameSlotForKeys')->willReturn(true);
        $this->mockNodeConnection->expects($this->exactly(2))->method('executeCommand')
            ->withConsecutive([$watch], [new UNWATCH()])
            ->willReturnOnConsecutiveCalls(new Error('NOPERM this user has no permissions'), 'OK');
        $this->mockNodeConnection->expects($this->never())->method('disconnect');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());

        $this->expectException(ServerException::class);
        $this->expectExceptionMessage('NOPERM this user has no permissions');

        $strategy->watch(['key1', 'key2']);
    }

    /**
     * @group disconnected
     */
    public function testCasReadErrorUnwatchesTheNodeAndThrows(): void
    {
        $this->mockNodeConnection->expects($this->exactly(2))->method('executeCommand')
            ->withConsecutive([new GET()], [new UNWATCH()])
            ->willReturnOnConsecutiveCalls(new Error('WRONGTYPE Operation against a key'), 'OK');
        $this->mockNodeConnection->expects($this->never())->method('disconnect');

        $state = new MultiExecState();
        $state->flag(MultiExecState::CAS);
        $strategy = new ClusterConnectionStrategy($this->mockConnection, $state);

        $this->expectException(ServerException::class);
        $this->expectExceptionMessage('WRONGTYPE Operation against a key');

        $strategy->executeCommand(new GET());
    }

    /**
     * @group disconnected
     * @dataProvider cleanupMethods
     */
    public function testCleanupUnwatchesTheNodeThatReceivedWatch(string $method): void
    {
        $this->mockStrategy->method('checkSameSlotForKeys')->willReturn(true);
        $this->mockStrategy->method('getSlotByKey')->willReturn(123);
        $this->mockConnection->expects($this->once())->method('getConnectionByCommand');
        $this->mockNodeConnection->expects($this->exactly(2))->method('executeCommand')
            ->withConsecutive([$this->isInstanceOf(WATCH::class)], [$this->callback(static function ($command) {
                return $command instanceof UNWATCH && $command->getSlot() === 123;
            })])
            ->willReturn('OK');
        $this->mockNodeConnection->expects($this->never())->method('disconnect');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->watch(['key1']);

        $this->assertEquals('OK', $strategy->$method());
        $this->assertEquals('OK', $strategy->$method());
    }

    public function cleanupMethods(): array
    {
        return [['discard'], ['unwatch']];
    }

    /**
     * @group disconnected
     */
    public function testFailedUnwatchDisconnectsTheNode(): void
    {
        $this->mockStrategy->method('checkSameSlotForKeys')->willReturn(true);
        $this->mockNodeConnection->expects($this->exactly(2))->method('executeCommand')
            ->withConsecutive([$this->isInstanceOf(WATCH::class)], [new UNWATCH()])
            ->willReturnOnConsecutiveCalls('OK', new Error('NOPERM'));
        $this->mockNodeConnection->expects($this->once())->method('disconnect');

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->watch(['key1']);

        $this->assertInstanceOf(ErrorInterface::class, $strategy->discard());
    }

    /**
     * @group disconnected
     */
    public function testDiscardClearsQueuedCommandsWithoutContactingTheNode(): void
    {
        $this->mockNodeConnection->expects($this->exactly(3))->method('executeCommand')
            ->withConsecutive([new MULTI()], [new GET()], [new EXEC()])
            ->willReturnOnConsecutiveCalls('OK', 'QUEUED', ['value']);

        $strategy = new ClusterConnectionStrategy($this->mockConnection, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());
        $this->assertEquals('OK', $strategy->discard());

        $strategy->initializeTransaction();
        $strategy->executeCommand(new GET());
        $this->assertSame(['value'], $strategy->executeTransaction());
    }

    /**
     * @group disconnected
     * @dataProvider movedStages
     */
    public function testMovedResponseUpdatesSlotMapAfterReleasingTheNode(string $stage, array $expected): void
    {
        $events = [];
        $cluster = $this->getMockRedisCluster();
        $cluster->expects($this->once())->method('applyMovedResponse')->with('123 127.0.0.1:6380')
            ->willReturnCallback(static function () use (&$events) {
                $events[] = 'applyMovedResponse';
            });
        $this->mockNodeConnection->method('executeCommand')->willReturnCallback(
            static function ($command) use ($stage, &$events) {
                $events[] = $id = $command->getId();

                if ($id === $stage) {
                    return new Error('MOVED 123 127.0.0.1:6380');
                }

                return $id === 'SET' ? 'QUEUED' : 'OK';
            }
        );

        $strategy = new ClusterConnectionStrategy($cluster, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        $this->assertNull($strategy->executeTransaction());
        $this->assertSame($expected, $events);
    }

    public function movedStages(): array
    {
        return [
            'queueing' => ['SET', ['MULTI', 'SET', 'DISCARD', 'applyMovedResponse']],
            'execution' => ['EXEC', ['MULTI', 'SET', 'EXEC', 'UNWATCH', 'applyMovedResponse']],
        ];
    }

    /**
     * @group disconnected
     */
    public function testMovedResponseToCasReadUpdatesSlotMapAndThrows(): void
    {
        $this->mockStrategy->method('checkSameSlotForKeys')->willReturn(true);
        $cluster = $this->getMockRedisCluster();
        $cluster->expects($this->once())->method('applyMovedResponse')->with('123 127.0.0.1:6380');
        $this->mockNodeConnection->expects($this->exactly(3))->method('executeCommand')
            ->withConsecutive([$this->isInstanceOf(WATCH::class)], [new GET()], [new UNWATCH()])
            ->willReturnOnConsecutiveCalls('OK', new Error('MOVED 123 127.0.0.1:6380'), 'OK');

        $state = new MultiExecState();
        $state->flag(MultiExecState::CAS);
        $strategy = new ClusterConnectionStrategy($cluster, $state);
        $strategy->watch(['key1']);

        $this->expectException(ServerException::class);
        $this->expectExceptionMessage('MOVED 123 127.0.0.1:6380');

        $strategy->executeCommand(new GET());
    }

    /**
     * @group disconnected
     */
    public function testReadOnlyResponseRefreshesTheTopologyAfterReleasingTheNode(): void
    {
        $events = [];
        $cluster = $this->getMockRedisCluster();
        $cluster->expects($this->never())->method('applyMovedResponse');
        $cluster->expects($this->once())->method('applyReadOnlyResponse')->with($this->mockNodeConnection)
            ->willReturnCallback(static function () use (&$events) {
                $events[] = 'applyReadOnlyResponse';
            });
        $this->mockNodeConnection->method('executeCommand')->willReturnCallback(
            static function ($command) use (&$events) {
                $events[] = $id = $command->getId();

                return $id === 'SET' ? new Error('READONLY You cannot write against a read only replica') : 'OK';
            }
        );

        $strategy = new ClusterConnectionStrategy($cluster, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        $this->assertNull($strategy->executeTransaction());
        $this->assertSame(['MULTI', 'SET', 'DISCARD', 'applyReadOnlyResponse'], $events);
    }

    /**
     * @group disconnected
     * @dataProvider errorsWithoutSlotOwnership
     */
    public function testOtherErrorsLeaveTheSlotMapUntouched(string $error): void
    {
        $cluster = $this->getMockRedisCluster();
        $cluster->expects($this->never())->method('applyMovedResponse');
        $cluster->expects($this->never())->method('applyReadOnlyResponse');
        $cluster->expects($this->never())->method('applyNodeFailure');
        $this->mockNodeConnection->expects($this->exactly(3))->method('executeCommand')
            ->withConsecutive([new MULTI()], [new SET()], [new DISCARD()])
            ->willReturnOnConsecutiveCalls('OK', new Error($error), 'OK');

        $strategy = new ClusterConnectionStrategy($cluster, new MultiExecState());
        $strategy->initializeTransaction();
        $strategy->executeCommand(new SET());

        $this->assertNull($strategy->executeTransaction());
    }

    public function errorsWithoutSlotOwnership(): array
    {
        return [
            ['OOM command not allowed'],
            ['ASK 123 127.0.0.1:6380'],
            ['MOVED'],
        ];
    }

    /**
     * Returns a mocked Redis cluster routing every command to the mocked node.
     *
     * @return RedisCluster|MockObject
     */
    private function getMockRedisCluster()
    {
        $cluster = $this->getMockBuilder(RedisCluster::class)->disableOriginalConstructor()->getMock();
        $cluster->method('getClusterStrategy')->willReturn($this->mockStrategy);
        $cluster->method('getConnectionByCommand')->willReturn($this->mockNodeConnection);
        $cluster->method('executeCommand')->willReturnCallback($this->routeToMockNode());

        return $cluster;
    }

    /**
     * Returns a callback emulating the cluster routing a command to the mocked
     * node, keeping track of the commands that went through the cluster.
     *
     * @return callable
     */
    private function routeToMockNode(): callable
    {
        return function (CommandInterface $command) {
            $this->clusterCommands[] = $command->getId();

            return $this->mockNodeConnection->executeCommand($command);
        };
    }
}