Merge pull request #18 from driftphp/feature/added-async-commands-test

Added async commands tests

Author: mmoreram

Async/AMQPAdapter.php

@@ -106,7 +106,7 @@ public function dropInfrastructure(OutputPrinter $outputPrinter): PromiseInterfa
     {
         return $this
             ->channel
-            ->queueDelete($this->queueName, false, true)
+            ->queueDelete($this->queueName, false, false)
             ->then(function () use ($outputPrinter) {
                 (new CommandBusLineMessage(sprintf('Queue with name %s deleted properly', $this->queueName)))->print($outputPrinter);
             }, function (ClientException $exception) use ($outputPrinter) {

Tests/Async/AsyncAdapterTest.php

@@ -25,6 +25,7 @@
 use Drift\CommandBus\Tests\CommandHandler\ChangeYetAnotherThingHandler;
 use Drift\CommandBus\Tests\Context;
 use Drift\CommandBus\Tests\Middleware\Middleware1;
+use Symfony\Component\Process\Process;
 use function Clue\React\Block\await;
 use function Clue\React\Block\awaitAll;
 
@@ -105,8 +106,6 @@ public function testInfrastructure()
 
     /**
      * Test by reading only 1 command.
-     *
-     * @group async1
      */
     public function test1Command()
     {
@@ -129,22 +128,21 @@ public function test1Command()
         await($promise2, $this->getLoop());
         await($promise3, $this->getLoop());
 
-        $this->assertNull($this->getContextValue('middleware1'));
+        if ($this instanceof InMemoryAsyncTest) {
+            $this->assertNull($this->getContextValue('middleware1'));
+        }
+
         $this->assertFalse(file_exists('/tmp/a.thing'));
-        $output = $this->runCommand([
-            'command-bus:consume-commands',
-            '--limit' => 1,
-        ]);
+        $output = $this->consumeCommands(1);
 
         $this->assertContains("\033[01;32mConsumed\033[0m ChangeAThing", $output);
         $this->assertNotContains("\033[01;32mConsumed\033[0m ChangeAnotherThing", $output);
-        $this->assertTrue($this->getContextValue('middleware1'));
-        $this->assertTrue(file_exists('/tmp/a.thing'));
+        if ($this instanceof InMemoryAsyncTest) {
+            $this->assertTrue($this->getContextValue('middleware1'));
+        }
 
-        $output2 = $this->runCommand([
-            'command-bus:consume-commands',
-            '--limit' => 1,
-        ]);
+        $this->assertTrue(file_exists('/tmp/a.thing'));
+        $output2 = $this->consumeCommands(1);
 
         $this->assertNotContains("\033[01;32mConsumed\033[0m ChangeAThing", $output2);
         $this->assertContains("\033[01;36mIgnored \033[0m ChangeBThing", $output2);
@@ -153,8 +151,6 @@ public function test1Command()
 
     /**
      * Test by reading 2 commands.
-     *
-     * @group async2
      */
     public function test2Commands()
     {
@@ -174,27 +170,60 @@ public function test2Commands()
 
         awaitAll($promises, $this->getLoop());
 
-        $output = $this->runCommand([
-            'command-bus:consume-commands',
-            '--limit' => 2,
-        ]);
+        $output = $this->consumeCommands(2);
 
         $this->assertContains("\033[01;32mConsumed\033[0m ChangeAThing", $output);
         $this->assertContains("\033[01;32mConsumed\033[0m ChangeAnotherThing", $output);
         $this->assertNotContains("\033[01;32mConsumed\033[0m ChangeYetAnotherThing", $output);
-
-        $output = $this->runCommand([
-            'command-bus:consume-commands',
-            '--limit' => 1,
-        ]);
+        $output = $this->consumeCommands(1);
 
         $this->assertNotContains("\033[01;32mConsumed\033[0m ChangeAThing", $output);
         $this->assertNotContains("\033[01;32mConsumed\033[0m ChangeAnotherThing", $output);
         $this->assertContains("\033[01;32mConsumed\033[0m ChangeYetAnotherThing", $output);
     }
 
+    /**
+     * Test async commands
+     */
+    public function testAsyncCommands()
+    {
+        $this->resetInfrastructure();
+
+        $process = $this->runAsyncCommand([
+            'command-bus:consume-commands'
+        ]);
+
+        usleep(200000);
+
+        $promise1 = $this
+            ->getCommandBus()
+            ->execute(new ChangeAThing('thing'));
+
+        await($promise1, $this->getLoop());
+        usleep(200000);
+
+        $promises = [];
+        $promises[] = $this
+            ->getCommandBus()
+            ->execute(new ChangeAnotherThing('thing'));
+
+        $promises[] = $this
+            ->getCommandBus()
+            ->execute(new ChangeYetAnotherThing('thing'));
+
+        awaitAll($promises, $this->getLoop());
+        usleep(200000);
+        $output = $process->getOutput();
+        $this->assertContains("\033[01;32mConsumed\033[0m ChangeAThing", $output);
+        $this->assertContains("\033[01;32mConsumed\033[0m ChangeAnotherThing", $output);
+        $this->assertContains("\033[01;32mConsumed\033[0m ChangeYetAnotherThing", $output);
+        $process->stop();
+    }
+
     /**
      * Reset infrastructure.
+     *
+     * We wait .1 second to sure that the infrastructure is properly created
      */
     private function resetInfrastructure()
     {
@@ -228,6 +257,27 @@ private function createInfrastructure(): string
         ]);
     }
 
+    /**
+     * Consume commands
+     *
+     * @param int $limit
+     *
+     * @return string
+     */
+    protected function consumeCommands(int $limit): string
+    {
+        $process = $this->runAsyncCommand([
+            'command-bus:consume-commands',
+            "--limit=$limit",
+        ]);
+
+        while($process->isRunning()) {
+            usleep(100000);
+        }
+
+        return $process->getOutput();
+    }
+
     /**
      * Check infrastructure.
      *

Tests/Async/InMemoryAsyncTest.php

@@ -18,9 +18,9 @@
 use Drift\CommandBus\Async\InMemoryAdapter;
 
 /**
- * Class FilesystemAsyncAdapterTest.
+ * Class InMemoryAsyncTest.
  */
-class FilesystemAsyncAdapterTest extends AsyncAdapterTest
+class InMemoryAsyncTest extends AsyncAdapterTest
 {
     /**
      * {@inheritdoc}
@@ -40,4 +40,27 @@ public function testAdapterIsPublic()
     {
         $this->assertInstanceOf(InMemoryAdapter::class, $this->get(InMemoryAdapter::class));
     }
+
+    /**
+     * Test async commands
+     */
+    public function testAsyncCommands()
+    {
+        $this->markTestSkipped('InMemory adapter should not fire this test');
+    }
+
+    /**
+     * Consume commands
+     *
+     * @param int $limit
+     *
+     * @return string
+     */
+    protected function consumeCommands(int $limit): string
+    {
+        return $this->runCommand([
+            'command-bus:consume-commands',
+            "--limit" => $limit,
+        ]);
+    }
 }

Tests/autowiring.yml

@@ -7,6 +7,7 @@ services:
 
   Drift\CommandBus\Tests\Middleware\:
     resource: 'Middleware/.'
+    exclude: 'Middleware/*Test.php'
 
   Drift\CommandBus\Tests\QueryHandler\:
     resource: 'QueryHandler/.'

Tests/docker-compose.yml

@@ -7,6 +7,16 @@ services:
       - "6379:6379"
 
   amqp:
-    image: "rabbitmq:3"
+    image: "rabbitmq:3-management"
     ports:
       - "5672:5672"
+      - "15672:15672"
+
+  postgres:
+    image: "postgres:alpine"
+    environment:
+      POSTGRES_PASSWORD: root
+      POSTGRES_USER: root
+      POSTGRES_DB: commands
+    ports:
+      - "5432:5432"

composer.json

@@ -16,16 +16,17 @@
         "symfony/console": "^5.0.0",
         "symfony/dependency-injection": "^5.0.0",
         "symfony/proxy-manager-bridge": "^5.0.0",
+        "symfony/process": "^5.0.0",
 
-        "mmoreram/base-bundle": "^2.1",
+        "mmoreram/base-bundle": "^2.1.2",
         "drift/http-kernel": "0.1.*, >=0.1.5",
         "drift/console-bridge": "0.1.*",
         "clue/block-react": "^1.3"
     },
     "require-dev": {
         "drift/redis-bundle": "0.1.*, >=0.1.2",
         "drift/amqp-bundle": "0.1.*, >=0.1.1",
-        "drift/postgresql-bundle": "0.1.*"
+        "drift/postgresql-bundle": "0.1.*, >=0.1.1"
     },
     "suggest": {
         "drift/redis-bundle": "Install this bundle to use Redis as a queue system",