postgrest/test/io/test_io.py

"Unit tests for Input/Ouput of PostgREST seen as a black box."

import os
import re
import signal
import time
import pytest
import requests

from config import SECRET
from util import (
    Thread,
    jwtauthheader,
    relativeSeconds,
    drain_stdout,
    match_log,
    psql_as_superuser,
)
from postgrest import (
    Admin,
    PostgrestTimedOut,
    freeport,
    is_ipv6,
    reset_statement_timeout,
    run,
    run_pgproxy,
    set_statement_timeout,
    wait_until_exit,
)


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


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_graceful_shutdown_waits_for_in_flight_request(defaultenv):
    "SIGTERM should allow in-flight requests to finish before exiting"

    with run(env=defaultenv, wait_max_seconds=5) as postgrest:

        def sleep():
            response = postgrest.session.get("/rpc/sleep?seconds=3", timeout=10)
            assert response.text == ""
            assert response.status_code == 204

        t = Thread(target=sleep)
        t.start()

        # Wait for the request to be in-flight before shutting down.
        time.sleep(1)

        postgrest.process.terminate()

        t.join()


def test_termination_unix_signal_logging(defaultenv):
    "Server logs when handling termination unix signals."

    with run(env=defaultenv) as postgrest:
        postgrest.process.send_signal(signal.SIGTERM)
        lines = postgrest.read_stdout(nlines=1)
        wait_until_exit(postgrest)

    assert any("SIGTERM" in line for line in lines)

    with run(env=defaultenv) as postgrest:
        postgrest.process.send_signal(signal.SIGINT)
        lines = postgrest.read_stdout(nlines=1)
        wait_until_exit(postgrest)

    assert any("SIGINT" in line for line in lines)


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."

    # set host to _all_ addresses to force port conflict without SO_REUSEPORT
    # setting to localhost (which is the default)
    # might allow running multiple instances on the same port
    # as the name might be resolved to many IP addresses
    host = "0.0.0.0"
    port = freeport()
    admin_port = freeport(used_ports=[port])
    failures = []
    # mutable location shared between threads
    keep_running = {"value": True}

    # 1. Start first PostgREST instance
    # 2. Start a "client" thread issuing requests in a loop
    #    remembering all received errors
    # 3. Start second PostgREST instance on the same port as the first one
    # 4. Wait a little and terminate the first instance
    #
    # We expect the client does not get any errors after stopping the first instance
    # and seamlessly migrate to the second instance.
    #
    # 5. Stop client thread
    # 6. Stop second PostgREST instance
    # 7. Verify client did not get any errors
    with run(
        env={**defaultenv, "PGRST_SERVER_REUSEPORT": "true"},
        port=port,
        host=host,
        admin_port=admin_port,
    ) as first:

        def continuously_request():
            while keep_running["value"]:
                try:
                    response = first.session.get("/projects", timeout=1)
                    assert response.status_code == 200
                except Exception as exc:
                    failures.append(exc)
                    break
                time.sleep(0.2)

        requester = Thread(target=continuously_request)
        requester.start()

        try:
            time.sleep(1)
            with run(
                env={**defaultenv, "PGRST_SERVER_REUSEPORT": "true"},
                port=port,
                host=host,
                # we do not set SO_REUSEPORT on admin socket
                admin_port=freeport(used_ports=[port, admin_port]),
            ):
                time.sleep(1)
                first.process.terminate()
                wait_until_exit(first, 2)

                time.sleep(1)
        finally:
            keep_running["value"] = False
            requester.join()

    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


