Add scheduler sample for PostgreSQL user synchronization #6045
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,302 @@ | ||
| #!/usr/bin/env python3 | ||
| """Run the PostgreSQL user synchronizer against real services.""" | ||
|
|
||
| import json | ||
| import os | ||
| import secrets | ||
| import subprocess | ||
| import sys | ||
| import tempfile | ||
| import uuid | ||
| from pathlib import Path | ||
|
|
||
| import pymysql | ||
|
gitar-bot[bot] marked this conversation as resolved.
|
||
|
|
||
|
|
||
| ROOT = Path(os.environ.get("WORKSPACE", Path(__file__).resolve().parents[3])) | ||
| ASSET_DIR = ROOT / "test/tap/pgsql_user_sync" | ||
| SCRIPT = ASSET_DIR / "proxysql_pgsql_user_sync.py" | ||
| PROFILE = "tap-real" | ||
|
|
||
|
|
||
| try: | ||
| import psycopg | ||
| from psycopg import sql | ||
| except ImportError: | ||
| raise SystemExit( | ||
| "psycopg is required for this test; install the test-image dependencies before running it" | ||
| ) from None | ||
|
|
||
|
|
||
| class Tap: | ||
| def __init__(self): | ||
| self.count = 0 | ||
| self.failures = 0 | ||
|
|
||
| def check(self, condition, description): | ||
| self.count += 1 | ||
| print(f"{'ok' if condition else 'not ok'} {self.count} - {description}") | ||
| if not condition: | ||
| self.failures += 1 | ||
|
|
||
|
|
||
| def env(name, default=None): | ||
| value = os.environ.get(name, default) | ||
| if not value: | ||
| raise RuntimeError(f"required environment variable {name} is not set") | ||
| return value | ||
|
|
||
|
|
||
| def source_connection(): | ||
| return psycopg.connect( | ||
| host=env("TAP_PGSQLSERVER_HOST"), | ||
| port=int(env("TAP_PGSQLSERVER_PORT", "5432")), | ||
| dbname="postgres", | ||
| user=env("TAP_PGSQLSERVER_USERNAME"), | ||
| password=env("TAP_PGSQLSERVER_PASSWORD"), | ||
| autocommit=True, | ||
| ) | ||
|
|
||
|
|
||
| def admin_connection(): | ||
| return pymysql.connect( | ||
| host=env("TAP_ADMINHOST", "127.0.0.1"), | ||
| port=int(env("TAP_ADMINPORT", "6032")), | ||
| user=env("TAP_ADMINUSERNAME", "radmin"), | ||
| password=env("TAP_ADMINPASSWORD", "radmin"), | ||
| database="main", | ||
| autocommit=True, | ||
| ) | ||
|
|
||
|
|
||
| def admin_execute(query, params=()): | ||
| with admin_connection() as connection: | ||
| with connection.cursor() as cursor: | ||
| cursor.execute(query, params) | ||
|
|
||
|
|
||
| def admin_row(username, runtime=False): | ||
| table = "runtime_pgsql_users" if runtime else "pgsql_users" | ||
| with admin_connection() as connection: | ||
| with connection.cursor() as cursor: | ||
| cursor.execute( | ||
| f"SELECT username,password,active,default_hostgroup,attributes " | ||
| f"FROM {table} WHERE username=%s AND backend=1", | ||
| (username,), | ||
| ) | ||
| return cursor.fetchone() | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
|
|
||
| def create_source_role(connection, schema, username, password): | ||
| with connection.cursor() as cursor: | ||
| cursor.execute( | ||
| sql.SQL("CREATE ROLE {} LOGIN PASSWORD {}").format( | ||
| sql.Identifier(username), sql.Literal(password) | ||
| ) | ||
| ) | ||
| cursor.execute( | ||
| sql.SQL("CREATE SCHEMA {} AUTHORIZATION CURRENT_USER").format( | ||
| sql.Identifier(schema) | ||
| ) | ||
| ) | ||
| cursor.execute( | ||
| sql.SQL( | ||
| "CREATE FUNCTION {}.export_login_role() " | ||
| "RETURNS TABLE(username text, password text) " | ||
| "LANGUAGE sql SECURITY DEFINER SET search_path = pg_catalog AS $$ " | ||
| "SELECT rolname::text, rolpassword FROM pg_catalog.pg_authid " | ||
| "WHERE rolname = {} AND rolcanlogin AND rolpassword IS NOT NULL $$" | ||
| ).format(sql.Identifier(schema), sql.Literal(username)) | ||
| ) | ||
|
|
||
|
|
||
| def source_verifier(connection, username): | ||
| with connection.cursor() as cursor: | ||
| cursor.execute( | ||
| "SELECT rolpassword FROM pg_catalog.pg_authid WHERE rolname=%s", | ||
| (username,), | ||
| ) | ||
| return cursor.fetchone()[0] | ||
|
|
||
|
|
||
| def write_config(path, schema, workdir): | ||
| path.write_text( | ||
| "\n".join( | ||
| ( | ||
| "[source]", | ||
| f"host = {env('TAP_PGSQLSERVER_HOST')}", | ||
| f"port = {env('TAP_PGSQLSERVER_PORT', '5432')}", | ||
| "database = postgres", | ||
| f"username = {env('TAP_PGSQLSERVER_USERNAME')}", | ||
| f"password = {env('TAP_PGSQLSERVER_PASSWORD')}", | ||
| f"function = {schema}.export_login_role", | ||
| "", | ||
| "[proxysql]", | ||
| f"host = {env('TAP_PGSQLADMIN_HOST', '127.0.0.1')}", | ||
| f"port = {env('TAP_PGSQLADMIN_PORT', '6132')}", | ||
| f"username = {env('TAP_ADMINUSERNAME', 'radmin')}", | ||
| f"password = {env('TAP_ADMINPASSWORD', 'radmin')}", | ||
| "", | ||
| "[sync]", | ||
| f"profile = {PROFILE}", | ||
| "default_hostgroup = 0", | ||
| "missing_role_action = disable", | ||
| "adopt_existing_users = false", | ||
| "allow_empty_snapshot = false", | ||
| "save_to_disk = true", | ||
| f"lock_file = {workdir / 'pgsql-user-sync.lock'}", | ||
| "", | ||
| ) | ||
| ), | ||
| encoding="utf-8", | ||
| ) | ||
| os.chmod(path, 0o600) | ||
|
|
||
|
|
||
| def direct_login(username, password): | ||
| with psycopg.connect( | ||
| host=env("TAP_PGSQLSERVER_HOST"), | ||
| port=int(env("TAP_PGSQLSERVER_PORT", "5432")), | ||
| dbname="postgres", | ||
| user=username, | ||
| password=password, | ||
| connect_timeout=5, | ||
| ) as connection: | ||
| connection.execute("SELECT 1") | ||
|
|
||
|
|
||
| def cleanup(connection, schema, username): | ||
| success = True | ||
| try: | ||
| admin_execute("DELETE FROM pgsql_users WHERE username=%s", (username,)) | ||
| admin_execute("LOAD PGSQL USERS TO RUNTIME") | ||
| admin_execute("SAVE PGSQL USERS TO DISK") | ||
| except Exception: | ||
| success = False | ||
| if connection is not None: | ||
| try: | ||
| with connection.cursor() as cursor: | ||
| cursor.execute( | ||
| sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format( | ||
| sql.Identifier(schema) | ||
| ) | ||
| ) | ||
| cursor.execute( | ||
| sql.SQL("DROP ROLE IF EXISTS {}").format(sql.Identifier(username)) | ||
| ) | ||
| except Exception: | ||
| success = False | ||
| return success | ||
|
|
||
|
|
||
| def main(): | ||
| tap = Tap() | ||
| suffix = uuid.uuid4().hex[:16] | ||
| username = f"tap_psync_{suffix}" | ||
| schema = f"tap_psync_schema_{suffix}" | ||
| password = secrets.token_urlsafe(24) | ||
| source = None | ||
|
|
||
| try: | ||
| source = source_connection() | ||
| create_source_role(source, schema, username, password) | ||
| direct_login(username, password) | ||
| tap.check(True, "login role exists on the real PostgreSQL backend") | ||
|
|
||
| with tempfile.TemporaryDirectory(prefix="pgsql-user-sync-") as temp: | ||
| workdir = Path(temp) | ||
| config = workdir / "pgsql-user-sync.ini" | ||
| write_config(config, schema, workdir) | ||
| result = subprocess.run( | ||
| [sys.executable, str(SCRIPT), "--config", str(config)], | ||
| text=True, | ||
| capture_output=True, | ||
| ) | ||
| if result.returncode != 0: | ||
| print(f"# synchronizer diagnostic: {result.stderr.strip()}") | ||
| repeated = subprocess.run( | ||
| [sys.executable, str(SCRIPT), "--config", str(config)], | ||
| text=True, | ||
| capture_output=True, | ||
| ) | ||
| if repeated.returncode != 0: | ||
| print(f"# repeated synchronizer diagnostic: {repeated.stderr.strip()}") | ||
|
|
||
| tap.check(result.returncode == 0, "real synchronizer run succeeds") | ||
| verifier = source_verifier(source, username) | ||
| main_row = admin_row(username) | ||
| runtime_row = admin_row(username, runtime=True) | ||
| if ( | ||
| main_row is None | ||
| or main_row[1] != verifier | ||
| or int(main_row[2]) != 1 | ||
| or int(main_row[3]) != 0 | ||
| ): | ||
| print( | ||
| "# main row diagnostic: " | ||
| f"present={main_row is not None} " | ||
| f"verifier_match={main_row is not None and main_row[1] == verifier} " | ||
| f"active={main_row[2] if main_row else None} " | ||
| f"hostgroup={main_row[3] if main_row else None} " | ||
| f"attributes={(main_row[4] if main_row else None)!r}" | ||
| ) | ||
| if ( | ||
| runtime_row is None | ||
| or runtime_row[1] != verifier | ||
| or int(runtime_row[2]) != 1 | ||
| or int(runtime_row[3]) != 0 | ||
| ): | ||
| print( | ||
| "# runtime row diagnostic: " | ||
| f"present={runtime_row is not None} " | ||
| f"verifier_match={runtime_row is not None and runtime_row[1] == verifier} " | ||
| f"active={runtime_row[2] if runtime_row else None} " | ||
| f"hostgroup={runtime_row[3] if runtime_row else None}" | ||
| ) | ||
| try: | ||
| ownership_marker = ( | ||
| json.loads(main_row[4] or "{}").get("proxysql_pgsql_user_sync") | ||
| if main_row is not None | ||
| else None | ||
| ) | ||
| except (TypeError, json.JSONDecodeError): | ||
| ownership_marker = None | ||
| tap.check( | ||
| main_row is not None | ||
| and main_row[1] == verifier | ||
| and int(main_row[2]) == 1 | ||
| and int(main_row[3]) == 0 | ||
| and ownership_marker == {"profile": PROFILE}, | ||
| "synchronizer automatically creates the real pgsql_users row", | ||
| ) | ||
| tap.check( | ||
| runtime_row is not None | ||
| and runtime_row[1] == verifier | ||
| and int(runtime_row[2]) == 1 | ||
| and int(runtime_row[3]) == 0, | ||
| "synchronizer loads the user into runtime_pgsql_users", | ||
| ) | ||
| repeated_is_noop = ( | ||
| repeated.returncode == 0 | ||
| and "loaded=false saved=false" in repeated.stdout | ||
| ) | ||
| if not repeated_is_noop: | ||
| print(f"# repeated synchronizer output: {repeated.stdout.strip()}") | ||
| tap.check( | ||
| repeated_is_noop, | ||
| "repeated synchronization detects no backend-runtime drift", | ||
| ) | ||
| except Exception as error: | ||
| print(f"# real integration failure class: {type(error).__name__}") | ||
| tap.check(False, "real PostgreSQL to ProxySQL synchronization completes") | ||
| finally: | ||
| tap.check(cleanup(source, schema, username), "test data is removed") | ||
| if source is not None: | ||
| source.close() | ||
|
|
||
| print(f"1..{tap.count}") | ||
| return 1 if tap.failures else 0 | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| raise SystemExit(main()) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| #!/usr/bin/env python3 | ||
| import os | ||
| import subprocess | ||
| import sys | ||
| from pathlib import Path | ||
|
|
||
| root = Path(os.environ.get("WORKSPACE", Path(__file__).resolve().parents[3])) | ||
| suite = root / "test/tap/pgsql_user_sync/tests/test_pgsql_user_sync.py" | ||
|
renecannao marked this conversation as resolved.
|
||
| if not suite.is_file(): | ||
| suite = root / "tools/pgsql_user_sync/tests/test_pgsql_user_sync.py" | ||
| raise SystemExit(subprocess.call( | ||
| [sys.executable, str(suite)], cwd=root | ||
| )) | ||
|
Comment on lines
+11
to
+13
There was a problem hiding this comment. 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
# Description: Determine whether the TAP runner parses output or only exit codes.
set -euo pipefail
rg -n -C6 'TEST_PY_TAP|tap_parser|1\.\.|not ok' test/infra test/tap --glob '!**/tests/*-t.cpp' | head -n 200Repository: sysown/proxysql Length of output: 11234 🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '%s\n' '--- candidate runner files ---'
fd -t f . test/infra test/tap | rg '(^|/)(run-tests|.*tap.*|.*test.*\.py)$' | head -n 200
printf '%s\n' '--- runner references to Python TAP tests and output handling ---'
rg -n -C8 'TEST_PY_TAP_INCL|subprocess|Popen|communicate|returncode|tap|TAP|unittest|pytest|\.py' \
test/infra test/tap \
--glob '*.bash' --glob '*.sh' --glob '*.py' --glob 'Makefile*' --glob '*.json' \
| head -n 400Repository: sysown/proxysql Length of output: 8309 🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '%s\n' '--- isolated runner paths ---'
find test -type f \( -name 'run-tests-isolated.bash' -o -name '*runner*' -o -name '*tap*.py' -o -name '*tap*.bash' \) -print
printf '%s\n' '--- all execution and result-handling references ---'
rg -n -C8 'TEST_PY_TAP_INCL|TEST_PY_TAP|subprocess|Popen|returncode|communicate|unittest|tap-parser|TAP' \
test --glob '*.bash' --glob '*.sh' --glob '*.py' --glob 'Makefile*' --glob '*.json' \
| head -n 500Repository: sysown/proxysql Length of output: 39415 🏁 Script executed: #!/bin/bash
set -euo pipefail
printf '%s\n' '--- isolated runner ---'
sed -n '1,260p' test/infra/control/run-tests-isolated.bash
printf '%s\n' '--- test discovery, invocation, and result handling ---'
rg -n -C12 'TEST_PY_TAP_INCL|tap_tests|Popen|subprocess\.run|subprocess\.call|returncode|exit\(|sys\.exit|1\.\.|not ok|ok [0-9]' \
test/scripts/bin/proxysql-tester.py \
| head -n 500
printf '%s\n' '--- wrapper and neighboring Python test files ---'
sed -n '1,180p' test/tap/tests/pgsql-user-sync-unit-t.py
sed -n '1,180p' test/tap/tests/pgsql-user-sync-t.py 2>/dev/null || trueRepository: sysown/proxysql Length of output: 40546 🏁 Script executed: #!/bin/bash
set -u
printf '%s\n' '--- matching files ---'
git ls-files | rg 'pgsql-user-sync|run-tests-isolated|proxysql-tester'
find test/tap/tests -maxdepth 1 -type f -name 'pgsql-user-sync*' -printf '%M %p\n'
printf '%s\n' '--- isolated runner remainder ---'
sed -n '240,520p' test/infra/control/run-tests-isolated.bash
printf '%s\n' '--- target wrapper ---'
target=$(git ls-files | rg '(^|/)pgsql-user-sync-unit-t\.py$' | head -n 1)
if [ -n "${target}" ]; then
cat -n "${target}"
else
echo 'target file not found in tracked files'
fi
printf '%s\n' '--- tester entry points ---'
rg -n -C10 'execute_tap_tests|proxysql-tester|tap_workdir|TEST_PY_TAP_INCL|python' \
test/infra/control/run-tests-isolated.bash test/scripts/bin/proxysql-tester.py test/tap/Makefile \
| head -n 500Repository: sysown/proxysql Length of output: 43029 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import os
import stat
from pathlib import Path
p = Path("test/tap/tests/pgsql-user-sync-unit-t.py")
st = p.stat()
print(f"path={p}")
print(f"mode={oct(stat.S_IMODE(st.st_mode))}")
print(f"executable={os.access(p, os.X_OK)}")
print(f"matches_runner_glob={p.match('*-t')}")
PY
printf '%s\n' '--- TAP workdir configuration ---'
rg -n -C8 'TAP_WORKDIR(S)?|test/tap/tests|pgsql-user-sync-unit-t' \
test/infra test/scripts/etc test/tap/groups test/tap/Makefile \
--glob '*.bash' --glob '*.sh' --glob '*.yml' --glob '*.yaml' --glob '*.json' --glob 'Makefile*' \
| head -n 300
printf '%s\n' '--- file modes for neighboring Python TAP wrappers ---'
find test/tap/tests -maxdepth 1 -type f -name '*-t.py' -printf '%M %f\n' | sortRepository: sysown/proxysql Length of output: 28481 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
import glob
from pathlib import Path
root = Path("test/tap/tests")
pattern = str(root / "*-t")
matches = sorted(Path(p) for p in glob.glob(pattern))
print(f"pattern={pattern}")
print("target_in_matches=" + str((root / "pgsql-user-sync-unit-t.py") in matches))
print("neighbor_in_matches=" + str((root / "pgsql-user-sync-t.py") in matches))
print("matching_names=" + repr([p.name for p in matches if "pgsql-user-sync" in p.name]))
PY
printf '%s\n' '--- TAP workdir values ---'
rg -n -C10 'TAP_WORKDIR(S)?|test/tap/tests' \
test/scripts/etc/proxysql-tester.yml test/infra/control/env-isolated.bash \
test/infra/common/env.sh test/tap/groups/no-infra-g1 \
2>/dev/null | head -n 300
printf '%s\n' '--- group registrations ---'
rg -n -C2 'pgsql-user-sync(-unit)?-t' test/tap/groups/groups.jsonRepository: sysown/proxysql Length of output: 6359 Rename the wrapper to match the runner glob. 🧰 Tools🪛 Ruff (0.16.1)[error] 9-9: (S603) 🤖 Prompt for AI Agents |
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.