data_pipelines-1.x-dev/tests/src/Kernel/DestinationQueueTest.php

<?php

declare(strict_types=1);

namespace Drupal\Tests\data_pipelines\Kernel;

use Drupal\Core\Queue\QueueWorkerManagerInterface;
use Drupal\Tests\data_pipelines\Traits\DatasetTestProcessingTrait;
use Drupal\data_pipelines\Destination\ProcessingOperation;
use Drupal\data_pipelines\Destination\ProcessingOperationEnum;
use Drupal\data_pipelines\Entity\DatasetInterface;
use Drupal\data_pipelines\Form\DatasetBatchContext;

/**
 * Defines a class for testing queue indexing.
 *
 * @group data_pipelines
 *
 * @covers \Drupal\data_pipelines\EntityHandlers\DatasetStorage
 * @covers \Drupal\data_pipelines\Plugin\QueueWorker\DestinationWorker
 */
class DestinationQueueTest extends DatasetKernelTestBase {
  use DatasetTestProcessingTrait;

  /**
   * Tests queue derivation.
   *
   * @covers \Drupal\data_pipelines\Plugin\Derivative\DatasetQueueWorkerDeriver
   */
  public function testQueueDerivation() {
    $queue = \Drupal::service('plugin.manager.queue_worker');
    assert($queue instanceof QueueWorkerManagerInterface);
    $this->assertCount(0, array_filter(array_keys($queue->getDefinitions()), function (string $id) {
      return strpos($id, 'data_pipelines_process') === 0;
    }));
    $dataset = $this->createTestDataset();
    $this->assertCount(1, array_filter(array_keys($queue->getDefinitions()), function (string $id) {
      return strpos($id, 'data_pipelines_process') === 0;
    }));
    $dataset->delete();
    $this->assertCount(0, array_filter(array_keys($queue->getDefinitions()), function (string $id) {
      return strpos($id, 'data_pipelines_process') === 0;
    }));
  }

  /**
   * Tests indexing.
   */
  public function testQueueWorker() {
    $destinationId = $this->randomMachineName();
    $dataset = $this->createTestDataset(['destinationId' => $destinationId]);
    $queue_id = $dataset->getProcessingQueueId();
    $process_queue = \Drupal::queue($queue_id);
    $dataset->queueProcessing($process_queue, new DatasetBatchContext($dataset->getBatchSize()));
    // The state destination wants items one at a time.
    $this->assertEquals(4, $process_queue->numberOfItems());
    $this->processQueue($queue_id, $destinationId, [
      ProcessingOperationEnum::Begin,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::End,
    ]);

    $data = \Drupal::state()->get('data_pipelines_test');
    $this->assertEquals([$dataset->id() => iterator_to_array($dataset->getDataIterator())], $data);
    $this->assertFalse($process_queue->claimItem());

    $expected = $data[$dataset->id()][0];

    $this->assertTrue($expected['should_we']);
    $this->assertEquals('bloggs, joe', $expected['full_name']);

    // Set unpublished.
    $dataset->setUnpublished();
    $dataset->save();
    $data = \Drupal::state()->get('data_pipelines_test');
    $this->assertEmpty($data);

    // Set published.
    $dataset->setPublished();
    $dataset->save();
    $dataset->queueProcessing($process_queue, new DatasetBatchContext($dataset->getBatchSize()));
    // The state destination wants items one at a time.
    $this->assertEquals(4, $process_queue->numberOfItems());
    $this->processQueue($queue_id, $destinationId, [
      ProcessingOperationEnum::Begin,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::End,
    ]);

    $data = \Drupal::state()->get('data_pipelines_test');
    $this->assertEquals([$dataset->id() => iterator_to_array($dataset->getDataIterator())], $data);

    // Clean up.
    $dataset->delete();
    $data = \Drupal::state()->get('data_pipelines_test');
    $this->assertEmpty($data);
  }

  /**
   * Tests queue items are per destination.
   */
  public function testMultipleDestinations(): void {
    $id1 = $this->randomMachineName();
    $id2 = $this->randomMachineName();
    $destinations = [
      $this->createTestMemoryDestination(['id' => $id1]),
      $this->createTestMemoryDestination(['id' => $id2]),
    ];
    $dataset = $this->createTestDataset(['destinations' => $destinations]);
    $process_queue = \Drupal::queue($dataset->getProcessingQueueId());
    $dataset->queueProcessing($process_queue, new DatasetBatchContext($dataset->getBatchSize()));

    // The state destination wants items one at a time and there is also an end
    // and begin task. So four items, and two destinations, so should be eight
    // in total.
    $this->assertEquals(8, $process_queue->numberOfItems());
  }

