summaryrefslogtreecommitdiffstats
path: root/tests/pq/test_pq.py
blob: 076c3b62b9cf42a84a6c7589c085d156334a9fbb (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import os

import pytest

import psycopg
from psycopg import pq

from ..utils import check_libpq_version


def test_version():
    rv = pq.version()
    assert rv > 90500
    assert rv < 200000  # you are good for a while


def test_build_version():
    assert pq.__build_version__ and pq.__build_version__ >= 70400


@pytest.mark.skipif("not os.environ.get('PSYCOPG_TEST_WANT_LIBPQ_BUILD')")
def test_want_built_version():
    want = os.environ["PSYCOPG_TEST_WANT_LIBPQ_BUILD"]
    got = pq.__build_version__
    assert not check_libpq_version(got, want)


@pytest.mark.skipif("not os.environ.get('PSYCOPG_TEST_WANT_LIBPQ_IMPORT')")
def test_want_import_version():
    want = os.environ["PSYCOPG_TEST_WANT_LIBPQ_IMPORT"]
    got = pq.version()
    assert not check_libpq_version(got, want)


# Note: These tests are here because test_pipeline.py tests are all skipped
# when pipeline mode is not supported.


@pytest.mark.libpq(">= 14")
def test_pipeline_supported(conn):
    assert psycopg.Pipeline.is_supported()
    assert psycopg.AsyncPipeline.is_supported()

    with conn.pipeline():
        pass


@pytest.mark.libpq("< 14")
def test_pipeline_not_supported(conn):
    assert not psycopg.Pipeline.is_supported()
    assert not psycopg.AsyncPipeline.is_supported()

    with pytest.raises(psycopg.NotSupportedError) as exc:
        with conn.pipeline():
            pass

    assert "too old" in str(exc.value)