pytest-databases/tests/test_mysql.py at ff6745330d200fe76bb1553fe2a73ff760a7cc18 · litestar-org/pytest-databases · GitHub

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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
from __future__ import annotations

import pytest


@pytest.mark.parametrize(
    "service_fixture",
    [
        "mysql_8_service",
        "mysql_56_service",
        "mysql_57_service",
    ],
)
def test_service_fixture(pytester: pytest.Pytester, service_fixture: str) -> None:
    pytester.makepyfile(f"""
    import pymysql
    pytest_plugins = ["pytest_databases.docker.mysql"]

    def test({service_fixture}):
        with pymysql.connect(
            host={service_fixture}.host,
            port={service_fixture}.port,
            user={service_fixture}.user,
            database={service_fixture}.db,
            password={service_fixture}.password,
        ) as conn, conn.cursor() as cursor:
            cursor.execute("select 1 as is_available")
            resp = cursor.fetchone()
        assert resp is not None and resp[0] == 1
    """)

    result = pytester.runpytest("-vv")
    result.assert_outcomes(passed=1)


@pytest.mark.parametrize(
    "connection_fixture",
    [
    "mysql_56_connection",
    "mysql_57_connection",
    ],
)
def test_connection_fixture(pytester: pytest.Pytester, connection_fixture: str) -> None:
    pytester.makepyfile(f"""
    pytest_plugins = ["pytest_databases.docker.mysql"]

    def test({connection_fixture}):
        with {connection_fixture}.cursor() as cursor:
            cursor.execute("CREATE TABLE if not exists simple_table as SELECT 1 as the_value")
            cursor.execute("select * from simple_table")
            result = cursor.fetchall()
            assert result is not None and result[0][0] == 1
    """)

    result = pytester.runpytest("-vv")
    result.assert_outcomes(passed=1)


def test_xdist_isolate_database(pytester: pytest.Pytester) -> None:
    pytester.makepyfile("""
    pytest_plugins = ["pytest_databases.docker.mysql"]

    def test_1(mysql_56_connection):
        with mysql_56_connection.cursor() as cursor:
            cursor.execute("CREATE TABLE simple_table as SELECT 1 as the_value;")

    def test_2(mysql_56_connection):
        with mysql_56_connection.cursor() as cursor:
            cursor.execute("CREATE TABLE simple_table as SELECT 1 as the_value;")
    """)

    result = pytester.runpytest("-n", "2")
    result.assert_outcomes(passed=2)


def test_xdist_isolate_server(pytester: pytest.Pytester) -> None:
    pytester.makepyfile("""
    import pytest
    pytest_plugins = ["pytest_databases.docker.mysql"]

    @pytest.fixture(scope="session")
    def xdist_mysql_isolation_level():
        return "server"

    def test_1(mysql_56_connection):
        with mysql_56_connection.cursor() as cursor:
            cursor.execute("CREATE DATABASE db_test")

    def test_2(mysql_56_connection):
        with mysql_56_connection.cursor() as cursor:
            cursor.execute("CREATE DATABASE db_test")
    """)

    result = pytester.runpytest("-n", "2")
    result.assert_outcomes(passed=2)