This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 08295bb130b [FLINK-40601][runtime-web] Add Vitest specs for remaining
web dashboard views. This closes #29138
08295bb130b is described below
commit 08295bb130bc98bf53468a6a205811d958d27f55
Author: Purushottam Sinha <[email protected]>
AuthorDate: Wed Sep 23 23:44:20 2026 +0530
[FLINK-40601][runtime-web] Add Vitest specs for remaining web dashboard
views. This closes #29138
- Core services: config, overview, status, job-manager and job
- Job-detail drawer views under pages/job/overview
- JobManager log, stdout, thread-dump and profiler pages
- TaskManager stdout, thread-dump and profiler pages
- job-local and application-local services
- Job and application list and detail shells
- Job chart, its resize service and the submit page
Generated-by: Claude Code (claude-sonnet-5)
---
.../job-chart/job-chart.component.spec.ts | 74 ++++++++
.../components/job-chart/job-chart.service.spec.ts | 33 ++++
.../application-detail.component.spec.ts | 70 +++++++
.../application/application-local.service.spec.ts | 50 +++++
.../application/application.component.spec.ts | 85 +++++++++
.../logs/job-manager-logs.component.spec.ts | 66 +++++++
.../job-manager-profiler.component.spec.ts | 77 ++++++++
.../stdout/job-manager-stdout.component.spec.ts | 64 +++++++
.../job-manager-thread-dump.component.spec.ts | 77 ++++++++
.../job/job-detail/job-detail.component.spec.ts | 74 ++++++++
.../src/app/pages/job/job-local.service.spec.ts | 72 +++++++
.../src/app/pages/job/job.component.spec.ts | 85 +++++++++
...-overview-drawer-accumulators.component.spec.ts | 80 ++++++++
...-overview-drawer-backpressure.component.spec.ts | 78 ++++++++
.../job-overview-drawer-chart.component.spec.ts | 80 ++++++++
.../job-overview-drawer-detail.component.spec.ts | 59 ++++++
...ob-overview-drawer-flamegraph.component.spec.ts | 92 +++++++++
.../job-overview-drawer-subtasks.component.spec.ts | 87 +++++++++
...-overview-drawer-taskmanagers.component.spec.ts | 66 +++++++
...ob-overview-drawer-watermarks.component.spec.ts | 71 +++++++
.../src/app/pages/submit/submit.component.spec.ts | 128 +++++++++++++
.../task-manager-profiler.component.spec.ts | 81 ++++++++
.../stdout/task-manager-stdout.component.spec.ts | 68 +++++++
.../task-manager-thread-dump.component.spec.ts | 89 +++++++++
.../src/app/services/config.service.spec.ts | 57 ++++++
.../src/app/services/job-manager.service.spec.ts | 208 +++++++++++++++++++++
.../src/app/services/job.service.spec.ts | 169 +++++++++++++++++
.../src/app/services/overview.service.spec.ts | 54 ++++++
.../src/app/services/status.service.spec.ts | 94 ++++++++++
29 files changed, 2388 insertions(+)
diff --git
a/flink-runtime-web/web-dashboard/src/app/components/job-chart/job-chart.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/components/job-chart/job-chart.component.spec.ts
new file mode 100644
index 00000000000..9220f5ff2d2
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/components/job-chart/job-chart.component.spec.ts
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+
+import type { Chart } from '@antv/g2';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobChartComponent } from './job-chart.component';
+import { JobChartService } from './job-chart.service';
+
+// Constructed directly: ngAfterViewInit builds a real G2 chart against a
canvas jsdom can't
+// provide, so these specs drive refresh()/close() with the chart instance
left as a fake.
+const cdr = { detectChanges: vi.fn(), detach: vi.fn(), markForCheck: vi.fn() }
as unknown as ChangeDetectorRef;
+
+describe('JobChartComponent', () => {
+ let component: JobChartComponent;
+
+ beforeEach(() => {
+ component = new JobChartComponent(cdr, new JobChartService());
+ component.title = 'cpu';
+ });
+
+ it('appends the metric value for its own title and forwards data to the
chart', () => {
+ const changeData = vi.fn();
+ component.chartInstance = { changeData } as unknown as Chart;
+
+ component.refresh({ timestamp: 1000, values: { cpu: 42, mem: 7 } });
+
+ expect(component.latestValue).toBe(42);
+ expect(component.data).toEqual([{ time: 1000, value: 42, type: 'cpu' }]);
+ expect(changeData).toHaveBeenCalledWith(component.data);
+ });
+
+ it('keeps only the most recent 20 samples', () => {
+ for (let i = 0; i < 25; i++) {
+ component.refresh({ timestamp: i, values: { cpu: i } });
+ }
+
+ expect(component.data).toHaveLength(20);
+ expect(component.data[0].time).toBe(5);
+ expect(component.data[19].time).toBe(24);
+ });
+
+ it('emits its title when closed', () => {
+ const closed = vi.fn();
+ component.closed.subscribe(closed);
+
+ component.close();
+
+ expect(closed).toHaveBeenCalledWith('cpu');
+ });
+
+ it('reports the big layout only when sized big', () => {
+ expect(component.isBig).toBe(false);
+ component.size = 'big';
+ expect(component.isBig).toBe(true);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/components/job-chart/job-chart.service.spec.ts
b/flink-runtime-web/web-dashboard/src/app/components/job-chart/job-chart.service.spec.ts
new file mode 100644
index 00000000000..ada54fbc708
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/components/job-chart/job-chart.service.spec.ts
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { describe, expect, it, vi } from 'vitest';
+
+import { JobChartService } from './job-chart.service';
+
+describe('JobChartService', () => {
+ it('broadcasts a resize event to subscribers', () => {
+ const service = new JobChartService();
+ const listener = vi.fn();
+ service.resize$.subscribe(listener);
+
+ service.resize();
+
+ expect(listener).toHaveBeenCalledTimes(1);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/application/application-detail/application-detail.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/application/application-detail/application-detail.component.spec.ts
new file mode 100644
index 00000000000..c321645f862
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/application/application-detail/application-detail.component.spec.ts
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { ActivatedRoute } from '@angular/router';
+import { of, throwError } from 'rxjs';
+
+import { ApplicationLocalService } from
'@flink-runtime-web/pages/application/application-local.service';
+import { ApplicationService, StatusService } from
'@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { ApplicationDetailComponent } from './application-detail.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const activatedRoute = { snapshot: { params: { id: 'app-1' } } } as unknown as
ActivatedRoute;
+
+describe('ApplicationDetailComponent', () => {
+ const loadApplication = vi.fn();
+ const setApplicationDetail = vi.fn();
+ let component: ApplicationDetailComponent;
+
+ beforeEach(() => {
+ loadApplication.mockReset();
+ setApplicationDetail.mockReset();
+ component = new ApplicationDetailComponent(
+ { loadApplication } as unknown as ApplicationService,
+ { setApplicationDetail } as unknown as ApplicationLocalService,
+ { refresh$: of(true) } as unknown as StatusService,
+ activatedRoute,
+ cdr
+ );
+ });
+
+ it('publishes the loaded application detail to the local service', () => {
+ const application = { id: 'app-1' };
+ loadApplication.mockReturnValue(of(application));
+
+ component.ngOnInit();
+
+ expect(loadApplication).toHaveBeenCalledWith('app-1');
+ expect(setApplicationDetail).toHaveBeenCalledWith(application);
+ expect(component.isLoading).toBe(false);
+ expect(component.isError).toBe(false);
+ });
+
+ it('flags an error when the application fails to load', () => {
+ loadApplication.mockReturnValue(throwError(() => new Error('boom')));
+
+ component.ngOnInit();
+
+ expect(component.isError).toBe(true);
+ expect(component.isLoading).toBe(false);
+ expect(setApplicationDetail).not.toHaveBeenCalled();
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/application/application-local.service.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/application/application-local.service.spec.ts
new file mode 100644
index 00000000000..a53bfaa9ffc
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/application/application-local.service.spec.ts
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { firstValueFrom } from 'rxjs';
+
+import { ApplicationDetail } from '@flink-runtime-web/interfaces';
+import { beforeEach, describe, expect, it } from 'vitest';
+
+import { ApplicationLocalService } from './application-local.service';
+
+describe('ApplicationLocalService', () => {
+ let service: ApplicationLocalService;
+
+ beforeEach(() => {
+ service = new ApplicationLocalService();
+ });
+
+ it('replays the latest application detail to a late subscriber', async () =>
{
+ const application = { id: 'app-1' } as ApplicationDetail;
+ service.setApplicationDetail(application);
+
+ const result = await firstValueFrom(service.applicationDetailChanges());
+
+ expect(result).toBe(application);
+ });
+
+ it('replays only the most recent application detail', async () => {
+ service.setApplicationDetail({ id: 'app-1' } as ApplicationDetail);
+ service.setApplicationDetail({ id: 'app-2' } as ApplicationDetail);
+
+ const result = await firstValueFrom(service.applicationDetailChanges());
+
+ expect(result.id).toBe('app-2');
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/application/application.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/application/application.component.spec.ts
new file mode 100644
index 00000000000..9f8bb6087ac
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/application/application.component.spec.ts
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { ActivatedRoute, NavigationEnd, Router } from '@angular/router';
+import { Subject } from 'rxjs';
+
+import { ApplicationItem } from '@flink-runtime-web/interfaces';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { ApplicationComponent } from './application.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const activatedRoute = {} as ActivatedRoute;
+const seg = (path: string): { path: string; toString: () => string } => ({
path, toString: () => path });
+
+function urlTree(paths: string[]): unknown {
+ return { root: { children: { primary: { segments: paths.map(seg) } } } };
+}
+
+describe('ApplicationComponent', () => {
+ const events = new Subject<unknown>();
+ const parseUrl = vi.fn();
+ const navigate = vi.fn().mockResolvedValue(true);
+ let component: ApplicationComponent;
+
+ beforeEach(() => {
+ parseUrl.mockReset();
+ navigate.mockClear();
+ const router = { events, url: '/application/running/app-1', parseUrl,
navigate } as unknown as Router;
+ component = new ApplicationComponent(activatedRoute, router, cdr);
+ });
+
+ it('reads the selected application id and running state from the url', () =>
{
+ parseUrl.mockReturnValue(urlTree(['application', 'running', 'app-1']));
+
+ component.ngOnInit();
+
+ expect(component.applicationIdSelected).toBe('app-1');
+ expect(component.isCompleted).toBe(false);
+ expect(component.cardTitle).toBe('Running Applications');
+ });
+
+ it('recognises the completed list with no application selected', () => {
+ parseUrl.mockReturnValue(urlTree(['application', 'completed']));
+
+ component.ngOnInit();
+
+ expect(component.applicationIdSelected).toBeUndefined();
+ expect(component.isCompleted).toBe(true);
+ expect(component.cardTitle).toBe('Completed Applications');
+ });
+
+ it('re-reads the url on navigation end', () => {
+ parseUrl.mockReturnValue(urlTree(['application', 'running', 'app-1']));
+ component.ngOnInit();
+
+ parseUrl.mockReturnValue(urlTree(['application', 'completed', 'app-2']));
+ events.next(new NavigationEnd(1, '/application/completed/app-2',
'/application/completed/app-2'));
+
+ expect(component.applicationIdSelected).toBe('app-2');
+ expect(component.isCompleted).toBe(true);
+ });
+
+ it('navigates to an application relative to the current route', () => {
+ component.navigateToApplication({ id: 'app-9' } as ApplicationItem);
+
+ expect(navigate).toHaveBeenCalledWith(['app-9'], { relativeTo:
activatedRoute });
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/logs/job-manager-logs.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/logs/job-manager-logs.component.spec.ts
new file mode 100644
index 00000000000..0f056e175e9
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/logs/job-manager-logs.component.spec.ts
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of, throwError } from 'rxjs';
+
+import { JOB_MANAGER_MODULE_DEFAULT_CONFIG } from
'@flink-runtime-web/pages/job-manager/job-manager.config';
+import { ConfigService, JobManagerService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobManagerLogsComponent } from './job-manager-logs.component';
+
+// The log/stdout/thread-dump pages embed a monaco code editor; constructing
them directly and
+// driving reload() exercises the controller without rendering that editor in
jsdom.
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+
+describe('JobManagerLogsComponent', () => {
+ const loadLogs = vi.fn();
+ let component: JobManagerLogsComponent;
+
+ beforeEach(() => {
+ loadLogs.mockReset();
+ component = new JobManagerLogsComponent(
+ { loadLogs } as unknown as JobManagerService,
+ { BASE_URL: '/api' } as unknown as ConfigService,
+ cdr,
+ JOB_MANAGER_MODULE_DEFAULT_CONFIG
+ );
+ });
+
+ it('loads the jobmanager log and exposes the download target', () => {
+ loadLogs.mockReturnValue(of('2026-01-01 INFO started\n'));
+
+ component.ngOnInit();
+
+ expect(loadLogs).toHaveBeenCalled();
+ expect(component.logs).toBe('2026-01-01 INFO started\n');
+ expect(component.downloadUrl).toBe('/api/jobmanager/log');
+ expect(component.downloadName).toBe('jobmanager_log');
+ expect(component.loading).toBe(false);
+ });
+
+ it('falls back to empty logs when the request fails', () => {
+ loadLogs.mockReturnValue(throwError(() => new Error('log unavailable')));
+
+ component.ngOnInit();
+
+ expect(component.logs).toBe('');
+ expect(component.loading).toBe(false);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.spec.ts
new file mode 100644
index 00000000000..443aad28f0c
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/profiler/job-manager-profiler.component.spec.ts
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of } from 'rxjs';
+
+import { ProfilingDetail } from '@flink-runtime-web/interfaces/job-profiler';
+import { JobManagerService, StatusService } from '@flink-runtime-web/services';
+import { NzMessageService } from 'ng-zorro-antd/message';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobManagerProfilerComponent } from './job-manager-profiler.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+
+describe('JobManagerProfilerComponent', () => {
+ const loadProfilingList = vi.fn();
+ const createProfilingInstance = vi.fn();
+ const warning = vi.fn();
+ let component: JobManagerProfilerComponent;
+
+ beforeEach(() => {
+ loadProfilingList.mockReset().mockReturnValue(of({ profilingList: [] }));
+ createProfilingInstance.mockReset();
+ warning.mockReset();
+ component = new JobManagerProfilerComponent(
+ { loadProfilingList, createProfilingInstance } as unknown as
JobManagerService,
+ { refresh$: of(true) } as unknown as StatusService,
+ { warning } as unknown as NzMessageService,
+ cdr
+ );
+ });
+
+ it('loads the profiling list and marks profiling enabled on refresh', () => {
+ loadProfilingList.mockReturnValue(of({ profilingList: [{ status:
'FINISHED' } as ProfilingDetail] }));
+
+ component.ngOnInit();
+
+ expect(component.profilingList).toEqual([{ status: 'FINISHED' }]);
+ expect(component.isEnabled).toBe(true);
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('prepends a newly created profiling instance', () => {
+ createProfilingInstance.mockReturnValue(of({ status: 'RUNNING' } as
ProfilingDetail));
+
+ component.createProfilingInstance();
+
+ expect(createProfilingInstance).toHaveBeenCalledWith('ITIMER', 30);
+ expect(component.profilingList[0]).toEqual({ status: 'RUNNING' });
+ expect(component.isCreating).toBe(false);
+ });
+
+ it('refuses to start a new profiling run while one is still running', () => {
+ component.profilingList = [{ status: 'RUNNING' } as ProfilingDetail];
+
+ component.createProfilingInstance();
+
+ expect(warning).toHaveBeenCalled();
+ expect(createProfilingInstance).not.toHaveBeenCalled();
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/stdout/job-manager-stdout.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/stdout/job-manager-stdout.component.spec.ts
new file mode 100644
index 00000000000..cf82d6b290d
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/stdout/job-manager-stdout.component.spec.ts
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of, throwError } from 'rxjs';
+
+import { JOB_MANAGER_MODULE_DEFAULT_CONFIG } from
'@flink-runtime-web/pages/job-manager/job-manager.config';
+import { ConfigService, JobManagerService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobManagerStdoutComponent } from './job-manager-stdout.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+
+describe('JobManagerStdoutComponent', () => {
+ const loadStdout = vi.fn();
+ let component: JobManagerStdoutComponent;
+
+ beforeEach(() => {
+ loadStdout.mockReset();
+ component = new JobManagerStdoutComponent(
+ { loadStdout } as unknown as JobManagerService,
+ { BASE_URL: '/api' } as unknown as ConfigService,
+ cdr,
+ JOB_MANAGER_MODULE_DEFAULT_CONFIG
+ );
+ });
+
+ it('loads the jobmanager stdout and exposes the download target', () => {
+ loadStdout.mockReturnValue(of('hello stdout\n'));
+
+ component.ngOnInit();
+
+ expect(loadStdout).toHaveBeenCalled();
+ expect(component.stdout).toBe('hello stdout\n');
+ expect(component.downloadUrl).toBe('/api/jobmanager/stdout');
+ expect(component.downloadName).toBe('jobmanager_stdout');
+ expect(component.loading).toBe(false);
+ });
+
+ it('falls back to empty stdout when the request fails', () => {
+ loadStdout.mockReturnValue(throwError(() => new Error('stdout
unavailable')));
+
+ component.ngOnInit();
+
+ expect(component.stdout).toBe('');
+ expect(component.loading).toBe(false);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.spec.ts
new file mode 100644
index 00000000000..8fc31a5310a
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job-manager/thread-dump/job-manager-thread-dump.component.spec.ts
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of, throwError } from 'rxjs';
+
+import { JOB_MANAGER_MODULE_DEFAULT_CONFIG } from
'@flink-runtime-web/pages/job-manager/job-manager.config';
+import { ConfigService, JobManagerService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobManagerThreadDumpComponent } from
'./job-manager-thread-dump.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+
+describe('JobManagerThreadDumpComponent', () => {
+ const loadThreadDump = vi.fn();
+ let component: JobManagerThreadDumpComponent;
+
+ beforeEach(() => {
+ loadThreadDump.mockReset().mockReturnValue(of('thread dump\n'));
+ component = new JobManagerThreadDumpComponent(
+ { loadThreadDump } as unknown as JobManagerService,
+ { BASE_URL: '/api' } as unknown as ConfigService,
+ cdr,
+ JOB_MANAGER_MODULE_DEFAULT_CONFIG
+ );
+ });
+
+ it('loads the thread dump with no mode by default', () => {
+ component.ngOnInit();
+
+ expect(loadThreadDump).toHaveBeenCalledWith(undefined);
+ expect(component.dump).toBe('thread dump\n');
+ expect(component.downloadUrl).toBe('/api/jobmanager/thread-dump');
+ expect(component.loading).toBe(false);
+ });
+
+ it('updates mode and download url on selection and reloads with that mode',
() => {
+ component.selectMode('full');
+ expect(component.mode).toBe('full');
+
expect(component.downloadUrl).toBe('/api/jobmanager/thread-dump?mode=full');
+
+ component.reload();
+ expect(loadThreadDump).toHaveBeenLastCalledWith('full');
+ });
+
+ it('ignores re-selecting the current mode', () => {
+ component.selectMode('lite');
+ component.selectMode('lite');
+
+ expect(component.mode).toBe('lite');
+ });
+
+ it('falls back to an empty dump when the request fails', () => {
+ loadThreadDump.mockReturnValue(throwError(() => new Error('unavailable')));
+
+ component.ngOnInit();
+
+ expect(component.dump).toBe('');
+ expect(component.loading).toBe(false);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/job-detail/job-detail.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/job-detail/job-detail.component.spec.ts
new file mode 100644
index 00000000000..5d541bb9063
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/job-detail/job-detail.component.spec.ts
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { ActivatedRoute } from '@angular/router';
+import { of, throwError } from 'rxjs';
+
+import { JobLocalService } from
'@flink-runtime-web/pages/job/job-local.service';
+import { JobService, StatusService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobDetailComponent } from './job-detail.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const activatedRoute = { snapshot: { params: { jid: 'job-1' } } } as unknown
as ActivatedRoute;
+
+describe('JobDetailComponent', () => {
+ const loadJob = vi.fn();
+ const loadExceptions = vi.fn();
+ const setJobDetail = vi.fn();
+ let component: JobDetailComponent;
+
+ beforeEach(() => {
+ loadJob.mockReset();
+ loadExceptions.mockReset().mockReturnValue(of({ 'root-exception': '' }));
+ setJobDetail.mockReset();
+ component = new JobDetailComponent(
+ { loadJob, loadExceptions } as unknown as JobService,
+ { setJobDetail } as unknown as JobLocalService,
+ { refresh$: of(true) } as unknown as StatusService,
+ activatedRoute,
+ cdr
+ );
+ });
+
+ it('publishes the loaded job detail to the local service', () => {
+ const job = { plan: { jid: 'job-1' } };
+ loadJob.mockReturnValue(of(job));
+
+ component.ngOnInit();
+
+ expect(loadJob).toHaveBeenCalledWith('job-1');
+ expect(setJobDetail).toHaveBeenCalledWith(job);
+ expect(component.isLoading).toBe(false);
+ expect(component.isError).toBe(false);
+ });
+
+ it('surfaces the root exception when the job fails to load', () => {
+ loadJob.mockReturnValue(throwError(() => new Error('boom')));
+ loadExceptions.mockReturnValue(of({ 'root-exception':
'java.lang.RuntimeException: boom' }));
+
+ component.ngOnInit();
+
+ expect(component.isError).toBe(true);
+ expect(component.isLoading).toBe(false);
+ expect(loadExceptions).toHaveBeenCalledWith('job-1', 10);
+ expect(component.errorDetails).toBe('java.lang.RuntimeException: boom');
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/job-local.service.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/job-local.service.spec.ts
new file mode 100644
index 00000000000..31564f0397c
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/job-local.service.spec.ts
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { firstValueFrom } from 'rxjs';
+
+import { JobDetailCorrect, NodesItemCorrect } from
'@flink-runtime-web/interfaces';
+import { beforeEach, describe, expect, it } from 'vitest';
+
+import { JobLocalService, JobWithVertex } from './job-local.service';
+
+describe('JobLocalService', () => {
+ let service: JobLocalService;
+ const job = { plan: { jid: 'job-1' } } as JobDetailCorrect;
+ const vertex = { id: 'v1' } as NodesItemCorrect;
+
+ beforeEach(() => {
+ service = new JobLocalService();
+ });
+
+ it('replays the latest job detail', async () => {
+ service.setJobDetail(job);
+
+ expect(await firstValueFrom(service.jobDetailChanges())).toBe(job);
+ });
+
+ it('replays the latest selected vertex', async () => {
+ service.setSelectedVertex(vertex);
+
+ expect(await firstValueFrom(service.selectedVertexChanges())).toBe(vertex);
+ });
+
+ it('pairs the selected vertex with the latest job detail', async () => {
+ service.setJobDetail(job);
+ service.setSelectedVertex(vertex);
+
+ expect(await firstValueFrom(service.jobWithVertexChanges())).toEqual({
job, vertex });
+ });
+
+ it('suppresses the paired stream while no vertex is selected', () => {
+ service.setJobDetail(job);
+ const emissions: JobWithVertex[] = [];
+ const sub = service.jobWithVertexChanges().subscribe(val =>
emissions.push(val));
+
+ // A null selection must be filtered out; the next real vertex is what
surfaces.
+ service.setSelectedVertex(null);
+ expect(emissions).toEqual([]);
+
+ service.setSelectedVertex(vertex);
+ expect(emissions).toEqual([{ job, vertex }]);
+
+ sub.unsubscribe();
+ });
+
+ it('exposes a metrics cache map', () => {
+ expect(service.metricsCacheMap).toBeInstanceOf(Map);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/job.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/job.component.spec.ts
new file mode 100644
index 00000000000..bfd6381cbe2
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/job.component.spec.ts
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { ActivatedRoute, NavigationEnd, Router } from '@angular/router';
+import { Subject } from 'rxjs';
+
+import { JobsItem } from '@flink-runtime-web/interfaces';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobComponent } from './job.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const activatedRoute = {} as ActivatedRoute;
+const seg = (path: string): { path: string; toString: () => string } => ({
path, toString: () => path });
+
+function urlTree(paths: string[]): unknown {
+ return { root: { children: { primary: { segments: paths.map(seg) } } } };
+}
+
+describe('JobComponent', () => {
+ const events = new Subject<unknown>();
+ const parseUrl = vi.fn();
+ const navigate = vi.fn().mockResolvedValue(true);
+ let component: JobComponent;
+
+ beforeEach(() => {
+ parseUrl.mockReset();
+ navigate.mockClear();
+ const router = { events, url: '/job/running/job-1', parseUrl, navigate }
as unknown as Router;
+ component = new JobComponent(activatedRoute, router, cdr);
+ });
+
+ it('reads the selected job id and running state from the url', () => {
+ parseUrl.mockReturnValue(urlTree(['job', 'running', 'job-1']));
+
+ component.ngOnInit();
+
+ expect(component.jobIdSelected).toBe('job-1');
+ expect(component.isCompleted).toBe(false);
+ expect(component.cardTitle).toBe('Running Jobs');
+ });
+
+ it('recognises the completed list with no job selected', () => {
+ parseUrl.mockReturnValue(urlTree(['job', 'completed']));
+
+ component.ngOnInit();
+
+ expect(component.jobIdSelected).toBeUndefined();
+ expect(component.isCompleted).toBe(true);
+ expect(component.cardTitle).toBe('Completed Jobs');
+ });
+
+ it('re-reads the url on navigation end', () => {
+ parseUrl.mockReturnValue(urlTree(['job', 'running', 'job-1']));
+ component.ngOnInit();
+
+ parseUrl.mockReturnValue(urlTree(['job', 'completed', 'job-2']));
+ events.next(new NavigationEnd(1, '/job/completed/job-2',
'/job/completed/job-2'));
+
+ expect(component.jobIdSelected).toBe('job-2');
+ expect(component.isCompleted).toBe(true);
+ });
+
+ it('navigates to a job relative to the current route', () => {
+ component.navigateToJob({ jid: 'job-9' } as JobsItem);
+
+ expect(navigate).toHaveBeenCalledWith(['job-9'], { relativeTo:
activatedRoute });
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/accumulators/job-overview-drawer-accumulators.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/accumulators/job-overview-drawer-accumulators.component.spec.ts
new file mode 100644
index 00000000000..8b61156b395
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/accumulators/job-overview-drawer-accumulators.component.spec.ts
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of, throwError } from 'rxjs';
+
+import { JobService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobOverviewDrawerAccumulatorsComponent } from
'./job-overview-drawer-accumulators.component';
+import { JobLocalService } from '../../job-local.service';
+
+// The drawer components pull their data through services in ngOnInit and,
when rendered, drag in
+// chart/dynamic-host children jsdom cannot draw. These specs construct the
component directly and
+// assert the controller logic, rather than rendering the template through
TestBed.
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const jobLocalService = {
+ jobWithVertexChanges: () => of({ job: { jid: 'job-1' }, vertex: { id: 'v1' }
})
+} as unknown as JobLocalService;
+
+describe('JobOverviewDrawerAccumulatorsComponent', () => {
+ const loadAccumulators = vi.fn();
+ let component: JobOverviewDrawerAccumulatorsComponent;
+
+ beforeEach(() => {
+ loadAccumulators.mockReset();
+ component = new JobOverviewDrawerAccumulatorsComponent(
+ { loadAccumulators } as unknown as JobService,
+ jobLocalService,
+ cdr
+ );
+ });
+
+ it('flattens the per-subtask user accumulators into individual rows', () => {
+ loadAccumulators.mockReturnValue(
+ of({
+ main: [{ name: 'records', type: 'Long', value: '5' }],
+ subtasks: [{ subtask: 0, 'user-accumulators': [{ name: 'records',
type: 'Long', value: '5' }] }]
+ })
+ );
+
+ component.ngOnInit();
+
+ expect(loadAccumulators).toHaveBeenCalledWith('job-1', 'v1');
+ expect(component.listOfAccumulator).toEqual([{ name: 'records', type:
'Long', value: '5' }]);
+ expect(component.listOfSubTaskAccumulator).toHaveLength(1);
+ expect(component.listOfSubTaskAccumulator[0]).toMatchObject({
+ subtask: 0,
+ name: 'records',
+ type: 'Long',
+ value: '5'
+ });
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('falls back to empty tables when the accumulators request fails', () => {
+ loadAccumulators.mockReturnValue(throwError(() => new Error('boom')));
+
+ component.ngOnInit();
+
+ expect(component.listOfAccumulator).toEqual([]);
+ expect(component.listOfSubTaskAccumulator).toEqual([]);
+ expect(component.isLoading).toBe(false);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.spec.ts
new file mode 100644
index 00000000000..09e32406cba
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.spec.ts
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of, throwError } from 'rxjs';
+
+import { JOB_OVERVIEW_MODULE_DEFAULT_CONFIG } from
'@flink-runtime-web/pages/job/overview/job-overview.config';
+import { JobService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobOverviewDrawerBackpressureComponent } from
'./job-overview-drawer-backpressure.component';
+import { JobLocalService } from '../../job-local.service';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const jobLocalService = {
+ jobWithVertexChanges: () => of({ job: { jid: 'job-1' }, vertex: { id: 'v1' }
})
+} as unknown as JobLocalService;
+
+describe('JobOverviewDrawerBackpressureComponent', () => {
+ const loadOperatorBackPressure = vi.fn();
+ const loadSubTasks = vi.fn();
+ let component: JobOverviewDrawerBackpressureComponent;
+
+ beforeEach(() => {
+ loadOperatorBackPressure.mockReset();
+ loadSubTasks.mockReset().mockReturnValue(of({ subtasks: [] }));
+ component = new JobOverviewDrawerBackpressureComponent(
+ { loadOperatorBackPressure, loadSubTasks } as unknown as JobService,
+ jobLocalService,
+ cdr,
+ JOB_OVERVIEW_MODULE_DEFAULT_CONFIG
+ );
+ });
+
+ it('renders backpressure ratios as rounded percentages and N/A for NaN', ()
=> {
+ expect(component.prettyPrint(NaN)).toBe('N/A');
+ expect(component.prettyPrint(0.1234)).toBe('12%');
+ expect(component.prettyPrint(1)).toBe('100%');
+ });
+
+ it('stores the backpressure response and indexes subtasks by index', () => {
+ loadOperatorBackPressure.mockReturnValue(of({ status: 'ok', subtasks: [{
subtask: 0, ratio: 0.5 }] }));
+ loadSubTasks.mockReturnValue(of({ subtasks: [{ subtask: 0 }, { subtask: 1
}] }));
+
+ component.ngOnInit();
+
+ expect(component.selectedVertex).toEqual({ id: 'v1' });
+ expect(component.backpressure).toEqual({ status: 'ok', subtasks: [{
subtask: 0, ratio: 0.5 }] });
+ expect(component.listOfSubTaskBackpressure).toEqual([{ subtask: 0, ratio:
0.5 }]);
+ expect(component.mapOfSubtask.get(1)).toEqual({ subtask: 1 });
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('falls back to an empty backpressure view when the request fails', () => {
+ loadOperatorBackPressure.mockReturnValue(throwError(() => new
Error('boom')));
+
+ component.ngOnInit();
+
+ expect(component.backpressure).toEqual({});
+ expect(component.listOfSubTaskBackpressure).toEqual([]);
+ expect(component.isLoading).toBe(false);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/chart/job-overview-drawer-chart.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/chart/job-overview-drawer-chart.component.spec.ts
new file mode 100644
index 00000000000..f53c8d84188
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/chart/job-overview-drawer-chart.component.spec.ts
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of } from 'rxjs';
+
+import { MetricsService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobOverviewDrawerChartComponent } from
'./job-overview-drawer-chart.component';
+import { JobLocalService } from '../../job-local.service';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+
+describe('JobOverviewDrawerChartComponent', () => {
+ const loadAllAvailableMetrics = vi.fn();
+ let metricsCacheMap: Map<string, string[]>;
+ let component: JobOverviewDrawerChartComponent;
+
+ beforeEach(() => {
+ loadAllAvailableMetrics.mockReset().mockReturnValue(of([{ id: 'm1' }, {
id: 'm2' }, { id: 'm3' }]));
+ metricsCacheMap = new Map();
+ const jobLocalService = {
+ jobWithVertexChanges: () => of({ job: { jid: 'job-1' }, vertex: { id:
'v1' } }),
+ metricsCacheMap
+ } as unknown as JobLocalService;
+ component = new JobOverviewDrawerChartComponent(
+ { loadAllAvailableMetrics, loadMetrics: vi.fn().mockReturnValue(of({}))
} as unknown as MetricsService,
+ jobLocalService,
+ cdr
+ );
+ });
+
+ it('lists all available metrics as unselected when nothing is cached', () =>
{
+ component.ngOnInit();
+
+ expect(component.cacheMetricKey).toBe('job-1/v1');
+ expect(component.listOfMetricName).toEqual(['m1', 'm2', 'm3']);
+ expect(component.listOfSelectedMetric).toEqual([]);
+ expect(component.listOfUnselectedMetric).toEqual(['m1', 'm2', 'm3']);
+ });
+
+ it('restores the previously selected metrics from the per-vertex cache', ()
=> {
+ metricsCacheMap.set('job-1/v1', ['m2']);
+
+ component.ngOnInit();
+
+ expect(component.listOfSelectedMetric).toEqual(['m2']);
+ expect(component.listOfUnselectedMetric).toEqual(['m1', 'm3']);
+ });
+
+ it('moves a metric between the selected and unselected lists and writes
through to the cache', () => {
+ component.ngOnInit();
+
+ component.updateMetric('m1');
+ expect(component.listOfSelectedMetric).toEqual(['m1']);
+ expect(component.listOfUnselectedMetric).toEqual(['m2', 'm3']);
+ expect(metricsCacheMap.get('job-1/v1')).toEqual(['m1']);
+
+ component.closeMetric('m1');
+ expect(component.listOfSelectedMetric).toEqual([]);
+ expect(component.listOfUnselectedMetric).toEqual(['m1', 'm2', 'm3']);
+ expect(metricsCacheMap.get('job-1/v1')).toEqual([]);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/detail/job-overview-drawer-detail.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/detail/job-overview-drawer-detail.component.spec.ts
new file mode 100644
index 00000000000..206664c180a
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/detail/job-overview-drawer-detail.component.spec.ts
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of } from 'rxjs';
+
+import { NodesItemCorrect } from '@flink-runtime-web/interfaces';
+import { JOB_OVERVIEW_MODULE_DEFAULT_CONFIG } from
'@flink-runtime-web/pages/job/overview/job-overview.config';
+import { describe, expect, it, vi } from 'vitest';
+
+import { JobOverviewDrawerDetailComponent } from
'./job-overview-drawer-detail.component';
+import { JobLocalService } from '../../job-local.service';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+
+function createComponent(node: NodesItemCorrect | null):
JobOverviewDrawerDetailComponent {
+ const jobLocalService = { selectedVertexChanges: () => of(node) } as unknown
as JobLocalService;
+ return new JobOverviewDrawerDetailComponent(jobLocalService, cdr,
JOB_OVERVIEW_MODULE_DEFAULT_CONFIG);
+}
+
+describe('JobOverviewDrawerDetailComponent', () => {
+ it('rewrites <br/> tags in the vertex description to newlines', () => {
+ const component = createComponent({ id: 'v1', description:
'line-a<br/>line-b<br/>line-c' } as NodesItemCorrect);
+
+ component.ngOnInit();
+
+ expect(component.node?.description).toBe('line-a\nline-b\nline-c');
+ });
+
+ it('leaves a description without <br/> untouched', () => {
+ const component = createComponent({ id: 'v1', description: 'plain
description' } as NodesItemCorrect);
+
+ component.ngOnInit();
+
+ expect(component.node?.description).toBe('plain description');
+ });
+
+ it('handles a null selected vertex without throwing', () => {
+ const component = createComponent(null);
+
+ expect(() => component.ngOnInit()).not.toThrow();
+ expect(component.node).toBeNull();
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.spec.ts
new file mode 100644
index 00000000000..f1d882a3665
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.spec.ts
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of, throwError } from 'rxjs';
+
+import { FlameGraphType } from '@flink-runtime-web/interfaces';
+import { JobService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobOverviewDrawerFlameGraphComponent } from
'./job-overview-drawer-flamegraph.component';
+import { JobLocalService } from '../../job-local.service';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const jobLocalService = {
+ jobWithVertexChanges: () => of({ job: { jid: 'job-1' }, vertex: { id: 'v1' }
})
+} as unknown as JobLocalService;
+
+describe('JobOverviewDrawerFlameGraphComponent', () => {
+ const loadOperatorFlameGraph = vi.fn();
+ const loadOperatorFlameGraphForSingleSubtask = vi.fn();
+ const loadSubTasks = vi.fn();
+ let component: JobOverviewDrawerFlameGraphComponent;
+
+ beforeEach(() => {
+ loadOperatorFlameGraph.mockReset().mockReturnValue(of({ endTimestamp: 123
}));
+ loadOperatorFlameGraphForSingleSubtask.mockReset().mockReturnValue(of({
endTimestamp: 456 }));
+ loadSubTasks.mockReset().mockReturnValue(of({ subtasks: [] }));
+ component = new JobOverviewDrawerFlameGraphComponent(
+ { loadOperatorFlameGraph, loadOperatorFlameGraphForSingleSubtask,
loadSubTasks } as unknown as JobService,
+ jobLocalService,
+ cdr
+ );
+ });
+
+ it('offers only running/initializing subtasks (plus "all") as sampleable',
() => {
+ loadSubTasks.mockReturnValue(
+ of({
+ subtasks: [
+ { subtask: 0, status: 'RUNNING' },
+ { subtask: 1, status: 'FINISHED' },
+ { subtask: 2, status: 'INITIALIZING' }
+ ]
+ })
+ );
+
+ component.ngOnInit();
+
+ expect(component.listOfSampleableSubtasks).toEqual(['all', '0', '2']);
+ });
+
+ it('loads the whole-vertex flame graph by default and stores the graph
type', () => {
+ component.ngOnInit();
+
+ expect(loadOperatorFlameGraph).toHaveBeenCalledWith('job-1', 'v1',
FlameGraphType.ON_CPU);
+ expect(component.flameGraph.endTimestamp).toBe(123);
+ expect(component.flameGraph.graphType).toBe(FlameGraphType.ON_CPU);
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('switches to the single-subtask flame graph once a subtask is selected',
() => {
+ component.ngOnInit();
+
+ component.selectSubtask('0');
+
+
expect(loadOperatorFlameGraphForSingleSubtask).toHaveBeenCalledWith('job-1',
'v1', FlameGraphType.ON_CPU, '0');
+ expect(component.flameGraph.endTimestamp).toBe(456);
+ });
+
+ it('keeps just "all" sampleable when the subtasks request fails', () => {
+ loadSubTasks.mockReturnValue(throwError(() => new Error('boom')));
+
+ component.ngOnInit();
+
+ expect(component.listOfSampleableSubtasks).toEqual(['all']);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/subtasks/job-overview-drawer-subtasks.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/subtasks/job-overview-drawer-subtasks.component.spec.ts
new file mode 100644
index 00000000000..6ea4d5f58b2
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/subtasks/job-overview-drawer-subtasks.component.spec.ts
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of, throwError } from 'rxjs';
+
+import { JobVertexStatusDuration } from '@flink-runtime-web/interfaces';
+import { JOB_OVERVIEW_MODULE_DEFAULT_CONFIG } from
'@flink-runtime-web/pages/job/overview/job-overview.config';
+import { JobService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobOverviewDrawerSubtasksComponent } from
'./job-overview-drawer-subtasks.component';
+import { JobLocalService } from '../../job-local.service';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const jobLocalService = {
+ jobWithVertexChanges: () => of({ job: { jid: 'job-1' }, vertex: { id: 'v1' }
})
+} as unknown as JobLocalService;
+
+describe('JobOverviewDrawerSubtasksComponent', () => {
+ const loadSubTasks = vi.fn();
+ let component: JobOverviewDrawerSubtasksComponent;
+
+ beforeEach(() => {
+ loadSubTasks.mockReset();
+ component = new JobOverviewDrawerSubtasksComponent(
+ { loadSubTasks } as unknown as JobService,
+ jobLocalService,
+ cdr,
+ JOB_OVERVIEW_MODULE_DEFAULT_CONFIG
+ );
+ });
+
+ it('loads the subtasks and aggregated metrics for the selected vertex', ()
=> {
+ loadSubTasks.mockReturnValue(of({ subtasks: [{ subtask: 0 }], aggregated:
{ 'read-bytes': { min: 1 } } }));
+
+ component.ngOnInit();
+
+ expect(loadSubTasks).toHaveBeenCalledWith('job-1', 'v1');
+ expect(component.listOfTask).toEqual([{ subtask: 0 }]);
+ expect(component.aggregated).toEqual({ 'read-bytes': { min: 1 } });
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('falls back to an empty task list when the request fails', () => {
+ loadSubTasks.mockReturnValue(throwError(() => new Error('boom')));
+
+ component.ngOnInit();
+
+ expect(component.listOfTask).toEqual([]);
+ expect(component.aggregated).toBeUndefined();
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('orders the status durations by lifecycle stage', () => {
+ const statusDuration = {
+ CREATED: 10,
+ SCHEDULED: 20,
+ DEPLOYING: 30,
+ INITIALIZING: 40,
+ RUNNING: 50
+ } as unknown as JobVertexStatusDuration<number>;
+
+ expect(component.convertStatusDuration(statusDuration)).toEqual([
+ { state: 'CREATED', duration: 10 },
+ { state: 'SCHEDULED', duration: 20 },
+ { state: 'DEPLOYING', duration: 30 },
+ { state: 'INITIALIZING', duration: 40 },
+ { state: 'RUNNING', duration: 50 }
+ ]);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/taskmanagers/job-overview-drawer-taskmanagers.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/taskmanagers/job-overview-drawer-taskmanagers.component.spec.ts
new file mode 100644
index 00000000000..2b628a91341
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/taskmanagers/job-overview-drawer-taskmanagers.component.spec.ts
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of, throwError } from 'rxjs';
+
+import { JOB_OVERVIEW_MODULE_DEFAULT_CONFIG } from
'@flink-runtime-web/pages/job/overview/job-overview.config';
+import { JobService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobOverviewDrawerTaskmanagersComponent } from
'./job-overview-drawer-taskmanagers.component';
+import { JobLocalService } from '../../job-local.service';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const jobLocalService = {
+ jobWithVertexChanges: () => of({ job: { jid: 'job-1' }, vertex: { id: 'v1' }
})
+} as unknown as JobLocalService;
+
+describe('JobOverviewDrawerTaskmanagersComponent', () => {
+ const loadTaskManagers = vi.fn();
+ let component: JobOverviewDrawerTaskmanagersComponent;
+
+ beforeEach(() => {
+ loadTaskManagers.mockReset();
+ component = new JobOverviewDrawerTaskmanagersComponent(
+ { loadTaskManagers } as unknown as JobService,
+ jobLocalService,
+ cdr,
+ JOB_OVERVIEW_MODULE_DEFAULT_CONFIG
+ );
+ });
+
+ it('extracts the taskmanagers list for the selected vertex', () => {
+ loadTaskManagers.mockReturnValue(of({ taskmanagers: [{ endpoint: 'host-a'
}, { endpoint: 'host-b' }] }));
+
+ component.ngOnInit();
+
+ expect(loadTaskManagers).toHaveBeenCalledWith('job-1', 'v1');
+ expect(component.listOfTaskManager).toEqual([{ endpoint: 'host-a' }, {
endpoint: 'host-b' }]);
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('falls back to an empty list when the request fails', () => {
+ loadTaskManagers.mockReturnValue(throwError(() => new Error('boom')));
+
+ component.ngOnInit();
+
+ expect(component.listOfTaskManager).toEqual([]);
+ expect(component.isLoading).toBe(false);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/watermarks/job-overview-drawer-watermarks.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/watermarks/job-overview-drawer-watermarks.component.spec.ts
new file mode 100644
index 00000000000..d20f25f14f3
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/watermarks/job-overview-drawer-watermarks.component.spec.ts
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { of, throwError } from 'rxjs';
+
+import { MetricsService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { JobOverviewDrawerWatermarksComponent } from
'./job-overview-drawer-watermarks.component';
+import { JobLocalService } from '../../job-local.service';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const jobLocalService = {
+ jobWithVertexChanges: () => of({ job: { jid: 'job-1' }, vertex: { id: 'v1' }
})
+} as unknown as JobLocalService;
+
+describe('JobOverviewDrawerWatermarksComponent', () => {
+ const loadWatermarks = vi.fn();
+ let component: JobOverviewDrawerWatermarksComponent;
+
+ beforeEach(() => {
+ loadWatermarks.mockReset();
+ component = new JobOverviewDrawerWatermarksComponent(
+ jobLocalService,
+ { loadWatermarks } as unknown as MetricsService,
+ cdr
+ );
+ });
+
+ it('turns the watermark map into indexed rows', () => {
+ loadWatermarks.mockReturnValue(of({ watermarks: { '0': 100, '1': 200 } }));
+
+ component.ngOnInit();
+
+ expect(loadWatermarks).toHaveBeenCalledWith('job-1', 'v1');
+ expect(component.listOfWaterMark).toEqual([
+ { subTaskIndex: 0, watermark: 100 },
+ { subTaskIndex: 1, watermark: 200 }
+ ]);
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('falls back to an empty table when the watermark request fails', () => {
+ loadWatermarks.mockReturnValue(throwError(() => new Error('boom')));
+
+ component.ngOnInit();
+
+ expect(component.listOfWaterMark).toEqual([]);
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('sorts watermarks ascending', () => {
+ expect(component.sortWatermark({ subTaskIndex: 0, watermark: 5 }, {
subTaskIndex: 1, watermark: 3 })).toBe(2);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/submit/submit.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/submit/submit.component.spec.ts
new file mode 100644
index 00000000000..7136bb34e38
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/submit/submit.component.spec.ts
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { HttpEventType } from '@angular/common/http';
+import { ChangeDetectorRef } from '@angular/core';
+import { UntypedFormBuilder } from '@angular/forms';
+import { Router } from '@angular/router';
+import { of, throwError } from 'rxjs';
+
+import { JarFilesItem } from '@flink-runtime-web/interfaces';
+import { JarService, StatusService } from '@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { SubmitComponent } from './submit.component';
+
+const cdr = { markForCheck: vi.fn(), detectChanges: vi.fn() } as unknown as
ChangeDetectorRef;
+
+describe('SubmitComponent', () => {
+ const loadJarList = vi.fn();
+ const uploadJar = vi.fn();
+ const deleteJar = vi.fn();
+ const runJob = vi.fn();
+ const forceRefresh = vi.fn();
+ const navigate = vi.fn().mockResolvedValue(true);
+ let component: SubmitComponent;
+
+ beforeEach(() => {
+ loadJarList.mockReset().mockReturnValue(of({ files: [], address:
'http://jm', error: undefined }));
+ uploadJar.mockReset();
+ deleteJar.mockReset().mockReturnValue(of(undefined));
+ runJob.mockReset();
+ forceRefresh.mockReset();
+ navigate.mockClear();
+ component = new SubmitComponent(
+ { loadJarList, uploadJar, deleteJar, runJob } as unknown as JarService,
+ { refresh$: of(true), forceRefresh } as unknown as StatusService,
+ new UntypedFormBuilder(),
+ { navigate } as unknown as Router,
+ cdr
+ );
+ });
+
+ it('loads the uploaded jars on init', () => {
+ loadJarList.mockReturnValue(of({ files: [{ id: 'jar-1' }], address:
'http://jm', error: undefined }));
+
+ component.ngOnInit();
+
+ expect(component.listOfJar).toEqual([{ id: 'jar-1' }]);
+ expect(component.address).toBe('http://jm');
+ expect(component.isLoading).toBe(false);
+ expect(component.noAccess).toBe(false);
+ expect(component.isYarn).toBe(false);
+ });
+
+ it('marks no-access when the jar list carries an error', () => {
+ loadJarList.mockReturnValue(of({ files: [], address: '', error:
'forbidden' }));
+
+ component.ngOnInit();
+
+ expect(component.noAccess).toBe(true);
+ });
+
+ it('marks no-access when the jar list request fails', () => {
+ loadJarList.mockReturnValue(throwError(() => new Error('boom')));
+
+ component.ngOnInit();
+
+ expect(component.noAccess).toBe(true);
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('expands a jar and prefills the entry class from its first entry', () => {
+ component.ngOnInit();
+
+ component.expandJar({ id: 'jar-1', entry: [{ name: 'com.example.Main' }] }
as JarFilesItem);
+
+ expect(component.expandedMap.get('jar-1')).toBe(true);
+
expect(component.validateForm.get('entryClass')!.value).toBe('com.example.Main');
+
+ // Expanding the same jar again collapses it.
+ component.expandJar({ id: 'jar-1', entry: [{ name: 'com.example.Main' }] }
as JarFilesItem);
+ expect(component.expandedMap.get('jar-1')).toBe(false);
+ });
+
+ it('refreshes and collapses after deleting a jar', () => {
+ component.ngOnInit();
+ component.expandedMap.set('jar-1', true);
+
+ component.deleteJar({ id: 'jar-1' } as JarFilesItem);
+
+ expect(deleteJar).toHaveBeenCalledWith('jar-1');
+ expect(forceRefresh).toHaveBeenCalled();
+ expect(component.expandedMap.get('jar-1')).toBe(false);
+ });
+
+ it('tracks upload progress from http events', () => {
+ uploadJar.mockReturnValue(of({ type: HttpEventType.UploadProgress, loaded:
50, total: 100 }));
+
+ component.uploadJar(new File([''], 'app.jar'));
+
+ expect(component.isUploading).toBe(true);
+ expect(component.progress).toBe(50);
+ });
+
+ it('navigates to the new job after submitting', () => {
+ component.ngOnInit();
+ runJob.mockReturnValue(of({ jobid: 'job-9' }));
+
+ component.submitJob({ id: 'jar-1' } as JarFilesItem);
+
+ expect(navigate).toHaveBeenCalledWith(['job', 'running', 'job-9']);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/profiler/task-manager-profiler.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/profiler/task-manager-profiler.component.spec.ts
new file mode 100644
index 00000000000..5ebaf56cb1d
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/profiler/task-manager-profiler.component.spec.ts
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { ActivatedRoute } from '@angular/router';
+import { of } from 'rxjs';
+
+import { ProfilingDetail } from '@flink-runtime-web/interfaces/job-profiler';
+import { StatusService, TaskManagerService } from
'@flink-runtime-web/services';
+import { NzMessageService } from 'ng-zorro-antd/message';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { TaskManagerProfilerComponent } from
'./task-manager-profiler.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const activatedRoute = { parent: { snapshot: { params: { taskManagerId: 'tm-1'
} } } } as unknown as ActivatedRoute;
+
+describe('TaskManagerProfilerComponent', () => {
+ const loadProfilingList = vi.fn();
+ const createProfilingInstance = vi.fn();
+ const warning = vi.fn();
+ let component: TaskManagerProfilerComponent;
+
+ beforeEach(() => {
+ loadProfilingList.mockReset().mockReturnValue(of({ profilingList: [] }));
+ createProfilingInstance.mockReset();
+ warning.mockReset();
+ component = new TaskManagerProfilerComponent(
+ { loadProfilingList, createProfilingInstance } as unknown as
TaskManagerService,
+ activatedRoute,
+ { refresh$: of(true) } as unknown as StatusService,
+ { warning } as unknown as NzMessageService,
+ cdr
+ );
+ });
+
+ it('loads the profiling list for the routed task manager on refresh', () => {
+ loadProfilingList.mockReturnValue(of({ profilingList: [{ status:
'FINISHED' } as ProfilingDetail] }));
+
+ component.ngOnInit();
+
+ expect(loadProfilingList).toHaveBeenCalledWith('tm-1');
+ expect(component.profilingList).toEqual([{ status: 'FINISHED' }]);
+ expect(component.isEnabled).toBe(true);
+ expect(component.isLoading).toBe(false);
+ });
+
+ it('creates a profiling instance for the routed task manager and prepends
it', () => {
+ createProfilingInstance.mockReturnValue(of({ status: 'RUNNING' } as
ProfilingDetail));
+
+ component.createProfilingInstance();
+
+ expect(createProfilingInstance).toHaveBeenCalledWith('tm-1', 'ITIMER', 30);
+ expect(component.profilingList[0]).toEqual({ status: 'RUNNING' });
+ expect(component.isCreating).toBe(false);
+ });
+
+ it('refuses to start a new profiling run while one is still running', () => {
+ component.profilingList = [{ status: 'RUNNING' } as ProfilingDetail];
+
+ component.createProfilingInstance();
+
+ expect(warning).toHaveBeenCalled();
+ expect(createProfilingInstance).not.toHaveBeenCalled();
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/stdout/task-manager-stdout.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/stdout/task-manager-stdout.component.spec.ts
new file mode 100644
index 00000000000..f4df0ff0e37
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/stdout/task-manager-stdout.component.spec.ts
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { ActivatedRoute } from '@angular/router';
+import { of, throwError } from 'rxjs';
+
+import { TASK_MANAGER_MODULE_DEFAULT_CONFIG } from
'@flink-runtime-web/pages/task-manager/task-manager.config';
+import { ConfigService, TaskManagerService } from
'@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { TaskManagerStdoutComponent } from './task-manager-stdout.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const activatedRoute = { parent: { snapshot: { params: { taskManagerId: 'tm-1'
} } } } as unknown as ActivatedRoute;
+
+describe('TaskManagerStdoutComponent', () => {
+ const loadStdout = vi.fn();
+ let component: TaskManagerStdoutComponent;
+
+ beforeEach(() => {
+ loadStdout.mockReset();
+ component = new TaskManagerStdoutComponent(
+ { loadStdout } as unknown as TaskManagerService,
+ { BASE_URL: '/api' } as unknown as ConfigService,
+ activatedRoute,
+ cdr,
+ TASK_MANAGER_MODULE_DEFAULT_CONFIG
+ );
+ });
+
+ it('loads stdout for the routed task manager and derives the download
target', () => {
+ loadStdout.mockReturnValue(of('hello stdout\n'));
+
+ component.ngOnInit();
+
+ expect(loadStdout).toHaveBeenCalledWith('tm-1');
+ expect(component.taskManagerId).toBe('tm-1');
+ expect(component.stdout).toBe('hello stdout\n');
+ expect(component.downloadUrl).toBe('/api/taskmanagers/tm-1/stdout');
+ expect(component.downloadName).toBe('taskmanager_tm-1_stdout');
+ expect(component.loading).toBe(false);
+ });
+
+ it('falls back to empty stdout when the request fails', () => {
+ loadStdout.mockReturnValue(throwError(() => new Error('unavailable')));
+
+ component.ngOnInit();
+
+ expect(component.stdout).toBe('');
+ expect(component.loading).toBe(false);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.spec.ts
new file mode 100644
index 00000000000..90994f532ca
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/thread-dump/task-manager-thread-dump.component.spec.ts
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { ActivatedRoute } from '@angular/router';
+import { of, throwError } from 'rxjs';
+
+import { TASK_MANAGER_MODULE_DEFAULT_CONFIG } from
'@flink-runtime-web/pages/task-manager/task-manager.config';
+import { ConfigService, TaskManagerService } from
'@flink-runtime-web/services';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { TaskManagerThreadDumpComponent } from
'./task-manager-thread-dump.component';
+
+const cdr = { markForCheck: vi.fn() } as unknown as ChangeDetectorRef;
+const activatedRoute = {
+ parent: { snapshot: { params: { taskManagerId: 'tm-1' } } },
+ queryParams: of({ vertexName: 'Source%3A%20generator' })
+} as unknown as ActivatedRoute;
+
+describe('TaskManagerThreadDumpComponent', () => {
+ const loadThreadDump = vi.fn();
+ let component: TaskManagerThreadDumpComponent;
+
+ beforeEach(() => {
+ loadThreadDump.mockReset().mockReturnValue(of('thread dump\n'));
+ component = new TaskManagerThreadDumpComponent(
+ { loadThreadDump } as unknown as TaskManagerService,
+ { BASE_URL: '/api' } as unknown as ConfigService,
+ activatedRoute,
+ cdr,
+ TASK_MANAGER_MODULE_DEFAULT_CONFIG
+ );
+ });
+
+ it('derives ids and download target and decodes the vertex name on init', ()
=> {
+ component.ngOnInit();
+
+ expect(component.taskManagerId).toBe('tm-1');
+ expect(component.downloadName).toBe('taskmanager_tm-1_thread_dump');
+ expect(component.downloadUrl).toBe('/api/taskmanagers/tm-1/thread-dump');
+ expect(component.vertexName).toBe('Source: generator');
+ });
+
+ it('loads the dump with the current mode via reload', () => {
+ component.ngOnInit();
+
+ component.reload();
+
+ expect(loadThreadDump).toHaveBeenCalledWith('tm-1', undefined);
+ expect(component.dump).toBe('thread dump\n');
+ expect(component.loading).toBe(false);
+ });
+
+ it('updates mode and download url on selection and reloads with that mode',
() => {
+ component.ngOnInit();
+
+ component.selectMode('full');
+ expect(component.mode).toBe('full');
+
expect(component.downloadUrl).toBe('/api/taskmanagers/tm-1/thread-dump?mode=full');
+
+ component.reload();
+ expect(loadThreadDump).toHaveBeenLastCalledWith('tm-1', 'full');
+ });
+
+ it('falls back to an empty dump when the request fails', () => {
+ loadThreadDump.mockReturnValue(throwError(() => new Error('unavailable')));
+ component.ngOnInit();
+
+ component.reload();
+
+ expect(component.dump).toBe('');
+ expect(component.loading).toBe(false);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/services/config.service.spec.ts
b/flink-runtime-web/web-dashboard/src/app/services/config.service.spec.ts
new file mode 100644
index 00000000000..4324649b16b
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/services/config.service.spec.ts
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { describe, expect, it } from 'vitest';
+
+import { ColorKey, ConfigService } from './config.service';
+
+// Every status the UI can render a badge/timeline bar for. A key missing from
COLOR_MAP would
+// render blank instead of coloured, so this list guards against a status
being added without one.
+const COLOR_KEYS: ColorKey[] = [
+ 'TOTAL',
+ 'RUNNING',
+ 'FAILED',
+ 'FINISHED',
+ 'CANCELED',
+ 'CANCELING',
+ 'CREATED',
+ 'DEPLOYING',
+ 'RECONCILING',
+ 'IN_PROGRESS',
+ 'SCHEDULED',
+ 'COMPLETED',
+ 'RESTARTING',
+ 'PENDING',
+ 'INITIALIZING',
+ 'IGNORED'
+];
+
+describe('ConfigService', () => {
+ const service = new ConfigService();
+
+ it('defaults the REST base url to the current origin', () => {
+ expect(service.BASE_URL).toBe('.');
+ });
+
+ it('maps every known status key to a hex colour with no extras', () => {
+
expect(Object.keys(service.COLOR_MAP).sort()).toEqual([...COLOR_KEYS].sort());
+ for (const key of COLOR_KEYS) {
+ expect(service.COLOR_MAP[key]).toMatch(/^#[0-9a-f]{6}$/i);
+ }
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.spec.ts
b/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.spec.ts
new file mode 100644
index 00000000000..c0e370a5b83
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/services/job-manager.service.spec.ts
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { HttpClient } from '@angular/common/http';
+import { firstValueFrom, of } from 'rxjs';
+
+import { ClusterConfiguration, EnvironmentInfo, JobManagerLogItem } from
'@flink-runtime-web/interfaces';
+import { ProfilingDetail, ProfilingList } from
'@flink-runtime-web/interfaces/job-profiler';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { ConfigService } from './config.service';
+import { JobManagerService } from './job-manager.service';
+
+describe('JobManagerService', () => {
+ let httpClient: { get: ReturnType<typeof vi.fn>; post: ReturnType<typeof
vi.fn> };
+ let service: JobManagerService;
+
+ beforeEach(() => {
+ httpClient = { get: vi.fn(), post: vi.fn() };
+ service = new JobManagerService(httpClient as unknown as HttpClient, new
ConfigService());
+ });
+
+ describe('loadConfig', () => {
+ it('passes the cluster configuration through', async () => {
+ const config = [{ key: 'k', value: 'v' }] as ClusterConfiguration[];
+ httpClient.get.mockReturnValue(of(config));
+
+ const result = await firstValueFrom(service.loadConfig());
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobmanager/config');
+ expect(result).toBe(config);
+ });
+ });
+
+ describe('loadEnvironment', () => {
+ it('passes the environment info through', async () => {
+ const env = { jvm: {} } as unknown as EnvironmentInfo;
+ httpClient.get.mockReturnValue(of(env));
+
+ const result = await firstValueFrom(service.loadEnvironment());
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobmanager/environment');
+ expect(result).toBe(env);
+ });
+ });
+
+ describe('loadLogs / loadStdout', () => {
+ it('requests the plain log text with a no-cache header', async () => {
+ httpClient.get.mockReturnValue(of('log contents'));
+
+ const result = await firstValueFrom(service.loadLogs());
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobmanager/log',
expect.anything());
+ expect(result).toBe('log contents');
+ });
+
+ it('requests the plain stdout text', async () => {
+ httpClient.get.mockReturnValue(of('stdout contents'));
+
+ const result = await firstValueFrom(service.loadStdout());
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobmanager/stdout',
expect.anything());
+ expect(result).toBe('stdout contents');
+ });
+ });
+
+ describe('loadLogList', () => {
+ it('extracts the logs array from the response', async () => {
+ const log = { name: 'jobmanager.log' } as JobManagerLogItem;
+ httpClient.get.mockReturnValue(of({ logs: [log] }));
+
+ const result = await firstValueFrom(service.loadLogList());
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobmanager/logs');
+ expect(result).toEqual([log]);
+ });
+ });
+
+ describe('loadLog', () => {
+ it('pairs the raw log text with the request url', async () => {
+ httpClient.get.mockReturnValue(of('log contents'));
+
+ const result = await firstValueFrom(service.loadLog('jobmanager.log'));
+
+ expect(result.data).toBe('log contents');
+ expect(result.url).toBe('./jobmanager/logs/jobmanager.log');
+ });
+ });
+
+ describe('loadThreadDump', () => {
+ it('joins the stringified thread infos into a single dump', async () => {
+ httpClient.get.mockReturnValue(
+ of({ threadInfos: [{ stringifiedThreadInfo: 'thread-A\n' }, {
stringifiedThreadInfo: 'thread-B\n' }] })
+ );
+
+ const result = await firstValueFrom(service.loadThreadDump());
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobmanager/thread-dump');
+ expect(result).toBe('thread-A\nthread-B\n');
+ });
+
+ it('appends the mode query parameter when provided', async () => {
+ httpClient.get.mockReturnValue(of({ threadInfos: [] }));
+
+ await firstValueFrom(service.loadThreadDump('full'));
+
+
expect(httpClient.get).toHaveBeenCalledWith('./jobmanager/thread-dump?mode=full');
+ });
+ });
+
+ describe('loadMetricsName', () => {
+ it('maps the metric descriptors down to their ids', async () => {
+ httpClient.get.mockReturnValue(of([{ id: 'Status.JVM.CPU.Load' }, { id:
'Status.JVM.Memory.Heap.Used' }]));
+
+ const result = await firstValueFrom(service.loadMetricsName());
+
+ expect(result).toEqual(['Status.JVM.CPU.Load',
'Status.JVM.Memory.Heap.Used']);
+ });
+ });
+
+ describe('loadMetrics', () => {
+ it('parses the metric values into a numeric map keyed by id', async () => {
+ httpClient.get.mockReturnValue(of([{ id: 'Status.JVM.CPU.Load', value:
'0.42' }]));
+
+ const result = await
firstValueFrom(service.loadMetrics(['Status.JVM.CPU.Load']));
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobmanager/metrics', {
params: { get: 'Status.JVM.CPU.Load' } });
+ expect(result).toEqual({ 'Status.JVM.CPU.Load': 0.42 });
+ });
+ });
+
+ describe('loadHistoryServerConfig / loadHistoryServerEnvironment /
loadHistoryServerJobManagerLogUrl', () => {
+ it('requests the history-server cluster configuration for a job', async ()
=> {
+ const config = [] as ClusterConfiguration[];
+ httpClient.get.mockReturnValue(of(config));
+
+ const result = await
firstValueFrom(service.loadHistoryServerConfig('job-1'));
+
+
expect(httpClient.get).toHaveBeenCalledWith('./jobs/job-1/jobmanager/config');
+ expect(result).toBe(config);
+ });
+
+ it('requests the history-server environment for a job', async () => {
+ const env = {} as EnvironmentInfo;
+ httpClient.get.mockReturnValue(of(env));
+
+ const result = await
firstValueFrom(service.loadHistoryServerEnvironment('job-1'));
+
+
expect(httpClient.get).toHaveBeenCalledWith('./jobs/job-1/jobmanager/environment');
+ expect(result).toBe(env);
+ });
+
+ it('extracts the history-server log url for a job', async () => {
+ httpClient.get.mockReturnValue(of({ url: 'http://history/log' }));
+
+ const result = await
firstValueFrom(service.loadHistoryServerJobManagerLogUrl('job-1'));
+
+
expect(httpClient.get).toHaveBeenCalledWith('./jobs/job-1/jobmanager/log-url');
+ expect(result).toBe('http://history/log');
+ });
+ });
+
+ describe('profiling', () => {
+ it('passes the profiling list through', async () => {
+ const list = { profiling: [] } as unknown as ProfilingList;
+ httpClient.get.mockReturnValue(of(list));
+
+ const result = await firstValueFrom(service.loadProfilingList());
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobmanager/profiler');
+ expect(result).toBe(list);
+ });
+
+ it('posts the requested profiling mode and duration', async () => {
+ const detail = { status: 'RUNNING' } as ProfilingDetail;
+ httpClient.post.mockReturnValue(of(detail));
+
+ const result = await
firstValueFrom(service.createProfilingInstance('ITIMER', 30));
+
+ expect(httpClient.post).toHaveBeenCalledWith('./jobmanager/profiler', {
mode: 'ITIMER', duration: 30 });
+ expect(result).toBe(detail);
+ });
+
+ it('pairs the raw profiling result text with the request url', async () =>
{
+ httpClient.get.mockReturnValue(of('profile contents'));
+
+ const result = await
firstValueFrom(service.loadProfilingResult('profile.jfr'));
+
+ expect(result.data).toBe('profile contents');
+ expect(result.url).toBe('./jobmanager/profiler/profile.jfr');
+ });
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/services/job.service.spec.ts
b/flink-runtime-web/web-dashboard/src/app/services/job.service.spec.ts
new file mode 100644
index 00000000000..9da4661ed15
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/services/job.service.spec.ts
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { HttpClient } from '@angular/common/http';
+import { firstValueFrom, of, throwError, toArray } from 'rxjs';
+
+import { JobDetail, JobOverview } from '@flink-runtime-web/interfaces';
+import { JobResourceRequirements } from
'@flink-runtime-web/interfaces/job-resource-requirements';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { ConfigService } from './config.service';
+import { JobService } from './job.service';
+
+describe('JobService', () => {
+ let httpClient: { get: ReturnType<typeof vi.fn>; put: ReturnType<typeof
vi.fn> };
+ let service: JobService;
+
+ beforeEach(() => {
+ httpClient = { get: vi.fn(), put: vi.fn() };
+ service = new JobService(httpClient as unknown as HttpClient, new
ConfigService());
+ });
+
+ describe('loadJobs', () => {
+ it('uppercases task-state keys, folds in pending operators, and derives
the completed flag', async () => {
+ const overview = {
+ jobs: [
+ { jid: 'job-1', state: 'RUNNING', tasks: { running: 2, finished: 1
}, 'pending-operators': 4 },
+ { jid: 'job-2', state: 'FINISHED', tasks: {} }
+ ]
+ } as unknown as JobOverview;
+ httpClient.get.mockReturnValue(of(overview));
+
+ const result = await firstValueFrom(service.loadJobs());
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobs/overview');
+ expect(result[0].tasks).toEqual({ RUNNING: 2, FINISHED: 1, PENDING: 4 });
+ expect(result[0].completed).toBe(false);
+ // No pending-operators field falls back to 0, and FINISHED counts as
completed.
+ expect(result[1].tasks).toEqual({ PENDING: 0 });
+ expect(result[1].completed).toBe(true);
+ });
+
+ it('swallows request errors into an empty stream', async () => {
+ httpClient.get.mockReturnValue(throwError(() => new Error('boom')));
+
+ const emissions = await
firstValueFrom(service.loadJobs().pipe(toArray()));
+
+ expect(emissions).toEqual([]);
+ });
+ });
+
+ describe('loadJob', () => {
+ it('builds graph nodes/links from the plan and stream graph and sorts
nodes by vertex order', async () => {
+ const job = {
+ jid: 'job-1',
+ vertices: [{ id: 'v1' }, { id: 'v2' }],
+ plan: {
+ jid: 'job-1',
+ // Deliberately out of vertex order so the sort has something to do.
+ nodes: [{ id: 'v2', inputs: [{ id: 'v1' }] }, { id: 'v1' }]
+ },
+ 'stream-graph': { nodes: [{ id: 'sn1', inputs: [{ id: 'sn0' }] }, {
id: 'sn0' }] },
+ 'status-counts': {},
+ 'pending-operators': 3
+ } as unknown as JobDetail;
+ httpClient.get.mockReturnValue(of(job));
+
+ const result = await firstValueFrom(service.loadJob('job-1'));
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobs/job-1');
+ expect(result.plan.nodes.map(n => n.id)).toEqual(['v1', 'v2']);
+ expect(result.plan.nodes[0]).toMatchObject({ id: 'v1', job_vertex_id:
'v1', detail: { id: 'v1' } });
+ expect(result.plan.links).toEqual([{ id: 'v1-v2', source: 'v1', target:
'v2' }]);
+ expect(result.plan.streamNodes.map(n => n.id)).toEqual(['sn1', 'sn0']);
+ expect(result.plan.streamLinks).toEqual([{ id: 'sn0-sn1', source: 'sn0',
target: 'sn1' }]);
+ // The stream graph carries pending operators into the status counts.
+ expect(result['status-counts']['PENDING']).toBe(3);
+ });
+
+ it('swallows request errors into an empty stream', async () => {
+ httpClient.get.mockReturnValue(throwError(() => new Error('boom')));
+
+ const emissions = await
firstValueFrom(service.loadJob('job-1').pipe(toArray()));
+
+ expect(emissions).toEqual([]);
+ });
+ });
+
+ describe('loadAccumulators', () => {
+ it('combines the user and subtask accumulator responses', async () => {
+ httpClient.get
+ .mockReturnValueOnce(of({ 'user-accumulators': [{ name: 'records' }]
}))
+ .mockReturnValueOnce(of({ subtasks: [{ subtask: 0 }] }));
+
+ const result = await firstValueFrom(service.loadAccumulators('job-1',
'v1'));
+
+ expect(httpClient.get).toHaveBeenNthCalledWith(1,
'./jobs/job-1/vertices/v1/accumulators');
+ expect(httpClient.get).toHaveBeenNthCalledWith(2,
'./jobs/job-1/vertices/v1/subtasks/accumulators');
+ expect(result).toEqual({ main: [{ name: 'records' }], subtasks: [{
subtask: 0 }] });
+ });
+ });
+
+ describe('changeDesiredParallelism', () => {
+ it('raises the upper bound only for the requested vertices and puts the
updated requirements', async () => {
+ const requirements: JobResourceRequirements = {
+ v1: { parallelism: { lowerBound: 1, upperBound: 2 } },
+ v2: { parallelism: { lowerBound: 1, upperBound: 2 } }
+ };
+ httpClient.get.mockReturnValue(of(requirements));
+ httpClient.put.mockReturnValue(of(undefined));
+
+ await firstValueFrom(service.changeDesiredParallelism('job-1', new
Map([['v1', 4]])));
+
+
expect(httpClient.put).toHaveBeenCalledWith('./jobs/job-1/resource-requirements',
{
+ v1: { parallelism: { lowerBound: 1, upperBound: 4 } },
+ v2: { parallelism: { lowerBound: 1, upperBound: 2 } }
+ });
+ });
+ });
+
+ describe('URL construction', () => {
+ it('cancels a job through the yarn-cancel handler', async () => {
+ httpClient.get.mockReturnValue(of(undefined));
+
+ await firstValueFrom(service.cancelJob('job-1'));
+
+ expect(httpClient.get).toHaveBeenCalledWith('./jobs/job-1/yarn-cancel');
+ });
+
+ it('passes the max-exceptions cap as a query parameter', async () => {
+ httpClient.get.mockReturnValue(of({}));
+
+ await firstValueFrom(service.loadExceptions('job-1', 25));
+
+
expect(httpClient.get).toHaveBeenCalledWith('./jobs/job-1/exceptions?maxExceptions=25');
+ });
+
+ it('embeds the checkpoint id in the details path', async () => {
+ httpClient.get.mockReturnValue(of({}));
+
+ await firstValueFrom(service.loadCheckpointDetails('job-1', 7));
+
+
expect(httpClient.get).toHaveBeenCalledWith('./jobs/job-1/checkpoints/details/7');
+ });
+
+ it('passes both the flame-graph type and subtask index', async () => {
+ httpClient.get.mockReturnValue(of({}));
+
+ await
firstValueFrom(service.loadOperatorFlameGraphForSingleSubtask('job-1', 'v1',
'ON_CPU', '3'));
+
+
expect(httpClient.get).toHaveBeenCalledWith('./jobs/job-1/vertices/v1/flamegraph?type=ON_CPU&subtaskindex=3');
+ });
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/services/overview.service.spec.ts
b/flink-runtime-web/web-dashboard/src/app/services/overview.service.spec.ts
new file mode 100644
index 00000000000..0f271ae62ab
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/services/overview.service.spec.ts
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { HttpClient } from '@angular/common/http';
+import { firstValueFrom, of, throwError, toArray } from 'rxjs';
+
+import { Overview } from '@flink-runtime-web/interfaces';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { ConfigService } from './config.service';
+import { OverviewService } from './overview.service';
+
+describe('OverviewService', () => {
+ let httpClient: { get: ReturnType<typeof vi.fn> };
+ let service: OverviewService;
+
+ beforeEach(() => {
+ httpClient = { get: vi.fn() };
+ service = new OverviewService(httpClient as unknown as HttpClient, new
ConfigService());
+ });
+
+ it('passes the cluster overview through on success', async () => {
+ const overview = { 'slots-total': 12 } as Overview;
+ httpClient.get.mockReturnValue(of(overview));
+
+ const result = await firstValueFrom(service.loadOverview());
+
+ expect(httpClient.get).toHaveBeenCalledWith('./overview');
+ expect(result).toBe(overview);
+ });
+
+ it('swallows request errors into an empty stream', async () => {
+ httpClient.get.mockReturnValue(throwError(() => new Error('cluster
unreachable')));
+
+ const emissions = await
firstValueFrom(service.loadOverview().pipe(toArray()));
+
+ expect(emissions).toEqual([]);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/services/status.service.spec.ts
b/flink-runtime-web/web-dashboard/src/app/services/status.service.spec.ts
new file mode 100644
index 00000000000..40186164efe
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/services/status.service.spec.ts
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { HttpClient } from '@angular/common/http';
+import { ChangeDetectorRef } from '@angular/core';
+import { TestBed } from '@angular/core/testing';
+import { Router } from '@angular/router';
+import { EMPTY, firstValueFrom, of } from 'rxjs';
+
+import { Configuration } from '@flink-runtime-web/interfaces';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { ConfigService } from './config.service';
+import { StatusService } from './status.service';
+
+const mockConfig = { 'refresh-interval': 1000, features: {}, 'flink-version':
'2.4', 'flink-revision': 'abc' };
+
+describe('StatusService', () => {
+ let service: StatusService;
+ const get = vi.fn();
+
+ beforeEach(() => {
+ get.mockReset().mockReturnValue(of(mockConfig));
+ TestBed.configureTestingModule({
+ providers: [
+ StatusService,
+ { provide: HttpClient, useValue: { get } },
+ { provide: ConfigService, useValue: new ConfigService() },
+ { provide: Router, useValue: { events: EMPTY } }
+ ]
+ });
+ service = TestBed.inject(StatusService);
+ });
+
+ it('starts with no cached errors and a zero network-failure count', () => {
+ expect(service.listOfErrorMessage).toEqual([]);
+ expect(service.networkFailureCount).toBe(0);
+ expect(service.networkFailureThreshold).toBe(5);
+ expect(service.networkErrorNotificationId).toBeNull();
+ });
+
+ it('forwards markAppForCheck to the registered change detector', () => {
+ const markForCheck = vi.fn();
+ service.registerAppCdr({ markForCheck } as unknown as ChangeDetectorRef);
+
+ service.markAppForCheck();
+
+ expect(markForCheck).toHaveBeenCalledOnce();
+ });
+
+ it('is a no-op when markAppForCheck runs before any detector is registered',
() => {
+ expect(() => service.markAppForCheck()).not.toThrow();
+ });
+
+ it('caches the fetched configuration on boot', async () => {
+ const result = await firstValueFrom(service.boot());
+
+ expect(get).toHaveBeenCalledWith('./config');
+ expect(result).toEqual(mockConfig);
+ expect(service.configuration).toEqual(mockConfig as unknown as
Configuration);
+ });
+
+ it('emits on the refresh stream once it is wired up by boot', async () => {
+ vi.useFakeTimers();
+ try {
+ await firstValueFrom(service.boot());
+
+ const emitted: boolean[] = [];
+ const sub = service.refresh$.subscribe(value => emitted.push(value));
+ // The merged trigger stream is debounced by 300ms before the interval
kicks in.
+ vi.advanceTimersByTime(300);
+ sub.unsubscribe();
+
+ expect(emitted).toContain(true);
+ } finally {
+ vi.useRealTimers();
+ }
+ });
+});
