198 lines
7.1 KiB
Python
198 lines
7.1 KiB
Python
import os
|
|
|
|
import infra.basetest
|
|
|
|
|
|
class TestPostgreSQL(infra.basetest.BRTest):
|
|
# We use a specific configuration for:
|
|
# - using Aarch64, to have more than 256MB memory,
|
|
# - to have an ext4 rootfs image exposed as a virtio storage
|
|
# (rather than cpio initrd). This will save some memory, as the
|
|
# rootfs image is big.
|
|
config = \
|
|
"""
|
|
BR2_aarch64=y
|
|
BR2_TOOLCHAIN_EXTERNAL=y
|
|
BR2_TARGET_GENERIC_GETTY_PORT="ttyAMA0"
|
|
BR2_LINUX_KERNEL=y
|
|
BR2_LINUX_KERNEL_CUSTOM_VERSION=y
|
|
BR2_LINUX_KERNEL_CUSTOM_VERSION_VALUE="6.1.72"
|
|
BR2_LINUX_KERNEL_USE_CUSTOM_CONFIG=y
|
|
BR2_LINUX_KERNEL_CUSTOM_CONFIG_FILE="board/qemu/aarch64-virt/linux.config"
|
|
BR2_PACKAGE_POSTGRESQL=y
|
|
BR2_TARGET_ROOTFS_EXT2=y
|
|
BR2_TARGET_ROOTFS_EXT2_4=y
|
|
BR2_TARGET_ROOTFS_EXT2_SIZE="256M"
|
|
# BR2_TARGET_ROOTFS_TAR is not set
|
|
"""
|
|
|
|
def __init__(self, names):
|
|
super(TestPostgreSQL, self).__init__(names)
|
|
self.db_admin = "postgres"
|
|
self.db_user = "br_user"
|
|
self.db_name = "br_database"
|
|
self.backup_file = "dump.sql"
|
|
self.pgdata_dir = "/var/lib/pgsql"
|
|
|
|
def run_user_db_query(self, user, database, query, opts=None):
|
|
cmd = f"psql --username={user} --dbname={database}"
|
|
cmd += f' --command="{query}"'
|
|
if opts is not None:
|
|
cmd += " " + opts
|
|
self.assertRunOk(cmd)
|
|
|
|
def run_admin_sql_query(self, query, opts=None):
|
|
self.run_user_db_query(self.db_admin, self.db_admin, query, opts)
|
|
|
|
def run_sql_query(self, query, opts=None):
|
|
self.run_user_db_query(self.db_user, self.db_name, query, opts)
|
|
|
|
def cleanup_database(self):
|
|
# This cleanup is useful when run-test -k is used. It makes
|
|
# this test idempotent. Since the drive storage is preserved
|
|
# between reboots, this cleanup will prevent errors during the
|
|
# user/db creation.
|
|
|
|
# Drop the test database, if it exists.
|
|
cmd = f"dropdb --username={self.db_admin} "
|
|
cmd += f"--if-exists {self.db_name}"
|
|
self.assertRunOk(cmd)
|
|
|
|
# Drop the test user, if it exists.
|
|
cmd = f"dropuser --username={self.db_admin} "
|
|
cmd += f"--if-exists {self.db_user}"
|
|
self.assertRunOk(cmd)
|
|
|
|
def create_tables(self):
|
|
sql_query = "CREATE TABLE fruits ("
|
|
sql_query += "id integer PRIMARY KEY, name varchar(16) NOT NULL);"
|
|
self.run_sql_query(sql_query)
|
|
|
|
sql_query = "CREATE TABLE colors ("
|
|
sql_query += "id integer PRIMARY KEY, name varchar(16) NOT NULL);"
|
|
self.run_sql_query(sql_query)
|
|
|
|
sql_query = "CREATE TABLE fruit_colors ("
|
|
sql_query += "fruit_id integer REFERENCES fruits(id), "
|
|
sql_query += "color_id integer REFERENCES colors(id), "
|
|
sql_query += "UNIQUE (fruit_id, color_id));"
|
|
self.run_sql_query(sql_query)
|
|
|
|
def insert_data(self):
|
|
fruits = ["Banana", "Blueberry", "Orange", "Raspberry"]
|
|
fruit_id = 1
|
|
for fruit in fruits:
|
|
sql_query = "INSERT INTO fruits (id, name) "
|
|
sql_query += f"VALUES ({fruit_id}, '{fruit}');"
|
|
self.run_sql_query(sql_query)
|
|
fruit_id += 1
|
|
|
|
colors = ["Blue", "Orange", "Red", "Yellow"]
|
|
color_id = 1
|
|
for color in colors:
|
|
sql_query = "INSERT INTO colors (id, name) "
|
|
sql_query += f"VALUES ({color_id}, '{color}');"
|
|
self.run_sql_query(sql_query)
|
|
color_id += 1
|
|
|
|
fruit_colors = [(1, 4), (2, 1), (3, 2), (4, 3)]
|
|
for fruit_color in fruit_colors:
|
|
fruit_id, color_id = fruit_color
|
|
sql_query = "INSERT INTO fruit_colors (fruit_id, color_id) "
|
|
sql_query += f"VALUES ({fruit_id}, {color_id});"
|
|
self.run_sql_query(sql_query)
|
|
|
|
def query_database(self):
|
|
sql_query = "SELECT "
|
|
sql_query += "fruits.name AS fruit, colors.name AS color "
|
|
sql_query += "FROM fruits, colors, fruit_colors "
|
|
sql_query += "WHERE fruits.id = fruit_colors.fruit_id "
|
|
sql_query += "AND colors.id = fruit_colors.color_id "
|
|
sql_query += "ORDER BY fruit;"
|
|
self.run_sql_query(sql_query)
|
|
|
|
def test_run(self):
|
|
drive = os.path.join(self.builddir, "images", "rootfs.ext4")
|
|
kern = os.path.join(self.builddir, "images", "Image")
|
|
self.emulator.boot(arch="aarch64",
|
|
kernel=kern,
|
|
kernel_cmdline=["root=/dev/vda console=ttyAMA0"],
|
|
options=["-M", "virt",
|
|
"-cpu", "cortex-a57",
|
|
"-m", "512M",
|
|
"-smp", "2",
|
|
"-drive", f"file={drive},if=virtio,format=raw"])
|
|
self.emulator.login()
|
|
|
|
# Check the server binary can execute.
|
|
self.assertRunOk("postgres --version")
|
|
|
|
# Check the client binary can execute.
|
|
self.assertRunOk("psql --version")
|
|
|
|
# Check the server is ready.
|
|
self.assertRunOk("pg_isready")
|
|
|
|
# Query the server version from the client.
|
|
sql_query = "SELECT version();"
|
|
self.run_admin_sql_query(sql_query,
|
|
opts="--tuples-only --no-align")
|
|
|
|
self.cleanup_database()
|
|
|
|
# Create a new user.
|
|
cmd = f"createuser --username={self.db_admin} "
|
|
cmd += "--no-superuser --no-createdb --no-createrole "
|
|
cmd += self.db_user
|
|
self.assertRunOk(cmd)
|
|
|
|
# Create a new database.
|
|
cmd = f"createdb --username={self.db_admin} "
|
|
cmd += f"--owner={self.db_user} "
|
|
cmd += f'{self.db_name} "Test Database for Buildroot Test"'
|
|
self.assertRunOk(cmd)
|
|
|
|
self.create_tables()
|
|
|
|
self.insert_data()
|
|
|
|
self.query_database()
|
|
|
|
# Update a table.
|
|
sql_query = "UPDATE fruits SET name = 'Lemon' WHERE id = 1;"
|
|
self.run_sql_query(sql_query)
|
|
|
|
# Backup the test database.
|
|
cmd = f"pg_dump --username={self.db_user} --dbname={self.db_name} "
|
|
cmd += f"--file={self.backup_file} --inserts"
|
|
self.assertRunOk(cmd)
|
|
|
|
# Drop all the tables.
|
|
sql_query = "DROP TABLE fruit_colors, fruits, colors;"
|
|
self.run_sql_query(sql_query)
|
|
|
|
# Query the server status.
|
|
cmd = f"su - {self.db_admin} -c 'pg_ctl status -D {self.pgdata_dir}'"
|
|
self.assertRunOk(cmd)
|
|
|
|
# Stop the server.
|
|
cmd = f"su - {self.db_admin} -c 'pg_ctl stop -D {self.pgdata_dir}'"
|
|
self.assertRunOk(cmd)
|
|
|
|
# Check the server is no longer ready.
|
|
_, exit_code = self.emulator.run("pg_isready")
|
|
self.assertNotEqual(exit_code, 0)
|
|
|
|
# Restart the server.
|
|
cmd = f"su - {self.db_admin} -c 'pg_ctl start -D {self.pgdata_dir}'"
|
|
self.assertRunOk(cmd)
|
|
|
|
# Restore the backup.
|
|
cmd = f"psql --username={self.db_user} --dbname={self.db_name} "
|
|
cmd += f"--file={self.backup_file}"
|
|
self.assertRunOk(cmd)
|
|
|
|
# Query on last time our data, to check the backup restoration
|
|
# succeeded.
|
|
self.query_database()
|