SchemaTest.php

<?php

declare (strict_types=1);
namespace Drupal\Tests\pgsql\Kernel\pgsql;

use Drupal\KernelTests\Core\Database\DriverSpecificSchemaTestBase;

// cSpell:ignore relkind objid refobjid regclass attname attrelid attnum
// cSpell:ignore refobjsubid

/**
 * Tests schema API for the PostgreSQL driver.
 *
 * @group Database
 */
class SchemaTest extends DriverSpecificSchemaTestBase {

  /**
   * {@inheritdoc}
   */
  public function checkSchemaComment(string $description, string $table, string $column = NULL) : void {
    $this
      ->assertSame($description, $this->schema
      ->getComment($table, $column), 'The comment matches the schema description.');
  }

  /**
   * {@inheritdoc}
   */
  protected function checkSequenceRenaming(string $tableName) : void {

    // For PostgreSQL, we also need to check that the sequence has been renamed.
    // The initial name of the sequence has been generated automatically by
    // PostgreSQL when the table was created, however, on subsequent table
    // renames the name is generated by Drupal and can not be easily
    // re-constructed. Hence we can only check that we still have a sequence on
    // the new table name.
    $sequenceExists = (bool) $this->connection
      ->query("SELECT pg_get_serial_sequence('{" . $tableName . "}', 'id')")
      ->fetchField();
    $this
      ->assertTrue($sequenceExists, 'Sequence was renamed.');

    // Rename the table again and repeat the check.
    $anotherTableName = strtolower($this
      ->getRandomGenerator()
      ->name(63 - strlen($this
      ->getDatabasePrefix())));
    $this->schema
      ->renameTable($tableName, $anotherTableName);
    $sequenceExists = (bool) $this->connection
      ->query("SELECT pg_get_serial_sequence('{" . $anotherTableName . "}', 'id')")
      ->fetchField();
    $this
      ->assertTrue($sequenceExists, 'Sequence was renamed.');
  }

  /**
   * {@inheritdoc}
   */
  public function testTableWithSpecificDataType() : void {
    $table_specification = [
      'description' => 'Schema table description.',
      'fields' => [
        'timestamp' => [
          'pgsql_type' => 'timestamp',
          'not null' => FALSE,
          'default' => NULL,
        ],
      ],
    ];
    $this->schema
      ->createTable('test_timestamp', $table_specification);
    $this
      ->assertTrue($this->schema
      ->tableExists('test_timestamp'));
  }

  /**
   * @covers \Drupal\pgsql\Driver\Database\pgsql\Schema::introspectIndexSchema
   */
  public function testIntrospectIndexSchema() : void {
    $table_specification = [
      'fields' => [
        'id' => [
          'type' => 'int',
          'not null' => TRUE,
          'default' => 0,
        ],
        'test_field_1' => [
          'type' => 'int',
          'not null' => TRUE,
          'default' => 0,
        ],
        'test_field_2' => [
          'type' => 'int',
          'default' => 0,
        ],
        'test_field_3' => [
          'type' => 'int',
          'default' => 0,
        ],
        'test_field_4' => [
          'type' => 'int',
          'default' => 0,
        ],
        'test_field_5' => [
          'type' => 'int',
          'default' => 0,
        ],
      ],
      'primary key' => [
        'id',
        'test_field_1',
      ],
      'unique keys' => [
        'test_field_2' => [
          'test_field_2',
        ],
        'test_field_3_test_field_4' => [
          'test_field_3',
          'test_field_4',
        ],
      ],
      'indexes' => [
        'test_field_4' => [
          'test_field_4',
        ],
        'test_field_4_test_field_5' => [
          'test_field_4',
          'test_field_5',
        ],
      ],
    ];
    $table_name = strtolower($this
      ->getRandomGenerator()
      ->name());
    $this->schema
      ->createTable($table_name, $table_specification);
    unset($table_specification['fields']);
    $introspect_index_schema = new \ReflectionMethod(get_class($this->schema), 'introspectIndexSchema');
    $index_schema = $introspect_index_schema
      ->invoke($this->schema, $table_name);

    // The PostgreSQL driver is using a custom naming scheme for its indexes, so
    // we need to adjust the initial table specification.
    $ensure_identifier_length = new \ReflectionMethod(get_class($this->schema), 'ensureIdentifiersLength');
    foreach ($table_specification['unique keys'] as $original_index_name => $columns) {
      unset($table_specification['unique keys'][$original_index_name]);
      $new_index_name = $ensure_identifier_length
        ->invoke($this->schema, $table_name, $original_index_name, 'key');
      $table_specification['unique keys'][$new_index_name] = $columns;
    }
    foreach ($table_specification['indexes'] as $original_index_name => $columns) {
      unset($table_specification['indexes'][$original_index_name]);
      $new_index_name = $ensure_identifier_length
        ->invoke($this->schema, $table_name, $original_index_name, 'idx');
      $table_specification['indexes'][$new_index_name] = $columns;
    }
    $this
      ->assertEquals($table_specification, $index_schema);
  }