  /**
   * Tests cleanup queue creation.
   */
  public function testCleanup(): void {
    $destinationId = $this->randomMachineName();
    $file = $this->getTestFile(dirname(__DIR__, 1) . '/fixtures/test-pipeline-1-invalid.csv');
    $dataset = $this->createTestDataset([
      'source' => 'csv:file',
      'name' => $this->randomMachineName(),
      'machine_name' => mb_strtolower($this->randomMachineName()),
      'pipeline' => 'test_pipeline_1',
      'destinationId' => $destinationId,
      'csv_file' => $file,
      'invalid_values' => DatasetInterface::INVALID_REMOVE,
    ]);
    $queue_id = $dataset->getProcessingQueueId();
    $process_queue = \Drupal::queue($queue_id);
    $dataset->queueProcessing($process_queue, new DatasetBatchContext($dataset->getBatchSize()));
    // The state destination wants items one at a time but two are invalid.
    // So we should have begin, process, end, cleanup totalling 4.
    $this->assertEquals(4, $process_queue->numberOfItems());
    $process_queue->claimItem();
    $process_queue->claimItem();
    $cleanup = $process_queue->claimItem()->data;
    assert($cleanup instanceof ProcessingOperation);
    $this->assertEquals(ProcessingOperationEnum::ProcessCleanup, $cleanup->getOperation());
    $this->assertEquals($destinationId, $cleanup->getDestinationId());
    $this->assertEquals([0, 1], $cleanup->getProcessingData());
  }

  /**
   * Tests outdated data removal.
   */
  public function testOutdatedCleanup(): void {
    $dataset = $this->createTestDataset([
      'invalid_values' => DatasetInterface::INVALID_REMOVE,
    ]);
    $destination = $dataset->getDestinations()[0];
    $destination_plugin = $destination->getDestinationPlugin();
    $this->assertEquals(0, $destination_plugin->getLastDelta($dataset));
    $queue_id = $dataset->getProcessingQueueId();
    /** @var \Drupal\Core\Queue\QueueWorkerInterface $queue_worker */
    $process_queue = \Drupal::queue($queue_id);
    $dataset->queueProcessing($process_queue, new DatasetBatchContext($dataset->getBatchSize()));
    // The state destination wants items one at a time.
    $this->assertEquals(4, $process_queue->numberOfItems());
    $this->processQueue($queue_id, $destination->id(), [
      ProcessingOperationEnum::Begin,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::End,
    ]);
    $this->assertEquals(2, $destination_plugin->getLastDelta($dataset));
    // Add some more data to the dataset including some invalid data.
    $csv = <<<CSV
should_we,firstname,lastname
Y,joe,bloggs
N,betty,bloggs
N,li,bloggs
Y,nick,bloggs
N,nora,bloggs
CSV;
    $dataset->csv_text->value = $csv;
    $dataset->save();
    $dataset->queueProcessing($process_queue, new DatasetBatchContext($dataset->getBatchSize()));
    // The state destination wants items one at a time. Item 3 is invalid.
    $this->assertEquals(7, $process_queue->numberOfItems());
    $this->processQueue($queue_id, $destination->id(), [
      ProcessingOperationEnum::Begin,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::ProcessCleanup,
      ProcessingOperationEnum::End,
    ]);
    $this->assertEquals(4, $destination_plugin->getLastDelta($dataset));
    // Remove some more data to the dataset.
    $csv = <<<CSV
should_we,firstname,lastname
N,li,bloggs
Y,nick,bloggs
N,nora,bloggs
CSV;
    $dataset->csv_text->value = $csv;
    $dataset->save();
    $dataset->queueProcessing($process_queue, new DatasetBatchContext($dataset->getBatchSize()));
    // The state destination wants items one at a time. Item 1 is valid.
    $this->assertEquals(6, $process_queue->numberOfItems());
    $this->processQueue($queue_id, $destination->id(), [
      ProcessingOperationEnum::Begin,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::ProcessChunk,
      ProcessingOperationEnum::ProcessCleanup,
      ProcessingOperationEnum::ProcessCleanup,
      ProcessingOperationEnum::End,
    ]);
    $this->assertEquals(2, $destination_plugin->getLastDelta($dataset));
  }

}