422 lines
19 KiB
PHP
422 lines
19 KiB
PHP
<?php
|
|
|
|
namespace Tests\Unit\EcommerceFinance;
|
|
|
|
use Illuminate\Database\Schema\Blueprint;
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Facades\Schema;
|
|
use Modules\EcommerceFinance\Repositories\GenericBusinessMirrorRepository;
|
|
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
|
|
use Modules\EcommerceFinance\Services\BusinessMirrorSchemaManager;
|
|
use Modules\EcommerceFinance\Services\BusinessMirrorSynchronizer;
|
|
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
|
|
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
|
|
use Tests\TestCase;
|
|
|
|
class BusinessMirrorSynchronizerTest extends TestCase
|
|
{
|
|
private FakeBusinessMirrorQuery $query;
|
|
private BusinessMirrorSynchronizer $synchronizer;
|
|
|
|
protected function setUp(): void
|
|
{
|
|
parent::setUp();
|
|
|
|
config([
|
|
'database.connections.business_mirror' => [
|
|
'driver' => 'sqlite',
|
|
'database' => ':memory:',
|
|
'prefix' => '',
|
|
],
|
|
'upstream_mirrors.connection' => 'business_mirror',
|
|
'upstream_mirrors.datasets' => [
|
|
'test.dbo.records' => [
|
|
'source_key' => 'test',
|
|
'schema' => 'dbo',
|
|
'object' => 'records',
|
|
'target' => 'test_records',
|
|
'keys' => ['Id'],
|
|
'watermark' => 'updateTime',
|
|
],
|
|
],
|
|
'upstream_mirrors.table_sync.overlap_hours' => 24,
|
|
'upstreams.sources.test' => [
|
|
'label' => 'Test source',
|
|
'config' => 'test',
|
|
],
|
|
]);
|
|
DB::purge('business_mirror');
|
|
|
|
Schema::connection('business_mirror')->create('business_mirror_table_states', function (Blueprint $table): void {
|
|
$table->string('dataset_key')->primary();
|
|
$table->string('source_key');
|
|
$table->string('source_object');
|
|
$table->string('target_table');
|
|
$table->string('status')->default('pending');
|
|
$table->string('run_mode')->nullable();
|
|
$table->string('generation')->nullable();
|
|
$table->json('cursor')->nullable();
|
|
$table->json('high_watermark')->nullable();
|
|
$table->dateTime('window_started_at')->nullable();
|
|
$table->dateTime('window_ends_at')->nullable();
|
|
$table->dateTime('initial_completed_at')->nullable();
|
|
$table->dateTime('last_full_succeeded_at')->nullable();
|
|
$table->dateTime('last_attempted_at')->nullable();
|
|
$table->dateTime('last_succeeded_at')->nullable();
|
|
$table->unsignedBigInteger('last_upserted_rows')->default(0);
|
|
$table->unsignedBigInteger('mirror_row_count')->default(0);
|
|
$table->string('last_error_code')->nullable();
|
|
$table->timestamps();
|
|
});
|
|
|
|
DB::table('upstream_schema_columns')->insert([
|
|
$this->column('Id', 1, 'int', null),
|
|
$this->column('updateTime', 2, 'datetime2', null),
|
|
$this->column('displayName', 3, 'nvarchar', 120),
|
|
$this->column('businessDate', 4, 'date', null),
|
|
]);
|
|
|
|
$this->query = new FakeBusinessMirrorQuery();
|
|
$factory = new class($this->query) extends UpstreamReadOnlyQueryFactory {
|
|
public function __construct(private readonly UpstreamReadOnlyQuery $query) {}
|
|
public function forSource(string $source): UpstreamReadOnlyQuery { return $this->query; }
|
|
};
|
|
$registry = new BusinessMirrorDatasetRegistry();
|
|
$this->synchronizer = new BusinessMirrorSynchronizer(
|
|
$registry,
|
|
new BusinessMirrorSchemaManager(),
|
|
new GenericBusinessMirrorRepository(app('db')),
|
|
$factory,
|
|
);
|
|
}
|
|
|
|
public function test_full_sync_materializes_complete_table_and_marks_state_ready(): void
|
|
{
|
|
$this->query->batches = [[
|
|
['Id' => 1, 'updateTime' => '2026-08-14 10:00:00.1234567', 'displayName' => '一号', 'businessDate' => '2026-08-14'],
|
|
['Id' => 2, 'updateTime' => '2026-08-14 11:00:00.0000000', 'displayName' => '二号', 'businessDate' => '2026-08-15'],
|
|
]];
|
|
|
|
$result = $this->synchronizer->sync('test.dbo.records', true, 10, 1);
|
|
|
|
$this->assertSame('succeeded', $result['status']);
|
|
$this->assertSame(2, DB::connection('business_mirror')->table('test_records')->count());
|
|
$this->assertSame(
|
|
'2026-08-14 10:00:00.123456',
|
|
DB::connection('business_mirror')->table('test_records')->where('Id', 1)->value('updateTime'),
|
|
);
|
|
$state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
|
|
$this->assertSame('succeeded', $state->status);
|
|
$this->assertNotNull($state->initial_completed_at);
|
|
$this->assertSame(2, (int) $state->mirror_row_count);
|
|
$this->assertStringContainsString('SELECT TOP (10)', $this->query->calls[0]['sql']);
|
|
$this->assertStringContainsString('CONVERT(varchar(27), [updateTime], 126) AS [updateTime]', $this->query->calls[0]['sql']);
|
|
$this->assertStringContainsString('CONVERT(varchar(10), [businessDate], 23) AS [businessDate]', $this->query->calls[0]['sql']);
|
|
$this->assertStringContainsString('ORDER BY [Id]', $this->query->calls[0]['sql']);
|
|
}
|
|
|
|
public function test_incremental_sync_uses_overlap_window_and_updates_existing_key(): void
|
|
{
|
|
$this->query->batches = [[
|
|
['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '旧值'],
|
|
]];
|
|
$this->synchronizer->sync('test.dbo.records', true, 10, 1);
|
|
|
|
$this->query->calls = [];
|
|
$this->query->batches = [[
|
|
['Id' => 1, 'updateTime' => '2026-08-14 12:00:00', 'displayName' => '新值'],
|
|
['Id' => 2, 'updateTime' => '2026-08-14 12:30:00', 'displayName' => '新增'],
|
|
]];
|
|
|
|
$result = $this->synchronizer->sync('test.dbo.records', false, 10, 1);
|
|
|
|
$this->assertSame('succeeded', $result['status']);
|
|
$this->assertSame(2, DB::connection('business_mirror')->table('test_records')->count());
|
|
$this->assertSame('新值', DB::connection('business_mirror')->table('test_records')->where('Id', 1)->value('displayName'));
|
|
$this->assertStringContainsString('[updateTime] >= ? AND [updateTime] < ?', $this->query->calls[0]['sql']);
|
|
$this->assertStringContainsString('ORDER BY [Id]', $this->query->calls[0]['sql']);
|
|
$this->assertStringNotContainsString('ORDER BY [updateTime]', $this->query->calls[0]['sql']);
|
|
$this->assertCount(2, $this->query->calls[0]['bindings']);
|
|
foreach ($this->query->calls[0]['bindings'] as $binding) {
|
|
$this->assertMatchesRegularExpression('/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}$/', $binding);
|
|
$this->assertStringNotContainsString(' ', $binding);
|
|
}
|
|
}
|
|
|
|
public function test_incremental_keyset_keeps_watermark_out_of_pagination_cursor(): void
|
|
{
|
|
$this->query->batches = [[
|
|
['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '初始'],
|
|
]];
|
|
$this->synchronizer->sync('test.dbo.records', true, 10, 1);
|
|
|
|
$this->query->calls = [];
|
|
$this->query->batches = [[
|
|
['Id' => 1, 'updateTime' => '2026-08-14 12:00:00', 'displayName' => '更新一'],
|
|
['Id' => 2, 'updateTime' => '2026-08-14 12:30:00', 'displayName' => '更新二'],
|
|
], []];
|
|
|
|
$result = $this->synchronizer->sync('test.dbo.records', false, 2, 0);
|
|
|
|
$this->assertSame('succeeded', $result['status']);
|
|
$this->assertCount(2, $this->query->calls);
|
|
$this->assertStringContainsString('[updateTime] >= ? AND [updateTime] < ?', $this->query->calls[1]['sql']);
|
|
$this->assertStringContainsString('[Id] > ?', $this->query->calls[1]['sql']);
|
|
$this->assertStringContainsString('ORDER BY [Id]', $this->query->calls[1]['sql']);
|
|
$this->assertStringNotContainsString('[updateTime] > ?', $this->query->calls[1]['sql']);
|
|
$this->assertCount(3, $this->query->calls[1]['bindings']);
|
|
$this->assertSame(2, $this->query->calls[1]['bindings'][2]);
|
|
}
|
|
|
|
public function test_incremental_sync_refuses_to_guess_before_first_full_sync(): void
|
|
{
|
|
$result = $this->synchronizer->sync('test.dbo.records', false, 10, 1);
|
|
|
|
$this->assertSame('requires_full_sync', $result['status']);
|
|
$this->assertSame([], $this->query->calls);
|
|
}
|
|
|
|
public function test_supervised_full_sync_caps_explicit_batch_size_at_five_thousand(): void
|
|
{
|
|
$this->query->batches = [[
|
|
['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '一号', 'businessDate' => '2026-08-14'],
|
|
]];
|
|
|
|
$this->synchronizer->sync('test.dbo.records', true, 50000, 1);
|
|
|
|
$this->assertStringContainsString('SELECT TOP (5000)', $this->query->calls[0]['sql']);
|
|
}
|
|
|
|
public function test_full_sync_sanitizes_invalid_utf8_text_without_changing_valid_text(): void
|
|
{
|
|
$this->query->batches = [[
|
|
[
|
|
'Id' => 1,
|
|
'updateTime' => '2026-08-14 10:00:00',
|
|
'displayName' => hex2bin('e80044495900'),
|
|
],
|
|
[
|
|
'Id' => 2,
|
|
'updateTime' => '2026-08-14 11:00:00',
|
|
'displayName' => '合法名称',
|
|
],
|
|
]];
|
|
|
|
$this->synchronizer->sync('test.dbo.records', true, 10, 1);
|
|
|
|
$rows = DB::connection('business_mirror')->table('test_records')->orderBy('Id')->pluck('displayName')->all();
|
|
$this->assertStringEndsWith('DIY', $rows[0]);
|
|
$this->assertNotSame(hex2bin('e80044495900'), $rows[0]);
|
|
$this->assertSame('合法名称', $rows[1]);
|
|
$this->assertTrue(mb_check_encoding($rows[0], 'UTF-8'));
|
|
$this->assertStringNotContainsString("\0", $rows[0]);
|
|
}
|
|
|
|
public function test_wide_source_schema_stores_only_short_strings_inline(): void
|
|
{
|
|
$datasets = (array) config('upstream_mirrors.datasets', []);
|
|
$datasets['test.dbo.wide_records'] = [
|
|
'source_key' => 'test',
|
|
'schema' => 'dbo',
|
|
'object' => 'wide_records',
|
|
'target' => 'test_wide_records',
|
|
'keys' => ['Id'],
|
|
'watermark' => 'updateTime',
|
|
];
|
|
config(['upstream_mirrors.datasets' => $datasets]);
|
|
|
|
$columns = [
|
|
$this->column('Id', 1, 'int', null),
|
|
$this->column('updateTime', 2, 'datetime2', null),
|
|
$this->column('shortCode', 3, 'nvarchar', 32),
|
|
$this->column('longLabel', 4, 'nvarchar', 120),
|
|
];
|
|
DB::table('upstream_schema_columns')->insert(array_map(
|
|
fn (array $column): array => [...$column, 'object_name' => 'wide_records'],
|
|
$columns,
|
|
));
|
|
|
|
$dataset = (new BusinessMirrorDatasetRegistry())->get('test.dbo.wide_records');
|
|
(new BusinessMirrorSchemaManager())->ensure($dataset);
|
|
|
|
$types = DB::connection('business_mirror')->select('PRAGMA table_info(test_wide_records)');
|
|
$types = collect($types)->pluck('type', 'name')->map(fn (string $type): string => strtolower($type));
|
|
|
|
$this->assertSame('varchar', $types['shortCode']);
|
|
$this->assertSame('text', $types['longLabel']);
|
|
}
|
|
|
|
public function test_small_payment_work_order_dataset_accepts_only_supported_indexes(): void
|
|
{
|
|
$datasets = (array) config('upstream_mirrors.datasets', []);
|
|
$datasets['jiansun.dbo.jst_workOrderInfo'] = [
|
|
'source_key' => 'jiansun',
|
|
'schema' => 'dbo',
|
|
'object' => 'jst_workOrderInfo',
|
|
'target' => 'test_small_payment_work_orders',
|
|
'keys' => ['Id'],
|
|
'watermark' => 'updateTime',
|
|
'columns' => [
|
|
'Id', 'woNo', 'woTypeId', 'title', 'content', 'status', 'grade',
|
|
'urging', 'createby', 'executors', 'beginTime', 'endTime',
|
|
'completer', 'completeTime', 'updated', 'created',
|
|
'businessFieldList', 'createTime', 'updateTime',
|
|
],
|
|
'indexes' => [
|
|
['woTypeId', 'woNo', 'Id'],
|
|
['created', 'Id'],
|
|
],
|
|
];
|
|
config(['upstream_mirrors.datasets' => $datasets]);
|
|
|
|
$columns = [
|
|
['name' => 'Id', 'type' => 'int', 'length' => null],
|
|
['name' => 'woNo', 'type' => 'nvarchar', 'length' => 100],
|
|
['name' => 'woTypeId', 'type' => 'nvarchar', 'length' => 100],
|
|
['name' => 'urging', 'type' => 'int', 'length' => null],
|
|
['name' => 'created', 'type' => 'datetime2', 'length' => null],
|
|
['name' => 'updateTime', 'type' => 'datetime2', 'length' => null],
|
|
];
|
|
$selected = $datasets['jiansun.dbo.jst_workOrderInfo']['columns'];
|
|
foreach ($selected as $position => $name) {
|
|
$definition = collect($columns)->firstWhere('name', $name) ?? [
|
|
'name' => $name,
|
|
'type' => 'nvarchar',
|
|
'length' => 120,
|
|
];
|
|
DB::table('upstream_schema_columns')->insert(array_replace($this->column(
|
|
$definition['name'],
|
|
$position + 1,
|
|
$definition['type'],
|
|
$definition['length'],
|
|
), [
|
|
'source_key' => 'jiansun',
|
|
'schema_name' => 'dbo',
|
|
'object_name' => 'jst_workOrderInfo',
|
|
]));
|
|
}
|
|
|
|
$dataset = (new BusinessMirrorDatasetRegistry())->get('jiansun.dbo.jst_workOrderInfo');
|
|
$this->assertSame([
|
|
['woTypeId', 'woNo', 'Id'],
|
|
['created', 'Id'],
|
|
], $dataset->indexDefinitions());
|
|
|
|
(new BusinessMirrorSchemaManager())->ensure($dataset);
|
|
|
|
$indexes = collect(Schema::connection('business_mirror')->getIndexes($dataset->targetTable()))
|
|
->map(static fn (array $index): array => array_map('strtolower', (array) ($index['columns'] ?? [])))
|
|
->all();
|
|
$types = collect(DB::connection('business_mirror')->select('PRAGMA table_info(test_small_payment_work_orders)'))
|
|
->pluck('type', 'name')
|
|
->map(static fn (string $type): string => strtolower($type));
|
|
$this->assertSame('varchar', $types['woNo']);
|
|
$this->assertSame('varchar', $types['woTypeId']);
|
|
$this->assertSame('text', $types['businessFieldList']);
|
|
$this->assertContains(['wotypeid', 'wono', 'id'], $indexes);
|
|
$this->assertContains(['created', 'id'], $indexes);
|
|
}
|
|
|
|
public function test_declared_255_character_lookup_preserves_the_value_and_builds_an_index(): void
|
|
{
|
|
DB::table('upstream_schema_columns')->where('source_key', 'test')->where('column_name', 'displayName')
|
|
->update(['max_length' => 255]);
|
|
$definitions = config('upstream_mirrors.datasets');
|
|
$definitions['test.dbo.records']['indexes'] = [['displayName', 'Id']];
|
|
config(['upstream_mirrors.datasets' => $definitions]);
|
|
$dataset = (new BusinessMirrorDatasetRegistry())->get('test.dbo.records');
|
|
(new BusinessMirrorSchemaManager())->ensure($dataset);
|
|
$indexes = collect(Schema::connection('business_mirror')->getIndexes($dataset->targetTable()))
|
|
->pluck('columns')->all();
|
|
$this->assertContains(['displayName', 'Id'], $indexes);
|
|
$value = str_repeat('x', 255);
|
|
$this->query->batches = [[['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => $value]]];
|
|
$this->synchronizer->sync('test.dbo.records', true);
|
|
$this->assertSame($value, DB::connection('business_mirror')->table($dataset->targetTable())->value('displayName'));
|
|
}
|
|
|
|
public function test_empty_full_read_does_not_delete_previous_mirror_rows(): void
|
|
{
|
|
$this->query->batches = [[
|
|
['Id' => 1, 'updateTime' => '2026-08-14 10:00:00', 'displayName' => '保留'],
|
|
]];
|
|
$this->synchronizer->sync('test.dbo.records', true, 10, 1);
|
|
|
|
$this->query->batches = [[]];
|
|
|
|
try {
|
|
$this->synchronizer->sync('test.dbo.records', true, 10, 1);
|
|
$this->fail('空的上游响应不应覆盖已有镜像。');
|
|
} catch (\RuntimeException $exception) {
|
|
$this->assertSame('business_mirror_unexpected_empty_source', $exception->getMessage());
|
|
}
|
|
|
|
$this->assertSame(1, DB::connection('business_mirror')->table('test_records')->count());
|
|
$this->assertSame('保留', DB::connection('business_mirror')->table('test_records')->value('displayName'));
|
|
$this->assertSame(
|
|
'failed',
|
|
DB::connection('business_mirror')->table('business_mirror_table_states')->value('status'),
|
|
);
|
|
}
|
|
|
|
public function test_project_dataset_registry_has_unique_safe_targets(): void
|
|
{
|
|
$configured = require base_path('config/upstream_mirrors.php');
|
|
config(['upstream_mirrors.datasets' => $configured['datasets']]);
|
|
$datasets = (new BusinessMirrorDatasetRegistry())->all();
|
|
$targets = array_map(fn ($dataset): string => $dataset->targetTable(), $datasets);
|
|
|
|
$this->assertGreaterThanOrEqual(38, count($datasets));
|
|
$this->assertCount(count($targets), array_unique($targets));
|
|
$this->assertContains('operations_replica', array_map(fn ($dataset): string => $dataset->sourceKey(), $datasets));
|
|
$this->assertContains('jiansun', array_map(fn ($dataset): string => $dataset->sourceKey(), $datasets));
|
|
$this->assertNotContains('Bulk_Price_Limit', array_map(fn ($dataset): string => $dataset->sourceObject(), $datasets));
|
|
$this->assertContains('fys_Ticketing_install_view', array_map(fn ($dataset): string => $dataset->sourceObject(), $datasets));
|
|
$this->assertContains('fys_Ticketing_install_item', array_map(fn ($dataset): string => $dataset->sourceObject(), $datasets));
|
|
}
|
|
|
|
/** @return array<string,mixed> */
|
|
private function column(string $name, int $position, string $type, ?int $length): array
|
|
{
|
|
$now = now();
|
|
|
|
return [
|
|
'source_key' => 'test',
|
|
'schema_name' => 'dbo',
|
|
'object_name' => 'records',
|
|
'object_type' => 'table',
|
|
'column_name' => $name,
|
|
'ordinal_position' => $position,
|
|
'data_type' => $type,
|
|
'max_length' => $length,
|
|
'numeric_precision' => null,
|
|
'numeric_scale' => null,
|
|
'is_nullable' => true,
|
|
'synchronized_at' => $now,
|
|
'created_at' => $now,
|
|
'updated_at' => $now,
|
|
];
|
|
}
|
|
}
|
|
|
|
class FakeBusinessMirrorQuery extends UpstreamReadOnlyQuery
|
|
{
|
|
/** @var list<list<array<string,mixed>>> */
|
|
public array $batches = [];
|
|
|
|
/** @var list<array{sql:string,bindings:array}> */
|
|
public array $calls = [];
|
|
|
|
public function __construct()
|
|
{
|
|
parent::__construct('test');
|
|
}
|
|
|
|
public function select(string $sql, array $bindings = []): array
|
|
{
|
|
$this->calls[] = compact('sql', 'bindings');
|
|
|
|
return array_shift($this->batches) ?? [];
|
|
}
|
|
}
|