nala/compras
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

commit14

Browse Source
This commit is contained in:
nala
2026-11-02 20:27:11 +02:00
parent b44378b899
commit badaada043
300 changed files with 17458 additions and 1056 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Unit/WarehouseHttpSourceTest.php
+151
View File
@@ -0,0 +1,151 @@
<?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();
}
}
});
tests/Unit/WarehouseInboxTest.php
+181
View File
@@ -0,0 +1,181 @@
<?php
use Tempo\Integration\WarehouseFeed;
use Tempo\Integration\WarehouseInbox;
use Tempo\Integration\WarehousePull;
require_once __DIR__.'/../Support/InboxFixture.php';
beforeEach(function () {
$this->db = inboxDatabase();
$this->inbox = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'warehouse-reader');
});
test('receiver persists original facts and checkpoint together and safely acknowledges an exact retry', function () {
expect($this->inbox->checkpoint())->toBe(['version' => 0, 'cursor' => null, 'last_event_id' => 0]);
$delivery = inboxEvent();
$receipt = inboxEvent(2, 'receive', $delivery['idempotency_key']);
$page = inboxPage([$delivery, $receipt]);
expect($this->inbox->accept($page, 0, null))->toMatchArray(['inserted' => 2, 'replayed' => false, 'version' => 1])
->and($this->inbox->fact($delivery['idempotency_key']))->toBe($delivery)
->and($this->inbox->fact($receipt['idempotency_key']))->toBe($receipt)
->and($this->inbox->checkpoint())->toBe(['version' => 1, 'cursor' => $page['next_cursor'], 'last_event_id' => 2]);
expect($this->inbox->accept($page, 0, null))->toMatchArray(['inserted' => 0, 'replayed' => true, 'version' => 1])
->and((int) $this->db->query('SELECT COUNT(*) FROM sgc_exchange_facts')->fetchColumn())->toBe(2);
$restarted = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'warehouse-reader');
expect($restarted->checkpoint())->toBe($this->inbox->checkpoint());
});
test('receiver rolls back an entire page and cursor when a later receipt lacks its delivery', function () {
$page = inboxPage([inboxEvent(), inboxEvent(3, 'receive', 'sgc:7:warehouse-event:2')]);
expect(fn () => $this->inbox->accept($page, 0, null))->toThrow(RuntimeException::class)
->and($this->inbox->checkpoint()['version'])->toBe(0)
->and($this->inbox->fact('sgc:7:warehouse-event:1'))->toBeNull();
$fixed = inboxPage([inboxEvent(), inboxEvent(2), inboxEvent(3, 'receive', 'sgc:7:warehouse-event:2')]);
expect($this->inbox->accept($fixed, 0, null)['inserted'])->toBe(3);
});
test('two readers cannot advance the same checkpoint with different pages', function () {
$second = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'warehouse-reader');
$before = $second->checkpoint();
$this->inbox->accept(inboxPage([inboxEvent()], 'cursor-a'), 0, null);
expect(fn () => $second->accept(inboxPage([inboxEvent(2)], 'cursor-b'), $before['version'], $before['cursor']))->toThrow(RuntimeException::class)
->and($this->inbox->checkpoint()['cursor'])->toBe('cursor-a')
->and($this->inbox->fact('sgc:7:warehouse-event:2'))->toBeNull();
});
test('receiver refuses stale pages mismatched request cursors and conflicting event identities', function () {
$this->inbox->accept(inboxPage([inboxEvent()]), 0, null);
expect(fn () => $this->inbox->accept(inboxPage([inboxEvent(2)]), 1, 'another-cursor'))->toThrow(RuntimeException::class)
->and(fn () => $this->inbox->accept(inboxPage([inboxEvent()]), 1, 'synthetic-opaque-cursor'))->toThrow(RuntimeException::class);
$otherStream = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'another-reader');
$conflict = inboxEvent();
$conflict['lines'][0]['quantity_milli'] = 2500;
expect(fn () => $otherStream->accept(inboxPage([inboxSign($conflict)]), 0, null))->toThrow(RuntimeException::class)
->and($otherStream->checkpoint()['version'])->toBe(0)
->and($this->inbox->fact('sgc:7:warehouse-event:1')['lines'][0]['quantity_milli'])->toBe(1500);
});
test('origins destination companies and consumer checkpoints remain separate', function () {
$page = inboxPage([inboxEvent()]);
$this->inbox->accept($page, 0, null);
foreach ([['synthetic-sgc-b', 7, 99], ['synthetic-sgc-a', 7, 100], ['synthetic-sgc-a', 8, 99]] as [$origin, $tenant, $destination]) {
$other = new WarehouseInbox($this->db, $origin, $tenant, $destination, 'warehouse-reader');
expect($other->fact('sgc:7:warehouse-event:1'))->toBeNull()->and($other->checkpoint()['version'])->toBe(0);
}
$otherStream = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'another-reader');
expect($otherStream->accept($page, 0, null)['inserted'])->toBe(0)
->and($otherStream->checkpoint()['version'])->toBe(1);
});
test('receipt must match the exact source request and cannot be acknowledged twice', function () {
$delivery = inboxEvent();
$this->inbox->accept(inboxPage([$delivery], 'cursor-1'), 0, null);
$receipt = inboxEvent(2, 'receive', $delivery['idempotency_key']);
$receipt['request']['destination'] = 'Outro destino';
expect(fn () => $this->inbox->accept(inboxPage([inboxSign($receipt)]), 1, 'cursor-1'))->toThrow(RuntimeException::class);
$this->inbox->accept(inboxPage([inboxEvent(2, 'receive', $delivery['idempotency_key'])], 'cursor-2'), 1, 'cursor-1');
expect(fn () => $this->inbox->accept(inboxPage([inboxEvent(3, 'receive', $delivery['idempotency_key'])]), 2, 'cursor-2'))->toThrow(RuntimeException::class)
->and($this->inbox->checkpoint()['version'])->toBe(2);
});
test('empty polling pages remain resumable and do not invent a physical fact', function () {
$result = $this->inbox->accept(inboxPage([], 'empty-cursor'), 0, null);
expect($result['inserted'])->toBe(0)->and($result['has_more'])->toBeFalse()
->and($this->inbox->checkpoint())->toBe(['version' => 1, 'cursor' => 'empty-cursor', 'last_event_id' => 0]);
expect($this->inbox->accept(inboxPage([inboxEvent()]), 1, 'empty-cursor')['inserted'])->toBe(1);
});
test('malformed or altered feed data is refused before any receiving write', function () {
$cases = [];
foreach (['tenant_id' => 8, 'event_id' => '1', 'event_version' => 0, 'action' => 'authorize', 'occurred_at' => '2026-02-30T10:00:00+00:00', 'idempotency_key' => 'forged'] as $key => $value) {
$event = inboxEvent();
$event[$key] = $value;
$cases[] = inboxPage([inboxSign($event)]);
}
$event = inboxEvent();
$event['lines'][0]['quantity_milli'] = -1000;
$cases[] = inboxPage([inboxSign($event)]);
$event = inboxEvent();
$event['lines'][0]['quantity_milli'] = 1001;
$cases[] = inboxPage([$event]); // hash was not updated
$event = inboxEvent();
$event['lines'][0]['use'] = 'equipment';
$cases[] = inboxPage([inboxSign($event)]);
$cases[] = inboxPage([inboxEvent(2), inboxEvent()]);
$cases[] = inboxPage([inboxEvent(), inboxEvent()]);
$cases[] = inboxPage([], 'cursor', true);
$cases[] = ['schema' => 'unknown', 'items' => [], 'next_cursor' => 'cursor', 'has_more' => false];
foreach ($cases as $page) {
expect(fn () => $this->inbox->accept($page, 0, null))->toThrow(InvalidArgumentException::class);
}
expect((int) $this->db->query('SELECT COUNT(*) FROM sgc_exchange_streams')->fetchColumn())->toBe(0)
->and((int) $this->db->query('SELECT COUNT(*) FROM sgc_exchange_facts')->fetchColumn())->toBe(0);
});
test('new compatible metadata is preserved and incompatible schemas are rejected', function () {
$event = inboxEvent();
$event['extra_metadata'] = ['reference' => 'Origem adicional sintética'];
$event = inboxSign($event);
expect($this->inbox->accept(inboxPage([$event]), 0, null)['inserted'])->toBe(1)
->and($this->inbox->fact($event['idempotency_key'])['extra_metadata'])->toBe($event['extra_metadata']);
$bad = inboxPage([]);
$bad['schema'] = 'sgc.warehouse.fulfilment-feed.v2';
expect(fn () => WarehouseFeed::validate($bad, 7))->toThrow(InvalidArgumentException::class);
});
test('receiver never installs schema implicitly and refuses an externally owned transaction', function () {
$empty = new PDO('sqlite::memory:');
$inbox = new WarehouseInbox($empty, 'origin', 7, 99, 'reader');
expect((int) $empty->query("SELECT COUNT(*) FROM sqlite_master WHERE type='table'")->fetchColumn())->toBe(0);
$this->db->beginTransaction();
try {
expect(fn () => $this->inbox->accept(inboxPage([inboxEvent()]), 0, null))->toThrow(InvalidArgumentException::class)
->and($this->db->inTransaction())->toBeTrue();
} finally {
$this->db->rollBack();
}
});
test('bounded pull continues from the persisted cursor on the next execution', function () {
$seen = [];
$fetch = function (?string $cursor, int $size) use (&$seen): array {
$seen[] = [$cursor, $size];
return $cursor === null ? inboxPage([inboxEvent()], 'cursor-one', true)
: inboxPage([inboxEvent(2, 'receive', 'sgc:7:warehouse-event:1')], 'cursor-two');
};
$pull = new WarehousePull($this->inbox);
expect($pull->run($fetch, 1, 1))->toMatchArray(['pages' => 1, 'inserted' => 1, 'has_more' => true]);
expect($pull->run($fetch, 1, 1))->toMatchArray(['pages' => 1, 'inserted' => 1, 'has_more' => false])
->and($seen)->toBe([[null, 1], ['cursor-one', 1]])
->and($this->inbox->checkpoint()['last_event_id'])->toBe(2);
});
test('transport failure preserves already accepted pages and later resumes the failed page', function () {
$fetch = fn (?string $cursor, int $size) => $cursor === null ? inboxPage([inboxEvent()], 'cursor-one', true) : throw new RuntimeException('Indisponibilidade sintética.');
$pull = new WarehousePull($this->inbox);
expect(fn () => $pull->run($fetch, 5))->toThrow(RuntimeException::class)
->and($this->inbox->checkpoint())->toBe(['version' => 1, 'cursor' => 'cursor-one', 'last_event_id' => 1]);
$retry = function (?string $cursor, int $size): array {
expect($cursor)->toBe('cursor-one');
return inboxPage([inboxEvent(2, 'receive', 'sgc:7:warehouse-event:1')]);
};
expect($pull->run($retry, 5)['inserted'])->toBe(1);
});
test('pull stops at the actual end and rejects invalid bounds before calling a transport', function () {
$calls = 0;
$fetch = function (?string $cursor, int $size) use (&$calls): array {
$calls++;
return inboxPage([]);
};
$pull = new WarehousePull($this->inbox);
expect($pull->run($fetch, 100)['pages'])->toBe(1)->and($calls)->toBe(1);
foreach ([[0, 1], [101, 1], [1, 0], [1, 101]] as [$pages, $size]) {
expect(fn () => $pull->run($fetch, $pages, $size))->toThrow(InvalidArgumentException::class);
}
expect($calls)->toBe(1);
});