sqlink/tests/test_insert.py at main · JSLEEKR/sqlink · 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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
"""Tests for INSERT query building."""

import pytest
from sqlink import Query, Table, F, Raw


class TestBasicInsert:
    def test_insert_dict(self):
        sql, params = (
            Query("users")
            .insert("name", "email")
            .values({"name": "John", "email": "john@example.com"})
            .build()
        )
        assert 'INSERT INTO "users"' in sql
        assert "VALUES" in sql
        assert params == ["John", "john@example.com"]

    def test_insert_multiple_rows(self):
        sql, params = (
            Query("users")
            .insert("name", "email")
            .values(
                {"name": "Alice", "email": "alice@example.com"},
                {"name": "Bob", "email": "bob@example.com"},
            )
            .build()
        )
        assert sql.count("(?, ?)") == 2
        assert len(params) == 4

    def test_insert_list_values(self):
        sql, params = (
            Query("users")
            .insert("name", "email")
            .values(["John", "john@example.com"])
            .build()
        )
        assert params == ["John", "john@example.com"]

    def test_insert_tuple_values(self):
        sql, params = (
            Query("users")
            .insert("name", "email")
            .values(("John", "john@example.com"))
            .build()
        )
        assert params == ["John", "john@example.com"]

    def test_insert_auto_columns_from_dict(self):
        sql, params = (
            Query("users")
            .insert()
            .values({"name": "John", "age": 30})
            .build()
        )
        assert '"name"' in sql
        assert '"age"' in sql
        assert params == ["John", 30]

    def test_insert_with_raw_expr(self):
        sql, params = (
            Query("users")
            .insert("name", "created_at")
            .values({"name": "John", "created_at": Raw("NOW()")})
            .build()
        )
        assert "NOW()" in sql
        assert params == ["John"]


class TestInsertReturning:
    def test_returning_single(self):
        sql, params = (
            Query("users")
            .insert("name")
            .values({"name": "John"})
            .returning("id")
            .build()
        )
        assert 'RETURNING "id"' in sql

    def test_returning_multiple(self):
        sql, params = (
            Query("users")
            .insert("name")
            .values({"name": "John"})
            .returning("id", "created_at")
            .build()
        )
        assert 'RETURNING "id", "created_at"' in sql


class TestUpsert:
    def test_on_conflict_do_nothing(self):
        sql, params = (
            Query("users")
            .insert("email", "name")
            .values({"email": "john@example.com", "name": "John"})
            .on_conflict_do_nothing(["email"])
            .build()
        )
        assert 'ON CONFLICT ("email") DO NOTHING' in sql

    def test_on_conflict_update(self):
        sql, params = (
            Query("users")
            .insert("email", "name")
            .values({"email": "john@example.com", "name": "John"})
            .on_conflict(["email"])
            .build()
        )
        assert "ON CONFLICT" in sql
        assert "DO UPDATE SET" in sql


class TestTableInsert:
    def test_table_insert(self):
        users = Table("users")
        sql, params = (
            users.insert("name", "email")
            .values({"name": "Jane", "email": "jane@example.com"})
            .build()
        )
        assert 'INSERT INTO "users"' in sql
        assert params == ["Jane", "jane@example.com"]