packages/ztd-query-pdo-adapter/tests/Integration/PostgreSql/CopyTest.php
1<?php
2
3declare(strict_types=1);
4
5namespace Tests\Integration\PostgreSql;
6
7use Container\PostgreSql16Container;
8use PDO;
9use PHPUnit\Framework\Attributes\CoversNothing;
10use PHPUnit\Framework\Attributes\Large;
11use PHPUnit\Framework\Attributes\TestWith;
12use PHPUnit\Framework\TestCase;
13use ZtdQuery\Adapter\Pdo\ZtdPdo;
14use ZtdQuery\Adapter\Pdo\ZtdPdoException;
15use ZtdQuery\Config\UnsupportedSqlBehavior;
16use ZtdQuery\Config\ZtdConfig;
17use ZtdQuery\Connection\Exception\DatabaseException;
18use ZtdQuery\Exception\UnsupportedSqlException;
19
20/**
21 * @requires extension pdo_pgsql
22 * @group integration
23 * @group postgres
24 */
25#[CoversNothing]
26#[Large]
27final class CopyTest extends TestCase
28{
29 #[TestWith(['FROM STDIN'])]
30 #[TestWith(['TO STDOUT'])]
31 public function testExecDelegatesUnsupportedCopyToTheSession(string $direction): void
32 {
33 $containerInstance = \Testcontainers\Testcontainers::run(PostgreSql16Container::class);
34 /** @var PDO $pdo */
35 $pdo = new PDO(
36 sprintf('pgsql:host=%s;port=%d;dbname=test', str_replace('localhost', '127.0.0.1', $containerInstance->getHost()), $containerInstance->getMappedPort(5432)),
37 'test',
38 'test',
39 [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC],
40 );
41
42 $schemaName = 'ztd_' . bin2hex(random_bytes(8));
43 $pdo->exec(sprintf('CREATE SCHEMA "%s"', $schemaName));
44 $pdo->exec(sprintf('SET search_path TO "%s"', $schemaName));
45
46
47 try {
48 $pdo->exec('CREATE TABLE copy_target (id INTEGER PRIMARY KEY)');
49 $pdo->exec('INSERT INTO copy_target VALUES (1)');
50 $ztdPdo = ZtdPdo::fromPdo($pdo);
51 $ztdPdo->exec('INSERT INTO copy_target VALUES (2)');
52 $sql = 'COPY copy_target ' . $direction;
53
54 try {
55 $ztdPdo->exec($sql);
56 self::fail('Expected the session to reject unsupported COPY SQL.');
57 } catch (ZtdPdoException $exception) {
58 $databaseException = $exception->getPrevious();
59 self::assertInstanceOf(DatabaseException::class, $databaseException);
60 $refusal = $databaseException->getPrevious();
61 self::assertInstanceOf(UnsupportedSqlException::class, $refusal);
62 self::assertSame($sql, $refusal->getSql());
63 }
64
65 $physical = $pdo->query('SELECT id FROM copy_target');
66 $shadow = $ztdPdo->query('SELECT id FROM copy_target');
67 self::assertNotFalse($physical);
68 self::assertNotFalse($shadow);
69 self::assertSame([1], $physical->fetchAll(PDO::FETCH_COLUMN));
70 self::assertSame([2], $shadow->fetchAll(PDO::FETCH_COLUMN));
71 } finally {
72 $pdo->exec(sprintf('DROP SCHEMA IF EXISTS "%s" CASCADE', $schemaName));
73 }
74 }
75
76 #[TestWith(['FROM STDIN'])]
77 #[TestWith(['TO STDOUT'])]
78 public function testQueryDelegatesUnsupportedCopyToTheSession(string $direction): void
79 {
80 $containerInstance = \Testcontainers\Testcontainers::run(PostgreSql16Container::class);
81 /** @var PDO $pdo */
82 $pdo = new PDO(
83 sprintf('pgsql:host=%s;port=%d;dbname=test', str_replace('localhost', '127.0.0.1', $containerInstance->getHost()), $containerInstance->getMappedPort(5432)),
84 'test',
85 'test',
86 [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC],
87 );
88
89 $schemaName = 'ztd_' . bin2hex(random_bytes(8));
90 $pdo->exec(sprintf('CREATE SCHEMA "%s"', $schemaName));
91 $pdo->exec(sprintf('SET search_path TO "%s"', $schemaName));
92
93
94 try {
95 $pdo->exec('CREATE TABLE copy_target (id INTEGER PRIMARY KEY)');
96 $pdo->exec('INSERT INTO copy_target VALUES (1)');
97 $ztdPdo = ZtdPdo::fromPdo($pdo);
98 $ztdPdo->exec('INSERT INTO copy_target VALUES (2)');
99 $sql = 'COPY copy_target ' . $direction;
100
101 try {
102 $ztdPdo->query($sql);
103 self::fail('Expected the session to reject unsupported COPY SQL.');
104 } catch (ZtdPdoException $exception) {
105 $databaseException = $exception->getPrevious();
106 self::assertInstanceOf(DatabaseException::class, $databaseException);
107 $refusal = $databaseException->getPrevious();
108 self::assertInstanceOf(UnsupportedSqlException::class, $refusal);
109 self::assertSame($sql, $refusal->getSql());
110 }
111
112 $physical = $pdo->query('SELECT id FROM copy_target');
113 $shadow = $ztdPdo->query('SELECT id FROM copy_target');
114 self::assertNotFalse($physical);
115 self::assertNotFalse($shadow);
116 self::assertSame([1], $physical->fetchAll(PDO::FETCH_COLUMN));
117 self::assertSame([2], $shadow->fetchAll(PDO::FETCH_COLUMN));
118 } finally {
119 $pdo->exec(sprintf('DROP SCHEMA IF EXISTS "%s" CASCADE', $schemaName));
120 }
121 }
122
123 #[TestWith(['FROM STDIN'])]
124 #[TestWith(['TO STDOUT'])]
125 public function testPrepareDelegatesUnsupportedCopyToTheSession(string $direction): void
126 {
127 $containerInstance = \Testcontainers\Testcontainers::run(PostgreSql16Container::class);
128 /** @var PDO $pdo */
129 $pdo = new PDO(
130 sprintf('pgsql:host=%s;port=%d;dbname=test', str_replace('localhost', '127.0.0.1', $containerInstance->getHost()), $containerInstance->getMappedPort(5432)),
131 'test',
132 'test',
133 [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC],
134 );
135
136 $schemaName = 'ztd_' . bin2hex(random_bytes(8));
137 $pdo->exec(sprintf('CREATE SCHEMA "%s"', $schemaName));
138 $pdo->exec(sprintf('SET search_path TO "%s"', $schemaName));
139
140
141 try {
142 $pdo->exec('CREATE TABLE copy_target (id INTEGER PRIMARY KEY)');
143 $pdo->exec('INSERT INTO copy_target VALUES (1)');
144 $ztdPdo = ZtdPdo::fromPdo($pdo);
145 $ztdPdo->exec('INSERT INTO copy_target VALUES (2)');
146 $sql = 'COPY copy_target ' . $direction;
147
148 try {
149 $ztdPdo->prepare($sql);
150 self::fail('Expected the session to reject unsupported COPY SQL.');
151 } catch (ZtdPdoException $exception) {
152 $databaseException = $exception->getPrevious();
153 self::assertInstanceOf(DatabaseException::class, $databaseException);
154 $refusal = $databaseException->getPrevious();
155 self::assertInstanceOf(UnsupportedSqlException::class, $refusal);
156 self::assertSame($sql, $refusal->getSql());
157 }
158
159 $physical = $pdo->query('SELECT id FROM copy_target');
160 $shadow = $ztdPdo->query('SELECT id FROM copy_target');
161 self::assertNotFalse($physical);
162 self::assertNotFalse($shadow);
163 self::assertSame([1], $physical->fetchAll(PDO::FETCH_COLUMN));
164 self::assertSame([2], $shadow->fetchAll(PDO::FETCH_COLUMN));
165 } finally {
166 $pdo->exec(sprintf('DROP SCHEMA IF EXISTS "%s" CASCADE', $schemaName));
167 }
168 }
169
170 #[TestWith(['FROM STDIN'])]
171 #[TestWith(['TO STDOUT'])]
172 public function testStandardPdoMethodsHonorTheSessionsIgnorePolicyForCopy(string $direction): void
173 {
174 $containerInstance = \Testcontainers\Testcontainers::run(PostgreSql16Container::class);
175 /** @var PDO $pdo */
176 $pdo = new PDO(
177 sprintf('pgsql:host=%s;port=%d;dbname=test', str_replace('localhost', '127.0.0.1', $containerInstance->getHost()), $containerInstance->getMappedPort(5432)),
178 'test',
179 'test',
180 [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC],
181 );
182
183 $schemaName = 'ztd_' . bin2hex(random_bytes(8));
184 $pdo->exec(sprintf('CREATE SCHEMA "%s"', $schemaName));
185 $pdo->exec(sprintf('SET search_path TO "%s"', $schemaName));
186
187
188 try {
189 $pdo->exec('CREATE TABLE copy_target (id INTEGER PRIMARY KEY)');
190 $pdo->exec('INSERT INTO copy_target VALUES (1)');
191 $ztdPdo = ZtdPdo::fromPdo($pdo, new ZtdConfig(unsupportedBehavior: UnsupportedSqlBehavior::Ignore));
192 $ztdPdo->exec('INSERT INTO copy_target VALUES (2)');
193
194 $sql = 'COPY copy_target ' . $direction;
195 self::assertSame(0, $ztdPdo->exec($sql));
196 self::assertFalse($ztdPdo->query($sql));
197 $statement = $ztdPdo->prepare($sql);
198 self::assertNotFalse($statement);
199 self::assertFalse($statement->execute());
200
201 $physical = $pdo->query('SELECT id FROM copy_target');
202 $shadow = $ztdPdo->query('SELECT id FROM copy_target');
203 self::assertNotFalse($physical);
204 self::assertNotFalse($shadow);
205 self::assertSame([1], $physical->fetchAll(PDO::FETCH_COLUMN));
206 self::assertSame([2], $shadow->fetchAll(PDO::FETCH_COLUMN));
207 } finally {
208 $pdo->exec(sprintf('DROP SCHEMA IF EXISTS "%s" CASCADE', $schemaName));
209 }
210 }
211}
212