AC-6762::Random Integration B2B failures

Author: glo82145

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

@@ -10,13 +10,12 @@
 
 use Magento\Framework\MessageQueue\PublisherInterface;
 use Magento\Framework\OsInfo;
-use Magento\TestFramework\Helper\Bootstrap;
 use Magento\TestFramework\Helper\Amqp;
 
 class PublisherConsumerController
 {
     /**
-     * @var string
+     * @var string[]
      */
     private $consumers = [];
 
@@ -50,11 +49,6 @@ class PublisherConsumerController
      */
     private $amqpHelper;
 
-    /**
-     * @var ClearQueueProcessor
-     */
-    private $clearQueueProcessor;
-
     /**
      * PublisherConsumerController constructor.
      * @param PublisherInterface $publisher
@@ -64,7 +58,6 @@ class PublisherConsumerController
      * @param array $consumers
      * @param array $appInitParams
      * @param null|int $maxMessages
-     * @param ClearQueueProcessor $clearQueueProcessor
      */
     public function __construct(
         PublisherInterface $publisher,
@@ -73,8 +66,7 @@ public function __construct(
         $logFilePath,
         $consumers,
         $appInitParams,
-        $maxMessages = null,
-        ClearQueueProcessor $clearQueueProcessor = null
+        $maxMessages = null
     ) {
         $this->consumers = $consumers;
         $this->publisher = $publisher;
@@ -83,8 +75,6 @@ public function __construct(
         $this->osInfo = $osInfo;
         $this->appInitParams = $appInitParams;
         $this->amqpHelper = $amqpHelper;
-        $this->clearQueueProcessor = $clearQueueProcessor
-            ?: Bootstrap::getObjectManager()->get(ClearQueueProcessor::class);
     }
 
     /**
@@ -97,7 +87,12 @@ public function initialize()
     {
         $this->validateEnvironmentPreconditions();
 
-        $this->clearQueueProcessor->execute("async.operations.all");
+        $connections = $this->amqpHelper->getConnections();
+        foreach (array_keys($connections) as $connectionName) {
+            $this->amqpHelper->deleteConnection($connectionName);
+        }
+        $this->amqpHelper->clearQueue("async.operations.all");
+
         $this->stopConsumers();
         $this->startConsumers();
 
@@ -125,6 +120,12 @@ private function validateEnvironmentPreconditions()
                 "This test relies on *nix shell and should be skipped in Windows environment."
             );
         }
+
+        if (!$this->amqpHelper->isAvailable()) {
+            throw new PreconditionFailedException(
+                'This test relies on RabbitMQ Management Plugin.'
+            );
+        }
     }
 
     /**