  /**
   * {@inheritdoc}
   */
  public function testReservedKeywordsForNaming() : void {
    $table_specification = [
      'description' => 'A test table with an ANSI reserved keywords for naming.',
      'fields' => [
        'primary' => [
          'description' => 'Simple unique ID.',
          'type' => 'int',
          'not null' => TRUE,
        ],
        'update' => [
          'description' => 'A column with reserved name.',
          'type' => 'varchar',
          'length' => 255,
        ],
      ],
      'primary key' => [
        'primary',
      ],
      'unique keys' => [
        'having' => [
          'update',
        ],
      ],
      'indexes' => [
        'in' => [
          'primary',
          'update',
        ],
      ],
    ];

    // Creating a table.
    $table_name = 'select';
    $this->schema
      ->createTable($table_name, $table_specification);
    $this
      ->assertTrue($this->schema
      ->tableExists($table_name));

    // Finding all tables.
    $tables = $this->schema
      ->findTables('%');
    sort($tables);
    $this
      ->assertEquals([
      'config',
      'select',
    ], $tables);

    // Renaming a table.
    $table_name_new = 'from';
    $this->schema
      ->renameTable($table_name, $table_name_new);
    $this
      ->assertFalse($this->schema
      ->tableExists($table_name));
    $this
      ->assertTrue($this->schema
      ->tableExists($table_name_new));

    // Adding a field.
    $field_name = 'delete';
    $this->schema
      ->addField($table_name_new, $field_name, [
      'type' => 'int',
      'not null' => TRUE,
    ]);
    $this
      ->assertTrue($this->schema
      ->fieldExists($table_name_new, $field_name));

    // Dropping a primary key.
    $this->schema
      ->dropPrimaryKey($table_name_new);

    // Adding a primary key.
    $this->schema
      ->addPrimaryKey($table_name_new, [
      $field_name,
    ]);

    // Check the primary key columns.
    $find_primary_key_columns = new \ReflectionMethod(get_class($this->schema), 'findPrimaryKeyColumns');
    $this
      ->assertEquals([
      $field_name,
    ], $find_primary_key_columns
      ->invoke($this->schema, $table_name_new));

    // Dropping a primary key.
    $this->schema
      ->dropPrimaryKey($table_name_new);

    // Changing a field.
    $field_name_new = 'where';
    $this->schema
      ->changeField($table_name_new, $field_name, $field_name_new, [
      'type' => 'int',
      'not null' => FALSE,
    ]);
    $this
      ->assertFalse($this->schema
      ->fieldExists($table_name_new, $field_name));
    $this
      ->assertTrue($this->schema
      ->fieldExists($table_name_new, $field_name_new));

    // Adding an unique key
    $unique_key_name = $unique_key_introspect_name = 'unique';
    $this->schema
      ->addUniqueKey($table_name_new, $unique_key_name, [
      $field_name_new,
    ]);

    // Check the unique key columns.
    $introspect_index_schema = new \ReflectionMethod(get_class($this->schema), 'introspectIndexSchema');
    $ensure_identifiers_length = new \ReflectionMethod(get_class($this->schema), 'ensureIdentifiersLength');
    $unique_key_introspect_name = $ensure_identifiers_length
      ->invoke($this->schema, $table_name_new, $unique_key_name, 'key');
    $this
      ->assertEquals([
      $field_name_new,
    ], $introspect_index_schema
      ->invoke($this->schema, $table_name_new)['unique keys'][$unique_key_introspect_name]);

    // Dropping an unique key
    $this->schema
      ->dropUniqueKey($table_name_new, $unique_key_name);

    // Dropping a field.
    $this->schema
      ->dropField($table_name_new, $field_name_new);
    $this
      ->assertFalse($this->schema
      ->fieldExists($table_name_new, $field_name_new));

    // Adding an index.
    $index_name = $index_introspect_name = 'index';
    $this->schema
      ->addIndex($table_name_new, $index_name, [
      'update',
    ], $table_specification);
    $this
      ->assertTrue($this->schema
      ->indexExists($table_name_new, $index_name));

    // Check the index columns.
    $index_introspect_name = $ensure_identifiers_length
      ->invoke($this->schema, $table_name_new, $index_name, 'idx');
    $this
      ->assertEquals([
      'update',
    ], $introspect_index_schema
      ->invoke($this->schema, $table_name_new)['indexes'][$index_introspect_name]);

    // Dropping an index.
    $this->schema
      ->dropIndex($table_name_new, $index_name);
    $this
      ->assertFalse($this->schema
      ->indexExists($table_name_new, $index_name));

    // Dropping a table.
    $this->schema
      ->dropTable($table_name_new);
    $this
      ->assertFalse($this->schema
      ->tableExists($table_name_new));
  }

  /**
   * @covers \Drupal\Core\Database\Driver\pgsql\Schema::extensionExists
   */
  public function testPgsqlExtensionExists() : void {

    // Test the method for a non existing extension.
    $this
      ->assertFalse($this->schema
      ->extensionExists('non_existing_extension'));

    // Test the method for an existing extension.
    $this
      ->assertTrue($this->schema
      ->extensionExists('pg_trgm'));
  }

