limeng
43390be327
1) 钉钉「线上代发业务报销」有 23 张已通过审批单的商品明细表格不叫“收款费用明细”（新版模板表格名为空、旧单叫“费用明细”）， 同步时整单只剩一行空明细。按名称找不到时改取行内含“商品ID”列的明细表格；料号兼容新版“商品编码”。上线后对该数据集全量重同步。 2) 新增报销成本核查：按聚水潭订单由本行商品编码拆出的子商品，取普通商品资料成本价×发货数量求和为系统成本， 与代发订单金额比较，高出 0.01 元以上判“多报销”，另有“成本缺失”“无发货明细”；页面新增筛选、计数、多报销金额汇总及四列，导出同步。 涉及：金额核查展示（只读）、镜像全量重同步；不涉及迁移、权限和上游写入。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VANWYsNUbu2piMtCKymoyd
362 lines
17 KiB
PHP
362 lines
17 KiB
PHP
<?php
|
||
|
||
namespace Tests\Unit\EcommerceFinance;
|
||
|
||
use Illuminate\Database\Schema\Blueprint;
|
||
use Illuminate\Support\Carbon;
|
||
use Illuminate\Support\Facades\DB;
|
||
use Illuminate\Support\Facades\Schema;
|
||
use Modules\EcommerceFinance\Services\DingPayableWorkflowMirrorSynchronizer;
|
||
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
|
||
use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
|
||
use Tests\TestCase;
|
||
|
||
class DingPayableWorkflowMirrorSynchronizerTest extends TestCase
|
||
{
|
||
private FakeDingPayableWorkflowQuery $query;
|
||
|
||
private DingPayableWorkflowMirrorSynchronizer $synchronizer;
|
||
|
||
protected function setUp(): void
|
||
{
|
||
parent::setUp();
|
||
|
||
config([
|
||
'database.connections.business_mirror' => [
|
||
'driver' => 'sqlite',
|
||
'database' => ':memory:',
|
||
'prefix' => '',
|
||
],
|
||
'upstream_mirrors.connection' => 'business_mirror',
|
||
'upstream_mirrors.table_sync.overlap_hours' => 48,
|
||
'upstream_mirrors.ding_payable_workflows' => [
|
||
'dataset_key' => 'operations_replica.rpt.ding_payable_workflows',
|
||
'source_key' => 'operations_replica',
|
||
'process_code' => 'PROC-TEST',
|
||
'target_table' => 'ding_payable_workflow_mirror',
|
||
'batch_size' => 100,
|
||
'max_batches_per_run' => 2,
|
||
],
|
||
]);
|
||
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();
|
||
});
|
||
|
||
$this->query = new FakeDingPayableWorkflowQuery();
|
||
$factory = new class($this->query) extends UpstreamReadOnlyQueryFactory {
|
||
public function __construct(private readonly UpstreamReadOnlyQuery $query) {}
|
||
|
||
public function forSource(string $source): UpstreamReadOnlyQuery
|
||
{
|
||
return $this->query;
|
||
}
|
||
};
|
||
$this->synchronizer = new DingPayableWorkflowMirrorSynchronizer($factory);
|
||
}
|
||
|
||
public function test_full_sync_resumes_with_update_time_and_process_id_tie_breaker(): void
|
||
{
|
||
$sameUpdateAt = '2026-08-29 10:00:00.1234567';
|
||
$this->query->workflowBatches = [[
|
||
$this->workflow('A-process', $sameUpdateAt),
|
||
$this->workflow('B-process', $sameUpdateAt),
|
||
]];
|
||
|
||
$result = $this->synchronizer->sync(true, 2, 1);
|
||
|
||
$this->assertSame('running', $result['status']);
|
||
$this->assertStringContainsString('ORDER BY COALESCE([updateAt], [createAt]), [processInstanceId]', $this->query->calls[0]['sql']);
|
||
$state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
|
||
$this->assertSame([
|
||
'updateAt' => $sameUpdateAt,
|
||
'processInstanceId' => 'B-process',
|
||
], json_decode((string) $state->cursor, true));
|
||
|
||
$this->query->calls = [];
|
||
$this->query->workflowBatches = [[
|
||
$this->workflow('C-process', $sameUpdateAt),
|
||
]];
|
||
|
||
$result = $this->synchronizer->sync(true, 2, 0);
|
||
|
||
$this->assertSame('succeeded', $result['status']);
|
||
$this->assertStringContainsString(
|
||
'COALESCE([updateAt], [createAt]) > ? OR (COALESCE([updateAt], [createAt]) = ? AND [processInstanceId] > ?',
|
||
$this->query->calls[0]['sql'],
|
||
);
|
||
$this->assertSame([$sameUpdateAt, $sameUpdateAt, 'B-process'], array_slice($this->query->calls[0]['bindings'], -3));
|
||
$this->assertSame(3, DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->count());
|
||
}
|
||
|
||
public function test_incremental_sync_uses_overlap_window_and_keeps_fixed_process_code(): void
|
||
{
|
||
$this->query->workflowBatches = [[
|
||
$this->workflow('existing-process', now()->subHour()->format('Y-m-d H:i:s.u')),
|
||
]];
|
||
$this->synchronizer->sync(true, 10, 0);
|
||
|
||
$this->query->calls = [];
|
||
$this->query->workflowBatches = [[
|
||
$this->workflow('new-process', now()->format('Y-m-d H:i:s.u'), 'RUNNING'),
|
||
]];
|
||
|
||
$result = $this->synchronizer->sync(false, 10, 0);
|
||
|
||
$this->assertSame('succeeded', $result['status']);
|
||
$sql = $this->query->calls[0]['sql'];
|
||
$this->assertStringContainsString('[processCode] = ?', $sql);
|
||
$this->assertStringContainsString('COALESCE([updateAt], [createAt]) >= ? AND COALESCE([updateAt], [createAt]) < ?', $sql);
|
||
$this->assertStringContainsString('ORDER BY COALESCE([updateAt], [createAt]), [processInstanceId]', $sql);
|
||
$this->assertCount(3, $this->query->calls[0]['bindings']);
|
||
$this->assertSame('PROC-TEST', $this->query->calls[0]['bindings'][0]);
|
||
$this->assertSame('RUNNING', json_decode(
|
||
(string) DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->where('process_instance_id', 'new-process')->value('workflow_payload'),
|
||
true,
|
||
)['status']);
|
||
|
||
$state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
|
||
$this->assertArrayHasKey('updateAt', json_decode((string) $state->high_watermark, true));
|
||
}
|
||
|
||
public function test_legacy_process_id_only_watermark_does_not_skip_new_processes(): void
|
||
{
|
||
$this->query->workflowBatches = [[
|
||
$this->workflow('existing-process', now()->subHour()->format('Y-m-d H:i:s.u')),
|
||
]];
|
||
$this->synchronizer->sync(true, 10, 0);
|
||
|
||
DB::connection('business_mirror')->table('business_mirror_table_states')->update([
|
||
'high_watermark' => json_encode(['processInstanceId' => 'z-last-process']),
|
||
]);
|
||
$this->query->calls = [];
|
||
$this->query->workflowBatches = [[
|
||
$this->workflow('a-new-process', now()->format('Y-m-d H:i:s.u')),
|
||
]];
|
||
|
||
$this->synchronizer->sync(false, 10, 0);
|
||
|
||
$this->assertStringNotContainsString('AND [processInstanceId] > ?', $this->query->calls[0]['sql']);
|
||
$this->assertSame(2, DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->count());
|
||
}
|
||
|
||
public function test_extended_workflow_update_removes_rows_that_no_longer_match(): void
|
||
{
|
||
config(['upstream_mirrors.ding_online_daifa_price' => [
|
||
'dataset_key' => 'operations_replica.rpt.ding_online_daifa_price',
|
||
'target_table' => 'ding_online_daifa_price_mirror',
|
||
'source_key' => 'operations_replica', 'process_code' => 'PROC-TEST',
|
||
'extend_columns' => true, 'only_status' => 'COMPLETED',
|
||
'column_map' => ['product_id' => '商品ID'],
|
||
]]);
|
||
$sync = $this->synchronizer->forConfigKey('ding_online_daifa_price');
|
||
$this->query->workflowBatches = [[$this->workflow('changed', now()->subHour()->format('Y-m-d H:i:s.u'))]];
|
||
$sync->sync(true, 10, 0);
|
||
$table = DB::connection('business_mirror')->table('ding_online_daifa_price_mirror');
|
||
$this->assertSame(1, $table->count());
|
||
$this->query->workflowBatches = [[$this->workflow('changed', now()->format('Y-m-d H:i:s.u'), 'RUNNING')]];
|
||
$sync->sync(false, 10, 0);
|
||
$this->assertSame(0, $table->count());
|
||
}
|
||
|
||
public function test_daifa_detail_table_falls_back_to_grid_with_product_id_column(): void
|
||
{
|
||
config(['upstream_mirrors.ding_online_daifa_price' => [
|
||
'dataset_key' => 'operations_replica.rpt.ding_online_daifa_price',
|
||
'target_table' => 'ding_online_daifa_price_mirror',
|
||
'source_key' => 'operations_replica', 'process_code' => 'PROC-TEST',
|
||
'extend_columns' => true, 'only_status' => 'COMPLETED',
|
||
'detail_table' => '收款费用明细', 'detail_table_match_label' => '商品ID',
|
||
'column_map' => ['product_id' => '商品ID', 'shop_trade_no' => '店铺交易单号', 'material_no' => ['料号', '商品编码'], 'payee_name' => '收款人全称'],
|
||
]]);
|
||
$cell = static fn (string $label, string $value): array => ['label' => $label, 'value' => $value];
|
||
$this->query->schemeRows = [
|
||
['processInstanceId' => 'blank-name', 'name' => '收款人全称', 'componentType' => 'TextField', 'value' => '宁波顾家望业家居有限公司'],
|
||
// 付款明细不含“商品ID”，不能被当成商品明细。
|
||
['processInstanceId' => 'blank-name', 'name' => '付款明细', 'componentType' => 'TableField', 'value' => json_encode([
|
||
['rowValue' => [$cell('付款金额', '1384')]],
|
||
], JSON_UNESCAPED_UNICODE)],
|
||
// 新版模板：商品明细表格名称为空，料号改叫商品编码。
|
||
['processInstanceId' => 'blank-name', 'name' => '', 'componentType' => 'TableField', 'value' => json_encode([
|
||
['rowValue' => [$cell('商品ID', '10035514372245'), $cell('店铺交易单号', '3630463003975610'), $cell('商品编码', 'DNZ-01')]],
|
||
['rowValue' => [$cell('商品ID', '10035514372246'), $cell('店铺交易单号', '3630463003975611'), $cell('商品编码', 'DNZ-02')]],
|
||
], JSON_UNESCAPED_UNICODE)],
|
||
];
|
||
$this->query->workflowBatches = [[$this->workflow('blank-name', now()->subHour()->format('Y-m-d H:i:s.u'))]];
|
||
$this->synchronizer->forConfigKey('ding_online_daifa_price')->sync(true, 10, 0);
|
||
|
||
$rows = DB::connection('business_mirror')->table('ding_online_daifa_price_mirror')->orderBy('detail_index')->get();
|
||
$this->assertCount(2, $rows);
|
||
$this->assertSame('10035514372245', $rows[0]->product_id);
|
||
$this->assertSame('3630463003975611', $rows[1]->shop_trade_no);
|
||
$this->assertSame('DNZ-01', $rows[0]->material_no);
|
||
$this->assertSame('宁波顾家望业家居有限公司', $rows[1]->payee_name);
|
||
}
|
||
|
||
public function test_daifa_repository_includes_end_date_and_counts_only_resolved_products(): void
|
||
{
|
||
config(['upstream_mirrors.ding_online_daifa_price' => [
|
||
'dataset_key' => 'operations_replica.rpt.ding_online_daifa_price',
|
||
'target_table' => 'ding_online_daifa_price_mirror',
|
||
'source_key' => 'operations_replica', 'process_code' => 'PROC-TEST',
|
||
'extend_columns' => true, 'column_map' => ['product_id' => '商品ID'],
|
||
]]);
|
||
$this->query->workflowBatches = [[
|
||
$this->workflow('A', '2026-08-01 00:00:00'),
|
||
$this->workflow('B', '2026-08-01 23:59:59'),
|
||
$this->workflow('C', '2026-08-02 00:00:00'),
|
||
]];
|
||
$this->synchronizer->forConfigKey('ding_online_daifa_price')->sync(true, 10, 0);
|
||
$db = DB::connection('business_mirror');
|
||
$db->table('ding_online_daifa_price_mirror')->where('process_instance_id', 'A')
|
||
->update(['product_id' => 'fixture-a', 'product_name' => '已匹配示例']);
|
||
$db->table('ding_online_daifa_price_mirror')->where('process_instance_id', 'B')
|
||
->update(['product_id' => 'fixture-b', 'product_name' => null]);
|
||
$db->enableQueryLog();
|
||
$result = app(\Modules\EcommerceFinance\Repositories\DingOnlineDaifaPriceMirrorRepository::class)
|
||
->paginate(['created_from' => '2026-08-01', 'created_to' => '2026-08-01'], 1, 25);
|
||
$this->assertSame(2, $result['total']);
|
||
$this->assertSame(1, $result['summary']['matched_count']);
|
||
$this->assertSame(['B', 'A'], array_column($result['data'], 'process_instance_id'));
|
||
$queries = array_filter($db->getQueryLog(), fn ($q) => str_contains($q['query'], 'workflow_created_at'));
|
||
$this->assertContains('2026-08-02 00:00:00', array_values($queries)[0]['bindings']);
|
||
}
|
||
|
||
public function test_daifa_repository_filters_supplier_and_material_no_for_page_and_export(): void
|
||
{
|
||
config(['upstream_mirrors.ding_online_daifa_price' => [
|
||
'dataset_key' => 'operations_replica.rpt.ding_online_daifa_price',
|
||
'target_table' => 'ding_online_daifa_price_mirror',
|
||
'source_key' => 'operations_replica', 'process_code' => 'PROC-TEST',
|
||
'extend_columns' => true, 'column_map' => ['product_id' => '商品ID'],
|
||
]]);
|
||
$this->query->workflowBatches = [[
|
||
$this->workflow('A', '2026-08-01 10:00:00'),
|
||
$this->workflow('B', '2026-08-01 11:00:00'),
|
||
$this->workflow('C', '2026-08-01 12:00:00'),
|
||
]];
|
||
$this->synchronizer->forConfigKey('ding_online_daifa_price')->sync(true, 10, 0);
|
||
|
||
$db = DB::connection('business_mirror');
|
||
$db->table('ding_online_daifa_price_mirror')->where('process_instance_id', 'A')->update([
|
||
'supplier_name' => '供应商%测试', 'material_no' => 'MAT-001',
|
||
]);
|
||
$db->table('ding_online_daifa_price_mirror')->where('process_instance_id', 'B')->update([
|
||
'supplier_name' => '供应商X测试', 'material_no' => 'MAT-001',
|
||
]);
|
||
$db->table('ding_online_daifa_price_mirror')->where('process_instance_id', 'C')->update([
|
||
'supplier_name' => '供应商%测试', 'material_no' => 'MAT-002',
|
||
]);
|
||
|
||
$filters = ['supplier_name' => '供应商%测试', 'material_no' => 'MAT-001'];
|
||
$repository = app(\Modules\EcommerceFinance\Repositories\DingOnlineDaifaPriceMirrorRepository::class);
|
||
|
||
$page = $repository->paginate($filters, 1, 25);
|
||
$this->assertSame(1, $page['total']);
|
||
$this->assertSame(['A'], array_column($page['data'], 'process_instance_id'));
|
||
|
||
$exported = iterator_to_array($repository->exportRows($filters));
|
||
$this->assertSame(['A'], array_column($exported, 'process_instance_id'));
|
||
$this->assertSame('供应商%测试', $exported[0]['supplier_name']);
|
||
$this->assertSame('MAT-001', $exported[0]['material_no']);
|
||
}
|
||
|
||
public function test_daifa_repository_pages_in_sql_and_labels_the_payee(): void
|
||
{
|
||
config(['upstream_mirrors.ding_online_daifa_price' => [
|
||
'dataset_key' => 'operations_replica.rpt.ding_online_daifa_price',
|
||
'target_table' => 'ding_online_daifa_price_mirror',
|
||
'source_key' => 'operations_replica', 'process_code' => 'PROC-TEST',
|
||
'extend_columns' => true, 'only_status' => 'COMPLETED',
|
||
'column_map' => ['product_id' => '商品ID'],
|
||
]]);
|
||
$this->query->workflowBatches = [[
|
||
$this->workflow('A', '2026-08-01 10:00:00'),
|
||
$this->workflow('B', '2026-08-01 11:00:00'),
|
||
$this->workflow('C', '2026-08-01 12:00:00'),
|
||
]];
|
||
$this->synchronizer->forConfigKey('ding_online_daifa_price')->sync(true, 10, 0);
|
||
|
||
$db = DB::connection('business_mirror');
|
||
$db->table('ding_online_daifa_price_mirror')->where('process_instance_id', 'B')->update([
|
||
'payee_name' => '宁波顾家望业家居有限公司',
|
||
]);
|
||
$db->flushQueryLog();
|
||
$db->enableQueryLog();
|
||
|
||
$result = app(\Modules\EcommerceFinance\Repositories\DingOnlineDaifaPriceMirrorRepository::class)
|
||
->paginate([], 2, 1);
|
||
|
||
$this->assertSame(3, $result['total']);
|
||
$this->assertSame(3, $result['summary']['count']);
|
||
$this->assertSame(['B'], array_column($result['data'], 'process_instance_id'));
|
||
$this->assertSame('刘萧代发', $result['data'][0]['tag']);
|
||
$this->assertTrue(collect($db->getQueryLog())->contains(
|
||
static fn (array $query): bool => str_contains($query['query'], 'limit 1 offset 1')
|
||
));
|
||
}
|
||
|
||
/** @return array<string,string> */
|
||
private function workflow(string $id, string $updateAt, string $status = 'COMPLETED'): array
|
||
{
|
||
return [
|
||
'processInstanceId' => $id,
|
||
'processCode' => 'PROC-TEST',
|
||
'updateAt' => $updateAt,
|
||
'createAt' => $updateAt,
|
||
'status' => $status,
|
||
'originatorUserId' => '',
|
||
];
|
||
}
|
||
}
|
||
|
||
class FakeDingPayableWorkflowQuery extends UpstreamReadOnlyQuery
|
||
{
|
||
/** @var list<list<array<string,mixed>>> */
|
||
public array $workflowBatches = [];
|
||
|
||
/** @var list<array<string,mixed>> */
|
||
public array $schemeRows = [];
|
||
|
||
/** @var list<array{sql:string,bindings:array}> */
|
||
public array $calls = [];
|
||
|
||
public function __construct()
|
||
{
|
||
parent::__construct('operations_replica');
|
||
}
|
||
|
||
public function select(string $sql, array $bindings = []): array
|
||
{
|
||
$this->calls[] = compact('sql', 'bindings');
|
||
|
||
if (str_contains($sql, 'FROM [rpt].[ding_Workflow]')) {
|
||
return array_shift($this->workflowBatches) ?? [];
|
||
}
|
||
if (str_contains($sql, 'FROM [rpt].[ding_WorkflowScheme]')) {
|
||
return $this->schemeRows;
|
||
}
|
||
|
||
return [];
|
||
}
|
||
}
|