packages/ztd-query-pdo-adapter/tests/Integration/PostgreSql/InsertOnConflictTest.php
1<?php
2
3declare(strict_types=1);
4
5namespace Tests\Integration\PostgreSql;
6
7use Container\Endpoint;
8use Container\PostgreSqlRelease;
9use PDO;
10use PHPUnit\Framework\Attributes\CoversNothing;
11use PHPUnit\Framework\Attributes\Large;
12use PHPUnit\Framework\TestCase;
13use ZtdQuery\Adapter\Pdo\ZtdPdo;
14
15/**
16 * @requires extension pdo_pgsql
17 * @group integration
18 * @group postgres
19 *
20 * @phpstan-type Row array<string, mixed>
21 */
22#[CoversNothing]
23#[Large]
24final class InsertOnConflictTest extends TestCase
25{
26 public function testPreparedOnConflictDoUpdateReplacesExistingRow(): void
27 {
28 $endpoint = \Testcontainers\Testcontainers::run(PostgreSqlRelease::fromEnvironment())->getData(Endpoint::class);
29 /** @var PDO $rawPdo */
30 $rawPdo = new PDO(
31 $endpoint->dsn(),
32 $endpoint->username,
33 $endpoint->password,
34 [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC],
35 );
36
37 $schemaName = 'ztd_' . bin2hex(random_bytes(8));
38 $rawPdo->exec(sprintf('CREATE SCHEMA "%s"', $schemaName));
39 $rawPdo->exec(sprintf('SET search_path TO "%s"', $schemaName));
40
41 $table = 'prefix_' . bin2hex(random_bytes(8));
42
43 try {
44 $rawPdo->exec("CREATE TABLE {$table} (id INTEGER PRIMARY KEY, name TEXT NOT NULL)");
45 $ztdPdo = ZtdPdo::fromPdo($rawPdo);
46 $ztdPdo->exec("INSERT INTO {$table} VALUES (1, 'original')");
47 $statement = $ztdPdo->prepare(
48 "INSERT INTO {$table} VALUES (?, ?) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name"
49 );
50 self::assertNotFalse($statement);
51
52 self::assertTrue($statement->execute(['1', 'updated']));
53
54 $rows = $ztdPdo->query("SELECT * FROM {$table} WHERE id = 1");
55 self::assertNotFalse($rows);
56 self::assertSame([['id' => 1, 'name' => 'updated']], $rows->fetchAll());
57 } finally {
58 $rawPdo->exec(sprintf('DROP SCHEMA IF EXISTS "%s" CASCADE', $schemaName));
59 }
60 }
61
62 public function testOnConflictDoNothing(): void
63 {
64 $endpoint = \Testcontainers\Testcontainers::run(PostgreSqlRelease::fromEnvironment())->getData(Endpoint::class);
65 /** @var PDO $rawPdo */
66 $rawPdo = new PDO(
67 $endpoint->dsn(),
68 $endpoint->username,
69 $endpoint->password,
70 [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC],
71 );
72
73 $schemaName = 'ztd_' . bin2hex(random_bytes(8));
74 $rawPdo->exec(sprintf('CREATE SCHEMA "%s"', $schemaName));
75 $rawPdo->exec(sprintf('SET search_path TO "%s"', $schemaName));
76
77 $table = 'prefix_' . bin2hex(random_bytes(8));
78
79 try {
80 $rawPdo->exec("CREATE TABLE {$table} (id INTEGER PRIMARY KEY, name TEXT NOT NULL, age INTEGER NOT NULL)");
81
82 $ztdPdo = ZtdPdo::fromPdo($rawPdo);
83
84 $rawPdo->exec("INSERT INTO {$table} (id, name, age) VALUES (1, 'Alice', 30)");
85 $ztdPdo->exec("INSERT INTO {$table} (id, name, age) VALUES (1, 'Alice', 30)");
86
87 $rawPdo->exec("INSERT INTO {$table} (id, name, age) VALUES (1, 'Duplicate', 99) ON CONFLICT (id) DO NOTHING");
88 $ztdPdo->exec("INSERT INTO {$table} (id, name, age) VALUES (1, 'Duplicate', 99) ON CONFLICT (id) DO NOTHING");
89
90 $stmt = $rawPdo->query("SELECT * FROM {$table} ORDER BY id");
91 self::assertNotFalse($stmt);
92 /** @var list<Row> */
93 $rawRows = $stmt->fetchAll();
94
95 $stmt = $ztdPdo->query("SELECT * FROM {$table} ORDER BY id");
96 self::assertNotFalse($stmt);
97 /** @var list<Row> */
98 $ztdRows = $stmt->fetchAll();
99
100 self::assertSame($rawRows, $ztdRows);
101 } finally {
102 $rawPdo->exec(sprintf('DROP SCHEMA IF EXISTS "%s" CASCADE', $schemaName));
103 }
104 }
105
106 public function testOnConflictDoUpdate(): void
107 {
108 $endpoint = \Testcontainers\Testcontainers::run(PostgreSqlRelease::fromEnvironment())->getData(Endpoint::class);
109 /** @var PDO $rawPdo */
110 $rawPdo = new PDO(
111 $endpoint->dsn(),
112 $endpoint->username,
113 $endpoint->password,
114 [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC],
115 );
116
117 $schemaName = 'ztd_' . bin2hex(random_bytes(8));
118 $rawPdo->exec(sprintf('CREATE SCHEMA "%s"', $schemaName));
119 $rawPdo->exec(sprintf('SET search_path TO "%s"', $schemaName));
120
121 $table = 'prefix_' . bin2hex(random_bytes(8));
122
123 try {
124 $rawPdo->exec("CREATE TABLE {$table} (id INTEGER PRIMARY KEY, name TEXT NOT NULL, age INTEGER NOT NULL)");
125
126 $ztdPdo = ZtdPdo::fromPdo($rawPdo);
127
128 $rawPdo->exec("INSERT INTO {$table} (id, name, age) VALUES (1, 'Alice', 30)");
129 $ztdPdo->exec("INSERT INTO {$table} (id, name, age) VALUES (1, 'Alice', 30)");
130
131 $rawPdo->exec("INSERT INTO {$table} (id, name, age) VALUES (1, 'Alice Updated', 31) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, age = EXCLUDED.age");
132 $ztdPdo->exec("INSERT INTO {$table} (id, name, age) VALUES (1, 'Alice Updated', 31) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, age = EXCLUDED.age");
133
134 $stmt = $rawPdo->query("SELECT * FROM {$table} ORDER BY id");
135 self::assertNotFalse($stmt);
136 /** @var list<Row> */
137 $rawRows = $stmt->fetchAll();
138
139 $stmt = $ztdPdo->query("SELECT * FROM {$table} ORDER BY id");
140 self::assertNotFalse($stmt);
141 /** @var list<Row> */
142 $ztdRows = $stmt->fetchAll();
143
144 self::assertSame($rawRows, $ztdRows);
145 } finally {
146 $rawPdo->exec(sprintf('DROP SCHEMA IF EXISTS "%s" CASCADE', $schemaName));
147 }
148 }
149
150 public function testConditionalOnConflictAndReturningMatchPostgres(): void
151 {
152 $endpoint = \Testcontainers\Testcontainers::run(PostgreSqlRelease::fromEnvironment())->getData(Endpoint::class);
153 /** @var PDO $rawPdo */
154 $rawPdo = new PDO(
155 $endpoint->dsn(),
156 $endpoint->username,
157 $endpoint->password,
158 [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC],
159 );
160
161 $schemaName = 'ztd_' . bin2hex(random_bytes(8));
162 $rawPdo->exec(sprintf('CREATE SCHEMA "%s"', $schemaName));
163 $rawPdo->exec(sprintf('SET search_path TO "%s"', $schemaName));
164
165 $nativeTable = 'prefix_' . bin2hex(random_bytes(8));
166 $shadowTable = 'prefix_' . bin2hex(random_bytes(8));
167
168 try {
169 $rawPdo->exec("CREATE TABLE {$nativeTable} (id INTEGER PRIMARY KEY, name TEXT, score INTEGER)");
170 $rawPdo->exec("CREATE TABLE {$shadowTable} (id INTEGER PRIMARY KEY, name TEXT, score INTEGER)");
171 $ztdPdo = ZtdPdo::fromPdo($rawPdo);
172 $rawPdo->exec("INSERT INTO {$nativeTable} VALUES (1, 'original', 50)");
173 $ztdPdo->exec("INSERT INTO {$shadowTable} VALUES (1, 'original', 50)");
174
175 $rawSkipped = $rawPdo->query("INSERT INTO {$nativeTable} VALUES (1, 'skipped', 95) ON CONFLICT(id) DO UPDATE SET name = EXCLUDED.name WHERE {$nativeTable}.score >= 80 RETURNING id, name, score");
176 $ztdSkipped = $ztdPdo->query("INSERT INTO {$shadowTable} VALUES (1, 'skipped', 95) ON CONFLICT(id) DO UPDATE SET name = EXCLUDED.name WHERE {$shadowTable}.score >= 80 RETURNING id, name, score");
177 self::assertNotFalse($rawSkipped);
178 self::assertNotFalse($ztdSkipped);
179 self::assertSame($rawSkipped->fetchAll(), $ztdSkipped->fetchAll());
180
181 $rawPdo->exec("UPDATE {$nativeTable} SET score = 85 WHERE id = 1");
182 $ztdPdo->exec("UPDATE {$shadowTable} SET score = 85 WHERE id = 1");
183 $rawUpdated = $rawPdo->query("INSERT INTO {$nativeTable} VALUES (1, 'updated', 95) ON CONFLICT(id) DO UPDATE SET name = EXCLUDED.name WHERE {$nativeTable}.score >= 80 RETURNING id, name, score");
184 $ztdUpdated = $ztdPdo->query("INSERT INTO {$shadowTable} VALUES (1, 'updated', 95) ON CONFLICT(id) DO UPDATE SET name = EXCLUDED.name WHERE {$shadowTable}.score >= 80 RETURNING id, name, score");
185 self::assertNotFalse($rawUpdated);
186 self::assertNotFalse($ztdUpdated);
187 self::assertSame($rawUpdated->fetchAll(), $ztdUpdated->fetchAll());
188 } finally {
189 $rawPdo->exec(sprintf('DROP SCHEMA IF EXISTS "%s" CASCADE', $schemaName));
190 }
191 }
192}
193