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() : void {
        $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() : void {
        // 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);
        $this->assertTrue($this->schema
            ->tableExists($table_name));
    }

}