B2B-2155: Use Redis as message queue

- fix Catalog module integration tests

Author: joanhe

dev/tests/integration/testsuite/Magento/Catalog/Model/Attribute/Backend/ConsumerWebsiteAssignTest.php

@@ -14,9 +14,9 @@
 use Magento\Framework\Bulk\OperationInterface;
 use Magento\Framework\DB\Adapter\DeadlockException;
 use Magento\Framework\Exception\NoSuchEntityException;
+use Magento\Framework\MessageQueue\ConsumerFactory;
 use Magento\Framework\MessageQueue\MessageEncoder;
 use Magento\Framework\ObjectManagerInterface;
-use Magento\MysqlMq\Model\Driver\Queue;
 use Magento\Store\Api\WebsiteRepositoryInterface;
 use Magento\TestFramework\Helper\Bootstrap;
 use Magento\TestFramework\MysqlMq\DeleteTopicRelatedMessages;
@@ -43,9 +43,6 @@ class ConsumerWebsiteAssignTest extends TestCase
     /** @var ConsumerWebsiteAssign */
     private $consumer;
 
-    /** @var Queue */
-    private $queue;
-
     /** @var MessageEncoder */
     private $messageEncoder;
 
@@ -58,6 +55,9 @@ class ConsumerWebsiteAssignTest extends TestCase
     /** @var CollectionFactory */
     private $operationCollectionFactory;
 
+    /** @var ConsumerFactory */
+    private $consumerFactory;
+
     /**
      * @inheritdoc
      */
@@ -79,14 +79,11 @@ protected function setUp(): void
 
         $this->objectManager = Bootstrap::getObjectManager();
         $this->consumer = $this->objectManager->get(ConsumerWebsiteAssign::class);
-        $this->queue = $this->objectManager->create(
-            Queue::class,
-            ['queueName' => 'product_action_attribute.website.update']
-        );
         $this->messageEncoder = $this->objectManager->get(MessageEncoder::class);
         $this->productRepository = $this->objectManager->get(ProductRepositoryInterface::class);
         $this->websiteRepository = $this->objectManager->get(WebsiteRepositoryInterface::class);
         $this->operationCollectionFactory = $this->objectManager->get(CollectionFactory::class);
+        $this->consumerFactory = $this->objectManager->get(ConsumerFactory::class);
     }
 
     /**
@@ -212,9 +209,8 @@ private function assertProductWebsites(string $sku, array $expectedWebsites): vo
      */
     private function processMessages(): void
     {
-        $envelope = $this->queue->dequeue();
-        $decodedMessage = $this->messageEncoder->decode(self::TOPIC_NAME, $envelope->getBody());
-        $this->consumer->process($decodedMessage);
+        $consumer = $this->consumerFactory->get('product_action_attribute.website.update');
+        $consumer->process(1);
     }
 
     /**