data_pipelines-1.x-dev/tests/src/Kernel/Form/DatasetBatchOperationsTest.php

<?php

declare(strict_types=1);

namespace Drupal\Tests\data_pipelines\Kernel\Form;

use Drupal\Core\Messenger\MessengerInterface;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\Tests\data_pipelines\Kernel\DatasetKernelTestBase;
use Drupal\data_pipelines\Entity\DatasetInterface;
use Drupal\data_pipelines\Form\DatasetBatchContext;
use Drupal\data_pipelines\Form\DatasetBatchOperations;

/**
 * Defines a class for testing DatasetBatchOperations.
 *
 * @group data_pipelines
 *
 * @covers \Drupal\data_pipelines\Form\DatasetBatchOperations
 */
class DatasetBatchOperationsTest extends DatasetKernelTestBase {

  /**
   * Tests batch builder.
   */
  public function testBatchForDataset(): void {
    $dataset = $this->createTestDataset();
    $batch = DatasetBatchOperations::batchForDataset($dataset);
    $this->assertEquals([
      [[DatasetBatchOperations::class, 'operationQueueItem'], [$dataset->id()]],
      [[DatasetBatchOperations::class, 'operationProcess'], [$dataset->id()]],
    ], $batch['operations']);
  }

  /**
   * Tests validation operation.
   */
  public function testValidation(): void {
    $dataset = $this->createTestDataset();
    $dataset->setPendingProcessing();
    $context = [
      'results' => [],
      'sandbox' => [],
      'finished' => 0,
    ];
    $dataset->queueProcessing(\Drupal::queue($dataset->getProcessingQueueId()), new DatasetBatchContext($dataset->getBatchSize()));
    while ($context['finished'] !== 1) {
      DatasetBatchOperations::operationProcess((int) $dataset->id(), $context);
    }
    $this->assertEquals('Processed dataset.', (string) $context['message']);
    $csv = <<<CSV
should_we,firstname,lastname
Y,jo,bloggs
N,betty,bloggs
CSV;
    $dataset = $this->createTestDataset([
      'csv_text' => $csv,
    ]);
    $dataset->setPendingProcessing();
    $context = [
      'results' => [],
      'sandbox' => [],
      'finished' => 0,
    ];
    $dataset->queueProcessing(\Drupal::queue($dataset->getProcessingQueueId()), new DatasetBatchContext($dataset->getBatchSize()));
    while ($context['finished'] !== 1) {
      DatasetBatchOperations::operationProcess((int) $dataset->id(), $context);
    }
    $dataset = $this->reloadDataset($dataset);
    $logs = $dataset->getLogs();
    $this->assertCount(1, $logs);
    $this->assertEquals([
      'Validation error in row 1: Firstname should be at least 3 characters',
    ], $logs);
    $this->assertFalse($dataset->isPendingValidation());
  }

  /**
   * Tests process operation with failed validation.
   */
  public function testProcessingWhenInvalid(): void {
    $dataset = $this->createTestDataset();
    $context = [
      'results' => [],
    ];
    DatasetBatchOperations::operationProcess((int) $dataset->id(), $context);
    $this->assertEquals('Skipped processing dataset, validation errors exist.', (string) $context['message']);
  }

  /**
   * Tests queue operation.
   */
  public function testQueuingValid(): void {
    $dataset = $this->createTestDataset([
      'status' => DatasetInterface::STATUS_PENDING_PROCESSING,
    ]);
    $queue = \Drupal::queue($dataset->getProcessingQueueId());
    // Clear any previous operations.
    $queue->deleteQueue();
    $context = [
      'results' => [],
      'sandbox' => [],
    ];
    DatasetBatchOperations::operationQueueItem((int) $dataset->id(), $context);
    $this->assertEquals('Data is successfully queued for processing.', (string) $context['message']);
    // 1 for begin, 1 for end, 1 for each row (state destination has a chunk
    // size of 1).
    $this->assertEquals(4, $queue->numberOfItems());
  }

  /**
   * Tests process operation.
   */
  public function testProcessingValid(): void {
    $dataset = $this->createTestDataset([
      'status' => DatasetInterface::STATUS_PENDING_PROCESSING,
    ]);
    $queue = \Drupal::queue($dataset->getProcessingQueueId());
    $dataset->queueProcessing($queue, new DatasetBatchContext($dataset->getBatchSize()));
    $this->assertEquals(4, $queue->numberOfItems());
    $context = [
      'results' => [],
      'finished' => 0,
    ];
    while ($context['finished'] !== 1) {
      DatasetBatchOperations::operationProcess((int) $dataset->id(), $context);
    }
    $this->assertEquals('Processed dataset.', (string) $context['message']);
    $this->assertEquals(0, $queue->numberOfItems());
  }

  /**
   * Tests finished with success.
   */
  public function testFinishedSuccess(): void {
    $dataset = $this->createTestDataset([
      'status' => DatasetInterface::STATUS_PENDING_PROCESSING,
    ]);
    $context = [
      'results' => [
        'dataset_id' => $dataset->id(),
      ],
      'sandbox' => [],
      'finished' => 0,
    ];
    while ($context['finished'] !== 1) {
      DatasetBatchOperations::operationProcess((int) $dataset->id(), $context);
    }
    DatasetBatchOperations::finished(TRUE, $context['results']);
    $this->assertEquals([new TranslatableMarkup('Dataset %name was successfully processed.', [
      '%name' => $dataset->label(),
    ]),
    ], \Drupal::messenger()->messagesByType(MessengerInterface::TYPE_STATUS));
  }

  /**
   * Tests finished without processing.
   */
  public function testFinishedWithoutProcessing(): void {
    $dataset = $this->createTestDataset([
      'status' => DatasetInterface::STATUS_PENDING_PROCESSING,
    ]);
    $context = [
      'results' => [
        'dataset_id' => $dataset->id(),
      ],
    ];
    DatasetBatchOperations::finished(TRUE, $context['results']);
    $this->assertEquals([new TranslatableMarkup('Dataset %name was processed but some rows failed validation or skipped and were not processed. Please check <a href=":url">the logs</a>.', [
      ':url' => $dataset->toUrl()->toString(),
      '%name' => $dataset->label(),
    ]),
    ], \Drupal::messenger()->messagesByType(MessengerInterface::TYPE_ERROR));
  }

  /**
   * Tests finished with error.
   */
  public function testFinishedWithError(): void {
    $context = [
      'results' => [],
    ];
    DatasetBatchOperations::finished(TRUE, $context['results']);
    $this->assertEquals([new TranslatableMarkup('The validation and processing operation did not complete.')], \Drupal::messenger()->messagesByType(MessengerInterface::TYPE_ERROR));
  }

}