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