229 lines
13 KiB
PHP
229 lines
13 KiB
PHP
<?php
|
|
|
|
namespace Tests\Unit\EcommerceFinance;
|
|
|
|
use App\Models\Settings\DatabaseConnection;
|
|
use Illuminate\Database\Schema\Blueprint;
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Facades\Schema;
|
|
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
|
|
use Modules\EcommerceFinance\Services\BusinessMirrorDataReader;
|
|
use Modules\EcommerceFinance\Services\UpstreamConnection;
|
|
use Modules\EcommerceFinance\Services\UpstreamSourceRegistry;
|
|
use Modules\EcommerceFinance\Services\UpstreamSchemaCatalogSynchronizer;
|
|
use Modules\EcommerceFinance\Services\UpstreamSqlDialect;
|
|
use PDO;
|
|
use PDOException;
|
|
use Tests\TestCase;
|
|
|
|
class GenericUpstreamPipelineTest extends TestCase
|
|
{
|
|
protected function setUp(): void
|
|
{
|
|
parent::setUp();
|
|
config([
|
|
'upstreams.stored_configuration.enabled' => true,
|
|
'upstreams.local_configuration.enabled' => false,
|
|
'database.connections.sqlite.password' => 'fixture-cipher-key',
|
|
'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''],
|
|
'upstream_mirrors.connection' => 'business_mirror',
|
|
'upstream_mirrors.datasets' => [],
|
|
'upstream_mirrors.refresh_enabled' => true,
|
|
]);
|
|
DB::purge('business_mirror');
|
|
Schema::connection('business_mirror')->create('business_mirror_table_states', function (Blueprint $t): void {
|
|
$t->string('dataset_key')->primary();
|
|
foreach (['source_key', 'source_object', 'target_table', 'status', 'run_mode', 'generation', 'last_error_code'] as $c) $t->string($c)->nullable();
|
|
foreach (['cursor', 'high_watermark'] as $c) $t->json($c)->nullable();
|
|
foreach (['window_started_at', 'window_ends_at', 'initial_completed_at', 'last_full_succeeded_at', 'last_attempted_at', 'last_succeeded_at'] as $c) $t->dateTime($c)->nullable();
|
|
$t->unsignedBigInteger('last_upserted_rows')->default(0);
|
|
$t->unsignedBigInteger('mirror_row_count')->default(0);
|
|
$t->timestamps();
|
|
});
|
|
}
|
|
|
|
protected function tearDown(): void
|
|
{
|
|
UpstreamConnection::reset();
|
|
parent::tearDown();
|
|
}
|
|
|
|
private function connection(string $driver = 'pgsql'): DatabaseConnection
|
|
{
|
|
return DatabaseConnection::create([
|
|
'company_id' => 0, 'name' => 'Isolated fixture', 'driver' => $driver, 'host' => 'fixture.invalid',
|
|
'port' => $driver === 'pgsql' ? 5432 : 3306, 'database_name' => 'fixture',
|
|
'username' => 'fixture', 'password' => 'fixture-password', 'enabled' => true,
|
|
'connect_timeout' => 5, 'query_timeout' => 15,
|
|
]);
|
|
}
|
|
|
|
public function test_new_connection_is_immediately_a_source_without_changing_config(): void
|
|
{
|
|
$row = $this->connection();
|
|
$key = $row->sourceIdentifier();
|
|
$sources = app(UpstreamSourceRegistry::class);
|
|
$this->assertNull($row->source_key);
|
|
$this->assertNull(config('upstreams.sources.' . $key));
|
|
$this->assertSame('pgsql', $sources->driver($key));
|
|
$this->assertTrue($sources->enabled($key));
|
|
$this->assertSame('fixture-password', $sources->value($key, 'connection.password'));
|
|
$this->assertArrayHasKey($key, $sources->all());
|
|
$this->assertSame('jiansun', $sources->definition('jiansun')['key']);
|
|
}
|
|
|
|
public function test_dialects_use_engine_specific_limits_and_quotes(): void
|
|
{
|
|
$this->assertSame('SELECT TOP (5) [id] FROM [dbo].[records]', (new UpstreamSqlDialect('sqlsrv'))->select('[id]', 'FROM [dbo].[records]', 5));
|
|
$this->assertSame('SELECT "id" FROM "public"."records" LIMIT 5', (new UpstreamSqlDialect('pgsql'))->select('"id"', 'FROM "public"."records"', 5));
|
|
$this->assertSame('SELECT `id` FROM `records` LIMIT 5', (new UpstreamSqlDialect('mysql'))->select('`id`', 'FROM `records`', 5));
|
|
$this->assertStringContainsString('c.table_schema = DATABASE()', (new UpstreamSqlDialect('mysql'))->catalogSql());
|
|
}
|
|
|
|
public function test_source_discovery_does_not_decrypt_credentials(): void
|
|
{
|
|
$row = $this->connection();
|
|
DB::table('system_database_connections')->where('id', $row->id)->update(['shared_password' => 'invalid-fixture-envelope']);
|
|
$sources = app(UpstreamSourceRegistry::class);
|
|
$this->assertArrayHasKey($row->sourceIdentifier(), $sources->all());
|
|
$this->assertSame('pgsql', $sources->driver($row->sourceIdentifier()));
|
|
}
|
|
|
|
public function test_declarations_require_fields_and_explicit_initial_sync_approval(): void
|
|
{
|
|
$row = $this->connection();
|
|
$source = $row->sourceIdentifier();
|
|
DB::table('upstream_schema_columns')->insert([
|
|
'source_key' => $source, 'schema_name' => 'public', 'object_name' => 'records',
|
|
'object_type' => 'table', 'column_name' => 'id', 'ordinal_position' => 1, 'data_type' => 'bigint',
|
|
'is_nullable' => false, 'synchronized_at' => now(), 'created_at' => now(), 'updated_at' => now(),
|
|
]);
|
|
$args = ['source' => $source, 'schema' => 'public', 'object' => 'records', '--keys' => 'id', '--columns' => 'id'];
|
|
$this->artisan('business-mirror:register', $args)->assertSuccessful();
|
|
$key = $source . '.public.records';
|
|
$registry = app(BusinessMirrorDatasetRegistry::class);
|
|
$this->assertFalse($registry->automaticInitialSyncAllowed($key));
|
|
$this->artisan('business-mirror:sync-tables')->assertSuccessful();
|
|
$this->assertSame(0, DB::connection('business_mirror')->table('business_mirror_table_states')->count());
|
|
$this->artisan('business-mirror:register', $args + ['--enable-sync' => true])->assertSuccessful();
|
|
$this->assertTrue($registry->automaticInitialSyncAllowed($key));
|
|
$this->assertSame(['id'], $registry->columnSelection($key));
|
|
$this->artisan('business-mirror:register', ['source' => $source, 'schema' => 'public', 'object' => 'records', '--disable-sync' => true])->assertSuccessful();
|
|
$this->assertFalse($registry->scheduled($key));
|
|
}
|
|
|
|
public function test_original_bill_dataset_uses_stable_cursor_and_explicit_shared_schema(): void
|
|
{
|
|
$configuration = require base_path('config/upstream_mirrors.php');
|
|
config(['upstream_mirrors.datasets' => $configuration['datasets']]);
|
|
|
|
$key = 'operations_replica.rpt.jst_originalBillInfo';
|
|
$registry = app(BusinessMirrorDatasetRegistry::class);
|
|
$dataset = $registry->get($key);
|
|
|
|
$this->assertSame('operations_replica', $dataset->sourceKey());
|
|
$this->assertSame('rpt', $dataset->sourceSchema());
|
|
$this->assertSame('jst_originalBillInfo', $dataset->sourceObject());
|
|
$this->assertSame('ops_jst_original_bill_info', $dataset->targetTable());
|
|
$this->assertSame(['Id'], $dataset->keyColumns());
|
|
$this->assertSame('updateTime', $dataset->watermarkColumn());
|
|
$this->assertSame([
|
|
['Id'],
|
|
['shop_id', 'fee_time', 'Id'],
|
|
['fee_time', 'Id'],
|
|
['channel', 'fee_time', 'Id', 'inout_type', 'amount'],
|
|
['channel', 'shop_id', 'fee_time', 'Id', 'inout_type', 'amount', 'fee_name'],
|
|
], $dataset->indexDefinitions());
|
|
$this->assertSame([
|
|
'Id', 'shop_id', 'channel', 'fee_time', 'del_flag', 'primary_key',
|
|
'createTime', 'updateTime', 'auto_id', 'fee_name', 'inout_type',
|
|
'oc_amount', 'p_auto_id', 'shop_i_id', 'shop_sku_id', 'sku_id',
|
|
'so_id', 'combine_type', 'amount',
|
|
], $registry->columnSelection($key));
|
|
}
|
|
|
|
public function test_original_bill_migration_creates_compact_shared_table(): void
|
|
{
|
|
$migration = require base_path('database/migrations/2026_09_07_000000_create_original_bill_info_mirror_table.php');
|
|
$migration->up();
|
|
|
|
$schema = Schema::connection('business_mirror');
|
|
$this->assertTrue($schema->hasTable('ops_jst_original_bill_info'));
|
|
$this->assertTrue($schema->hasColumn('ops_jst_original_bill_info', 'Id'));
|
|
$this->assertTrue($schema->hasColumn('ops_jst_original_bill_info', 'updateTime'));
|
|
$this->assertTrue($schema->hasColumn('ops_jst_original_bill_info', '_mirror_synced_at'));
|
|
$this->assertSame('string', $schema->getColumnType('ops_jst_original_bill_info', 'channel'));
|
|
$this->assertSame('string', $schema->getColumnType('ops_jst_original_bill_info', 'so_id'));
|
|
}
|
|
|
|
/** @dataProvider nativeDrivers */
|
|
public function test_native_tls_catalog_first_sync_pagination_and_incremental(string $driver, bool $timezoneAware = false, ?string $fixtureHost = null): void
|
|
{
|
|
if (getenv('CAIWU_PIPELINE_INTEGRATION') !== '1') $this->markTestSkipped('Requires isolated pipeline fixture containers.');
|
|
$host = $fixtureHost ?? ($driver === 'pgsql' ? 'caiwu-pipeline-pg' : 'caiwu-pipeline-mysql');
|
|
$database = $driver === 'pgsql' ? 'postgres' : 'pipeline_fixture';
|
|
$user = $driver === 'pgsql' ? 'postgres' : 'root';
|
|
$schema = $driver === 'pgsql' ? 'public' : $database;
|
|
$dsn = $driver === 'pgsql' ? "pgsql:host={$host};dbname={$database};sslmode=require" : "mysql:host={$host};dbname={$database}";
|
|
$options = [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION];
|
|
if ($driver === 'mysql') {
|
|
$options[PDO::MYSQL_ATTR_SSL_CIPHER] = 'DEFAULT';
|
|
$options[PDO::MYSQL_ATTR_SSL_VERIFY_SERVER_CERT] = false;
|
|
}
|
|
$admin = new PDO($dsn, $user, 'fixture-password', $options);
|
|
$admin->exec('DROP TABLE IF EXISTS pipeline_records');
|
|
$type = $driver === 'pgsql' ? ($timezoneAware ? 'timestamp with time zone' : 'timestamp without time zone') : 'datetime(6)';
|
|
$admin->exec('CREATE TABLE pipeline_records (id bigint primary key, modified_at ' . $type . ', amount decimal(18,2), label varchar(50), private_note text)');
|
|
$date = now()->subMinute()->format($timezoneAware ? 'Y-m-d H:i:sP' : 'Y-m-d H:i:s');
|
|
$insert = $admin->prepare('INSERT INTO pipeline_records (id, modified_at, amount, label) VALUES (?, ?, ?, ?)');
|
|
$insert->execute([1, $date, '12.30', 'first']);
|
|
$insert->execute([2, $date, '45.60', 'second']);
|
|
$row = $this->connection($driver);
|
|
$row->update(['host' => $host, 'database_name' => $database, 'username' => $user]);
|
|
$source = $row->sourceIdentifier();
|
|
$result = app(UpstreamSchemaCatalogSynchronizer::class)->sync($source);
|
|
$this->assertGreaterThanOrEqual(5, $result['columns']);
|
|
$this->artisan('business-mirror:register', [
|
|
'source' => $source, 'schema' => $schema, 'object' => 'pipeline_records', '--keys' => 'id',
|
|
'--columns' => 'id,modified_at,amount,label', '--watermark' => 'modified_at', '--enable-sync' => true,
|
|
])->assertSuccessful();
|
|
$this->artisan('business-mirror:sync-tables')->assertSuccessful();
|
|
$dataset = app(BusinessMirrorDatasetRegistry::class)->get($source . '.' . $schema . '.pipeline_records');
|
|
$mirror = DB::connection('business_mirror')->table($dataset->targetTable());
|
|
$this->assertSame(2, $mirror->count());
|
|
$this->assertSame(2, (int) $mirror->orderBy('id')->offset(1)->value('id'));
|
|
$this->assertFalse(Schema::connection('business_mirror')->hasColumn($dataset->targetTable(), 'private_note'));
|
|
$admin->exec("UPDATE pipeline_records SET amount=99.50 WHERE id=1");
|
|
$insert->execute([3, $date, '6.70', 'third']);
|
|
$insert->execute([4, now()->addHour()->format($timezoneAware ? 'Y-m-d H:i:sP' : 'Y-m-d H:i:s'), '1.00', 'future']);
|
|
$this->artisan('business-mirror:sync-tables')->assertSuccessful();
|
|
$this->assertSame(3, DB::connection('business_mirror')->table($dataset->targetTable())->count());
|
|
$this->assertEquals(99.5, DB::connection('business_mirror')->table($dataset->targetTable())->where('id', 1)->value('amount'));
|
|
$pdo = UpstreamConnection::get($source);
|
|
$row->update(['enabled' => false]);
|
|
$page = app(BusinessMirrorDataReader::class)->paginate($dataset->key(), [], 2, 2);
|
|
$this->assertSame(3, $page['total']);
|
|
$this->assertSame(3, (int) $page['rows'][0]['id']);
|
|
$this->assertArrayNotHasKey('_mirror_key', $page['rows'][0]);
|
|
$filtered = app(BusinessMirrorDataReader::class)->paginate($dataset->key(), ['label' => 'first']);
|
|
$this->assertSame(1, $filtered['total']);
|
|
if ($driver === 'pgsql') {
|
|
$this->assertSame('on', $pdo->query('SHOW default_transaction_read_only')->fetchColumn());
|
|
$this->assertTrue((bool) $pdo->query('SELECT ssl FROM pg_stat_ssl WHERE pid=pg_backend_pid()')->fetchColumn());
|
|
} else {
|
|
$this->assertNotEmpty($pdo->query("SHOW SESSION STATUS LIKE 'Ssl_cipher'")->fetch(PDO::FETCH_NUM)[1]);
|
|
}
|
|
try {
|
|
$pdo->exec("INSERT INTO pipeline_records (id) VALUES (99)");
|
|
$this->fail('The native upstream connection must reject writes.');
|
|
} catch (PDOException $e) {
|
|
$this->assertStringContainsString('read', strtolower($e->getMessage()));
|
|
}
|
|
}
|
|
|
|
public static function nativeDrivers(): array
|
|
{
|
|
return [['pgsql'], ['mysql'], ['pgsql', true], ['mysql', false, 'caiwu-pipeline-mariadb']];
|
|
}
|
|
}
|