init

dev/tests/integration/framework/Magento/TestFramework/MessageQueue/ClearQueueProcessor.php

@@ -0,0 +1,69 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+declare(strict_types=1);
+
+namespace Magento\TestFramework\MessageQueue;
+
+use Magento\Framework\Exception\LocalizedException;
+use Magento\Framework\MessageQueue\Consumer\Config\ConsumerConfigItemInterface;
+use Magento\Framework\MessageQueue\Consumer\ConfigInterface as ConsumerConfig;
+use Magento\Framework\MessageQueue\ConsumerFactory;
+use Magento\Framework\MessageQueue\QueueRepository;
+
+/**
+ * The processor to clear message queue
+ */
+class ClearQueueProcessor
+{
+    /**
+     * @var ConsumerConfig
+     */
+    private $consumerConfig;
+
+    /**
+     * @var ConsumerFactory
+     */
+    private $consumerFactory;
+
+    /**
+     * @var QueueRepository
+     */
+    private $queueRepository;
+
+    /**
+     * ClearQueueProcessor constructor.
+     *
+     * @param ConsumerConfig $consumerConfig
+     * @param ConsumerFactory $consumerFactory
+     * @param QueueRepository $queueRepository
+     */
+    public function __construct(
+        ConsumerConfig $consumerConfig,
+        ConsumerFactory $consumerFactory,
+        QueueRepository $queueRepository
+    ) {
+        $this->consumerConfig = $consumerConfig;
+        $this->consumerFactory = $consumerFactory;
+        $this->queueRepository = $queueRepository;
+    }
+
+    /**
+     * Clear queue
+     *
+     * @param string $consumerName
+     * @throws LocalizedException
+     * return void
+     */
+    public function execute(string $consumerName): void
+    {
+        /** @var ConsumerConfigItemInterface $consumerConfig */
+        $consumerConfig = $this->consumerConfig->getConsumer($consumerName);
+        $queue = $this->queueRepository->get($consumerConfig->getConnection(), $consumerConfig->getQueue());
+        while ($message = $queue->dequeue()) {
+            $queue->acknowledge($message);
+        }
+    }
+}

dev/tests/integration/framework/Magento/TestFramework/MessageQueue/EnvironmentPreconditionException.php

@@ -0,0 +1,11 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+namespace Magento\TestFramework\MessageQueue;
+
+class EnvironmentPreconditionException extends \Exception
+{
+
+}

dev/tests/integration/framework/Magento/TestFramework/MessageQueue/PreconditionFailedException.php

@@ -0,0 +1,11 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+namespace Magento\TestFramework\MessageQueue;
+
+class PreconditionFailedException extends \Exception
+{
+
+}

dev/tests/integration/framework/Magento/TestFramework/MessageQueue/PublisherConsumerController.php