  /**
   * Tests if the new sequences get the right ownership.
   */
  public function testPgsqlSequences() : void {
    $table_specification = [
      'description' => 'A test table with an ANSI reserved keywords for naming.',
      'fields' => [
        'uid' => [
          'description' => 'Simple unique ID.',
          'type' => 'serial',
          'not null' => TRUE,
        ],
        'update' => [
          'description' => 'A column with reserved name.',
          'type' => 'varchar',
          'length' => 255,
        ],
      ],
      'primary key' => [
        'uid',
      ],
      'unique keys' => [
        'having' => [
          'update',
        ],
      ],
      'indexes' => [
        'in' => [
          'uid',
          'update',
        ],
      ],
    ];

    // Creating a table.
    $table_name = 'sequence_test';
    $this->schema
      ->createTable($table_name, $table_specification);
    $this
      ->assertTrue($this->schema
      ->tableExists($table_name));

    // Retrieves a sequence name that is owned by the table and column.
    $sequence_name = $this->connection
      ->query("SELECT pg_get_serial_sequence(:table, :column)", [
      ':table' => $this->connection
        ->getPrefix() . 'sequence_test',
      ':column' => 'uid',
    ])
      ->fetchField();
    $schema = $this->connection
      ->getConnectionOptions()['schema'] ?? 'public';
    $this
      ->assertEquals($schema . '.' . $this->connection
      ->getPrefix() . 'sequence_test_uid_seq', $sequence_name);

    // Checks if the sequence exists.
    $this
      ->assertTrue((bool) \Drupal::database()
      ->query("SELECT c.relname FROM pg_class as c WHERE c.relkind = 'S' AND c.relname = :name", [
      ':name' => $this->connection
        ->getPrefix() . 'sequence_test_uid_seq',
    ])
      ->fetchField());

    // Retrieves the sequence owner object.
    $sequence_owner = \Drupal::database()
      ->query("SELECT d.refobjid::regclass as table_name, a.attname as field_name\n      FROM pg_depend d\n      JOIN pg_attribute a ON a.attrelid = d.refobjid AND a.attnum = d.refobjsubid\n      WHERE d.objid = :seq_name::regclass\n      AND d.refobjsubid > 0\n      AND d.classid = 'pg_class'::regclass", [
      ':seq_name' => $sequence_name,
    ])
      ->fetchObject();
    $this
      ->assertEquals($this->connection
      ->getPrefix() . 'sequence_test', $sequence_owner->table_name);
    $this
      ->assertEquals('uid', $sequence_owner->field_name, 'New sequence is owned by its table.');
  }

  /**
   * Tests the method tableExists().
   */
  public function testTableExists() {
    $table_name = 'test_table';
    $table_specification = [
      'fields' => [
        'id' => [
          'type' => 'int',
          'default' => NULL,
        ],
      ],
    ];
    $this->schema
      ->createTable($table_name, $table_specification);
    $prefixed_table_name = $this->connection
      ->getPrefix($table_name) . $table_name;

    // Three different calls to the method Schema::tableExists() with an
    // unprefixed table name.
    $this
      ->assertTrue($this->schema
      ->tableExists($table_name));
    $this
      ->assertTrue($this->schema
      ->tableExists($table_name, TRUE));
    $this
      ->assertFalse($this->schema
      ->tableExists($table_name, FALSE));

    // Three different calls to the method Schema::tableExists() with a
    // prefixed table name.
    $this
      ->assertFalse($this->schema
      ->tableExists($prefixed_table_name));
    $this
      ->assertFalse($this->schema
      ->tableExists($prefixed_table_name, TRUE));
    $this
      ->assertTrue($this->schema
      ->tableExists($prefixed_table_name, FALSE));
  }

  /**
   * Tests renaming a table where the new index name is equal to the table name.
   */
  public function testRenameTableWithNewIndexNameEqualsTableName() {

    // Special table names for colliding with the PostgreSQL new index name.
    $table_name_old = 'some_new_table_name__id__idx';
    $table_name_new = 'some_new_table_name';
    $table_specification = [
      'fields' => [
        'id' => [
          'type' => 'int',
          'default' => NULL,
        ],
      ],
      'indexes' => [
        'id' => [
          'id',
        ],
      ],
    ];
    $this->schema
      ->createTable($table_name_old, $table_specification);

    // Renaming the table can fail for PostgreSQL, when a new index name is
    // equal to the old table name.
    $this->schema
      ->renameTable($table_name_old, $table_name_new);
    $this
      ->assertTrue($this->schema
      ->tableExists($table_name_new));
  }

  /**
   * Tests column name escaping in field constraints.
   */
  public function testUnsignedField() : void {
    $table_name = 'unsigned_table';
    $table_spec = [
      'fields' => [
        'order' => [
          'type' => 'int',
          'unsigned' => TRUE,
          'not null' => TRUE,
        ],
      ],
      'primary key' => [
        'order',
      ],
    ];
    $this->schema
      ->createTable($table_name, $table_spec);
  }

}