diff --git a/tests/integration/new_relations/test_relations_coherence.py b/tests/integration/new_relations/test_relations_coherence.py new file mode 100644 index 0000000000..00395ddf85 --- /dev/null +++ b/tests/integration/new_relations/test_relations_coherence.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +# Copyright 2022 Canonical Ltd. +# See LICENSE file for licensing details. +import logging +import secrets +import string + +import psycopg2 +import pytest +from pytest_operator.plugin import OpsTest + +from ..helpers import CHARM_SERIES, DATABASE_APP_NAME +from .helpers import build_connection_string +from .test_new_relations import DATA_INTEGRATOR_APP_NAME + +logger = logging.getLogger(__name__) + +APPLICATION_APP_NAME = "postgresql-test-app" +APP_NAMES = [DATABASE_APP_NAME, DATA_INTEGRATOR_APP_NAME] +FIRST_DATABASE_RELATION_NAME = "first-database" + + +@pytest.mark.group(1) +@pytest.mark.abort_on_fail +async def test_relations(ops_test: OpsTest, charm): + """Test that check relation data.""" + async with ops_test.fast_forward(): + await ops_test.model.deploy( + charm, + application_name=DATABASE_APP_NAME, + num_units=1, + series=CHARM_SERIES, + config={"profile": "testing"}, + ) + await ops_test.model.wait_for_idle(apps=[DATABASE_APP_NAME], status="active", timeout=3000) + + # Creating first time relation with user role + await ops_test.model.deploy(DATA_INTEGRATOR_APP_NAME) + await ops_test.model.applications[DATA_INTEGRATOR_APP_NAME].set_config({ + "database-name": DATA_INTEGRATOR_APP_NAME.replace("-", "_"), + }) + await ops_test.model.wait_for_idle(apps=[DATA_INTEGRATOR_APP_NAME], status="blocked") + await ops_test.model.add_relation(DATA_INTEGRATOR_APP_NAME, DATABASE_APP_NAME) + await ops_test.model.wait_for_idle(apps=APP_NAMES, status="active") + + connection_string = await build_connection_string( + ops_test, + DATA_INTEGRATOR_APP_NAME, + "postgresql", + database=DATA_INTEGRATOR_APP_NAME.replace("-", "_"), + ) + + connection = psycopg2.connect(connection_string) + connection.autocommit = True + cursor = connection.cursor() + try: + random_name = ( + f"test_{''.join(secrets.choice(string.ascii_lowercase) for _ in range(10))}" + ) + cursor.execute(f"CREATE DATABASE {random_name};") + assert False, "user role was able to create database" + except psycopg2.errors.InsufficientPrivilege: + pass + finally: + connection.close() + + with psycopg2.connect(connection_string) as connection: + connection.autocommit = True + with connection.cursor() as cursor: + # Check that it's possible to write and read data from the database that + # was created for the application. + cursor.execute("DROP TABLE IF EXISTS test;") + cursor.execute("CREATE TABLE test(data TEXT);") + cursor.execute("INSERT INTO test(data) VALUES('some data');") + cursor.execute("SELECT data FROM test;") + data = cursor.fetchone() + assert data[0] == "some data" + connection.close() + + await ops_test.model.applications[DATABASE_APP_NAME].remove_relation( + f"{DATABASE_APP_NAME}:database", f"{DATA_INTEGRATOR_APP_NAME}:postgresql" + ) + + # Re-relation with admin role with checking permission + await ops_test.model.applications[DATA_INTEGRATOR_APP_NAME].set_config({ + "database-name": DATA_INTEGRATOR_APP_NAME.replace("-", "_"), + "extra-user-roles": "admin", + }) + await ops_test.model.wait_for_idle(apps=[DATA_INTEGRATOR_APP_NAME], status="blocked") + await ops_test.model.add_relation(DATA_INTEGRATOR_APP_NAME, DATABASE_APP_NAME) + await ops_test.model.wait_for_idle(apps=APP_NAMES, status="active") + + connection_string = await build_connection_string( + ops_test, + DATA_INTEGRATOR_APP_NAME, + "postgresql", + database=DATA_INTEGRATOR_APP_NAME.replace("-", "_"), + ) + try: + connection = psycopg2.connect(connection_string) + connection.autocommit = True + cursor = connection.cursor() + random_name = ( + f"test_{''.join(secrets.choice(string.ascii_lowercase) for _ in range(10))}" + ) + cursor.execute(f"CREATE DATABASE {random_name};") + except psycopg2.errors.InsufficientPrivilege: + assert ( + False + ), f"failed connect to {random_name} or run a statement in the following database" + finally: + connection.close() + + await ops_test.model.applications[DATABASE_APP_NAME].remove_relation( + f"{DATABASE_APP_NAME}:database", f"{DATA_INTEGRATOR_APP_NAME}:postgresql" + ) + + # Re-relation again with user role and checking write data + await ops_test.model.applications[DATA_INTEGRATOR_APP_NAME].set_config({ + "database-name": DATA_INTEGRATOR_APP_NAME.replace("-", "_"), + "extra-user-roles": "", + }) + await ops_test.model.wait_for_idle(apps=[DATA_INTEGRATOR_APP_NAME], status="blocked") + await ops_test.model.add_relation(DATA_INTEGRATOR_APP_NAME, DATABASE_APP_NAME) + await ops_test.model.wait_for_idle(apps=APP_NAMES, status="active") + + for database in [ + DATA_INTEGRATOR_APP_NAME.replace("-", "_"), + "postgres", + ]: + logger.info(f"connecting to the following database: {database}") + connection_string = await build_connection_string( + ops_test, DATA_INTEGRATOR_APP_NAME, "postgresql", database=database + ) + connection = None + should_fail = database == "postgres" + try: + with psycopg2.connect( + connection_string + ) as connection, connection.cursor() as cursor: + cursor.execute("SELECT data FROM test;") + data = cursor.fetchone() + assert data[0] == "some data" + + # Write some data (it should fail in the "postgres" database). + random_name = f"test_{''.join(secrets.choice(string.ascii_lowercase) for _ in range(10))}" + cursor.execute(f"CREATE TABLE {random_name}(data TEXT);") + if should_fail: + assert ( + False + ), f"failed to run a statement in the following database: {database}" + except psycopg2.errors.InsufficientPrivilege as e: + if not should_fail: + logger.exception(e) + assert False, f"failed to connect to or run a statement in the following database: {database}" + except psycopg2.OperationalError as e: + if not should_fail: + logger.exception(e) + finally: + if connection is not None: + connection.close()