@@ -0,0 +1,241 @@
+<?php
+/**
+ * Copyright 2018 Adobe.
+ * All rights reserved.
+ */
+
+declare(strict_types=1);
+
+namespace Magento\TestFramework\MessageQueue;
+
+use Magento\Framework\MessageQueue\PublisherInterface;
+use Magento\Framework\OsInfo;
+use Magento\TestFramework\Helper\Bootstrap;
+use Magento\TestFramework\Helper\Amqp;
+
+class PublisherConsumerController
+{
+    /**
+     * @var string[]
+     */
+    private $consumers = [];
+
+    /**
+     * @var PublisherInterface
+     */
+    private $publisher;
+
+    /**
+     * @var string
+     */
+    private $logFilePath;
+
+    /**
+     * @var int|null
+     */
+    private $maxMessages = null;
+
+    /**
+     * @var OsInfo
+     */
+    private $osInfo;
+
+    /**
+     * @var array
+     */
+    private $appInitParams;
+
+    /**
+     * @var Amqp
+     */
+    private $amqpHelper;
+
+    /**
+     * @var ClearQueueProcessor
+     */
+    private $clearQueueProcessor;
+
+    /**
+     * @param PublisherInterface $publisher
+     * @param OsInfo $osInfo
+     * @param Amqp $amqpHelper
+     * @param string $logFilePath
+     * @param array $consumers
+     * @param array $appInitParams
+     * @param null|int $maxMessages
+     * @param ClearQueueProcessor $clearQueueProcessor
+     */
+    public function __construct(
+        PublisherInterface $publisher,
+        OsInfo $osInfo,
+        Amqp $amqpHelper,
+        string $logFilePath = TESTS_TEMP_DIR . '/MessageQueueTestLog.txt',
+        array $consumers = [],
+        array $appInitParams = [],
+        ?int $maxMessages = null,
+        ?ClearQueueProcessor $clearQueueProcessor = null
+    ) {
+        $this->consumers = $consumers;
+        $this->publisher = $publisher;
+        $this->logFilePath = $logFilePath;
+        $this->maxMessages = $maxMessages;
+        $this->osInfo = $osInfo;
+        $this->appInitParams = $appInitParams ?: Bootstrap::getInstance()->getAppInitParams();
+        $this->amqpHelper = $amqpHelper;
+        $this->clearQueueProcessor = $clearQueueProcessor
+            ?: Bootstrap::getObjectManager()->get(ClearQueueProcessor::class);
+    }
+
+    /**
+     * Initialize Environment and Consumers
+     *
+     * @throws EnvironmentPreconditionException
+     * @throws PreconditionFailedException
+     */
+    public function initialize()
+    {
+        $this->validateEnvironmentPreconditions();
+
+        $this->clearQueueProcessor->execute("async.operations.all");
+        $this->stopConsumers();
+        $this->startConsumers();
+
+        if (file_exists($this->logFilePath)) {
+            // try to remove before failing the test
+            unlink($this->logFilePath);
+            if (file_exists($this->logFilePath)) {
+                throw new PreconditionFailedException(
+                    "Precondition failed: test log ({$this->logFilePath}) cannot be deleted before test execution."
+                );
+            }
+        }
+    }
+
+    /**
+     * Validate environment preconditions
+     *
+     * @throws EnvironmentPreconditionException
+     * @throws PreconditionFailedException
+     */
+    private function validateEnvironmentPreconditions()
+    {
+        if ($this->osInfo->isWindows()) {
+            throw new EnvironmentPreconditionException(
+                "This test relies on *nix shell and should be skipped in Windows environment."
+            );
+        }
+    }
+
+    /**
+     * Stop Consumers
+     */
+    public function stopConsumers()
+    {
+        foreach ($this->consumers as $consumer) {
+            foreach ($this->getConsumerProcessIds($consumer) as $consumerProcessId) {
+                // exec() have to be here since this is test.
+                // phpcs:ignore Magento2.Security.InsecureFunction
+                exec("kill {$consumerProcessId}");
+            }
+        }
+    }
+
+    /**
+     * Get Consumers ProcessIds
+     *
+     * @return array
+     */
+    public function getConsumersProcessIds()
+    {
+        $consumers = [];
+        foreach ($this->consumers as $consumer) {
+            $consumers[$consumer] = $this->getConsumerProcessIds($consumer);
+        }
+        return $consumers;
+    }
+
+    /**
+     * Get Consumer ProcessIds
+     *
+     * @param string $consumer
+     * @return string[]
+     */
+    private function getConsumerProcessIds($consumer)
+    {
+        // exec() have to be here since this is test.
+        // phpcs:ignore Magento2.Security.InsecureFunction
+        exec("ps ax | grep -v grep | grep '{$this->getConsumerStartCommand($consumer)}' | awk '{print $1}'", $output);
+        return $output;
+    }
+
+    /**
+     * Get CLI command for starting specified consumer.
+     *
+     * @param string $consumer
+     * @param bool $withEnvVariables
+     * @return string
+     */
+    private function getConsumerStartCommand($consumer, $withEnvVariables = false)
+    {
+        $binDirectory = realpath(INTEGRATION_TESTS_DIR . '/bin/');
+        $magentoCli = $binDirectory . '/magento';
+        $consumerStartCommand = "php {$magentoCli} queue:consumers:start -vvv " . $consumer;
+        if ($this->maxMessages) {
+            $consumerStartCommand .= " --max-messages={$this->maxMessages}";
+        }
+        if ($withEnvVariables) {
+            $params = $this->appInitParams;
+            $params['MAGE_DIRS']['base']['path'] = BP;
+            $params = 'INTEGRATION_TEST_PARAMS="' . urldecode(http_build_query($params)) . '"';
+            $consumerStartCommand = $params . ' ' . $consumerStartCommand;
+        }
+        return $consumerStartCommand;
+    }
+
+    /**
+     * Wait for asynchronous result
+     *
+     * @param callable $condition
+     * @param array $params
+     * @throws PreconditionFailedException
+     */
+    public function waitForAsynchronousResult(callable $condition, $params = [])
+    {
+        $i = 0;
+        do {
+            sleep(10);
+            $assertion = call_user_func_array($condition, $params);
+        } while (!$assertion && ($i++ < 50));
+
+        if (!$assertion) {
+            throw new PreconditionFailedException("No asynchronous messages were processed.");
+        }
+    }
+
+    /**
+     * Get publisher
+     *
+     * @return PublisherInterface
+     */
+    public function getPublisher()
+    {
+        return $this->publisher;
+    }
+
+    /**
+     * Start consumers
+     *
+     * @return void
+     */
+    public function startConsumers(): void
+    {
+        foreach ($this->consumers as $consumer) {
+            if (!$this->getConsumerProcessIds($consumer)) {
+                // exec() have to be here since this is test.
+                // phpcs:ignore Magento2.Security.InsecureFunction
+                exec("{$this->getConsumerStartCommand($consumer, true)} > /dev/null &");
+            }
+            sleep(5);
+        }
+    }
+}