297 lines
11 KiB
PHP
297 lines
11 KiB
PHP
<?php
|
|
|
|
namespace Tests\Unit\EcommerceFinance;
|
|
|
|
use Illuminate\Support\Carbon;
|
|
use Illuminate\Support\Facades\Cache;
|
|
use InvalidArgumentException;
|
|
use Modules\EcommerceFinance\Exceptions\UpstreamSnapshotPendingException;
|
|
use Modules\EcommerceFinance\Services\UpstreamOdbcConnection;
|
|
use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
|
|
use Symfony\Component\HttpKernel\Exception\ServiceUnavailableHttpException;
|
|
use Tests\TestCase;
|
|
|
|
class UpstreamQueryCacheTest extends TestCase
|
|
{
|
|
protected function setUp(): void
|
|
{
|
|
parent::setUp();
|
|
|
|
config()->set('upstreams.snapshots.allow_inline_refresh', true);
|
|
config()->set('upstreams.development.direct_read', false);
|
|
|
|
Cache::setDefaultDriver('array');
|
|
Cache::flush();
|
|
Carbon::setTestNow('2026-07-22 12:00:00');
|
|
UpstreamOdbcConnection::reset();
|
|
config()->set('upstreams.snapshots.cache_store', 'array');
|
|
config()->set('upstreams.runtime.max_concurrency', 4);
|
|
config()->set('upstreams.runtime.slot_wait_ms', 0);
|
|
|
|
$this->configureSource('source_a');
|
|
$this->configureSource('source_b');
|
|
}
|
|
|
|
protected function tearDown(): void
|
|
{
|
|
Carbon::setTestNow();
|
|
Cache::flush();
|
|
UpstreamOdbcConnection::reset();
|
|
|
|
parent::tearDown();
|
|
}
|
|
|
|
public function test_identical_scope_and_parameters_are_isolated_by_source(): void
|
|
{
|
|
$factory = app(UpstreamQueryCacheFactory::class);
|
|
$sourceA = $factory->forSource('source_a');
|
|
$sourceB = $factory->forSource('source_b');
|
|
$callsA = 0;
|
|
$callsB = 0;
|
|
|
|
$valueA = $sourceA->remember('orders.list', ['page' => 1], function () use (&$callsA): array {
|
|
$callsA++;
|
|
|
|
return ['source' => 'a'];
|
|
});
|
|
$valueB = $sourceB->remember('orders.list', ['page' => 1], function () use (&$callsB): array {
|
|
$callsB++;
|
|
|
|
return ['source' => 'b'];
|
|
});
|
|
|
|
$this->assertSame('a', $valueA['source']);
|
|
$this->assertSame('b', $valueB['source']);
|
|
$this->assertSame(1, $callsA);
|
|
$this->assertSame(1, $callsB);
|
|
$this->assertNotEmpty(Cache::get('source_a:snapshots:active:v2'));
|
|
$this->assertNotEmpty(Cache::get('source_b:snapshots:active:v2'));
|
|
}
|
|
|
|
public function test_separate_runtime_instances_reuse_the_same_shared_snapshot(): void
|
|
{
|
|
$firstRuntime = app(UpstreamQueryCacheFactory::class)->forSource('source_a');
|
|
$secondRuntime = app(UpstreamQueryCacheFactory::class)->forSource('source_a');
|
|
$calls = 0;
|
|
|
|
$first = $firstRuntime->remember('orders.list', ['page' => 1], function () use (&$calls): array {
|
|
$calls++;
|
|
|
|
return ['total' => 8];
|
|
});
|
|
$second = $secondRuntime->remember('orders.list', ['page' => 1], function () use (&$calls): array {
|
|
$calls++;
|
|
|
|
return ['total' => 99];
|
|
});
|
|
|
|
$this->assertSame(8, $first['total']);
|
|
$this->assertSame($first, $second);
|
|
$this->assertSame(1, $calls);
|
|
}
|
|
|
|
public function test_page_cold_miss_only_registers_demand_without_querying_upstream(): void
|
|
{
|
|
config()->set('upstreams.snapshots.allow_inline_refresh', false);
|
|
$calls = 0;
|
|
|
|
try {
|
|
app(UpstreamQueryCacheFactory::class)
|
|
->forSource('source_a')
|
|
->remember('orders.list', ['page' => 1], function () use (&$calls): array {
|
|
$calls++;
|
|
|
|
return ['total' => 1];
|
|
});
|
|
$this->fail('页面冷查询不应直接访问上游数据库。');
|
|
} catch (UpstreamSnapshotPendingException $exception) {
|
|
$this->assertSame(0, $calls);
|
|
$this->assertStringContainsString('后台正在准备', $exception->getMessage());
|
|
$this->assertSame(15, $exception->getHeaders()['Retry-After']);
|
|
$this->assertNotEmpty(Cache::get('source_a:snapshots:active:v2'));
|
|
}
|
|
}
|
|
|
|
public function test_busy_source_does_not_consume_another_sources_slots(): void
|
|
{
|
|
$factory = app(UpstreamQueryCacheFactory::class);
|
|
$sourceA = $factory->forSource('source_a');
|
|
$sourceB = $factory->forSource('source_b');
|
|
$slotA = Cache::lock('source_a:upstream-slot:0', 30);
|
|
|
|
$this->assertTrue($slotA->get());
|
|
|
|
try {
|
|
$valueB = $sourceB->remember('orders.list', [], fn (): array => ['total' => 1]);
|
|
$this->assertSame(1, $valueB['total']);
|
|
|
|
try {
|
|
$sourceA->remember('orders.list', [], fn (): array => ['total' => 2]);
|
|
$this->fail('来源 A 的唯一并发槽被占用时不应继续访问来源 A。');
|
|
} catch (ServiceUnavailableHttpException) {
|
|
$this->assertTrue(true);
|
|
}
|
|
} finally {
|
|
$slotA->release();
|
|
}
|
|
}
|
|
|
|
public function test_scope_allowlists_are_independent_for_each_source(): void
|
|
{
|
|
$this->configureSource('source_a', ['finance.*']);
|
|
$this->configureSource('source_b', ['orders.*']);
|
|
$factory = app(UpstreamQueryCacheFactory::class);
|
|
$callsA = 0;
|
|
|
|
try {
|
|
$factory->forSource('source_a')->remember('orders.list', [], function () use (&$callsA): array {
|
|
$callsA++;
|
|
|
|
return [];
|
|
});
|
|
$this->fail('来源 A 未开放 orders 范围，不应调用解析器。');
|
|
} catch (ServiceUnavailableHttpException) {
|
|
$this->assertSame(0, $callsA);
|
|
}
|
|
|
|
$value = $factory->forSource('source_b')->remember(
|
|
'orders.list',
|
|
[],
|
|
fn (): array => ['allowed' => true]
|
|
);
|
|
|
|
$this->assertTrue($value['allowed']);
|
|
}
|
|
|
|
public function test_local_direct_read_does_not_wait_for_scope_or_shared_snapshot(): void
|
|
{
|
|
config()->set('upstreams.development.direct_read', true);
|
|
$this->configureSource('source_a', []);
|
|
$calls = 0;
|
|
$cache = app(UpstreamQueryCacheFactory::class)->forSource('source_a');
|
|
|
|
$first = $cache->remember('new-feature.list', [], function () use (&$calls): array {
|
|
return ['call' => ++$calls];
|
|
});
|
|
$second = $cache->remember('new-feature.list', [], function () use (&$calls): array {
|
|
return ['call' => ++$calls];
|
|
});
|
|
|
|
$this->assertSame(1, $first['call']);
|
|
$this->assertSame(2, $second['call']);
|
|
$this->assertSame('direct', $cache->responseHeaders()['X-Caiwu-Data-Status']);
|
|
$this->assertNull(Cache::get('source_a:snapshots:active:v2'));
|
|
}
|
|
|
|
public function test_global_cold_query_limit_protects_the_local_runtime(): void
|
|
{
|
|
config()->set('upstreams.runtime.max_concurrency', 1);
|
|
$globalSlot = Cache::lock('upstream:global-slot:0', 30);
|
|
|
|
$this->assertTrue($globalSlot->get());
|
|
|
|
try {
|
|
$this->expectException(ServiceUnavailableHttpException::class);
|
|
|
|
app(UpstreamQueryCacheFactory::class)
|
|
->forSource('source_b')
|
|
->remember('orders.list', [], fn (): array => ['total' => 1]);
|
|
} finally {
|
|
$globalSlot->release();
|
|
}
|
|
}
|
|
|
|
public function test_unknown_source_is_rejected_before_any_query(): void
|
|
{
|
|
$this->expectException(InvalidArgumentException::class);
|
|
|
|
app(UpstreamQueryCacheFactory::class)->forSource('unknown_source');
|
|
}
|
|
|
|
public function test_disabled_source_does_not_attempt_an_odbc_connection(): void
|
|
{
|
|
config()->set('test_upstream_source_a.enabled', false);
|
|
$this->expectException(ServiceUnavailableHttpException::class);
|
|
|
|
UpstreamOdbcConnection::get('source_a');
|
|
}
|
|
|
|
public function test_disabled_source_does_not_expose_an_existing_snapshot(): void
|
|
{
|
|
$cache = app(UpstreamQueryCacheFactory::class)->forSource('source_a');
|
|
$cache->remember('orders.list', [], fn (): array => ['sensitive' => true]);
|
|
config()->set('test_upstream_source_a.enabled', false);
|
|
|
|
$this->expectException(ServiceUnavailableHttpException::class);
|
|
|
|
$cache->remember('orders.list', [], fn (): array => ['sensitive' => false]);
|
|
}
|
|
|
|
public function test_non_read_only_connection_intent_is_rejected_before_connecting(): void
|
|
{
|
|
config()->set('test_upstream_source_a.connection', [
|
|
'host' => 'example.invalid',
|
|
'port' => 1433,
|
|
'database' => 'finance',
|
|
'username' => 'reader',
|
|
'password' => str_repeat('x', 8),
|
|
'odbc_driver' => 'FreeTDS',
|
|
'tds_version' => '7.4',
|
|
'application_intent' => 'ReadWrite',
|
|
]);
|
|
$this->expectException(ServiceUnavailableHttpException::class);
|
|
$this->expectExceptionMessage('上游数据库只允许只读连接意图。');
|
|
|
|
UpstreamOdbcConnection::create('source_a');
|
|
}
|
|
|
|
public function test_removed_scope_does_not_expose_an_existing_snapshot(): void
|
|
{
|
|
$cache = app(UpstreamQueryCacheFactory::class)->forSource('source_a');
|
|
$cache->remember('orders.list', [], fn (): array => ['sensitive' => true]);
|
|
config()->set('test_upstream_source_a.allowed_scopes', ['finance.*']);
|
|
|
|
$this->expectException(ServiceUnavailableHttpException::class);
|
|
|
|
$cache->remember('orders.list', [], fn (): array => ['sensitive' => false]);
|
|
}
|
|
|
|
private function configureSource(string $source, array $allowedScopes = ['*']): void
|
|
{
|
|
$configKey = 'test_upstream_' . $source;
|
|
config()->set('upstreams.sources.' . $source, [
|
|
'label' => strtoupper($source),
|
|
'config' => $configKey,
|
|
]);
|
|
config()->set($configKey, [
|
|
'enabled' => true,
|
|
'allowed_scopes' => $allowedScopes,
|
|
'connection' => [
|
|
'connect_timeout' => 1,
|
|
'query_timeout' => 1,
|
|
],
|
|
'snapshots' => [
|
|
'ttl_seconds' => 60,
|
|
'summary_ttl_seconds' => 60,
|
|
'options_ttl_seconds' => 3600,
|
|
'stale_ttl_seconds' => 3600,
|
|
'active_ttl_seconds' => 3600,
|
|
'registry_limit' => 300,
|
|
'refresh_limit' => 2,
|
|
'miss_budget_per_window' => 1000,
|
|
'budget_window_seconds' => 60,
|
|
'lock_seconds' => 15,
|
|
'lock_wait_seconds' => 1,
|
|
'max_concurrency' => 1,
|
|
'slot_wait_ms' => 0,
|
|
'failure_cooldown_seconds' => 60,
|
|
'query_failure_cooldown_seconds' => 60,
|
|
'slow_query_ms' => 5000,
|
|
'slow_query_cooldown_seconds' => 60,
|
|
'refresh_time_budget_seconds' => 10,
|
|
'refresh_pause_ms' => 0,
|
|
],
|
|
]);
|
|
}
|
|
}
|