From e0b90236779519b041f86b6c48ca6e6c3124d868 Mon Sep 17 00:00:00 2001 From: Taimoor Zaeem Date: Wed, 12 Aug 2026 13:48:46 +0500 Subject: [PATCH] test(io): move config related behavior tests to test_config.py Signed-off-by: Taimoor Zaeem (cherry picked from commit e89e0bc2552ca0b3a589c6455f695c83b928c11d) --- test/io/test_auth.py | 98 ---------------- test/io/test_config.py | 251 +++++++++++++++++++++++++++++++++++++++++ test/io/test_io.py | 146 ------------------------ 3 files changed, 251 insertions(+), 244 deletions(-) delete mode 100644 test/io/test_auth.py create mode 100644 test/io/test_config.py diff --git a/test/io/test_auth.py b/test/io/test_auth.py deleted file mode 100644 index 66e75a438..000000000 --- a/test/io/test_auth.py +++ /dev/null @@ -1,98 +0,0 @@ -"Auth related IO tests for PostgREST" - -from operator import attrgetter -import pytest - -from config import BASEDIR, FIXTURES, SECRET -from util import authheader, jwtauthheader -from postgrest import run - - -@pytest.mark.parametrize( - "secretpath", - [path for path in (BASEDIR / "secrets").iterdir() if path.suffix != ".jwt"], - ids=attrgetter("name"), -) -def test_read_secret_from_file(secretpath, defaultenv): - "Authorization should succeed when the secret is read from a file." - - env = {**defaultenv, "PGRST_JWT_SECRET": f"@{secretpath}"} - - if secretpath.suffix == ".b64": - env["PGRST_JWT_SECRET_IS_BASE64"] = "true" - - secret = secretpath.read_bytes() - headers = authheader(secretpath.with_suffix(".jwt").read_text()) - - with run(stdin=secret, env=env) as postgrest: - response = postgrest.session.get("/authors_only", headers=headers) - print(response.text) - assert response.status_code == 200 - - -def test_read_secret_from_stdin(defaultenv): - "Authorization should succeed when the secret is read from stdin." - - env = {**defaultenv, "PGRST_DB_CONFIG": "false", "PGRST_JWT_SECRET": "@/dev/stdin"} - - headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET) - - with run(stdin=SECRET.encode(), env=env) as postgrest: - response = postgrest.session.get("/authors_only", headers=headers) - print(response.text) - assert response.status_code == 200 - - -# TODO: This test would fail right now, because of -# https://github.com/PostgREST/postgrest/issues/2126 -@pytest.mark.skip -def test_read_secret_from_stdin_dbconfig(defaultenv): - "Authorization should succeed when the secret is read from stdin with db-config=true." - - env = {**defaultenv, "PGRST_DB_CONFIG": "true", "PGRST_JWT_SECRET": "@/dev/stdin"} - - headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET) - - with run(stdin=SECRET.encode(), env=env) as postgrest: - response = postgrest.session.get("/authors_only", headers=headers) - print(response.text) - assert response.status_code == 200 - - -@pytest.mark.parametrize( - "roleclaim", FIXTURES["roleclaims"], ids=lambda claim: claim["key"] -) -def test_role_claim_key(roleclaim, defaultenv): - "Authorization should depend on a correct role-claim-key and JWT claim." - env = { - **defaultenv, - "PGRST_JWT_ROLE_CLAIM_KEY": roleclaim["key"], - "PGRST_JWT_SECRET": SECRET, - } - headers = jwtauthheader(roleclaim["data"], SECRET) - - with run(env=env) as postgrest: - response = postgrest.session.get("/authors_only", headers=headers) - assert response.status_code == roleclaim["expected_status"] - - -@pytest.mark.parametrize( - "jwtaudroleclaim", - FIXTURES["jwtaudroleclaims"], - ids=lambda claim: claim["key"] + "_" + str(claim["expected_status"]), -) -def test_jwt_aud_in_role_claim_key(jwtaudroleclaim, defaultenv): - "Allows authorization with JWT aud claim in role-claim-key" - - env = { - **defaultenv, - "PGRST_JWT_AUD": "postgrest_test_author", - "PGRST_JWT_ROLE_CLAIM_KEY": jwtaudroleclaim["key"], - "PGRST_JWT_SECRET": SECRET, - } - - headers = jwtauthheader(jwtaudroleclaim["data"], SECRET) - - with run(env=env) as postgrest: - response = postgrest.session.get("/authors_only", headers=headers) - assert response.status_code == jwtaudroleclaim["expected_status"] diff --git a/test/io/test_config.py b/test/io/test_config.py new file mode 100644 index 000000000..7dd4c8bf2 --- /dev/null +++ b/test/io/test_config.py @@ -0,0 +1,251 @@ +"Test PostgREST configuration related behavior" + +import time +import pytest +from operator import attrgetter + +from config import BASEDIR, FIXTURES, SECRET +from util import ( + Thread, + authheader, + jwtauthheader, +) +from postgrest import ( + PostgrestTimedOut, + freeport, + run, +) + + +def test_pool_size(defaultenv, metapostgrest): + "Verify that PGRST_DB_POOL setting allows the correct number of parallel requests" + + env = { + **defaultenv, + "PGRST_DB_POOL": "2", + } + + with run(env=env) as postgrest: + start = time.time() + threads = [] + for i in range(4): + + def sleep(i=i): + response = postgrest.session.get("/rpc/sleep?seconds=0.5") + assert response.text == "" + assert response.status_code == 204, "thread {}".format(i) + + t = Thread(target=sleep) + t.start() + threads.append(t) + for t in threads: + t.join() + end = time.time() + delta = end - start + + # sleep 4 times for 0.5s each, with 2 requests in parallel + # => total time roughly 1s + assert delta > 1 and delta < 1.5 + + +@pytest.mark.parametrize( + "secretpath", + [path for path in (BASEDIR / "secrets").iterdir() if path.suffix != ".jwt"], + ids=attrgetter("name"), +) +def test_read_secret_from_file(secretpath, defaultenv): + "Authorization should succeed when the secret is read from a file." + + env = {**defaultenv, "PGRST_JWT_SECRET": f"@{secretpath}"} + + if secretpath.suffix == ".b64": + env["PGRST_JWT_SECRET_IS_BASE64"] = "true" + + secret = secretpath.read_bytes() + headers = authheader(secretpath.with_suffix(".jwt").read_text()) + + with run(stdin=secret, env=env) as postgrest: + response = postgrest.session.get("/authors_only", headers=headers) + print(response.text) + assert response.status_code == 200 + + +def test_read_secret_from_stdin(defaultenv): + "Authorization should succeed when the secret is read from stdin." + + env = {**defaultenv, "PGRST_DB_CONFIG": "false", "PGRST_JWT_SECRET": "@/dev/stdin"} + + headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET) + + with run(stdin=SECRET.encode(), env=env) as postgrest: + response = postgrest.session.get("/authors_only", headers=headers) + print(response.text) + assert response.status_code == 200 + + +# TODO: This test would fail right now, because of +# https://github.com/PostgREST/postgrest/issues/2126 +@pytest.mark.skip +def test_read_secret_from_stdin_dbconfig(defaultenv): + "Authorization should succeed when the secret is read from stdin with db-config=true." + + env = {**defaultenv, "PGRST_DB_CONFIG": "true", "PGRST_JWT_SECRET": "@/dev/stdin"} + + headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET) + + with run(stdin=SECRET.encode(), env=env) as postgrest: + response = postgrest.session.get("/authors_only", headers=headers) + print(response.text) + assert response.status_code == 200 + + +@pytest.mark.parametrize( + "roleclaim", FIXTURES["roleclaims"], ids=lambda claim: claim["key"] +) +def test_role_claim_key(roleclaim, defaultenv): + "Authorization should depend on a correct role-claim-key and JWT claim." + env = { + **defaultenv, + "PGRST_JWT_ROLE_CLAIM_KEY": roleclaim["key"], + "PGRST_JWT_SECRET": SECRET, + } + headers = jwtauthheader(roleclaim["data"], SECRET) + + with run(env=env) as postgrest: + response = postgrest.session.get("/authors_only", headers=headers) + assert response.status_code == roleclaim["expected_status"] + + +@pytest.mark.parametrize( + "jwtaudroleclaim", + FIXTURES["jwtaudroleclaims"], + ids=lambda claim: claim["key"] + "_" + str(claim["expected_status"]), +) +def test_jwt_aud_in_role_claim_key(jwtaudroleclaim, defaultenv): + "Allows authorization with JWT aud claim in role-claim-key" + + env = { + **defaultenv, + "PGRST_JWT_AUD": "postgrest_test_author", + "PGRST_JWT_ROLE_CLAIM_KEY": jwtaudroleclaim["key"], + "PGRST_JWT_SECRET": SECRET, + } + + headers = jwtauthheader(jwtaudroleclaim["data"], SECRET) + + with run(env=env) as postgrest: + response = postgrest.session.get("/authors_only", headers=headers) + assert response.status_code == jwtaudroleclaim["expected_status"] + + +def test_random_port_bound(defaultenv): + "PostgREST should bind to a random port when PGRST_SERVER_PORT is 0." + + with run(env=defaultenv, port="0"): + assert True # liveness check is done by run(), so we just need to check that it doesn't fail + + +def test_so_reuseport_defaults_to_false(defaultenv): + "A second PostgREST instance should not bind to the same port by default." + + host = "0.0.0.0" + port = freeport() + admin_port = freeport(used_ports=[port]) + + with run( + env={**defaultenv}, + port=port, + host=host, + admin_port=admin_port, + ): + with pytest.raises(PostgrestTimedOut): + with run( + env={**defaultenv}, + port=port, + host=host, + admin_port=freeport(used_ports=[port, admin_port]), + wait_max_seconds=1, + ): + pass + + +def test_schema_cache_startup_load_with_in_db_config(defaultenv, metapostgrest): + "verify that the Schema Cache loads correctly at startup, using the in-db `pgrst.db_schemas` config" + + response = metapostgrest.session.post("/rpc/change_db_schemas_config") + assert response.text == "" + assert response.status_code == 204 + + with run(env=defaultenv) as postgrest: + response = postgrest.session.get("/rpc/get_current_schema") + assert response.text == '"test"' + assert response.status_code == 200 + + response = metapostgrest.session.post("/rpc/reset_db_schemas_config") + assert response.text == "" + assert response.status_code == 204 + + +def test_allow_configs_to_be_set_to_empty(defaultenv): + 'configs that are explicitly set to empty (= "") should not throw parse error' + + env = { + **defaultenv, + "PGRST_DB_EXTRA_SEARCH_PATH": "", + } + + with run(env=env) as postgrest: + response = postgrest.session.get("/projects") + assert response.status_code == 200 + + +def test_connection_error_message_does_not_claim_retry(defaultenv): + "The connection error message should not claim retrying, since PostgREST stops on fatal errors." + uri = f'postgresql://?dbname={defaultenv["PGDATABASE"]}&host={defaultenv["PGHOST"]}&user=some_protected_user&password=invalid_pass' + env = {**defaultenv, "PGRST_DB_URI": uri} + with run(env=env, no_startup_stdout=False, wait_for=None) as postgrest: + output = postgrest.read_stdout(nlines=8) + assert any('"message":"Database connection error."' in line for line in output) + + +def test_db_pre_config_with_non_existent_function(defaultenv): + "Log error when db-pre-config is set to non-existent function" + + env = { + **defaultenv, + "PGRST_DB_PRE_CONFIG": "select", # no "select" function in our fixtures, fail gracefully at startup + } + + with run(env=env, no_startup_stdout=False, wait_for=None) as postgrest: + output = postgrest.read_stdout(nlines=8) + assert any("function select() does not exist" in line for line in output) + + +@pytest.mark.parametrize("enabled", ["true", "false"]) +def test_use_legacy_target_names(enabled, defaultenv): + "Show a warning when a target name is used instead of an alias, only when config is enabled" + + env = { + **defaultenv, + "PGRST_URL_USE_LEGACY_TARGET_NAMES": enabled, + } + + with run(env=env) as postgrest: + response = postgrest.session.get( + "/directors?select=name,all_films:films(title),awards_2026:awards(name)&films.order=title&awards.year=eq.2026" + ) + + output = postgrest.read_stdout(nlines=10) + + log_err_warning = "WARNING: Embedded resource was referenced by relation name even though it has an alias. This is deprecated and will stop working in a future release." + log_err_hint = "Update filters, orders or limits that use `films` to `all_films`, `awards` to `awards_2026` in `GET /directors?select=name,all_films:films(title),awards_2026:awards(name)&films.order=title&awards.year=eq.2026`" + + has_warning_log = any(log_err_warning in line for line in output) + has_hint_log = any(log_err_hint in line for line in output) + + if enabled == "true": + assert response.status_code == 200 + assert has_warning_log and has_hint_log + else: + assert response.status_code == 400 + assert not has_warning_log and not has_hint_log diff --git a/test/io/test_io.py b/test/io/test_io.py index 158e05b33..57c3fd322 100644 --- a/test/io/test_io.py +++ b/test/io/test_io.py @@ -1,14 +1,12 @@ "Unit tests for Input/Ouput of PostgREST seen as a black box." import time -import pytest from util import ( Thread, psql_as_superuser, ) from postgrest import ( - PostgrestTimedOut, freeport, run, wait_until_exit, @@ -36,13 +34,6 @@ def test_graceful_shutdown_waits_for_in_flight_request(defaultenv): t.join() -def test_random_port_bound(defaultenv): - "PostgREST should bind to a random port when PGRST_SERVER_PORT is 0." - - with run(env=defaultenv, port="0"): - assert True # liveness check is done by run(), so we just need to check that it doesn't fail - - def test_so_reuseport_zero_downtime_handover(defaultenv): "A second PostgREST instance should take over on the same main/admin ports without request failures." @@ -110,61 +101,6 @@ def test_so_reuseport_zero_downtime_handover(defaultenv): assert failures == [] -def test_so_reuseport_defaults_to_false(defaultenv): - "A second PostgREST instance should not bind to the same port by default." - - host = "0.0.0.0" - port = freeport() - admin_port = freeport(used_ports=[port]) - - with run( - env={**defaultenv}, - port=port, - host=host, - admin_port=admin_port, - ): - with pytest.raises(PostgrestTimedOut): - with run( - env={**defaultenv}, - port=port, - host=host, - admin_port=freeport(used_ports=[port, admin_port]), - wait_max_seconds=1, - ): - pass - - -def test_pool_size(defaultenv, metapostgrest): - "Verify that PGRST_DB_POOL setting allows the correct number of parallel requests" - - env = { - **defaultenv, - "PGRST_DB_POOL": "2", - } - - with run(env=env) as postgrest: - start = time.time() - threads = [] - for i in range(4): - - def sleep(i=i): - response = postgrest.session.get("/rpc/sleep?seconds=0.5") - assert response.text == "" - assert response.status_code == 204, "thread {}".format(i) - - t = Thread(target=sleep) - t.start() - threads.append(t) - for t in threads: - t.join() - end = time.time() - delta = end - start - - # sleep 4 times for 0.5s each, with 2 requests in parallel - # => total time roughly 1s - assert delta > 1 and delta < 1.5 - - def test_listener_query_is_visible_in_pg_stat_activity(defaultenv): "The listener connection should show the LISTEN pgrst statement in pg_stat_activity" @@ -187,85 +123,3 @@ def test_listener_query_is_visible_in_pg_stat_activity(defaultenv): ).strip() assert output == 'LISTEN "pgrst"' - - -def test_schema_cache_startup_load_with_in_db_config(defaultenv, metapostgrest): - "verify that the Schema Cache loads correctly at startup, using the in-db `pgrst.db_schemas` config" - - response = metapostgrest.session.post("/rpc/change_db_schemas_config") - assert response.text == "" - assert response.status_code == 204 - - with run(env=defaultenv) as postgrest: - response = postgrest.session.get("/rpc/get_current_schema") - assert response.text == '"test"' - assert response.status_code == 200 - - response = metapostgrest.session.post("/rpc/reset_db_schemas_config") - assert response.text == "" - assert response.status_code == 204 - - -def test_allow_configs_to_be_set_to_empty(defaultenv): - 'configs that are explicitly set to empty (= "") should not throw parse error' - - env = { - **defaultenv, - "PGRST_DB_EXTRA_SEARCH_PATH": "", - } - - with run(env=env) as postgrest: - response = postgrest.session.get("/projects") - assert response.status_code == 200 - - -def test_connection_error_message_does_not_claim_retry(defaultenv): - "The connection error message should not claim retrying, since PostgREST stops on fatal errors." - uri = f'postgresql://?dbname={defaultenv["PGDATABASE"]}&host={defaultenv["PGHOST"]}&user=some_protected_user&password=invalid_pass' - env = {**defaultenv, "PGRST_DB_URI": uri} - with run(env=env, no_startup_stdout=False, wait_for=None) as postgrest: - output = postgrest.read_stdout(nlines=8) - assert any('"message":"Database connection error."' in line for line in output) - - -def test_db_pre_config_with_non_existent_function(defaultenv): - "Log error when db-pre-config is set to non-existent function" - - env = { - **defaultenv, - "PGRST_DB_PRE_CONFIG": "select", # no "select" function in our fixtures, fail gracefully at startup - } - - with run(env=env, no_startup_stdout=False, wait_for=None) as postgrest: - output = postgrest.read_stdout(nlines=8) - assert any("function select() does not exist" in line for line in output) - - -@pytest.mark.parametrize("enabled", ["true", "false"]) -def test_use_legacy_target_names(enabled, defaultenv): - "Show a warning when a target name is used instead of an alias, only when config is enabled" - - env = { - **defaultenv, - "PGRST_URL_USE_LEGACY_TARGET_NAMES": enabled, - } - - with run(env=env) as postgrest: - response = postgrest.session.get( - "/directors?select=name,all_films:films(title),awards_2026:awards(name)&films.order=title&awards.year=eq.2026" - ) - - output = postgrest.read_stdout(nlines=10) - - log_err_warning = "WARNING: Embedded resource was referenced by relation name even though it has an alias. This is deprecated and will stop working in a future release." - log_err_hint = "Update filters, orders or limits that use `films` to `all_films`, `awards` to `awards_2026` in `GET /directors?select=name,all_films:films(title),awards_2026:awards(name)&films.order=title&awards.year=eq.2026`" - - has_warning_log = any(log_err_warning in line for line in output) - has_hint_log = any(log_err_hint in line for line in output) - - if enabled == "true": - assert response.status_code == 200 - assert has_warning_log and has_hint_log - else: - assert response.status_code == 400 - assert not has_warning_log and not has_hint_log