@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, wait_max_seconds=3
    ) 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_pool_acquisition_timeout_logs_are_debounced(defaultenv):
    "Pool acquisition timeout diagnostic logs should be debounced over a burst of failures"

    env = {
        **defaultenv,
        "PGRST_DB_POOL": "1",
        "PGRST_DB_POOL_ACQUISITION_TIMEOUT": "1",
        "PGRST_LOG_LEVEL": "error",
    }
    total_requests = 6

    with run(
        env=env, no_pool_connection_available=True, wait_max_seconds=3
    ) as postgrest:

        def request_timeout():
            response = postgrest.session.get("/projects")
            assert response.status_code == 504
            assert (
                response.json()["message"]
                == "Timed out acquiring connection from connection pool."
            )
            return response

        request_timeout()

        threads = [Thread(target=request_timeout) for _ in range(total_requests - 1)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()

        # Logger debouncing logs the first timeout immediately and, if more
        # timeouts happen during the cooldown, logs one more time afterwards.
        time.sleep(6)
        output = drain_stdout(postgrest)

    access_logs = [line for line in output if ' "GET /projects HTTP/1.1" 504 ' in line]
    timeout_logs = [
        line
        for line in output
        if "Timed out acquiring connection from connection pool." in line
    ]

    assert len(access_logs) == total_requests
    assert len(timeout_logs) == 2


def test_listener_query_is_visible_in_pg_stat_activity(defaultenv):
    "The listener connection should show the LISTEN pgrst statement in pg_stat_activity"

    env = {
        **defaultenv,
        "PGRST_DB_CHANNEL_ENABLED": "true",
        "PGAPPNAME": "listener-query-test",
    }

    with run(env=env):
        output = psql_as_superuser(
            """
        select query
        from pg_stat_activity
        where application_name = 'listener-query-test'
          and query = 'LISTEN "pgrst"'
        limit 1;
        """,
            capture_output=True,
        ).strip()

        assert output == 'LISTEN "pgrst"'


@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 = postgrest.read_stdout(nlines=9)

        if level == "crit":
            assert len(output) == 0
        elif level == "error":
            match_log(
                output,
                [r'- - - \[.+\] "GET / HTTP/1.1" 500 \d+ "" "python-requests/.+"'],
            )
            assert len(output) == 1
        elif level == "warn":
            match_log(
                output,
                [
                    r'- - - \[.+\] "GET / HTTP/1.1" 500 \d+ "" "python-requests/.+"',
                    r'- - postgrest_test_anonymous \[.+\] "GET /unknown HTTP/1.1" 404 \d+ "" "python-requests/.+"',
                ],
            )
            assert len(output) == 2
        elif level == "info":
            match_log(
                output,
                [
                    r'- - - \[.+\] "GET / HTTP/1.1" 500 \d+ "" "python-requests/.+"',
                    r'- - postgrest_test_anonymous \[.+\] "GET /unknown HTTP/1.1" 404 \d+ "" "python-requests/.+"',
                    r'- - postgrest_test_anonymous \[.+\] "GET / HTTP/1.1" 200 \d+ "" "python-requests/.+"',
                ],
            )
            assert len(output) == 3
        elif level == "debug":
            match_log(
                output,
                [
                    r'- - - \[.+\] "GET / HTTP/1.1" 500 \d+ "" "python-requests/.+"',
                    r'- - postgrest_test_anonymous \[.+\] "GET /unknown HTTP/1.1" 404 \d+ "" "python-requests/.+"',
                    r'- - postgrest_test_anonymous \[.+\] "GET / HTTP/1.1" 200 \d+ "" "python-requests/.+"',
                ],
            )
            assert len(output) == 9
            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"

    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 == 200

        response = postgrest.session.get(
            "/projects", headers={"Prefer": "count=planned"}
        )
        assert response.status_code == 200

        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.+pn\.nspname AS proc_schema.+FROM pg_proc p.+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_expired_jwt_log_lacks_role(defaultenv):
    "Expired JWT requests are logged without a role."

    env = {**defaultenv, "PGRST_JWT_SECRET": SECRET}
    headers = jwtauthheader({"exp": relativeSeconds(-35)}, SECRET)

    with run(env=env) as postgrest:
        response = postgrest.session.get("/authors_only", headers=headers)
        assert response.status_code == 401

        output = postgrest.read_stdout(nlines=1)

    assert len(output) == 1
    assert re.match(
        r'- - - \[.+\] "GET /authors_only HTTP/1.1" 401 \d+ "" "python-requests/.+"',
        output[0],
    )


def test_log_lacks_role_with_empty_anon_role(defaultenv):
    "Requests are logged without a role when db-anon-role is empty."

    env = {
        **defaultenv,
        "PGRST_DB_CONFIG": "false",
        "PGRST_DB_ANON_ROLE": "",
    }

    with run(env=env) as postgrest:
        response = postgrest.session.get("/projects")
        assert response.status_code == 401

        output = postgrest.read_stdout(nlines=1)

    assert len(output) == 1
    assert re.match(
        r'- - - \[.+\] "GET /projects HTTP/1.1" 401 \d+ "" "python-requests/.+"',
        output[0],
    )


def test_invalid_rpc_method_log_contains_role(defaultenv):
    "Invalid RPC method requests are logged with the anonymous role."

    with run(env=defaultenv) as postgrest:
        response = postgrest.session.put("/rpc/sleep")
        assert response.status_code == 405

        output = postgrest.read_stdout(nlines=1)

    assert len(output) == 1
    assert re.match(
        r'- - postgrest_test_anonymous \[.+\] "PUT /rpc/sleep HTTP/1.1" 405 \d+ "" "python-requests/.+"',
        output[0],
    )


def test_empty_schema_cache_log_contains_jwt_role(defaultenv):
    "Requests are logged with the role when the schema cache is empty on startup"

    env = {
        **defaultenv,
        "PGRST_DB_SCHEMAS": "non_existent_schema_aaaa",
        "PGRST_JWT_SECRET": SECRET,
    }
    headers = jwtauthheader({"role": "postgrest_test_author"}, SECRET)

    with run(env=env, wait_for=None) as postgrest:
        postgrest.wait_until_scache_starts_loading()

        response = postgrest.session.get("/authors_only", headers=headers)
        assert response.status_code == 503

        output = drain_stdout(postgrest)

    assert any(
        re.match(
            r'- - postgrest_test_author \[.+\] "GET /authors_only HTTP/1.1" 503 \d+ "" "python-requests/.+"',
            line,
        )
        for line in output
    )


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_options_request_logs_but_cors_preflight_does_not(defaultenv):
    "Plain OPTIONS requests should be logged, but CORS preflight requests should not."

    env = {
        **defaultenv,
        "PGRST_LOG_LEVEL": "info",
        "PGRST_SERVER_CORS_ALLOWED_ORIGINS": "http://example.com",
    }
    preflight_headers = {
        "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("/projects")
        assert response.status_code == 200

        response = postgrest.session.options("/projects", headers=preflight_headers)
        assert response.status_code == 200
        assert response.headers["Access-Control-Allow-Origin"] == "http://example.com"

        output = drain_stdout(postgrest)

    assert len(output) == 1
    assert re.match(
        r'- - postgrest_test_anonymous \[.+\] "OPTIONS /projects HTTP/1.1" 200 \d+ "" "python-requests/.+"',
        output[0],
    )


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
        wrong_secret = "This is the most wrong secret of all secrets"
        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


def test_schema_cache_query_sleep_logs(defaultenv):
    """Schema cache sleep should be reflected in the logged query duration."""

    env = {
        **defaultenv,
        "PGRST_INTERNAL_SCHEMA_CACHE_QUERY_SLEEP": "1000",
    }
    log_pattern = re.compile(r"Schema cache queried in ([\d.]+) milliseconds")

    with run(env=env, wait_max_seconds=3, no_startup_stdout=False) as postgrest:
        observed_ms = None
        collected = []

        lines = postgrest.read_stdout(nlines=10)
        collected.extend(lines)
        for line in lines:
            match = log_pattern.search(line)
            if match:
                observed_ms = float(match.group(1))
                break

        assert observed_ms is not None
        assert 1000 < observed_ms < 2000


@pytest.mark.parametrize("level", ["crit", "error", "warn", "info", "debug"])
def test_schema_cache_query_timings_log(level, defaultenv):
    "Schema cache query timings should be logged on log-level=debug."

    env = {
        **defaultenv,
        "PGRST_LOG_LEVEL": level,
    }
    log_pattern = re.compile(
        r".+: tables: [\d.]+ ms, keydeps: [\d.]+ ms, rels: [\d.]+ ms, funcs: [\d.]+ ms, comprels: [\d.]+ ms, dreps: [\d.]+ ms, mhandlers: [\d.]+ ms"
    )

    with run(env=env, no_startup_stdout=False) as postgrest:
        output = drain_stdout(postgrest)
        timing_matches = [
            match for line in output if (match := log_pattern.match(line))
        ]

        if level == "debug":
            assert len(timing_matches) == 1
        else:
            assert not timing_matches


@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"
    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=11)

        # Cannot assume a particular log entry order
        # Listening on a socket happens after schema querying
        # but is concurrent to the schema loading process
        # and migh happen before or after writing of the
        # "Schema cache loaded" log entry
        if is_unix:
            match_log(output, [r".*API server listening on .*/tmp/.*\.sock"])
        elif is_ipv6(host):
            match_log(output, [r".*API server listening on \[.+]:\d+"])
        else:  # IPv4
            match_log(output, [r".*API server listening on .+:\d+"])


@pytest.mark.parametrize(
    "host", ["127.0.0.1", "::1", None], ids=["IPv4", "IPv6", "Unix"]
)
def test_log_postgrest_admin_server_host_and_port(host, defaultenv):
    "PostgREST should log the admin server host and port"

    # We run admin server on unix socket when host and admin_port are set to None
    is_unix = host is None
    port = None if is_unix else freeport()
    admin_port = None if is_unix else freeport(used_ports=[port])

    with run(
        env=defaultenv,
        host=host,
        port=port,
        admin_port=admin_port,
        no_startup_stdout=False,
        wait_for=Admin.ready,
    ) as postgrest:
        output = postgrest.read_stdout(nlines=11)

        # Cannot assume a particular log entry order
        # Listening on a socket happens after schema querying
        # but is concurrent to the schema loading process
        # and migh happen before or after writing of the
        # "Schema cache loaded" log entry
        if is_unix:
            match_log(output, [r".*Admin server listening on .*/tmp/.*\.sock"])
        elif is_ipv6(host):
            match_log(output, [r".*Admin server listening on \[.+]:\d+"])
        else:  # IPv4
            match_log(output, [r".*Admin server listening on .+:\d+"])


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=None) 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


