sqlit/tests/test_postgresql.py at main · VietDuc19/sqlit · 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
96
97
98
99
100
101
102
103
104
105
"""Integration tests for PostgreSQL database operations."""

from __future__ import annotations

from .test_database_base import BaseDatabaseTestsWithLimit, DatabaseTestConfig


class TestPostgreSQLIntegration(BaseDatabaseTestsWithLimit):
    """Integration tests for PostgreSQL database operations via CLI.

    These tests require a running PostgreSQL instance (via Docker).
    Tests are skipped if PostgreSQL is not available.
    """

    @property
    def config(self) -> DatabaseTestConfig:
        return DatabaseTestConfig(
            db_type="postgresql",
            display_name="PostgreSQL",
            connection_fixture="postgres_connection",
            db_fixture="postgres_db",
            create_connection_args=lambda: [],  # Uses fixtures
            timezone_datetime_type="TIMESTAMPTZ",
        )

    def test_create_postgres_connection(self, postgres_db, cli_runner):
        """Test creating a PostgreSQL connection via CLI."""
        from .conftest import (
            POSTGRES_HOST,
            POSTGRES_PASSWORD,
            POSTGRES_PORT,
            POSTGRES_USER,
        )

        connection_name = "test_create_postgres"

        try:
            # Create connection
            result = cli_runner(
                "connections",
                "add",
                "postgresql",
                "--name",
                connection_name,
                "--server",
                POSTGRES_HOST,
                "--port",
                str(POSTGRES_PORT),
                "--database",
                postgres_db,
                "--username",
                POSTGRES_USER,
                "--password",
                POSTGRES_PASSWORD,
            )
            assert result.returncode == 0
            assert "created successfully" in result.stdout

            # Verify it appears in list
            result = cli_runner("connection", "list")
            assert connection_name in result.stdout
            assert "PostgreSQL" in result.stdout

        finally:
            # Cleanup
            cli_runner("connection", "delete", connection_name, check=False)

    def test_delete_postgres_connection(self, postgres_db, cli_runner):
        """Test deleting a PostgreSQL connection."""
        from .conftest import (
            POSTGRES_HOST,
            POSTGRES_PASSWORD,
            POSTGRES_PORT,
            POSTGRES_USER,
        )

        connection_name = "test_delete_postgres"

        # Create connection first
        cli_runner(
            "connections",
            "add",
            "postgresql",
            "--name",
            connection_name,
            "--server",
            POSTGRES_HOST,
            "--port",
            str(POSTGRES_PORT),
            "--database",
            postgres_db,
            "--username",
            POSTGRES_USER,
            "--password",
            POSTGRES_PASSWORD,
        )

        # Delete it
        result = cli_runner("connection", "delete", connection_name)
        assert result.returncode == 0
        assert "deleted successfully" in result.stdout

        # Verify it's gone
        result = cli_runner("connection", "list")
        assert connection_name not in result.stdout