wJYT5XT1

1"""
2This example has not been tested. Use it as a reference only.
3"""
4
5import psycopg2
6import pytest
7
8
9USER = "postgres"
10PASS = None
11HOST = "localhost"
12PORT = 5432
13
14DATABASE = "my_schema"
15TABLE = "my_table"
16CREATE_DB = "CREATE DATABASE %s;"
17TRUNCATE = "TRUNCATE TABLE %s CASCADE"
18DROP_DB = "DROP DATABASE %s;"
19
20
21# A simple fixture
22def db_conf():
23    return dict(user=USER, password=PASS, host=HOST, port=PORT)
24
25
26# __create_db and _drop_db should come from another package or from a library
27def _create_db(configuration):
28    connection = psycopg2.connect(**configuration)
29    connection.autocommit = True
30    try:
31        connection.cursor().execute(CREATE_DB, [psycopg2.extensions.AsIs(DATABASE)])
32    except Exception as error:
33        if not hasattr(error, "pgerror") or "already exists" not in error.pgerror:
34            raise error
35        print("Database '%s' already exists.", DATABASE)
36    connection.close()
37
38
39def _drop_db(configuration):
40    connection = psycopg2.connect(**configuration)
41    connection.autocommit = True
42    try:
43        connection.cursor().execute(DROP_DB, [psycopg2.extensions.AsIs(DATABASE)])
44    except Exception as error:
45        if not hasattr(error, "pgerror") or "already exists" not in error.pgerror:
46            raise error
47        print("Database '%s' already exists.", DATABASE)
48    connection.close()
49
50
51@pytest.fixture(scope="session", autouse=True)
52def test_db(db_conf):
53    """
54    This fixture will be executed once in the entire suite, independently of the filters you use
55    running pytest. It will create the test_db at the beginning and droping the table at the end
56    of all tests.
57    """
58    _create_db(**db_conf)
59    yield
60    _drop_db(**db_conf)
61
62
63# A function scoped fixture
64@pytest.fixture(scope="function", autouse=True)
65def clear_tables(db_conf):
66    """
67    This fixture will be executed after every test, clearing the
68    given table.
69    You can remove the autouse, and specify when it executed whenever you want.
70    """
71
72    yield
73    connection = psycopg2.connect(**db_conf)
74    connection.autocommit = True
75    connection.cursor().execute(TRUNCATE, [psycopg2.extensions.AsIs(TABLE)])
76    connection.close()