152 lines
7.9 KiB
PHP
152 lines
7.9 KiB
PHP
<?php
|
|
|
|
use Symfony\Component\Process\Process;
|
|
use Tempo\Integration\WarehouseHttpSource;
|
|
use Tempo\Integration\WarehouseInbox;
|
|
use Tempo\Integration\WarehousePull;
|
|
use Tempo\Integration\WarehouseTransportFailure;
|
|
|
|
require_once __DIR__.'/../Support/InboxFixture.php';
|
|
|
|
function warehouseHttpFixture(): array
|
|
{
|
|
static $fixture;
|
|
if ($fixture !== null) {
|
|
return $fixture;
|
|
}
|
|
$socket = stream_socket_server('tcp://127.0.0.1:0');
|
|
if ($socket === false) {
|
|
throw new RuntimeException('Não foi possível reservar a porta de teste.');
|
|
}
|
|
$address = stream_socket_get_name($socket, false);
|
|
fclose($socket);
|
|
$server = new Process([PHP_BINARY, '-S', $address, 'tests/Support/warehouse-http-router.php'], dirname(__DIR__, 2), ['WAREHOUSE_HTTP_FIXTURE' => '1'], timeout: null);
|
|
$server->start();
|
|
register_shutdown_function(fn () => $server->stop());
|
|
$deadline = microtime(true) + 5;
|
|
do {
|
|
$probe = @stream_socket_client('tcp://'.$address, $errno, $error, 0.1);
|
|
if (is_resource($probe)) {
|
|
fclose($probe);
|
|
|
|
return $fixture = ['base' => 'http://'.$address, 'server' => $server];
|
|
}
|
|
usleep(20000);
|
|
} while ($server->isRunning() && microtime(true) < $deadline);
|
|
throw new RuntimeException('Servidor descartável HTTP indisponível.');
|
|
}
|
|
|
|
function warehouseRemote(string $scenario = 'ok', ?Closure $credentials = null, int $timeout = 5000, int $maxBytes = 8388608): WarehouseHttpSource
|
|
{
|
|
return new WarehouseHttpSource(warehouseHttpFixture()['base'].'/'.$scenario, 7,
|
|
$credentials ?? fn () => 'SGC_TEST_SESSION=synthetic-authorized', allowLoopbackHttp: true,
|
|
timeoutMs: $timeout, connectTimeoutMs: min(1000, $timeout), maxResponseBytes: $maxBytes);
|
|
}
|
|
|
|
test('HTTP source fixes the endpoint safely encodes query and obtains credentials for each read', function () {
|
|
$calls = 0;
|
|
$source = warehouseRemote(credentials: function () use (&$calls): string {
|
|
$calls++;
|
|
|
|
return 'SGC_TEST_SESSION=synthetic-authorized';
|
|
});
|
|
$cursor = 'opaque&tenant_id=8#? +/=';
|
|
$page = $source($cursor, 7);
|
|
expect($page['schema'])->toBe('sgc.warehouse.fulfilment-feed.v1')
|
|
->and($page['fixture_query'])->toBe(['cursor' => $cursor, 'limit' => '7']);
|
|
$source(null, 1);
|
|
expect($calls)->toBe(2)->and(print_r($source, true))->not->toContain('synthetic-authorized');
|
|
});
|
|
|
|
test('real HTTP pages can be received durably in bounded runs without a stock engine', function () {
|
|
$inbox = new WarehouseInbox(inboxDatabase(), 'synthetic-http-source', 7, 99, 'reader');
|
|
$pull = new WarehousePull($inbox);
|
|
$source = warehouseRemote('paged');
|
|
expect($pull->run($source, 1, 1))->toMatchArray(['inserted' => 1, 'pages' => 1, 'has_more' => true]);
|
|
expect($pull->run($source, 5, 1))->toMatchArray(['inserted' => 1, 'pages' => 1, 'has_more' => false])
|
|
->and($inbox->checkpoint()['last_event_id'])->toBe(2)
|
|
->and($inbox->fact('sgc:7:warehouse-event:2')['delivery_event_key'])->toBe('sgc:7:warehouse-event:1');
|
|
});
|
|
|
|
test('source refuses insecure or ambiguous origins and invalid parameters before credential retrieval', function () {
|
|
$credentials = fn () => 'synthetic-private-secret';
|
|
foreach (['http://example.invalid', 'http://localhost:8005', 'file:///tmp/source', 'https://user:synthetic-private-secret@example.invalid', 'https://example.invalid?token=synthetic-private-secret', 'https://example.invalid#token', 'https://example.invalid/../host', 'https://example.invalid/%2e%2e/host', 'https://example.invalid/a\\b', "https://example.invalid/\nheader"] as $url) {
|
|
expect(fn () => new WarehouseHttpSource($url, 7, $credentials, allowLoopbackHttp: true))->toThrow(InvalidArgumentException::class);
|
|
}
|
|
expect(fn () => new WarehouseHttpSource('http://127.0.0.1:8005', 7, $credentials))->toThrow(InvalidArgumentException::class);
|
|
$calls = 0;
|
|
$source = warehouseRemote(credentials: function () use (&$calls): string {
|
|
$calls++;
|
|
|
|
return 'SGC_TEST_SESSION=synthetic-authorized';
|
|
});
|
|
foreach ([[null, 0], [null, 101], ['', 1], [str_repeat('c', 4097), 1]] as [$cursor, $limit]) {
|
|
expect(fn () => $source($cursor, $limit))->toThrow(InvalidArgumentException::class);
|
|
}
|
|
expect($calls)->toBe(0);
|
|
});
|
|
|
|
test('expired sessions forbidden scopes and redirects return sanitised diagnostics', function (string $scenario, string $category, ?int $status, bool $retryable) {
|
|
try {
|
|
warehouseRemote($scenario)(null, 1);
|
|
$this->fail('Era esperado erro de transporte.');
|
|
} catch (WarehouseTransportFailure $e) {
|
|
expect($e->category)->toBe($category)->and($e->httpStatus)->toBe($status)->and($e->retryable())->toBe($retryable)
|
|
->and($e->getMessage())->not->toContain('synthetic-private')->not->toContain('SGC_TEST_SESSION')
|
|
->and($e->getPrevious())->toBeNull();
|
|
}
|
|
})->with([
|
|
['status_401', 'authentication', 401, false], ['status_419', 'authentication', 419, false],
|
|
['status_403', 'authorization', 403, false], ['redirect', 'redirect', 302, false],
|
|
['status_409', 'cursor', 409, false], ['status_422', 'cursor', 422, false],
|
|
['status_429', 'remote_unavailable', 429, true], ['status_503', 'remote_unavailable', 503, true],
|
|
['status_404', 'http', 404, false],
|
|
]);
|
|
|
|
test('invalid or altered responses never advance the receiving checkpoint', function (string $scenario) {
|
|
$inbox = new WarehouseInbox(inboxDatabase(), 'synthetic-http-source', 7, 99, 'reader');
|
|
expect(fn () => (new WarehousePull($inbox))->run(warehouseRemote($scenario)))->toThrow(WarehouseTransportFailure::class)
|
|
->and($inbox->checkpoint())->toBe(['version' => 0, 'cursor' => null, 'last_event_id' => 0])
|
|
->and($inbox->fact('sgc:7:warehouse-event:1'))->toBeNull();
|
|
})->with(['html', 'json_error', 'wrong_tenant', 'bad_hash']);
|
|
|
|
test('credential failures and header injection expose no supplied secrets', function () {
|
|
foreach ([fn () => throw new RuntimeException('synthetic-private-secret'), fn () => "session=synthetic-private-secret\r\nOther: injected", fn () => '', fn () => ['invalid' => 'synthetic-private-secret']] as $provider) {
|
|
try {
|
|
warehouseRemote(credentials: $provider)(null, 1);
|
|
$this->fail('Era esperada recusa da credencial.');
|
|
} catch (WarehouseTransportFailure $e) {
|
|
expect($e->category)->toBe('credentials')->and($e->getMessage())->not->toContain('synthetic-private-secret')
|
|
->and($e->getPrevious())->toBeNull();
|
|
}
|
|
}
|
|
expect(fn () => warehouseRemote(credentials: fn () => 'SGC_TEST_SESSION=expired')(null, 1))->toThrow(WarehouseTransportFailure::class);
|
|
});
|
|
|
|
test('connection refusal is diagnosed without losing the last receiving checkpoint', function () {
|
|
$socket = stream_socket_server('tcp://127.0.0.1:0');
|
|
$address = stream_socket_get_name($socket, false);
|
|
fclose($socket);
|
|
$source = new WarehouseHttpSource('http://'.$address, 7, fn () => 'SGC_TEST_SESSION=synthetic-authorized', allowLoopbackHttp: true, timeoutMs: 1000, connectTimeoutMs: 200);
|
|
$inbox = new WarehouseInbox(inboxDatabase(), 'synthetic-http-source', 7, 99, 'reader');
|
|
try {
|
|
(new WarehousePull($inbox))->run($source);
|
|
$this->fail('Era esperada recusa da ligação.');
|
|
} catch (WarehouseTransportFailure $e) {
|
|
expect($e->category)->toBe('connection')->and($e->retryable())->toBeTrue()
|
|
->and($inbox->checkpoint()['version'])->toBe(0);
|
|
}
|
|
});
|
|
|
|
|
|
test('oversized responses and timeouts stop the HTTP read without committing a page', function () {
|
|
foreach ([warehouseRemote('oversized', maxBytes: 1024), warehouseRemote('slow', timeout: 200)] as $source) {
|
|
try {
|
|
$source(null, 1);
|
|
$this->fail('Era esperado limite de transporte.');
|
|
} catch (WarehouseTransportFailure $e) {
|
|
expect(in_array($e->category, ['response_size', 'timeout'], true))->toBeTrue();
|
|
}
|
|
}
|
|
});
|