datahub/backend/test/Cases/Integration/Materialization/AdminMaterializationControllerTest.php

<?php

declare(strict_types=1);

namespace HyperfTest\Cases\Integration\Materialization;

use App\Model\AggregateRefreshQueue;
use App\Model\Role;
use App\Model\User;
use Carbon\Carbon;
use HyperfTest\TestCase;
use Qbhy\HyperfAuth\AuthManager;

use function Hyperf\Support\make;

/**
 * AdminMaterializationController 集成测试
 *
 * @internal
 * @coversNothing
 */
class AdminMaterializationControllerTest extends TestCase
{
    protected function fetchAdminRole(): Role
    {
        return Role::query()->where('name', 'administrator')->firstOrFail();
    }

    protected function getAdminAuthToken(): string
    {
        $admin_role = $this->fetchAdminRole();
        $user = User::query()
            ->where('status', 1)
            ->where('role_id', $admin_role->id)
            ->first();
        if (!$user) {
            $this->markTestSkipped('没有可用的 administrator 用户，无法测试');
        }

        $auth = make(AuthManager::class);
        return $auth->guard('jwt')->login($user);
    }

    protected function adminHeaders(): array
    {
        return ['Authorization' => 'Bearer ' . $this->getAdminAuthToken()];
    }

    protected function getNonAdminToken(): array
    {
        $suffix = 'mat_nonadmin_' . uniqid();
        $user = User::query()->create([
            'username' => $suffix,
            'password' => 'Pass_' . $suffix,
            'email' => $suffix . '@example.com',
            'status' => 1,
            'api_key_enabled' => true,
        ]);
        $auth = make(AuthManager::class);
        $token = $auth->guard('jwt')->login($user);
        return ['Authorization' => 'Bearer ' . $token];
    }

    public function test_queue_lists_pending(): void
    {
        $date = '2030-12-31';
        $view = 'orders_daily_by_created';

        AggregateRefreshQueue::query()->insertOrIgnore([[
            'refresh_date' => $date,
            'aggregate_view' => $view,
            'created_at' => Carbon::now(),
        ]]);

        try {
            $response = $this->get(
                '/api/v1/admin/materialization/queue',
                ['view' => $view, 'from' => $date, 'to' => $date],
                $this->adminHeaders()
            );

            $response->assertStatus(200);
            $response->assertJsonPath('code', 0);

            $body = json_decode($response->getBody()->getContents(), true);
            $this->assertArrayHasKey('items', $body['data']);
            $this->assertArrayHasKey('total', $body['data']);
            $this->assertArrayHasKey('page', $body['data']);
            $this->assertArrayHasKey('per_page', $body['data']);
            $this->assertGreaterThanOrEqual(1, $body['data']['total']);

            $found = false;
            foreach ($body['data']['items'] as $item) {
                if ($item['refresh_date'] === $date && $item['aggregate_view'] === $view) {
                    $found = true;
                    break;
                }
            }
            $this->assertTrue($found, '应在 queue 列表中找到刚插入的 fixture 行');
        } finally {
            AggregateRefreshQueue::query()
                ->where('refresh_date', $date)
                ->where('aggregate_view', $view)
                ->delete();
        }
    }

    public function test_refresh_validates_view_whitelist(): void
    {
        $response = $this->post(
            '/api/v1/admin/materialization/refresh',
            [
                'view' => 'evil_view',
                'from' => '2026-01-01 00:00:00+00',
                'to' => '2026-01-02 00:00:00+00',
            ],
            $this->adminHeaders()
        );

        $response->assertStatus(400);
        $body = json_decode($response->getBody()->getContents(), true);
        $this->assertSame(400, $body['code']);
        $this->assertStringContainsString('view', $body['message']);
    }

    public function test_aggregates_returns_lag_seconds(): void
    {
        $response = $this->get('/api/v1/admin/materialization/aggregates', [], $this->adminHeaders());

        $response->assertStatus(200);
        $response->assertJsonPath('code', 0);

        $body = json_decode($response->getBody()->getContents(), true);
        $this->assertArrayHasKey('items', $body['data']);
        $items = $body['data']['items'];
        $this->assertNotEmpty($items, 'aggregates 应至少返回一条连续聚合记录（orders_daily_by_created）');
        $this->assertArrayHasKey('view_name', $items[0]);
        $this->assertArrayHasKey('lag_seconds', $items[0]);
    }

    public function test_jobs_lists_refresh_policy(): void
    {
        $response = $this->get('/api/v1/admin/materialization/jobs', [], $this->adminHeaders());

        $response->assertStatus(200);
        $response->assertJsonPath('code', 0);

        $body = json_decode($response->getBody()->getContents(), true);
        $this->assertArrayHasKey('items', $body['data']);
        // by_created 注册了 1 条 policy_refresh_continuous_aggregate；by_paid 由 Hyperf Crontab 调度，不入此表
        $this->assertGreaterThanOrEqual(1, count($body['data']['items']));
        $this->assertSame('policy_refresh_continuous_aggregate', $body['data']['items'][0]['proc_name']);
    }

    public function test_non_admin_blocked(): void
    {
        $headers = $this->getNonAdminToken();
        $response = $this->get('/api/v1/admin/materialization/queue', [], $headers);
        $response->assertStatus(403);
    }
}