@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 = postgrest.read_stdout(nlines=8)

        if level == "crit":
            assert len(output) == 0
        elif level == "debug":
            match_log(
                output,
                [
                    r".*canceling statement due to statement timeout.*",
                    r".*500.*",
                ],
            )
        else:
            assert " 500 " in output[1]
            assert "canceling statement due to statement timeout" in output[0]

    reset_statement_timeout(metapostgrest, role)


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_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)


def test_log_error_when_schema_cache_load_error_on_startup_to_stderr(defaultenv):
    "Should log the 503 error message when there is an error loading schema cache on startup"

    env = {
        **defaultenv,
        "PGRST_INTERNAL_SCHEMA_CACHE_QUERY_SLEEP_BEFORE_QUERIES": "1000",
        "PGRST_DB_SCHEMAS": "non_existent_schema_aaaa",
    }

    with run(env=env, wait_for=None) as postgrest:
        postgrest.wait_until_scache_starts_loading()

        # First call should fail with connection refused
        with pytest.raises(requests.ConnectionError):
            postgrest.session.get("/projects")

        # Next call should return 503
        time.sleep(1)
        response = postgrest.session.get("/projects")
        assert response.status_code == 503

        output_start = postgrest.read_stdout(nlines=10)

        log_err_message = '{"code":"PGRST002","details":null,"hint":null,"message":"Could not query the database for the schema cache. Retrying."}'

        assert any(log_err_message in line for line in output_start)


