diff --git a/.coverage b/.coverage deleted file mode 100644 index 89085e7..0000000 Binary files a/.coverage and /dev/null differ diff --git a/.github/workflows/check_python.yml b/.github/workflows/quality_gates.yml similarity index 78% rename from .github/workflows/check_python.yml rename to .github/workflows/quality_gates.yml index 3d2ddd9..efd58b5 100644 --- a/.github/workflows/check_python.yml +++ b/.github/workflows/quality_gates.yml @@ -1,4 +1,4 @@ -name: Python Check +name: Quality Gates on: pull_request: @@ -8,7 +8,7 @@ on: workflow_dispatch: concurrency: - group: static-python-check-${{ github.ref }} + group: quality-gates-${{ github.ref }} cancel-in-progress: true permissions: @@ -21,6 +21,7 @@ jobs: runs-on: ubuntu-latest outputs: python_changed: ${{ steps.changes.outputs.python_changed }} + database_changed: ${{ steps.changes.outputs.database_changed }} steps: - name: Checkout repository uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 @@ -28,28 +29,35 @@ jobs: persist-credentials: false fetch-depth: 0 - - name: Check if Python files changed + - name: Check if Python or database files changed id: changes shell: bash env: GH_TOKEN: ${{ github.token }} run: | set -euo pipefail - + if [[ "${{ github.event_name }}" == "pull_request" ]]; then CHANGED_FILES=$(gh api \ "repos/${{ github.repository }}/pulls/${{ github.event.pull_request.number }}/files" \ - --jq '.[].filename | select(endswith(".py") or (startswith("requirements") and endswith(".txt")))') + --paginate \ + --jq '.[].filename') else - CHANGED_FILES=$(git diff --name-only "${{ github.sha }}~1" "${{ github.sha }}" -- '*.py' 'requirements*.txt') + CHANGED_FILES=$(git diff --name-only "${{ github.sha }}~1" "${{ github.sha }}") fi - if [[ -n "$CHANGED_FILES" ]]; then + if grep -Eq '(^|/)[^/]+\.py$|^requirements[^/]*\.txt$' <<< "$CHANGED_FILES"; then echo "python_changed=true" >> "$GITHUB_OUTPUT" else echo "python_changed=false" >> "$GITHUB_OUTPUT" fi + if grep -Eq '^database/|^flyway\.toml$' <<< "$CHANGED_FILES"; then + echo "database_changed=true" >> "$GITHUB_OUTPUT" + else + echo "database_changed=false" >> "$GITHUB_OUTPUT" + fi + pylint-analysis: name: Pylint Static Code Analysis needs: detect @@ -139,7 +147,7 @@ jobs: integration-tests: name: Pytest Integration Tests needs: detect - if: needs.detect.outputs.python_changed == 'true' + if: needs.detect.outputs.python_changed == 'true' || needs.detect.outputs.database_changed == 'true' runs-on: ubuntu-latest timeout-minutes: 15 steps: @@ -152,13 +160,26 @@ jobs: - name: Set up dev Python environment uses: ./.github/actions/setup-dev-python-env + - name: Set up Java + uses: actions/setup-java@b6effb05e454b25005698d916606bdc6ffcbf961 + with: + distribution: temurin + java-version: '21' + + - name: Set up Flyway + uses: red-gate/setup-flyway@e024a17cd0890383f6996ed7edbded24c54ed86c + with: + version: '13.3.0' + edition: community + i-agree-to-the-eula: true + - name: Run integration tests run: pytest tests/integration/ -v --tb=short --log-cli-level=INFO noop: name: No Operation needs: detect - if: needs.detect.outputs.python_changed != 'true' + if: needs.detect.outputs.python_changed != 'true' && needs.detect.outputs.database_changed != 'true' runs-on: ubuntu-latest steps: - run: echo "No changes in the *.py files — passing." diff --git a/database/README.md b/database/README.md new file mode 100644 index 0000000..a16f9c5 --- /dev/null +++ b/database/README.md @@ -0,0 +1,79 @@ +# EventGate Database + +All database code lives here and is deployed with [Flyway](https://documentation.red-gate.com/flyway). +The migrations are the single source of truth for the schema, roles, and grants — the +same migrations build local, CI (integration tests), and real environments. + +## Layout + +```text +flyway.toml # Flyway configuration (locations, baseline, placeholders) — repo root +database/ +├── README.md +└── migrations/ + ├── 00_databases.ddl # One-off DB bootstrap (NOT a Flyway migration; no `V` prefix) + ├── V1.4.0.1__create_roles.ddl # owner / writer / reader roles + ├── V1.4.0.2__initial_schema.ddl # tables + └── V1.4.0.3__grants.ddl # ownership + least-privilege grants + ... +``` + +## Conventions + +- Versioned migrations follow Flyway's `V...__description.ext` format, + where `..` tracks the EventGate release the migration ships in and `` + increments per migration within that release. +- Extensions carry intent: `.ddl` for structural changes (tables, roles, constraints, indexes), + `.sql` for DML / data. + +## Roles + +| Role | Purpose | Used by | +|--------------------|-----------------------------------------------------|---------------------| +| master (superuser) | Runs the migrations | Flyway (deployment) | +| `eventgate_owner` | Owns the schema objects, may run DDL | Migrations | +| `eventgate_writer` | `SELECT` / `INSERT` / `UPDATE` on data tables | EventGate Lambda | +| `eventgate_reader` | `SELECT` only | EventStats Lambda | + +Role passwords are required Flyway placeholders (`eventgate_owner_password`, +`eventgate_writer_password`, `eventgate_reader_password`). Supply them from secrets in real +environments. + +## Local setup + +Requires the Flyway CLI (needs a JDK 17+) and Docker. + +```zsh +# 1. Start a local Postgres docker container +docker run --name=eventgate_db -e POSTGRES_PASSWORD=changeme -e POSTGRES_DB=eventgate_db -p 5432:5432 -d postgres:16 + +# 2. Apply the migrations (run from the repo root, where flyway.toml lives) +export FLYWAY_PLACEHOLDERS_EVENTGATE_OWNER_PASSWORD=changeme +export FLYWAY_PLACEHOLDERS_EVENTGATE_WRITER_PASSWORD=changeme +export FLYWAY_PLACEHOLDERS_EVENTGATE_READER_PASSWORD=changeme +flyway migrate + +# Inspect state / clean up +flyway info +docker kill eventgate_db && docker rm eventgate_db +``` + +## Adopting an existing database + +On a database that already contains the tables but has no Flyway history (i.e. production), a +plain `flyway migrate` fails because Flyway sees existing objects it didn't create. The first +migration against such a database must instead pass baseline flags explicitly, one time only: + +```zsh +flyway -baselineOnMigrate=true -baselineVersion=1.4.0.0 migrate +``` + +This records a baseline at `1.4.0.0` in `flyway_schema_history` and then applies `V1.4.0.1+` on +top. + +Before the first production migration: + +1. Compare the deployed schema with `V1.4.0.2__initial_schema.ddl`. +2. Back up the database and cluster roles. +3. Confirm the migration account can create roles and change ownership of every EventGate table. +4. Run `flyway info`, then the baseline command above with all role-password placeholders supplied from secrets. diff --git a/database/migrations/00_databases.ddl b/database/migrations/00_databases.ddl new file mode 100644 index 0000000..710ed5d --- /dev/null +++ b/database/migrations/00_databases.ddl @@ -0,0 +1,24 @@ +/* + * Copyright 2026 ABSA Group Limited + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +-- Database bootstrap (NOT a Flyway migration). +-- +-- Flyway connects to an existing database, so it cannot create the database it migrates. +-- This script is intentionally NOT prefixed with `V`, so Flyway ignores it. + +CREATE DATABASE eventgate_db + WITH + ENCODING = 'UTF8' + CONNECTION LIMIT = -1; diff --git a/database/migrations/V1.4.0.1__create_roles.ddl b/database/migrations/V1.4.0.1__create_roles.ddl new file mode 100644 index 0000000..ddc32ec --- /dev/null +++ b/database/migrations/V1.4.0.1__create_roles.ddl @@ -0,0 +1,81 @@ +/* + * Copyright 2026 ABSA Group Limited + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +-- Application database roles. +-- +-- eventgate_owner - owns the schema objects and may run DDL. +-- eventgate_writer - inserts/updates event data (main EventGate Lambda). +-- eventgate_reader - read-only access (EventStats Lambda). + +DO +$do$ + BEGIN + IF EXISTS ( + SELECT FROM pg_catalog.pg_roles + WHERE rolname = 'eventgate_owner') THEN + + RAISE NOTICE 'Role "eventgate_owner" already exists. Skipping.'; + ELSE + CREATE ROLE eventgate_owner WITH + LOGIN + NOSUPERUSER + INHERIT + NOCREATEDB + NOCREATEROLE + NOREPLICATION + PASSWORD '${eventgate_owner_password}'; + END IF; + END +$do$; + +DO +$do$ + BEGIN + IF EXISTS ( + SELECT FROM pg_catalog.pg_roles + WHERE rolname = 'eventgate_writer') THEN + RAISE NOTICE 'Role "eventgate_writer" already exists. Skipping.'; + ELSE + CREATE ROLE eventgate_writer WITH + LOGIN + NOSUPERUSER + INHERIT + NOCREATEDB + NOCREATEROLE + NOREPLICATION + PASSWORD '${eventgate_writer_password}'; + END IF; + END +$do$; + +DO +$do$ + BEGIN + IF EXISTS ( + SELECT FROM pg_catalog.pg_roles + WHERE rolname = 'eventgate_reader') THEN + RAISE NOTICE 'Role "eventgate_reader" already exists. Skipping.'; + ELSE + CREATE ROLE eventgate_reader WITH + LOGIN + NOSUPERUSER + INHERIT + NOCREATEDB + NOCREATEROLE + NOREPLICATION + PASSWORD '${eventgate_reader_password}'; + END IF; + END +$do$; diff --git a/tests/integration/schemas/postgres_schema.py b/database/migrations/V1.4.0.2__initial_schema.ddl similarity index 72% rename from tests/integration/schemas/postgres_schema.py rename to database/migrations/V1.4.0.2__initial_schema.ddl index 1a286e8..3740825 100644 --- a/tests/integration/schemas/postgres_schema.py +++ b/database/migrations/V1.4.0.2__initial_schema.ddl @@ -1,23 +1,21 @@ -# -# Copyright 2026 ABSA Group Limited -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# +/* + * Copyright 2026 ABSA Group Limited + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ -"""PostgreSQL schema for integration tests.""" +-- Initial EventGate schema. -SCHEMA_SQL = """ --- Table matching WriterPostgres._postgres_run_write columns +-- Run header rows for the runs topic. CREATE TABLE IF NOT EXISTS public_cps_za_runs ( event_id VARCHAR(255) NOT NULL, job_ref VARCHAR(255) NOT NULL, @@ -29,7 +27,7 @@ timestamp_end BIGINT ); --- Table matching WriterPostgres._postgres_run_write job rows +-- Per-job rows belonging to a run. CREATE TABLE IF NOT EXISTS public_cps_za_runs_jobs ( internal_id SERIAL PRIMARY KEY, event_id VARCHAR(255) NOT NULL, @@ -42,7 +40,7 @@ additional_info JSONB ); --- Table matching WriterPostgres._postgres_edla_write columns +-- Data lake change events. CREATE TABLE IF NOT EXISTS public_cps_za_dlchange ( event_id VARCHAR(255) NOT NULL, tenant_id VARCHAR(255) NOT NULL, @@ -59,7 +57,7 @@ additional_info JSONB ); --- Table matching WriterPostgres._postgres_test_write columns +-- Test topic events. CREATE TABLE IF NOT EXISTS public_cps_za_test ( event_id VARCHAR(255) NOT NULL, tenant_id VARCHAR(255) NOT NULL, @@ -69,7 +67,7 @@ additional_info JSONB ); --- Table for test_status_change_writer +-- Aggregated latest status per job (see ADR 001). CREATE TABLE IF NOT EXISTS public_cps_za_status_change_aggregated_job ( job_id UUID PRIMARY KEY, job_group_id UUID, @@ -97,4 +95,3 @@ finished_at TIMESTAMPTZ, last_updated_at TIMESTAMPTZ NOT NULL ); -""" diff --git a/database/migrations/V1.4.0.3__grants.ddl b/database/migrations/V1.4.0.3__grants.ddl new file mode 100644 index 0000000..7282e9a --- /dev/null +++ b/database/migrations/V1.4.0.3__grants.ddl @@ -0,0 +1,58 @@ +/* + * Copyright 2026 ABSA Group Limited + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +-- Object ownership and least-privilege grants for the application roles. + +-- Owner: owns every table (and its sequences) in the public schema. +ALTER TABLE public.public_cps_za_runs OWNER TO eventgate_owner; +ALTER TABLE public.public_cps_za_runs_jobs OWNER TO eventgate_owner; +ALTER TABLE public.public_cps_za_dlchange OWNER TO eventgate_owner; +ALTER TABLE public.public_cps_za_test OWNER TO eventgate_owner; +ALTER TABLE public.public_cps_za_status_change_aggregated_job OWNER TO eventgate_owner; + +-- Owner also needs CREATE on the schema so it can create future tables directly +GRANT CREATE ON SCHEMA public TO eventgate_owner; + +-- Both application roles (writer and reader) need to access the public schema. +GRANT USAGE ON SCHEMA public TO eventgate_writer, eventgate_reader; + +-- Reader: read-only access to EventGate data tables. +GRANT SELECT ON TABLE + public.public_cps_za_runs, + public.public_cps_za_runs_jobs, + public.public_cps_za_dlchange, + public.public_cps_za_test, + public.public_cps_za_status_change_aggregated_job +TO eventgate_reader; + +-- Writer: read and write EventGate data, but no DDL or migration metadata. +GRANT SELECT, INSERT, UPDATE ON TABLE + public.public_cps_za_runs, + public.public_cps_za_runs_jobs, + public.public_cps_za_dlchange, + public.public_cps_za_test, + public.public_cps_za_status_change_aggregated_job +TO eventgate_writer; + +-- Writer needs the SERIAL sequence (public_cps_za_runs_jobs.internal_id) to insert. +GRANT USAGE, SELECT ON SEQUENCE public.public_cps_za_runs_jobs_internal_id_seq TO eventgate_writer; + +-- Default privileges +ALTER DEFAULT PRIVILEGES FOR ROLE eventgate_owner IN SCHEMA public + GRANT SELECT ON TABLES TO eventgate_reader; +ALTER DEFAULT PRIVILEGES FOR ROLE eventgate_owner IN SCHEMA public + GRANT SELECT, INSERT, UPDATE ON TABLES TO eventgate_writer; +ALTER DEFAULT PRIVILEGES FOR ROLE eventgate_owner IN SCHEMA public + GRANT USAGE, SELECT ON SEQUENCES TO eventgate_writer; diff --git a/tests/integration/schemas/__init__.py b/flyway.toml similarity index 58% rename from tests/integration/schemas/__init__.py rename to flyway.toml index ebfbdd3..e252d13 100644 --- a/tests/integration/schemas/__init__.py +++ b/flyway.toml @@ -13,3 +13,16 @@ # See the License for the specific language governing permissions and # limitations under the License. # + +# Flyway configuration for EventGate database migrations. +# Lives at the project root alongside pyproject.toml, per project convention +# for top-level config. Paths below are relative to the repo root. + +[flyway] +locations = ["filesystem:database/migrations"] +sqlMigrationSuffixes = [".ddl", ".sql"] + +[environments.default] +url = "jdbc:postgresql://localhost:5432/eventgate_db" +user = "postgres" +password = "changeme" diff --git a/src/utils/config_loader.py b/src/utils/config_loader.py index 3093da4..c7565c2 100644 --- a/src/utils/config_loader.py +++ b/src/utils/config_loader.py @@ -59,7 +59,9 @@ def _load_json_from_path(path: str, aws_s3: ServiceResource) -> dict[str, Any]: name_parts = path.split("/") bucket_name = name_parts[2] bucket_object_key = "/".join(name_parts[3:]) - return json.loads(aws_s3.Bucket(bucket_name).Object(bucket_object_key).get()["Body"].read().decode("utf-8")) + bucket = aws_s3.Bucket(bucket_name) # type: ignore[attr-defined] + s3_object = bucket.Object(bucket_object_key) + return json.loads(s3_object.get()["Body"].read().decode("utf-8")) with open(path, "r", encoding="utf-8") as file: return json.load(file) diff --git a/src/writers/writer_eventbridge.py b/src/writers/writer_eventbridge.py index 076f007..d4908d4 100644 --- a/src/writers/writer_eventbridge.py +++ b/src/writers/writer_eventbridge.py @@ -36,7 +36,8 @@ class WriterEventBridge(Writer): def __init__(self, config: dict[str, Any]) -> None: super().__init__(config) - self._client: Optional["boto3.client"] = None + # boto3 clients are generated dynamically, so no precise static type exists. + self._client: Optional[Any] = None self._entries: list[dict[str, Any]] = [] self.event_bus_arn: str = config.get("event_bus_arn", "") logger.debug("Initialized EventBridge writer.") diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 0ffe7f5..c62bf4e 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -20,6 +20,7 @@ import logging import os import shutil +import subprocess import time from concurrent.futures import ThreadPoolExecutor, as_completed from http.server import HTTPServer, BaseHTTPRequestHandler @@ -37,12 +38,13 @@ from testcontainers.kafka import KafkaContainer from testcontainers.postgres import PostgresContainer -from tests.integration.schemas.postgres_schema import SCHEMA_SQL from tests.integration.utils.jwt_helper import create_test_jwt_keypair, generate_token logger = logging.getLogger(__name__) PROJECT_ROOT = Path(__file__).parent.parent.parent +FLYWAY_CONFIG = PROJECT_ROOT / "flyway.toml" +TEST_ROLE_PASSWORD = "changeme" # Mock JWT Provider (runs in-process via threading) @@ -165,6 +167,40 @@ def _convert_dsn(dsn: str) -> str: return dsn.replace("postgresql+psycopg2://", "postgresql://") +def _run_flyway_migrate(dsn: str) -> None: + """Apply Flyway migrations from `database/migrations` to the given database. + Runs the same migrations used for real environments so integration tests + validate the migrations as the single source of truth for the schema. + Args: + dsn: psycopg2-style DSN of the target database. + Raises: + RuntimeError: If the `flyway migrate` command fails. + """ + parsed = urlparse(dsn) + jdbc_url = f"jdbc:postgresql://{parsed.hostname}:{parsed.port}{parsed.path}" + command = [ + "flyway", + f"-configFiles={FLYWAY_CONFIG}", + f"-workingDirectory={PROJECT_ROOT}", + f"-url={jdbc_url}", + f"-user={parsed.username}", + f"-password={parsed.password}", + "migrate", + ] + environment = os.environ.copy() + environment.update( + { + "FLYWAY_PLACEHOLDERS_EVENTGATE_OWNER_PASSWORD": TEST_ROLE_PASSWORD, + "FLYWAY_PLACEHOLDERS_EVENTGATE_WRITER_PASSWORD": TEST_ROLE_PASSWORD, + "FLYWAY_PLACEHOLDERS_EVENTGATE_READER_PASSWORD": TEST_ROLE_PASSWORD, + } + ) + flyway_process = subprocess.run(command, capture_output=True, text=True, check=False, env=environment) + if flyway_process.returncode != 0: + raise RuntimeError(f"Flyway migrate failed:\n{flyway_process.stdout}\n{flyway_process.stderr}") + logger.debug("Flyway migrate output:\n%s", flyway_process.stdout) + + @pytest.fixture(scope="session") def postgres_container() -> Generator[str, None, None]: """PostgreSQL container with initialized schema.""" @@ -192,11 +228,10 @@ def postgres_container() -> Generator[str, None, None]: if conn is None: raise TimeoutError(f"Timed out waiting for Postgres to become available after 5 attempts: {last_exc}") - conn.autocommit = True - with conn.cursor() as cursor: - cursor.execute(SCHEMA_SQL) conn.close() - logger.debug("PostgreSQL schema initialized.") + logger.debug("Postgres ready, applying Flyway migrations.") + _run_flyway_migrate(dsn) + logger.debug("PostgreSQL schema initialized via Flyway.") yield dsn diff --git a/tests/integration/test_db_roles.py b/tests/integration/test_db_roles.py new file mode 100644 index 0000000..992ebf4 --- /dev/null +++ b/tests/integration/test_db_roles.py @@ -0,0 +1,128 @@ +# +# Copyright 2026 ABSA Group Limited +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +from urllib.parse import urlparse + +import psycopg2 +import pytest +from psycopg2 import errors + +from tests.integration.conftest import TEST_ROLE_PASSWORD + +SELECTABLE_TABLE = "public_cps_za_test" +OWNED_TABLES = ( + "public_cps_za_runs", + "public_cps_za_runs_jobs", + "public_cps_za_dlchange", + "public_cps_za_test", + "public_cps_za_status_change_aggregated_job", +) + + +def _connect_as(dsn: str, role: str) -> "psycopg2.extensions.connection": + parsed = urlparse(dsn) + return psycopg2.connect( + host=parsed.hostname, + port=parsed.port, + dbname=parsed.path.lstrip("/"), + user=role, + password=TEST_ROLE_PASSWORD, + ) + + +class TestReaderRole: + def test_reader_can_select(self, postgres_container: str) -> None: + conn = _connect_as(postgres_container, "eventgate_reader") + try: + with conn.cursor() as cursor: + cursor.execute(f"SELECT 1 FROM {SELECTABLE_TABLE} LIMIT 1") + finally: + conn.close() + + def test_reader_cannot_insert(self, postgres_container: str) -> None: + conn = _connect_as(postgres_container, "eventgate_reader") + try: + with conn.cursor() as cursor, pytest.raises(errors.InsufficientPrivilege): + cursor.execute( + f"INSERT INTO {SELECTABLE_TABLE} " + "(event_id, tenant_id, source_app, environment, timestamp_event) " + "VALUES ('e', 't', 'app', 'env', 1)" + ) + finally: + conn.close() + + def test_reader_cannot_read_flyway_history(self, postgres_container: str) -> None: + conn = _connect_as(postgres_container, "eventgate_reader") + try: + with conn.cursor() as cursor, pytest.raises(errors.InsufficientPrivilege): + cursor.execute("SELECT version FROM flyway_schema_history") + finally: + conn.close() + + +class TestWriterRole: + def test_writer_can_insert_and_select(self, postgres_container: str) -> None: + conn = _connect_as(postgres_container, "eventgate_writer") + try: + with conn.cursor() as cursor: + cursor.execute( + f"INSERT INTO {SELECTABLE_TABLE} " + "(event_id, tenant_id, source_app, environment, timestamp_event) " + "VALUES ('writer-e', 't', 'app', 'env', 1)" + ) + cursor.execute(f"SELECT 1 FROM {SELECTABLE_TABLE} LIMIT 1") + conn.commit() + finally: + conn.close() + + def test_writer_cannot_drop_table(self, postgres_container: str) -> None: + conn = _connect_as(postgres_container, "eventgate_writer") + try: + with conn.cursor() as cursor, pytest.raises(errors.InsufficientPrivilege): + cursor.execute(f"DROP TABLE {SELECTABLE_TABLE}") + finally: + conn.close() + + def test_writer_cannot_modify_flyway_history(self, postgres_container: str) -> None: + conn = _connect_as(postgres_container, "eventgate_writer") + try: + with conn.cursor() as cursor, pytest.raises(errors.InsufficientPrivilege): + cursor.execute("UPDATE flyway_schema_history SET description = description") + finally: + conn.close() + + +class TestOwnerRole: + def test_owner_owns_tables(self, postgres_container: str) -> None: + conn = _connect_as(postgres_container, "eventgate_owner") + try: + with conn.cursor() as cursor: + cursor.execute( + "SELECT tablename FROM pg_tables " "WHERE schemaname = 'public' AND tableowner = 'eventgate_owner'" + ) + owned = {row[0] for row in cursor.fetchall()} + finally: + conn.close() + assert set(OWNED_TABLES) == owned + + def test_owner_can_alter_table(self, postgres_container: str) -> None: + conn = _connect_as(postgres_container, "eventgate_owner") + try: + with conn.cursor() as cursor: + cursor.execute(f"ALTER TABLE {SELECTABLE_TABLE} ADD COLUMN tmp_col TEXT") + conn.rollback() + finally: + conn.close()