magento-engcom/bulk-api#4 Support for Async operations in WebAPI

- Fixed integration tests

Author: vrann

dev/tests/integration/framework/Magento/TestFramework/Helper/Amqp.php

@@ -3,9 +3,10 @@
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
-namespace Magento\TestFramework\Helper;
 
-use Magento\Framework\HTTP\Client\Curl;
+declare(strict_types=1);
+
+namespace Magento\TestFramework\Helper;
 
 /**
  * Helper class to access RabbitMQ server configuration
@@ -62,4 +63,51 @@ public function getExchangeBindings($name)
         $data = $this->curl->getBody();
         return json_decode($data, true);
     }
+
+    /**
+     * Get All available connections
+     *
+     * @return array
+     */
+    public function getConnections()
+    {
+        $this->curl->get($this->host . 'connections');
+        $data = $this->curl->getBody();
+        $data = json_decode($data, true);
+        $output = [];
+        foreach ($data as $value) {
+            $output[$value['name']] = $value;
+        }
+        return $output;
+    }
+
+    /**
+     * @param string $name
+     * @param int $numMessages
+     * @return string
+     */
+    public function clearQueue(string $name, int $numMessages = 50)
+    {
+        $body = [
+            "count" => $numMessages,
+            "ackmode" => "ack_requeue_false",
+            "encoding" => "auto",
+            "truncate" => 50000
+        ];
+        $this->curl->post($this->host . 'queue/%2f/' . $name . '/get', json_encode($body));
+        return $this->curl->getBody();
+    }
+
+    /**
+     * Delete connection
+     *
+     * @param $name
+     * @return string $data
+     */
+    public function deleteConnection($name)
+    {
+        $this->curl->delete($this->host . 'conections/' . urlencode($name));
+        $data = $this->curl->getBody();
+        return $data;
+    }
 }

dev/tests/integration/framework/Magento/TestFramework/Helper/Curl.php

@@ -0,0 +1,30 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+
+declare(strict_types=1);
+
+namespace Magento\TestFramework\Helper;
+
+use Magento\Framework\HTTP\Client\Curl as CurlLibrary;
+
+class Curl extends CurlLibrary
+{
+    /**
+     * Make DELETE request
+     *
+     * String type was added to parameter $param in order to support sending JSON or XML requests.
+     * This feature was added base on Community Pull Request https://github.com/magento/magento2/pull/8373
+     *
+     * @param string $uri
+     * @return void
+     *
+     * @see \Magento\Framework\HTTP\Client#post($uri, $params)
+     */
+    public function delete($uri)
+    {
+        $this->makeRequest("DELETE", $uri);
+    }
+}

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

@@ -3,10 +3,14 @@
  * Copyright © Magento, Inc. All rights reserved.
  * See COPYING.txt for license details.
  */
+
+declare(strict_types=1);
+
 namespace Magento\TestFramework\MessageQueue;
 
 use Magento\Framework\MessageQueue\PublisherInterface;
 use Magento\Framework\OsInfo;
+use Magento\TestFramework\Helper\Amqp;
 
 class PublisherConsumerController
 {
@@ -40,9 +44,15 @@ class PublisherConsumerController
      */
     private $appInitParams;
 
+    /**
+     * @var Amqp
+     */
+    private $amqpHelper;
+
     public function __construct(
         PublisherInterface $publisher,
         OsInfo $osInfo,
+        Amqp $amqpHelper,
         $logFilePath,
         $consumers,
         $appInitParams,
@@ -54,6 +64,7 @@ public function __construct(
         $this->maxMessages = $maxMessages;
         $this->osInfo = $osInfo;
         $this->appInitParams = $appInitParams;
+        $this->amqpHelper = $amqpHelper;
     }
 
     /**
@@ -69,10 +80,16 @@ public function initialize()
                 "This test relies on *nix shell and should be skipped in Windows environment."
             );
         }
+        $connections = $this->amqpHelper->getConnections();
+        foreach ($connections as $connectionName => $connectionData) {
+            $this->amqpHelper->deleteConnection($connectionName);
+        }
+        $this->amqpHelper->clearQueue("async.operations.all");
         foreach ($this->consumers as $consumer) {
             foreach ($this->getConsumerProcessIds($consumer) as $consumerProcessId) {
                 exec("kill {$consumerProcessId}");
             }
+
         }
         foreach ($this->consumers as $consumer) {
             if (!$this->getConsumerProcessIds($consumer)) {