@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 = r".*Trying to borrow a connection from pool.*"
    pool_req_fullfill = r".*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=7)
            assert len(output) == 7
            match_log(output, [pool_req, pool_req_fullfill])
        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_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=None) 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[6]
        )


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=None) 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
        )


def test_log_listener_connection_start(defaultenv):
    "The logs should show the listener connection start message in a single line"

    env = {
        **defaultenv,
        "PGRST_DB_CHANNEL_ENABLED": "true",
    }

    with run(env=env, no_startup_stdout=False, wait_for=Admin.ready) as postgrest:
        output = postgrest.read_stdout(nlines=10)
        # Check for the listener start message containing host and port
        # Do not check if pg version is displayed properly as it is tricky to test it
        assert any(
            f'"{defaultenv["PGHOST"]}:5432" and listening for database notifications on the "pgrst" channel'
            in line
            for line in output
        )


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)


def test_positive_pool_metric(defaultenv):
    "When a network failure is caused on the pg connection, pgrst_db_pool_available stays positive"

    with run_pgproxy(defaultenv, proxy_timeout="1ms") as pgproxyhost:
        env = {**defaultenv, "PGHOST": pgproxyhost}

        with run(env=env, wait_for=Admin.live) as postgrest:
            response = postgrest.admin.get("/metrics", timeout=1)
            assert response.status_code == 200

            metrics = float(
                re.search(
                    r"pgrst_db_pool_available (-?\d+(?:\.\d+)?)", response.text
                ).group(1)
            )
            assert metrics >= 0


@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