2011 lines
70 KiB
Python
2011 lines
70 KiB
Python
"Unit tests for Input/Ouput of PostgREST seen as a black box."
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from operator import attrgetter
|
|
import os
|
|
import re
|
|
import signal
|
|
import time
|
|
import pytest
|
|
|
|
from config import BASEDIR, CONFIGSDIR, FIXTURES, SECRET
|
|
from util import Thread, authheader, jwtauthheader, parse_server_timings_header
|
|
from postgrest import (
|
|
freeport,
|
|
is_ipv6,
|
|
reset_statement_timeout,
|
|
run,
|
|
set_statement_timeout,
|
|
sleep_until_postgrest_config_reload,
|
|
sleep_until_postgrest_full_reload,
|
|
sleep_until_postgrest_scache_reload,
|
|
wait_until_exit,
|
|
)
|
|
|
|
|
|
@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
|
|
|
|
|
|
def test_jwt_errors(defaultenv):
|
|
"invalid JWT should throw error"
|
|
|
|
env = {**defaultenv, "PGRST_JWT_SECRET": SECRET, "PGRST_JWT_AUD": "io tests"}
|
|
|
|
relativeSeconds = lambda sec: int(
|
|
(datetime.now(timezone.utc) + timedelta(seconds=sec)).timestamp()
|
|
)
|
|
|
|
with run(env=env) as postgrest:
|
|
headers = jwtauthheader({}, "other secret")
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "No suitable key or wrong key type"
|
|
assert (
|
|
response.json()["details"] == "None of the keys was able to decode the JWT"
|
|
)
|
|
|
|
headers = jwtauthheader({"role": "not_existing"}, SECRET)
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == 'role "not_existing" does not exist'
|
|
|
|
# -35 seconds, because we allow clock skew of 30 seconds
|
|
headers = jwtauthheader({"exp": relativeSeconds(-35)}, SECRET)
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "JWT expired"
|
|
|
|
# 35 seconds, because we allow clock skew of 30 seconds
|
|
headers = jwtauthheader({"nbf": relativeSeconds(35)}, SECRET)
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "JWT not yet valid"
|
|
|
|
# 35 seconds, because we allow clock skew of 35 seconds
|
|
headers = jwtauthheader({"iat": relativeSeconds(35)}, SECRET)
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "JWT issued at future"
|
|
|
|
headers = jwtauthheader({"aud": "not set"}, SECRET)
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "JWT not in audience"
|
|
|
|
# partial token, no signature
|
|
headers = authheader("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.bm90IGFuIG9iamVjdA")
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "Expected 3 parts in JWT; got 2"
|
|
|
|
# complete token but random characters
|
|
headers = authheader("quifquirndsjagnrgniur.fonvoienqhhdj.iuqvnvhojah")
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "JWT cryptographic operation failed"
|
|
|
|
# token with algorithm "none"
|
|
headers = authheader(
|
|
"eyJ0eXAiOiJKV1QiLCJhbGciOiJub25lIn0.e30.yOBhlOIqn56T-4NvyEXCjfi3UmyQZ-BzXtePMO2NgRI"
|
|
)
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "Wrong or unsupported encoding algorithm"
|
|
assert (
|
|
response.json()["details"]
|
|
== "JWT is unsecured but expected 'alg' was not 'none'"
|
|
)
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_SERVER_TIMING_ENABLED": "true",
|
|
"PGRST_JWT_CACHE_MAX_ENTRIES": "86400",
|
|
"PGRST_JWT_SECRET": SECRET,
|
|
}
|
|
|
|
# for code coverage with cache enabled and server-timing enabled
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/authors_only")
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "permission denied for table authors_only"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_SERVER_TIMING_ENABLED": "false",
|
|
"PGRST_JWT_CACHE_MAX_ENTRIES": "86400",
|
|
"PGRST_JWT_SECRET": SECRET,
|
|
}
|
|
|
|
# for code coverage with cache enabled and server-timing disabled
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/authors_only")
|
|
assert response.status_code == 401
|
|
assert response.json()["message"] == "permission denied for table authors_only"
|
|
|
|
|
|
def test_fail_with_invalid_password(defaultenv):
|
|
"Connecting with an invalid password should fail without retries."
|
|
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, wait_for_readiness=False) as postgrest:
|
|
exitCode = wait_until_exit(postgrest)
|
|
assert exitCode == 1
|
|
|
|
|
|
def test_connect_with_dburi(dburi, defaultenv):
|
|
"Connecting with db-uri instead of LIPQ* environment variables should work."
|
|
defaultenv_without_libpq = {
|
|
key: value
|
|
for key, value in defaultenv.items()
|
|
if key not in ["PGDATABASE", "PGHOST", "PGUSER"]
|
|
}
|
|
env = {**defaultenv_without_libpq, "PGRST_DB_URI": dburi.decode()}
|
|
with run(env=env):
|
|
pass
|
|
|
|
|
|
def test_read_dburi_from_stdin_without_eol(dburi, defaultenv):
|
|
"Reading the dburi from stdin with a single line should work."
|
|
defaultenv_without_libpq = {
|
|
key: value
|
|
for key, value in defaultenv.items()
|
|
if key not in ["PGDATABASE", "PGHOST", "PGUSER"]
|
|
}
|
|
env = {**defaultenv_without_libpq, "PGRST_DB_URI": "@/dev/stdin"}
|
|
|
|
with run(env=env, stdin=dburi):
|
|
pass
|
|
|
|
|
|
def test_read_dburi_from_stdin_with_eol(dburi, defaultenv):
|
|
"Reading the dburi from stdin containing a newline should work."
|
|
defaultenv_without_libpq = {
|
|
key: value
|
|
for key, value in defaultenv.items()
|
|
if key not in ["PGDATABASE", "PGHOST", "PGUSER"]
|
|
}
|
|
env = {**defaultenv_without_libpq, "PGRST_DB_URI": "@/dev/stdin"}
|
|
|
|
with run(env=env, stdin=dburi + b"\n"):
|
|
pass
|
|
|
|
|
|
@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_iat_claim(defaultenv):
|
|
"""
|
|
A claim with an 'iat' (issued at) attribute should be successful.
|
|
|
|
The PostgREST time cache leads to issues here, see:
|
|
https://github.com/PostgREST/postgrest/issues/1139
|
|
|
|
"""
|
|
|
|
env = {**defaultenv, "PGRST_JWT_SECRET": SECRET}
|
|
|
|
claim = {"role": "postgrest_test_author", "iat": datetime.now(timezone.utc)}
|
|
headers = jwtauthheader(claim, SECRET)
|
|
|
|
with run(env=env) as postgrest:
|
|
for _ in range(10):
|
|
response = postgrest.session.get("/authors_only", headers=headers)
|
|
assert response.status_code == 200
|
|
|
|
time.sleep(0.1)
|
|
|
|
|
|
def test_app_settings_flush_pool(defaultenv):
|
|
"""
|
|
App settings should not reset when the db pool is flushed.
|
|
|
|
See: https://github.com/PostgREST/postgrest/issues/1141
|
|
|
|
"""
|
|
|
|
env = {**defaultenv, "PGRST_APP_SETTINGS_EXTERNAL_API_SECRET": "0123456789abcdef"}
|
|
|
|
with run(env=env) as postgrest:
|
|
uri = "/rpc/get_guc_value?name=app.settings.external_api_secret"
|
|
response = postgrest.session.get(uri)
|
|
assert response.text == '"0123456789abcdef"'
|
|
|
|
# SIGUSR1 causes the postgres connection pool to be flushed
|
|
postgrest.process.send_signal(signal.SIGUSR1)
|
|
sleep_until_postgrest_scache_reload()
|
|
|
|
uri = "/rpc/get_guc_value?name=app.settings.external_api_secret"
|
|
response = postgrest.session.get(uri)
|
|
assert response.text == '"0123456789abcdef"'
|
|
|
|
|
|
def test_flush_pool_no_interrupt(defaultenv):
|
|
"Flushing the pool via SIGUSR1 doesn't interrupt ongoing requests"
|
|
|
|
with run(env=defaultenv) as postgrest:
|
|
|
|
def sleep():
|
|
response = postgrest.session.get("/rpc/sleep?seconds=0.5")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
t = Thread(target=sleep)
|
|
t.start()
|
|
|
|
# make sure the request has started
|
|
time.sleep(0.1)
|
|
|
|
# SIGUSR1 causes the postgres connection pool to be flushed
|
|
postgrest.process.send_signal(signal.SIGUSR1)
|
|
|
|
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") as postgrest:
|
|
assert True # liveness check is done by run(), so we just need to check that it doesn't fail
|
|
|
|
|
|
def test_app_settings_reload(tmp_path, defaultenv):
|
|
"App settings should be reloaded from file when PostgREST is sent SIGUSR2."
|
|
config = (CONFIGSDIR / "sigusr2-settings.config").read_text()
|
|
configfile = tmp_path / "test.config"
|
|
configfile.write_text(config)
|
|
uri = "/rpc/get_guc_value?name=app.settings.name_var"
|
|
|
|
with run(configfile, env=defaultenv) as postgrest:
|
|
response = postgrest.session.get(uri)
|
|
assert response.text == '"John"'
|
|
|
|
# change setting
|
|
configfile.write_text(config.replace("John", "Jane"))
|
|
# reload
|
|
postgrest.process.send_signal(signal.SIGUSR2)
|
|
|
|
sleep_until_postgrest_config_reload()
|
|
|
|
response = postgrest.session.get(uri)
|
|
assert response.text == '"Jane"'
|
|
|
|
|
|
def test_jwt_secret_reload(tmp_path, defaultenv):
|
|
"JWT secret should be reloaded from file when PostgREST is sent SIGUSR2."
|
|
config = (CONFIGSDIR / "sigusr2-settings.config").read_text()
|
|
configfile = tmp_path / "test.config"
|
|
configfile.write_text(config)
|
|
|
|
headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET)
|
|
|
|
with run(configfile, env=defaultenv) as postgrest:
|
|
response = postgrest.session.get("/authors_only", headers=headers)
|
|
assert response.status_code == 401
|
|
|
|
# change setting
|
|
configfile.write_text(config.replace("invalid" * 5, SECRET))
|
|
|
|
# reload config
|
|
postgrest.process.send_signal(signal.SIGUSR2)
|
|
|
|
sleep_until_postgrest_config_reload()
|
|
|
|
response = postgrest.session.get("/authors_only", headers=headers)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_jwt_secret_external_file_reload(tmp_path, defaultenv):
|
|
"JWT secret external file should be reloaded when PostgREST is sent a SIGUSR2 or a NOTIFY."
|
|
headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET)
|
|
|
|
external_secret_file = tmp_path / "jwt-secret-config"
|
|
external_secret_file.write_text("invalid" * 5)
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_JWT_SECRET": f"@{external_secret_file}",
|
|
"PGRST_DB_CHANNEL_ENABLED": "true",
|
|
"PGRST_DB_CONFIG": "false",
|
|
"PGRST_DB_ANON_ROLE": "postgrest_test_anonymous", # required for NOTIFY
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/authors_only", headers=headers)
|
|
assert response.status_code == 401
|
|
|
|
# change external file
|
|
external_secret_file.write_text(SECRET)
|
|
|
|
# SIGUSR1 doesn't reload external files, at least when db-config=false
|
|
postgrest.process.send_signal(signal.SIGUSR1)
|
|
sleep_until_postgrest_scache_reload()
|
|
|
|
response = postgrest.session.get("/authors_only", headers=headers)
|
|
assert response.status_code == 401
|
|
|
|
# reload config and external file with SIGUSR2
|
|
postgrest.process.send_signal(signal.SIGUSR2)
|
|
sleep_until_postgrest_config_reload()
|
|
|
|
response = postgrest.session.get("/authors_only", headers=headers)
|
|
assert response.status_code == 200
|
|
|
|
# change external file to wrong value again
|
|
external_secret_file.write_text("invalid" * 5)
|
|
|
|
# reload config and external file with NOTIFY
|
|
response = postgrest.session.post("/rpc/reload_pgrst_config")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
sleep_until_postgrest_config_reload()
|
|
|
|
response = postgrest.session.get("/authors_only", headers=headers)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_db_schema_reload(tmp_path, defaultenv):
|
|
"DB schema should be reloaded from file when PostgREST is sent SIGUSR2."
|
|
config = (CONFIGSDIR / "sigusr2-settings.config").read_text()
|
|
configfile = tmp_path / "test.config"
|
|
configfile.write_text(config)
|
|
|
|
with run(configfile, env=defaultenv) as postgrest:
|
|
response = postgrest.session.get("/rpc/get_guc_value?name=search_path")
|
|
assert response.text == '"\\"public\\", \\"public\\""'
|
|
|
|
# change setting
|
|
configfile.write_text(
|
|
config.replace('db-schemas = "public"', 'db-schemas = "v1"')
|
|
)
|
|
|
|
# reload config
|
|
postgrest.process.send_signal(signal.SIGUSR2)
|
|
sleep_until_postgrest_config_reload()
|
|
|
|
# reload schema cache to verify that the config reload actually happened
|
|
postgrest.process.send_signal(signal.SIGUSR1)
|
|
sleep_until_postgrest_scache_reload()
|
|
|
|
response = postgrest.session.get("/rpc/get_guc_value?name=search_path")
|
|
assert response.text == '"\\"v1\\", \\"public\\""'
|
|
|
|
|
|
def test_db_schema_notify_reload(defaultenv):
|
|
"DB schema and config should be reloaded when PostgREST is sent a NOTIFY"
|
|
|
|
env = {**defaultenv, "PGRST_DB_CONFIG": "true", "PGRST_DB_CHANNEL_ENABLED": "true"}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/rpc/get_guc_value?name=search_path")
|
|
assert response.text == '"\\"public\\", \\"public\\""'
|
|
|
|
# change db-schemas config on the db and reload config and cache with notify
|
|
postgrest.session.post(
|
|
"/rpc/change_db_schema_and_full_reload", data={"schemas": "v1"}
|
|
)
|
|
|
|
sleep_until_postgrest_full_reload()
|
|
|
|
response = postgrest.session.get("/rpc/get_guc_value?name=search_path")
|
|
assert response.text == '"\\"v1\\", \\"public\\""'
|
|
|
|
# reset db-schemas config on the db
|
|
response = postgrest.session.post("/rpc/reset_db_schema_config")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_max_rows_reload(defaultenv):
|
|
"max-rows should be reloaded from role settings when PostgREST receives a SIGUSR2."
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_CONFIG": "true",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.head("/projects")
|
|
assert response.status_code == 200
|
|
assert response.headers["Content-Range"] == "0-4/*"
|
|
|
|
# change max-rows config on the db
|
|
postgrest.session.post("/rpc/change_max_rows_config", data={"val": 1})
|
|
|
|
# reload config
|
|
postgrest.process.send_signal(signal.SIGUSR2)
|
|
|
|
sleep_until_postgrest_config_reload()
|
|
|
|
response = postgrest.session.head("/projects")
|
|
assert response.status_code == 200
|
|
assert response.headers["Content-Range"] == "0-0/*"
|
|
|
|
# reset max-rows config on the db
|
|
response = postgrest.session.post("/rpc/reset_max_rows_config")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_max_rows_notify_reload(defaultenv):
|
|
"max-rows should be reloaded from role settings when PostgREST receives a NOTIFY"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_CONFIG": "true",
|
|
"PGRST_DB_CHANNEL_ENABLED": "true",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.head("/projects")
|
|
assert response.status_code == 200
|
|
assert response.headers["Content-Range"] == "0-4/*"
|
|
|
|
# change max-rows config on the db and reload with notify
|
|
postgrest.session.post(
|
|
"/rpc/change_max_rows_config", data={"val": 1, "notify": True}
|
|
)
|
|
|
|
sleep_until_postgrest_config_reload()
|
|
|
|
response = postgrest.session.head("/projects")
|
|
assert response.status_code == 200
|
|
assert response.headers["Content-Range"] == "0-0/*"
|
|
|
|
# reset max-rows config on the db
|
|
response = postgrest.session.post("/rpc/reset_max_rows_config")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_invalid_role_claim_key_notify_reload(defaultenv):
|
|
"NOTIFY reload config should show an error if role-claim-key is invalid"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_CONFIG": "true",
|
|
"PGRST_DB_CHANNEL_ENABLED": "true",
|
|
"PGRST_LOG_LEVEL": "crit",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
postgrest.session.post("/rpc/invalid_role_claim_key_reload")
|
|
|
|
output = postgrest.read_stdout()
|
|
assert 'Received a config reload message on the "pgrst" channel' in output[0]
|
|
output = postgrest.read_stdout()
|
|
assert "failed to parse role-claim-key value" in output[0]
|
|
|
|
response = postgrest.session.post("/rpc/reset_invalid_role_claim_key")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_notify_do_nothing(defaultenv):
|
|
"NOTIFY with unknown message should do nothing"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_CONFIG": "true",
|
|
"PGRST_DB_CHANNEL_ENABLED": "true",
|
|
"PGRST_LOG_LEVEL": "crit",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.post("/rpc/notify_do_nothing")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
output = postgrest.read_stdout()
|
|
assert output == []
|
|
|
|
|
|
def test_db_prepared_statements_enable(defaultenv):
|
|
"Should use prepared statements when the setting is enabled."
|
|
|
|
with run(env=defaultenv) as postgrest:
|
|
response = postgrest.session.post("/rpc/uses_prepared_statements")
|
|
assert response.text == "true"
|
|
|
|
|
|
def test_db_prepared_statements_disable(defaultenv):
|
|
"Should not use any prepared statements when the setting is disabled."
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_PREPARED_STATEMENTS": "false",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.post("/rpc/uses_prepared_statements")
|
|
assert response.text == "false"
|
|
|
|
|
|
def test_statement_timeout(defaultenv, metapostgrest):
|
|
"Statement timeout times out slow statements"
|
|
|
|
role = "timeout_authenticator"
|
|
set_statement_timeout(metapostgrest, role, 1000) # 1 second
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGUSER": role,
|
|
"PGRST_DB_ANON_ROLE": role,
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/rpc/sleep?seconds=0.5")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
response = postgrest.session.get("/rpc/sleep?seconds=2")
|
|
assert response.status_code == 500
|
|
data = response.json()
|
|
assert data["message"] == "canceling statement due to statement timeout"
|
|
|
|
reset_statement_timeout(metapostgrest, role)
|
|
|
|
|
|
def test_change_statement_timeout(defaultenv, metapostgrest):
|
|
"Statement timeout changes take effect immediately"
|
|
|
|
role = "timeout_authenticator"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGUSER": role,
|
|
"PGRST_DB_ANON_ROLE": role,
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
# no limit initially
|
|
response = postgrest.session.get("/rpc/sleep?seconds=1")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
set_statement_timeout(metapostgrest, role, 500) # 0.5s
|
|
|
|
# trigger schema refresh
|
|
postgrest.process.send_signal(signal.SIGUSR1)
|
|
sleep_until_postgrest_scache_reload()
|
|
|
|
response = postgrest.session.get("/rpc/sleep?seconds=1")
|
|
assert response.status_code == 500
|
|
data = response.json()
|
|
assert data["message"] == "canceling statement due to statement timeout"
|
|
|
|
set_statement_timeout(metapostgrest, role, 2000) # 2s
|
|
|
|
# trigger role setting refresh
|
|
postgrest.process.send_signal(signal.SIGUSR1)
|
|
sleep_until_postgrest_scache_reload()
|
|
|
|
response = postgrest.session.get("/rpc/sleep?seconds=1")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
reset_statement_timeout(metapostgrest, role)
|
|
|
|
|
|
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("level", ["crit", "error", "warn", "info", "debug"])
|
|
def test_pool_acquisition_timeout(level, defaultenv, metapostgrest):
|
|
"Verify that PGRST_DB_POOL_ACQUISITION_TIMEOUT times out when the pool is empty"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_POOL": "1",
|
|
"PGRST_DB_POOL_ACQUISITION_TIMEOUT": "1", # 1 second
|
|
"PGRST_LOG_LEVEL": level,
|
|
}
|
|
|
|
with run(env=env, no_pool_connection_available=True) as postgrest:
|
|
response = postgrest.session.get("/projects")
|
|
assert response.status_code == 504
|
|
data = response.json()
|
|
assert data["message"] == "Timed out acquiring connection from connection pool."
|
|
|
|
# ensure the message appears on the logs as well
|
|
output = sorted(postgrest.read_stdout(nlines=10))
|
|
|
|
if level == "crit":
|
|
assert len(output) == 0
|
|
else:
|
|
assert any(" 504 " in line for line in output)
|
|
assert any(
|
|
"Timed out acquiring connection from connection pool." in line
|
|
for line in output
|
|
)
|
|
|
|
|
|
def test_change_statement_timeout_held_connection(defaultenv, metapostgrest):
|
|
"Statement timeout changes take effect immediately, even with a request outliving the reconfiguration"
|
|
|
|
role = "timeout_authenticator"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGUSER": role,
|
|
"PGRST_DB_ANON_ROLE": role,
|
|
"PGRST_DB_POOL": "2",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
# start a slow request that holds a pool connection
|
|
def hold_connection():
|
|
response = postgrest.session.get("/rpc/sleep?seconds=1")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
hold = Thread(target=hold_connection)
|
|
hold.start()
|
|
# give the request time to start before SIGUSR1 flushes the pool
|
|
time.sleep(0.1)
|
|
|
|
set_statement_timeout(metapostgrest, role, 500) # 0.5s
|
|
# trigger schema refresh; flushes pool and establishes a new connection
|
|
postgrest.process.send_signal(signal.SIGUSR1)
|
|
|
|
# wait for the slow request's connection to be returned to the pool
|
|
hold.join()
|
|
|
|
# subsequent requests should fail due to the lowered timeout; run several in parallel
|
|
# to ensure we use the full pool
|
|
threads = []
|
|
for i in range(2):
|
|
|
|
def sleep(i=i):
|
|
response = postgrest.session.get("/rpc/sleep?seconds=1")
|
|
assert response.status_code == 500, "thread {}".format(i)
|
|
data = response.json()
|
|
assert data["message"] == "canceling statement due to statement timeout"
|
|
|
|
thread = Thread(target=sleep)
|
|
thread.start()
|
|
threads.append(thread)
|
|
|
|
for t in threads:
|
|
t.join()
|
|
|
|
reset_statement_timeout(metapostgrest, role)
|
|
|
|
|
|
def test_admin_schema_cache(defaultenv):
|
|
"Should get a success response from the admin server containing current schema cache"
|
|
|
|
with run(env=defaultenv) as postgrest:
|
|
response = postgrest.admin.get("/schema_cache")
|
|
assert response.status_code == 200
|
|
assert '"dbTables":[[{"qiName":"authors_only"' in response.text
|
|
|
|
|
|
def test_admin_ready_w_channel(defaultenv):
|
|
"Should get a success response from the admin server ready endpoint when the LISTEN channel is enabled"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_CHANNEL_ENABLED": "true",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.admin.get("/ready")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_admin_ready_wo_channel(defaultenv):
|
|
"Should get a success response from the admin server ready endpoint when the LISTEN channel is disabled"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_CHANNEL_ENABLED": "false",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.admin.get("/ready")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_admin_ready_includes_schema_cache_state(defaultenv, metapostgrest):
|
|
"Should get a failed response from the admin server ready endpoint when the schema cache is not loaded"
|
|
|
|
role = "timeout_authenticator"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGUSER": role,
|
|
"PGRST_DB_ANON_ROLE": role,
|
|
"PGRST_INTERNAL_SCHEMA_CACHE_SLEEP": "500",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
# The schema cache query takes at least 500ms, due to PGRST_INTERNAL_SCHEMA_CACHE_SLEEP above.
|
|
# Make it impossible to load the schema cache, by setting statement timeout to 400ms.
|
|
set_statement_timeout(metapostgrest, role, 400)
|
|
|
|
# force a reconnection so the new role setting is picked up
|
|
postgrest.process.send_signal(signal.SIGUSR1)
|
|
|
|
postgrest.wait_until_scache_starts_loading()
|
|
|
|
response = postgrest.admin.get("/ready", timeout=1)
|
|
assert response.status_code == 503
|
|
|
|
response = postgrest.session.get("/projects", timeout=1)
|
|
assert response.status_code == 503
|
|
|
|
reset_statement_timeout(metapostgrest, role)
|
|
|
|
|
|
def test_metrics_include_schema_cache_fails(defaultenv, metapostgrest):
|
|
"Should get shema cache fails from the metrics endpoint"
|
|
|
|
role = "timeout_authenticator"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGUSER": role,
|
|
"PGRST_INTERNAL_SCHEMA_CACHE_SLEEP": "50",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
# The schema cache query takes at least 20ms, due to PGRST_INTERNAL_SCHEMA_CACHE_SLEEP above.
|
|
# Make it impossible to load the schema cache, by setting statement timeout to 100ms.
|
|
set_statement_timeout(metapostgrest, role, 20)
|
|
|
|
# force a reconnection so the new role setting is picked up
|
|
postgrest.process.send_signal(signal.SIGUSR1)
|
|
|
|
# wait for some schema cache retries
|
|
time.sleep(1)
|
|
|
|
response = postgrest.admin.get("/ready", timeout=1)
|
|
assert response.status_code == 503
|
|
|
|
response = postgrest.admin.get("/metrics", timeout=1)
|
|
assert response.status_code == 200
|
|
|
|
metrics = float(
|
|
re.search(
|
|
r'pgrst_schema_cache_loads_total{status="FAIL"} (\d+)', response.text
|
|
).group(1)
|
|
)
|
|
assert metrics == 1.0
|
|
|
|
reset_statement_timeout(metapostgrest, role)
|
|
|
|
|
|
def test_admin_not_found(defaultenv):
|
|
"Should get a not found from a undefined endpoint on the admin server"
|
|
|
|
with run(env=defaultenv) as postgrest:
|
|
response = postgrest.admin.get("/notfound")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_admin_ready_dependent_on_main_app(defaultenv):
|
|
"Should get a failure from the admin ready endpoint if the main app also fails"
|
|
|
|
with run(env=defaultenv) as postgrest:
|
|
# delete the unix socket to make the main app fail
|
|
os.remove(defaultenv["PGRST_SERVER_UNIX_SOCKET"])
|
|
response = postgrest.admin.get("/ready")
|
|
assert response.status_code == 500
|
|
|
|
|
|
def test_admin_live_good(defaultenv):
|
|
"Should get a success from the admin live endpoint if the main app is running"
|
|
|
|
with run(env=defaultenv, port=freeport()) as postgrest:
|
|
response = postgrest.admin.get("/live")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_admin_live_dependent_on_main_app(defaultenv):
|
|
"Should get a failure from the admin live endpoint if the main app also fails"
|
|
|
|
with run(env=defaultenv) as postgrest:
|
|
# delete the unix socket to make the main app fail
|
|
os.remove(defaultenv["PGRST_SERVER_UNIX_SOCKET"])
|
|
response = postgrest.admin.get("/live")
|
|
assert response.status_code == 500
|
|
|
|
|
|
@pytest.mark.parametrize("specialhostvalue", FIXTURES["specialhostvalues"])
|
|
def test_admin_works_with_host_special_values(specialhostvalue, defaultenv):
|
|
"Should get a success from the admin live and ready endpoints when using special host values for the main app"
|
|
|
|
with run(env=defaultenv, port=freeport(), host=specialhostvalue) as postgrest:
|
|
response = postgrest.admin.get("/live")
|
|
assert response.status_code == 200
|
|
|
|
response = postgrest.admin.get("/ready")
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize("level", ["crit", "error", "warn", "info", "debug"])
|
|
def test_log_level(level, defaultenv):
|
|
"log_level should filter request logging"
|
|
|
|
env = {**defaultenv, "PGRST_LOG_LEVEL": level}
|
|
|
|
# any token to test 500 response for "Server lacks JWT secret"
|
|
claim = {"role": "postgrest_test_author"}
|
|
headers = jwtauthheader(claim, SECRET)
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/", headers=headers)
|
|
assert response.status_code == 500
|
|
|
|
response = postgrest.session.get("/unknown")
|
|
assert response.status_code == 404
|
|
|
|
response = postgrest.session.get("/")
|
|
assert response.status_code == 200
|
|
|
|
output = sorted(postgrest.read_stdout(nlines=7))
|
|
|
|
if level == "crit":
|
|
assert len(output) == 0
|
|
elif level == "error":
|
|
assert re.match(
|
|
r'- - - \[.+\] "GET / HTTP/1.1" 500 \d+ "" "python-requests/.+"',
|
|
output[0],
|
|
)
|
|
assert len(output) == 1
|
|
elif level == "warn":
|
|
assert re.match(
|
|
r'- - - \[.+\] "GET / HTTP/1.1" 500 \d+ "" "python-requests/.+"',
|
|
output[0],
|
|
)
|
|
assert re.match(
|
|
r'- - postgrest_test_anonymous \[.+\] "GET /unknown HTTP/1.1" 404 \d+ "" "python-requests/.+"',
|
|
output[1],
|
|
)
|
|
assert len(output) == 2
|
|
elif level == "info":
|
|
assert re.match(
|
|
r'- - - \[.+\] "GET / HTTP/1.1" 500 \d+ "" "python-requests/.+"',
|
|
output[0],
|
|
)
|
|
assert re.match(
|
|
r'- - postgrest_test_anonymous \[.+\] "GET / HTTP/1.1" 200 \d+ "" "python-requests/.+"',
|
|
output[1],
|
|
)
|
|
assert re.match(
|
|
r'- - postgrest_test_anonymous \[.+\] "GET /unknown HTTP/1.1" 404 \d+ "" "python-requests/.+"',
|
|
output[2],
|
|
)
|
|
assert len(output) == 3
|
|
elif level == "debug":
|
|
assert re.match(
|
|
r'- - - \[.+\] "GET / HTTP/1.1" 500 \d+ "" "python-requests/.+"',
|
|
output[0],
|
|
)
|
|
assert re.match(
|
|
r'- - postgrest_test_anonymous \[.+\] "GET / HTTP/1.1" 200 \d+ "" "python-requests/.+"',
|
|
output[1],
|
|
)
|
|
assert re.match(
|
|
r'- - postgrest_test_anonymous \[.+\] "GET /unknown HTTP/1.1" 404 \d+ "" "python-requests/.+"',
|
|
output[2],
|
|
)
|
|
|
|
assert len(output) == 7
|
|
assert any("Connection" and "is available" in line for line in output)
|
|
assert any("Connection" and "is used" in line for line in output)
|
|
|
|
|
|
@pytest.mark.parametrize("level", ["crit", "error", "warn", "info", "debug"])
|
|
def test_log_query(level, defaultenv):
|
|
"log_query=true should log the SQL query according to the log_level"
|
|
|
|
def drain_stdout(proc):
|
|
lines = []
|
|
while True:
|
|
chunk = proc.read_stdout(nlines=20)
|
|
if not chunk:
|
|
break
|
|
lines.extend(chunk)
|
|
return lines
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_LOG_LEVEL": level,
|
|
"PGRST_LOG_QUERY": "true",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/")
|
|
assert response.status_code == 200
|
|
|
|
response = postgrest.session.get("/projects")
|
|
assert response.status_code == 200
|
|
|
|
response = postgrest.session.get(
|
|
"/projects", headers={"Prefer": "count=estimated"}
|
|
)
|
|
assert response.status_code == 206
|
|
|
|
response = postgrest.session.get(
|
|
"/projects", headers={"Prefer": "count=planned"}
|
|
)
|
|
assert response.status_code == 206
|
|
|
|
response = postgrest.session.get("/infinite_recursion")
|
|
assert response.status_code == 500
|
|
|
|
get_2xx_regx = r'.+: WITH pgrst_source AS.+SELECT "public"\."projects"\.\* FROM "public"\."projects".+_postgrest_t'
|
|
get_2xx_count_regx = (
|
|
r'.+: EXPLAIN \(FORMAT JSON\) SELECT 1 FROM "public"."projects"'
|
|
)
|
|
infinite_recursion_5xx_regx = r'.+: WITH pgrst_source AS.+SELECT "public"\."infinite_recursion"\.\* FROM "public"\."infinite_recursion".+_postgrest_t'
|
|
root_tables_regx = r".+: SELECT n.nspname AS table_schema, .+ FROM pg_class c .+ ORDER BY table_schema, table_name"
|
|
root_procs_regx = r".+: WITH base_types AS \(.+\) SELECT pn.nspname AS proc_schema, .+ FROM pg_proc p.+AND p.pronamespace = \$1::regnamespace"
|
|
root_descr_regx = r".+: SELECT pg_catalog\.obj_description\(\$1::regnamespace, 'pg_namespace'\)"
|
|
set_config_regx = (
|
|
r".+: select set_config\('search_path', \$1, true\), set_config\("
|
|
)
|
|
|
|
output = drain_stdout(postgrest)
|
|
|
|
project_queries = [line for line in output if re.match(get_2xx_regx, line)]
|
|
project_counts = [line for line in output if re.match(get_2xx_count_regx, line)]
|
|
infinite_queries = [
|
|
line for line in output if re.match(infinite_recursion_5xx_regx, line)
|
|
]
|
|
root_tables = [line for line in output if re.match(root_tables_regx, line)]
|
|
root_procs = [line for line in output if re.match(root_procs_regx, line)]
|
|
root_descr = [line for line in output if re.match(root_descr_regx, line)]
|
|
set_configs = [line for line in output if re.match(set_config_regx, line)]
|
|
|
|
if level == "crit":
|
|
assert not set_configs
|
|
assert not project_queries
|
|
assert not project_counts
|
|
assert not infinite_queries
|
|
assert not root_tables
|
|
assert not root_procs
|
|
assert not root_descr
|
|
elif level in {"error", "warn"}:
|
|
assert len(set_configs) == 1
|
|
assert len(infinite_queries) == 1
|
|
assert not project_queries
|
|
assert not project_counts
|
|
assert not root_tables
|
|
assert not root_procs
|
|
assert not root_descr
|
|
elif level == "info":
|
|
assert len(set_configs) == 5
|
|
assert len(project_queries) == 3
|
|
assert len(project_counts) == 2
|
|
assert len(infinite_queries) == 1
|
|
assert len(root_tables) == 1
|
|
assert len(root_procs) == 1
|
|
assert len(root_descr) == 1
|
|
elif level == "debug":
|
|
assert len(set_configs) == 5
|
|
assert len(project_queries) == 3
|
|
assert len(project_counts) == 2
|
|
assert len(infinite_queries) == 1
|
|
assert len(root_tables) == 1
|
|
assert len(root_procs) == 1
|
|
assert len(root_descr) == 1
|
|
|
|
pre_req_env = {
|
|
**env,
|
|
"PGRST_DB_PRE_REQUEST": "do_nothing",
|
|
}
|
|
|
|
with run(env=pre_req_env) as postgrest:
|
|
response = postgrest.session.get("/projects")
|
|
assert response.status_code == 200
|
|
|
|
output = drain_stdout(postgrest)
|
|
|
|
pre_request_regx = r'.+: select "do_nothing"()'
|
|
pre_reqs = [line for line in output if re.match(pre_request_regx, line)]
|
|
|
|
if level == "crit":
|
|
assert not pre_reqs
|
|
elif level in {"error", "warn"}:
|
|
assert not pre_reqs
|
|
elif level == "info":
|
|
assert len(pre_reqs) == 1
|
|
elif level == "debug":
|
|
assert len(pre_reqs) == 1
|
|
|
|
|
|
def test_no_pool_connection_required_on_bad_http_logic(defaultenv):
|
|
"no pool connection should be consumed for failing on invalid http logic"
|
|
|
|
with run(env=defaultenv, no_pool_connection_available=True) as postgrest:
|
|
# not found nested route shouldn't require opening a connection
|
|
response = postgrest.session.head("/path/notfound")
|
|
assert response.status_code == 404
|
|
|
|
# an invalid http method on a resource shouldn't require opening a connection
|
|
response = postgrest.session.request("TRACE", "/projects")
|
|
assert response.status_code == 405
|
|
response = postgrest.session.patch("/rpc/hello")
|
|
assert response.status_code == 405
|
|
|
|
|
|
def test_no_pool_connection_required_on_options(defaultenv):
|
|
"no pool connection should be consumed for OPTIONS requests"
|
|
|
|
with run(env=defaultenv, no_pool_connection_available=True) as postgrest:
|
|
# OPTIONS on a table shouldn't require opening a connection
|
|
response = postgrest.session.options("/projects")
|
|
assert response.status_code == 200
|
|
|
|
# OPTIONS on RPC shouldn't require opening a connection
|
|
response = postgrest.session.options("/rpc/hello")
|
|
assert response.status_code == 200
|
|
|
|
# OPTIONS on root shouldn't require opening a connection
|
|
response = postgrest.session.options("/")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_no_pool_connection_required_on_bad_jwt_claim(defaultenv):
|
|
"no pool connection should be consumed for failing on invalid jwt"
|
|
|
|
env = {**defaultenv, "PGRST_JWT_SECRET": SECRET}
|
|
|
|
with run(env=env, no_pool_connection_available=True) as postgrest:
|
|
# A JWT with an invalid signature shouldn't open a connection
|
|
headers = jwtauthheader({"role": "postgrest_test_author"}, "Wrong Secret")
|
|
response = postgrest.session.get("/projects", headers=headers)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_no_pool_connection_required_on_bad_embedding(defaultenv):
|
|
"no pool connection should be consumed for failing to embed"
|
|
|
|
with run(env=defaultenv, no_pool_connection_available=True) as postgrest:
|
|
# OPTIONS on a table shouldn't require opening a connection
|
|
response = postgrest.session.get("/projects?select=*,unexistent(*)")
|
|
assert response.status_code == 400
|
|
|
|
|
|
# https://github.com/PostgREST/postgrest/issues/2620
|
|
def test_notify_reloading_catalog_cache(defaultenv):
|
|
"notify should reload the connection catalog cache"
|
|
|
|
with run(env=defaultenv) as postgrest:
|
|
# first the id col is an uuid
|
|
response = postgrest.session.get(
|
|
"/cats?id=eq.dea27321-f988-4a57-93e4-8eeb38f3cf1e"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# change it to a bigint
|
|
response = postgrest.session.post("/rpc/drop_change_cats")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
sleep_until_postgrest_scache_reload()
|
|
|
|
# next request should succeed with a bigint value
|
|
response = postgrest.session.get("/cats?id=eq.1")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_role_settings(defaultenv):
|
|
"statement_timeout should be set per role"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_JWT_SECRET": SECRET,
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
# statement_timeout for postgrest_test_anonymous
|
|
response = postgrest.session.get("/rpc/get_guc_value?name=statement_timeout")
|
|
assert response.text == '"2s"'
|
|
|
|
# reload statement_timeout with NOTIFY
|
|
response = postgrest.session.post(
|
|
"/rpc/change_role_statement_timeout", data={"timeout": "5s"}
|
|
)
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
response = postgrest.session.get("/rpc/reload_pgrst_config")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
sleep_until_postgrest_config_reload()
|
|
|
|
response = postgrest.session.get("/rpc/get_guc_value?name=statement_timeout")
|
|
assert response.text == '"5s"'
|
|
|
|
# statement_timeout for postgrest_test_author
|
|
headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET)
|
|
response = postgrest.session.get(
|
|
"/rpc/get_guc_value?name=statement_timeout", headers=headers
|
|
)
|
|
assert response.text == '"10s"'
|
|
|
|
# reset statement timeout to original value
|
|
response = postgrest.session.post(
|
|
"/rpc/change_role_statement_timeout", data={"timeout": "2s"}
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_isolation_level(defaultenv):
|
|
"isolation_level should be set per role and per function"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_JWT_SECRET": SECRET,
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
# default isolation level for postgrest_test_anonymous
|
|
response = postgrest.session.get(
|
|
"/items_w_isolation_level?select=isolation_level&limit=1"
|
|
)
|
|
assert response.text == '[{"isolation_level":"read committed"}]'
|
|
|
|
# isolation level for postgrest_test_repeatable_read on GET
|
|
headers = jwtauthheader({"role": "postgrest_test_repeatable_read"}, SECRET)
|
|
response = postgrest.session.get(
|
|
"/items_w_isolation_level?select=isolation_level&limit=1", headers=headers
|
|
)
|
|
assert response.text == '[{"isolation_level":"repeatable read"}]'
|
|
|
|
# isolation level for postgrest_test_serializable on POST
|
|
headers = jwtauthheader({"role": "postgrest_test_serializable"}, SECRET)
|
|
headers["Prefer"] = "return=representation"
|
|
response = postgrest.session.post(
|
|
"/items_w_isolation_level?select=isolation_level",
|
|
json={"id": "666"},
|
|
headers=headers,
|
|
)
|
|
assert response.text == '[{"isolation_level":"serializable"}]'
|
|
|
|
# isolation level for postgrest_test_serializable on PATCH
|
|
headers = jwtauthheader({"role": "postgrest_test_serializable"}, SECRET)
|
|
headers["Prefer"] = "return=representation"
|
|
response = postgrest.session.patch(
|
|
"/items_w_isolation_level?select=isolation_level&id=eq.666",
|
|
json={"id": "666"},
|
|
headers=headers,
|
|
)
|
|
assert response.text == '[{"isolation_level":"serializable"}]'
|
|
|
|
# isolation level for postgrest_test_serializable on DELETE
|
|
headers = jwtauthheader({"role": "postgrest_test_serializable"}, SECRET)
|
|
headers["Prefer"] = "return=representation"
|
|
response = postgrest.session.delete(
|
|
"/items_w_isolation_level?select=isolation_level&id=eq.666", headers=headers
|
|
)
|
|
assert response.text == '[{"isolation_level":"serializable"}]'
|
|
|
|
# default isolation level for function
|
|
response = postgrest.session.get("/rpc/default_isolation_level")
|
|
assert response.text == '"read committed"'
|
|
|
|
# changes with role isolation level
|
|
headers = jwtauthheader({"role": "postgrest_test_repeatable_read"}, SECRET)
|
|
response = postgrest.session.get(
|
|
"/rpc/default_isolation_level", headers=headers
|
|
)
|
|
assert response.text == '"repeatable read"'
|
|
|
|
# isolation level can be set per function
|
|
response = postgrest.session.get("/rpc/serializable_isolation_level")
|
|
assert response.text == '"serializable"'
|
|
response = postgrest.session.get("/rpc/repeatable_read_isolation_level")
|
|
assert response.text == '"repeatable read"'
|
|
|
|
# isolation level for a function overrides the role isolation level
|
|
headers = jwtauthheader({"role": "postgrest_test_repeatable_read"}, SECRET)
|
|
response = postgrest.session.get("/rpc/serializable_isolation_level")
|
|
assert response.text == '"serializable"'
|
|
|
|
|
|
def test_schema_cache_concurrent_notifications(slow_schema_cache_env):
|
|
"schema cache should be up-to-date whenever a notification is sent while another reload is in progress, see https://github.com/PostgREST/postgrest/issues/2791"
|
|
|
|
internal_sleep = (
|
|
int(slow_schema_cache_env["PGRST_INTERNAL_SCHEMA_CACHE_SLEEP"]) / 1000
|
|
)
|
|
|
|
with run(env=slow_schema_cache_env, wait_for_readiness=False) as postgrest:
|
|
time.sleep(2 * internal_sleep + 0.1) # wait for readiness manually
|
|
|
|
# first request, create a function and set a schema cache reload in progress
|
|
response = postgrest.session.post("/rpc/create_function")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
time.sleep(
|
|
internal_sleep / 2
|
|
) # wait to be inside the schema cache reload process
|
|
|
|
# second request, change the same function and do another schema cache reload
|
|
response = postgrest.session.post("/rpc/migrate_function")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
time.sleep(
|
|
2 * internal_sleep
|
|
) # wait enough time to get the final schema cache state
|
|
|
|
# confirm the schema cache is up-to-date and the 2nd reload wasn't lost
|
|
response = postgrest.session.get("/rpc/mult_them?c=3&d=4")
|
|
assert response.text == "12"
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.parametrize("dburi_type", ["no_params", "no_params_qmark", "with_params"])
|
|
def test_get_pgrst_version_with_uri_connection_string(dburi_type, dburi, defaultenv):
|
|
"The fallback_application_name should be added to the db-uri if it has a URI format"
|
|
defaultenv_without_libpq = {
|
|
key: value
|
|
for key, value in defaultenv.items()
|
|
if key not in ["PGDATABASE", "PGHOST", "PGUSER"]
|
|
}
|
|
|
|
env = {
|
|
"no_params": {**defaultenv, "PGRST_DB_URI": "postgresql://"},
|
|
"no_params_qmark": {**defaultenv, "PGRST_DB_URI": "postgresql://?"},
|
|
"with_params": {**defaultenv_without_libpq, "PGRST_DB_URI": dburi.decode()},
|
|
}
|
|
|
|
with run(env=env[dburi_type]) as postgrest:
|
|
response = postgrest.session.post("/rpc/get_pgrst_version")
|
|
version = '"%s"' % response.headers["Server"].replace(
|
|
"postgrest/", "PostgREST "
|
|
)
|
|
assert response.text == version
|
|
|
|
|
|
def test_get_pgrst_version_with_keyval_connection_string(defaultenv):
|
|
"The fallback_application_name should be added to the db-uri if it has a keyword/value format"
|
|
uri = f'dbname={defaultenv["PGDATABASE"]} host={defaultenv["PGHOST"]} user={defaultenv["PGUSER"]}'
|
|
defaultenv_without_libpq = {
|
|
key: value
|
|
for key, value in defaultenv.items()
|
|
if key not in ["PGDATABASE", "PGHOST", "PGUSER"]
|
|
}
|
|
env = {**defaultenv_without_libpq, "PGRST_DB_URI": uri}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.post("/rpc/get_pgrst_version")
|
|
version = '"%s"' % response.headers["Server"].replace(
|
|
"postgrest/", "PostgREST "
|
|
)
|
|
assert response.text == version
|
|
|
|
|
|
def test_log_postgrest_version(defaultenv):
|
|
"Should show the PostgREST version in the logs"
|
|
|
|
env = {**defaultenv, "PGRST_LOG_LEVEL": "crit"}
|
|
|
|
with run(env=defaultenv, no_startup_stdout=False) as postgrest:
|
|
version = postgrest.session.head("/").headers["Server"].split("/")[1]
|
|
|
|
output = postgrest.read_stdout(nlines=1)
|
|
|
|
assert "Starting PostgREST %s..." % version in output[0]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"host", ["127.0.0.1", "::1", None], ids=["IPv4", "IPv6", "Unix"]
|
|
)
|
|
def test_log_postgrest_host_and_port(host, defaultenv):
|
|
"PostgREST should output the host and port it is bound to."
|
|
|
|
# We run postgrest on unix socket when host and port are set to None
|
|
is_unix = host is None
|
|
port = None if is_unix else freeport()
|
|
|
|
with run(
|
|
env=defaultenv, host=host, port=port, no_startup_stdout=False
|
|
) as postgrest:
|
|
output = postgrest.read_stdout(nlines=10)
|
|
|
|
if is_unix:
|
|
re.match(r'API server listening on "/tmp/.*\.sock"', output[2])
|
|
elif is_ipv6(host):
|
|
assert f"API server listening on [{host}]:{port}" in output[2]
|
|
else: # IPv4
|
|
assert f"API server listening on {host}:{port}" in output[2]
|
|
|
|
|
|
def test_succeed_w_role_having_superuser_settings(defaultenv):
|
|
"Should succeed when having superuser settings on the impersonated role"
|
|
|
|
env = {**defaultenv, "PGRST_DB_CONFIG": "true", "PGRST_JWT_SECRET": SECRET}
|
|
|
|
with run(stdin=SECRET.encode(), env=env) as postgrest:
|
|
headers = jwtauthheader({"role": "postgrest_test_w_superuser_settings"}, SECRET)
|
|
response = postgrest.session.get("/projects", headers=headers)
|
|
print(response.text)
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_get_granted_superuser_setting(defaultenv):
|
|
"Should succeed when the impersonated role has granted superuser settings"
|
|
|
|
env = {**defaultenv, "PGRST_DB_CONFIG": "true", "PGRST_JWT_SECRET": SECRET}
|
|
|
|
with run(stdin=SECRET.encode(), env=env) as postgrest:
|
|
response_ver = postgrest.session.get("/rpc/get_postgres_version")
|
|
pg_ver = eval(response_ver.text)
|
|
if pg_ver >= 150000:
|
|
headers = jwtauthheader(
|
|
{"role": "postgrest_test_w_superuser_settings"}, SECRET
|
|
)
|
|
response = postgrest.session.get(
|
|
"/rpc/get_guc_value?name=log_min_duration_sample", headers=headers
|
|
)
|
|
assert response.text == '"12345ms"'
|
|
|
|
|
|
def test_fail_with_invalid_dbname_and_automatic_recovery_disabled(defaultenv):
|
|
"Should fail without retries when automatic recovery is disabled and dbname is invalid"
|
|
dbname = "INVALID"
|
|
uri = f'postgresql://?dbname={dbname}&host={defaultenv["PGHOST"]}&user={defaultenv["PGUSER"]}'
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_URI": uri,
|
|
"PGRST_DB_POOL_AUTOMATIC_RECOVERY": "false",
|
|
}
|
|
|
|
with run(env=env, wait_for_readiness=False) as postgrest:
|
|
exitCode = wait_until_exit(postgrest)
|
|
assert exitCode == 1
|
|
|
|
|
|
def test_fail_with_automatic_recovery_disabled_and_terminated_using_query(defaultenv):
|
|
"Should fail without retries when automatic recovery is disabled and pg_terminate_backend(pid) is called"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_POOL_AUTOMATIC_RECOVERY": "false",
|
|
"PGAPPNAME": "target",
|
|
}
|
|
|
|
app_name = "'{}'".format(env["PGAPPNAME"])
|
|
|
|
with run(env=env) as postgrest:
|
|
os.system(
|
|
f'psql -d {env["PGDATABASE"]} -U {env["PGUSER"]} -h {env["PGHOST"]} --set ON_ERROR_STOP=1 -a -c "SELECT terminate_pgrst({app_name})"'
|
|
)
|
|
|
|
exitCode = wait_until_exit(postgrest)
|
|
assert exitCode == 1
|
|
|
|
|
|
def test_jwt_cache_server_timing(defaultenv):
|
|
"server-timing duration is exposed for JWT with expiry"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_SERVER_TIMING_ENABLED": "true",
|
|
"PGRST_JWT_CACHE_MAX_ENTRIES": "86400",
|
|
"PGRST_JWT_SECRET": SECRET,
|
|
"PGRST_DB_CONFIG": "false",
|
|
}
|
|
|
|
headers = jwtauthheader(
|
|
{
|
|
"role": "postgrest_test_author",
|
|
"exp": int(
|
|
(datetime.now(timezone.utc) + timedelta(minutes=30)).timestamp()
|
|
),
|
|
},
|
|
SECRET,
|
|
)
|
|
|
|
with run(env=env) as postgrest:
|
|
first = postgrest.session.get("/authors_only", headers=headers)
|
|
second = postgrest.session.get("/authors_only", headers=headers)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
|
|
first_dur = parse_server_timings_header(first.headers["Server-Timing"])["jwt"]
|
|
second_dur = parse_server_timings_header(second.headers["Server-Timing"])["jwt"]
|
|
|
|
# with jwt caching the parse time of second request with the same token
|
|
# should be at least as fast as the first one
|
|
assert second_dur <= first_dur
|
|
|
|
|
|
def test_jwt_cache_without_server_timing(defaultenv):
|
|
"JWT cache does not break requests with server-timing disabled"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_SERVER_TIMING_ENABLED": "false",
|
|
"PGRST_JWT_CACHE_MAX_ENTRIES": "86400",
|
|
"PGRST_JWT_SECRET": SECRET,
|
|
"PGRST_DB_CONFIG": "false",
|
|
}
|
|
|
|
headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET)
|
|
|
|
with run(env=env) as postgrest:
|
|
first = postgrest.session.get("/authors_only", headers=headers)
|
|
second = postgrest.session.get("/authors_only", headers=headers)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
|
|
|
|
def test_jwt_cache_without_exp_claim(defaultenv):
|
|
"server-timing duration is exposed for JWT without expiry"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_SERVER_TIMING_ENABLED": "true",
|
|
"PGRST_JWT_CACHE_MAX_ENTRIES": "86400",
|
|
"PGRST_JWT_SECRET": SECRET,
|
|
"PGRST_DB_CONFIG": "false",
|
|
}
|
|
|
|
headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET) # no exp
|
|
|
|
with run(env=env) as postgrest:
|
|
first = postgrest.session.get("/authors_only", headers=headers)
|
|
second = postgrest.session.get("/authors_only", headers=headers)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
|
|
first_dur = parse_server_timings_header(first.headers["Server-Timing"])["jwt"]
|
|
second_dur = parse_server_timings_header(second.headers["Server-Timing"])["jwt"]
|
|
|
|
assert first_dur >= 0
|
|
assert second_dur >= 0
|
|
|
|
|
|
def test_preflight_request_with_cors_allowed_origin_config(defaultenv):
|
|
"OPTIONS preflight request should return Access-Control-Allow-Origin equal to origin"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_SERVER_CORS_ALLOWED_ORIGINS": "http://example.com, http://example2.com",
|
|
}
|
|
|
|
headers = {
|
|
"Accept": "*/*",
|
|
"Origin": "http://example.com",
|
|
"Access-Control-Request-Method": "POST",
|
|
"Access-Control-Request-Headers": "Content-Type",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.options("/items", headers=headers)
|
|
assert (
|
|
response.headers["Access-Control-Allow-Origin"] == "http://example.com"
|
|
and response.headers["Access-Control-Allow-Credentials"] == "true"
|
|
)
|
|
|
|
|
|
def test_preflight_request_with_empty_cors_allowed_origin_config(defaultenv):
|
|
"OPTIONS preflight request should allow all origins when config is present but empty"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_SERVER_CORS_ALLOWED_ORIGINS": "",
|
|
}
|
|
|
|
headers = {
|
|
"Accept": "*/*",
|
|
"Origin": "http://anyorigin.com",
|
|
"Access-Control-Request-Method": "POST",
|
|
"Access-Control-Request-Headers": "Content-Type",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.options("/items", headers=headers)
|
|
assert response.headers["Access-Control-Allow-Origin"] == "*"
|
|
assert "POST" in response.headers["Access-Control-Allow-Methods"]
|
|
|
|
|
|
def test_no_preflight_request_with_CORS_config_should_return_header(defaultenv):
|
|
"GET no preflight request should return Access-Control-Allow-Origin equal to origin"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_SERVER_CORS_ALLOWED_ORIGINS": "http://example.com, http://example2.com",
|
|
}
|
|
|
|
headers = {
|
|
"Accept": "*/*",
|
|
"Origin": "http://example.com",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/items", headers=headers)
|
|
assert response.headers["Access-Control-Allow-Origin"] == "http://example.com"
|
|
|
|
|
|
def test_no_preflight_request_with_CORS_config_should_not_return_header(defaultenv):
|
|
"GET no preflight request should not return Access-Control-Allow-Origin"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_SERVER_CORS_ALLOWED_ORIGINS": "http://example.com, http://example2.com",
|
|
}
|
|
|
|
headers = {
|
|
"Accept": "*/*",
|
|
"Origin": "http://invalid.com",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/items", headers=headers)
|
|
assert "Access-Control-Allow-Origin" not in response.headers
|
|
|
|
|
|
@pytest.mark.parametrize("level", ["crit", "error", "warn", "info", "debug"])
|
|
def test_db_error_logging_to_stderr(level, defaultenv, metapostgrest):
|
|
"verify that DB errors are logged to stderr"
|
|
|
|
role = "timeout_authenticator"
|
|
set_statement_timeout(metapostgrest, role, 500)
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGUSER": role,
|
|
"PGRST_DB_ANON_ROLE": role,
|
|
"PGRST_LOG_LEVEL": level,
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/rpc/sleep?seconds=1")
|
|
assert response.status_code == 500
|
|
|
|
# ensure the message appears on the logs
|
|
output = sorted(postgrest.read_stdout(nlines=6))
|
|
|
|
if level == "crit":
|
|
assert len(output) == 0
|
|
elif level == "debug":
|
|
assert " 500 " in output[0]
|
|
assert "canceling statement due to statement timeout" in output[5]
|
|
else:
|
|
assert " 500 " in output[0]
|
|
assert "canceling statement due to statement timeout" in output[1]
|
|
|
|
reset_statement_timeout(metapostgrest, role)
|
|
|
|
|
|
def test_function_setting_statement_timeout_fails(defaultenv):
|
|
"statement that takes three seconds to execute should fail with one second timeout"
|
|
|
|
with run(env=defaultenv) as postgrest:
|
|
response = postgrest.session.post("/rpc/one_sec_timeout")
|
|
|
|
assert response.status_code == 500
|
|
assert (
|
|
response.text
|
|
== '{"code":"57014","details":null,"hint":null,"message":"canceling statement due to statement timeout"}'
|
|
)
|
|
|
|
|
|
def test_function_setting_statement_timeout_passes(defaultenv):
|
|
"statement that takes three seconds to execute should succeed with four second timeout"
|
|
|
|
with run(env=defaultenv) as postgrest:
|
|
response = postgrest.session.post("/rpc/four_sec_timeout")
|
|
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_function_setting_work_mem(defaultenv):
|
|
"check function setting work_mem is applied"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_HOISTED_TX_SETTINGS": "work_mem",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/rpc/rpc_work_mem?select=get_work_mem")
|
|
|
|
assert response.text == '{"get_work_mem":"6000kB"}'
|
|
|
|
|
|
def test_multiple_func_settings(defaultenv):
|
|
"check multiple function settings are applied"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_HOISTED_TX_SETTINGS": "work_mem,statement_timeout",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get(
|
|
"/rpc/rpc_with_two_hoisted?select=get_work_mem,get_statement_timeout"
|
|
)
|
|
|
|
assert (
|
|
response.text == '{"get_work_mem":"5000kB","get_statement_timeout":"10s"}'
|
|
)
|
|
|
|
|
|
def test_first_hoisted_setting_is_applied(defaultenv):
|
|
"test that work_mem is applied and statement_timeout is not applied"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_HOISTED_TX_SETTINGS": "work_mem", # only work_mem is hoisted
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get(
|
|
"/rpc/rpc_with_one_hoisted?select=get_work_mem,get_statement_timeout"
|
|
)
|
|
|
|
assert response.text == '{"get_work_mem":"3000kB","get_statement_timeout":"2s"}'
|
|
|
|
|
|
def test_second_hoisted_setting_is_applied(defaultenv):
|
|
"test that statement_timeout is applied and work_mem is not applied"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_HOISTED_TX_SETTINGS": "statement_timeout",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get(
|
|
"/rpc/rpc_with_one_hoisted?select=get_work_mem,get_statement_timeout"
|
|
)
|
|
|
|
assert response.text == '{"get_work_mem":"4MB","get_statement_timeout":"7s"}'
|
|
|
|
|
|
def test_admin_metrics(defaultenv):
|
|
"Should get metrics from the admin endpoint"
|
|
|
|
with run(env=defaultenv, port=freeport()) as postgrest:
|
|
response = postgrest.admin.get("/metrics")
|
|
assert response.status_code == 200
|
|
assert response.headers["Content-Type"] == "text/plain; charset=utf-8"
|
|
assert "pgrst_schema_cache_query_time_seconds" in response.text
|
|
assert 'pgrst_schema_cache_loads_total{status="SUCCESS"}' in response.text
|
|
assert "pgrst_db_pool_max" in response.text
|
|
assert "pgrst_db_pool_waiting" in response.text
|
|
assert "pgrst_db_pool_available" in response.text
|
|
assert "pgrst_db_pool_timeouts_total" in response.text
|
|
|
|
|
|
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_jwt_cache_purges_expired_entries(defaultenv):
|
|
"test expired cache entries are purged on cache miss"
|
|
|
|
# The verification of actual cache size reduction is done manually, see https://github.com/PostgREST/postgrest/pull/3801#issuecomment-2620776041
|
|
# This test is written for code coverage of purgeExpired function
|
|
|
|
relativeSeconds = lambda sec: int(
|
|
(datetime.now(timezone.utc) + timedelta(seconds=sec)).timestamp()
|
|
)
|
|
|
|
headers = lambda sec: jwtauthheader(
|
|
{"role": "postgrest_test_author", "exp": relativeSeconds(sec)},
|
|
SECRET,
|
|
)
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_JWT_CACHE_MAX_ENTRIES": "86400",
|
|
"PGRST_JWT_SECRET": SECRET,
|
|
"PGRST_DB_CONFIG": "false",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
|
|
# Generate two unique JWT tokens
|
|
# The 1 second sleep is needed for it generate a unique token
|
|
hdrs1 = headers(5)
|
|
postgrest.session.get("/authors_only", headers=hdrs1)
|
|
|
|
time.sleep(1)
|
|
|
|
hdrs2 = headers(5)
|
|
postgrest.session.get("/authors_only", headers=hdrs2)
|
|
|
|
# Wait 5 seconds for the tokens to expire
|
|
time.sleep(5)
|
|
|
|
hdrs3 = headers(5)
|
|
|
|
# Make another request which should cause a cache miss and so
|
|
# the purgeExpired function will be triggered.
|
|
#
|
|
# This should remove the 2 expired tokens but adds another to cache
|
|
response = postgrest.session.get("/authors_only", headers=hdrs3)
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_pgrst_log_503_client_error_to_stderr(defaultenv):
|
|
"PostgREST should log 503 errors to stderr"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGAPPNAME": "test-io",
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
|
|
postgrest.session.get("/rpc/terminate_pgrst?appname=test-io")
|
|
|
|
output = postgrest.read_stdout(nlines=6)
|
|
|
|
log_message = '{"code":"PGRST001","details":"no connection to the server\\n","hint":null,"message":"Database client error. Retrying the connection."}\n'
|
|
|
|
assert any(log_message in line for line in output)
|
|
|
|
|
|
@pytest.mark.parametrize("level", ["crit", "error", "warn", "info", "debug"])
|
|
def test_log_pool_req_observation(level, defaultenv):
|
|
"PostgREST should log PoolRequest and PoolRequestFullfilled observation when log-level=debug"
|
|
|
|
env = {**defaultenv, "PGRST_LOG_LEVEL": level, "PGRST_JWT_SECRET": SECRET}
|
|
|
|
headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET)
|
|
|
|
pool_req = "Trying to borrow a connection from pool"
|
|
pool_req_fullfill = "Borrowed a connection from the pool"
|
|
|
|
with run(env=env) as postgrest:
|
|
|
|
postgrest.session.get("/authors_only", headers=headers)
|
|
|
|
if level == "debug":
|
|
output = postgrest.read_stdout(nlines=5)
|
|
assert pool_req in output[1]
|
|
assert pool_req_fullfill in output[4]
|
|
assert len(output) == 5
|
|
elif level == "info":
|
|
output = postgrest.read_stdout(nlines=4)
|
|
assert len(output) == 1
|
|
else:
|
|
output = postgrest.read_stdout(nlines=4)
|
|
assert len(output) == 0
|
|
|
|
|
|
def test_proxy_status_header(defaultenv, metapostgrest):
|
|
"Test Proxy-Status header in statement timeout error"
|
|
|
|
role = "timeout_authenticator"
|
|
set_statement_timeout(metapostgrest, role, 1000) # 1 second
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGUSER": role,
|
|
"PGRST_DB_ANON_ROLE": role,
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/rpc/sleep?seconds=2")
|
|
assert response.status_code == 500
|
|
assert response.headers["Proxy-Status"] == "PostgREST; error=57014"
|
|
data = response.json()
|
|
assert data["message"] == "canceling statement due to statement timeout"
|
|
|
|
|
|
def test_invalidate_jwt_cache_when_secret_changes(tmp_path, defaultenv):
|
|
"JWT cache should be emptied after jwt-secret is changed in a config reload"
|
|
|
|
headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET)
|
|
|
|
external_secret_file = tmp_path / "jwt-secret-config"
|
|
external_secret_file.write_text(SECRET)
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_JWT_SECRET": f"@{external_secret_file}",
|
|
"PGRST_DB_CHANNEL_ENABLED": "true",
|
|
"PGRST_JWT_CACHE_MAX_ENTRIES": "86400", # enable cache
|
|
"PGRST_DB_ANON_ROLE": "postgrest_test_anonymous", # required for NOTIFY
|
|
}
|
|
|
|
with run(env=env) as postgrest:
|
|
response = postgrest.session.get("/authors_only", headers=headers)
|
|
assert response.status_code == 200 # jwt gets cached
|
|
|
|
# change external file
|
|
external_secret_file.write_text("invalid" * 5)
|
|
|
|
# reload config and external file with NOTIFY
|
|
# jwt-cache should get empty
|
|
response = postgrest.session.post("/rpc/reload_pgrst_config")
|
|
assert response.text == ""
|
|
assert response.status_code == 204
|
|
sleep_until_postgrest_config_reload()
|
|
|
|
# now the request should fail because the cached token is removed
|
|
response = postgrest.session.get("/authors_only", headers=headers)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_allow_configs_to_be_set_to_empty(defaultenv):
|
|
'configs that are explicitly set to empty (= "<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_schema_cache_error_observation(defaultenv):
|
|
"schema cache error observation should be logged with invalid db-schemas or db-extra-search-path"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGRST_DB_EXTRA_SEARCH_PATH": "x",
|
|
}
|
|
|
|
with run(env=env, no_startup_stdout=False, wait_for_readiness=False) as postgrest:
|
|
# TODO: postgrest should exit here, instead it keeps retrying
|
|
# exitCode = wait_until_exit(postgrest)
|
|
# assert exitCode == 1
|
|
|
|
output = postgrest.read_stdout(nlines=9)
|
|
assert (
|
|
"Failed to load the schema cache using db-schemas=public and db-extra-search-path=x"
|
|
in output[7]
|
|
)
|
|
|
|
|
|
def test_log_listener_connection_errors(defaultenv):
|
|
"The logs should show the listener connection error message in a single line"
|
|
|
|
env = {
|
|
**defaultenv,
|
|
"PGHOST": "no_host",
|
|
"PGRST_DB_CHANNEL_ENABLED": "true",
|
|
}
|
|
|
|
with run(env=env, no_startup_stdout=False, wait_for_readiness=False) as postgrest:
|
|
output = postgrest.read_stdout(nlines=5)
|
|
assert any(
|
|
'Failed listening for database notifications on the "pgrst" channel. could not translate host name "no_host" to address:'
|
|
in line
|
|
for line in output
|
|
)
|