package com.battery.business.service.impl;
|
|
import static org.junit.jupiter.api.Assertions.assertFalse;
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
import static org.mockito.ArgumentMatchers.any;
|
import static org.mockito.ArgumentMatchers.anyInt;
|
import static org.mockito.Mockito.mock;
|
import static org.mockito.Mockito.never;
|
import static org.mockito.Mockito.verify;
|
import static org.mockito.Mockito.when;
|
|
import java.io.IOException;
|
import java.math.BigDecimal;
|
import java.time.Duration;
|
import java.time.Instant;
|
import java.util.List;
|
import org.awaitility.Awaitility;
|
import org.junit.jupiter.api.Test;
|
import org.springframework.web.context.request.async.AsyncRequestNotUsableException;
|
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
|
import com.battery.device.api.TestDataObservationAdapter;
|
import com.battery.business.web.TestLiveStream;
|
import com.battery.business.service.ITestLiveService;
|
import com.battery.device.api.discovery.DeviceDiscoveryCandidateStore;
|
import com.battery.device.api.model.ChannelIdentity;
|
import com.battery.device.api.model.DeviceWorkStatus;
|
import com.battery.device.api.model.RealtimeSnapshot;
|
import com.battery.business.domain.TTest;
|
import com.battery.business.domain.TTestBattery;
|
import com.battery.business.domain.TTestChannelBinding;
|
import com.battery.business.domain.TTestWarningException;
|
import com.battery.business.mapper.TTestMapper;
|
import com.battery.business.service.ITTestService;
|
import com.battery.business.service.ITestLiveService.LiveState;
|
|
/** 验证当前测量、实时提醒与历史待办独立，以及实时流断连后的资源清理。 */
|
class TTestLiveServiceImplTest
|
{
|
@Test
|
void thresholdWarningsRespectDeviceTimeAndReceptionBoundary()
|
{
|
Instant now = Instant.parse("2026-09-05T12:00:00Z");
|
TTest test = test("RECEIVING");
|
test.setStartedAtUtc(java.util.Date.from(now));
|
test.setReceptionBoundaryAtUtc(java.util.Date.from(now));
|
var rule = new com.battery.business.domain.TTestWarningThreshold();
|
rule.setThresholdId(11L); rule.setMetricCode("VOLTAGE"); rule.setMaxValue(new BigDecimal("3"));
|
test.setThresholds(List.of(rule));
|
assertTrue(liveService(test, now, now.minusSeconds(1)).current(10L).exceptions().isEmpty());
|
assertEquals(1, liveService(test, now, now).current(10L).exceptions().size());
|
test.setReceptionBoundaryAtUtc(java.util.Date.from(now.plusSeconds(1)));
|
assertTrue(liveService(test, now, now).current(10L).exceptions().isEmpty());
|
|
var battery = test.getBatteries().get(0); battery.setCollectorId(7L); battery.setBinding(null);
|
var binding = new com.battery.business.domain.TTestCollectorChannel();
|
binding.setChannelCode("S-01"); binding.setMeasurementCode("PRIMARY_TEMPERATURE");
|
battery.setCollectorChannels(List.of(binding));
|
rule.setMetricCode("TEMPERATURE"); rule.setMaxValue(new BigDecimal("33"));
|
var tests = mock(ITTestService.class); when(tests.selectTestById(10L)).thenReturn(test);
|
var mapper = mock(TTestMapper.class); when(mapper.selectCollectorCode(7L)).thenReturn("collector-7");
|
var adapter = mock(TestDataObservationAdapter.class);
|
var service = new TTestLiveServiceImpl(tests,
|
new com.battery.device.api.discovery.InMemoryDeviceDiscoveryCandidateStore(), adapter, mapper, null);
|
test.setReceptionBoundaryAtUtc(java.util.Date.from(now));
|
for (Instant deviceTime : List.of(now.minusSeconds(1), now)) {
|
var measurement = new TestDataObservationAdapter.CollectorMeasurement("S-01", 1L, deviceTime, deviceTime.toString(),
|
now, "TEMPERATURE", new BigDecimal("38"), "℃");
|
when(adapter.latestCollector(org.mockito.ArgumentMatchers.nullable(String.class), org.mockito.ArgumentMatchers.eq("collector-7"))).thenReturn(new TestDataObservationAdapter.CollectorRealtime(
|
"collector-7", true, 1000, List.of(measurement), null));
|
assertEquals(deviceTime.equals(now) ? 1 : 0, service.current(10L).exceptions().size());
|
}
|
test.setReceptionBoundaryAtUtc(java.util.Date.from(now.plusSeconds(1)));
|
assertTrue(service.current(10L).exceptions().isEmpty());
|
}
|
|
@Test
|
void currentTemperatureAlarmIsIndependentOfHistoricalTodoHandling() throws Exception
|
{
|
TTest test = test("RECEIVING");
|
var battery = test.getBatteries().get(0); battery.setCollectorId(7L);
|
var primary = new com.battery.business.domain.TTestCollectorChannel();
|
primary.setChannelCode("S-01"); primary.setMeasurementCode("PRIMARY_TEMPERATURE");
|
var secondary = new com.battery.business.domain.TTestCollectorChannel();
|
secondary.setChannelCode("S-03"); secondary.setMeasurementCode("SECONDARY_TEMPERATURE_2");
|
battery.setCollectorChannels(List.of(primary, secondary));
|
var rule = new com.battery.business.domain.TTestWarningThreshold();
|
rule.setThresholdId(11L); rule.setMetricCode("TEMPERATURE");
|
rule.setMinValue(new BigDecimal("8")); rule.setMaxValue(new BigDecimal("33"));
|
test.setThresholds(List.of(rule));
|
var tests = mock(ITTestService.class); when(tests.selectTestById(10L)).thenReturn(test);
|
var mapper = mock(TTestMapper.class); when(mapper.selectCollectorCode(7L)).thenReturn("dev-sim-collector-1");
|
var adapter = mock(TestDataObservationAdapter.class);
|
Instant now = Instant.now();
|
// 采集器来源交出主、辅助两路测量：只有绑定了的 S-01 主温度进预警。
|
when(adapter.latestCollector(org.mockito.ArgumentMatchers.nullable(String.class),
|
org.mockito.ArgumentMatchers.eq("dev-sim-collector-1"))).thenReturn(collectorRealtime(now, "38"));
|
var service = new TTestLiveServiceImpl(tests,
|
new com.battery.device.api.discovery.InMemoryDeviceDiscoveryCandidateStore(), adapter, mapper, null);
|
var alarms = service.current(10L).exceptions().stream().filter(item -> "THRESHOLD".equals(item.get("exceptionType"))).toList();
|
assertEquals(1, alarms.size());
|
assertEquals(new BigDecimal("38"), alarms.get(0).get("value"));
|
assertEquals("7:S-01", alarms.get(0).get("sourceIdentity"));
|
assertFalse(alarms.get(0).containsKey("exceptionId"));
|
|
var history = new TTestWarningException(); history.setExceptionStatus("OPEN");
|
history.setExceptionType("THRESHOLD"); history.setMetricCode("temperature");
|
history.setAbnormalValue(new BigDecimal("38"));
|
when(mapper.selectWarningExceptions(10L, "OPEN")).thenReturn(List.of(history));
|
when(adapter.latestCollector(org.mockito.ArgumentMatchers.nullable(String.class),
|
org.mockito.ArgumentMatchers.eq("dev-sim-collector-1"))).thenReturn(collectorRealtime(now, "25"));
|
assertTrue(service.current(10L).exceptions().stream().noneMatch(item -> "THRESHOLD".equals(item.get("exceptionType"))));
|
assertEquals("OPEN", history.getExceptionStatus());
|
}
|
|
private static TestDataObservationAdapter.CollectorRealtime collectorRealtime(Instant now, String temperature)
|
{
|
return new TestDataObservationAdapter.CollectorRealtime("dev-sim-collector-1", true, 1000,
|
List.of(new TestDataObservationAdapter.CollectorMeasurement("S-01", 1L, now, now.toString(), now,
|
"TEMPERATURE", new BigDecimal(temperature), "℃")), null);
|
}
|
|
@Test
|
void consoleReadsConfiguredPlanAtCurrentChannelPosition() throws Exception
|
{
|
var test = test("RECEIVING");
|
var binding = test.getBatteries().get(0).getBinding();
|
binding.setDevtype(22); binding.setSubdevid(1);
|
var tests = mock(ITTestService.class);
|
when(tests.selectTestById(10L)).thenReturn(test);
|
var adapter = mock(TestDataObservationAdapter.class);
|
// 来源交出它自己的方案：当前循环 2、当前工步 2、计划 3 个循环，工步行按循环位置排列。
|
when(adapter.testPlan(any(ChannelIdentity.class))).thenReturn(new TestDataObservationAdapter.TestPlan(2, 2, 3,
|
List.of(new TestDataObservationAdapter.PlanRow("STEP", 1, 1, "充电", 60, 4.2, "已完成", null, null),
|
new TestDataObservationAdapter.PlanRow("STEP", 2, 2, "放电", 90, 3.0, "进行中", null, null))));
|
var service = new TTestLiveServiceImpl(tests,
|
new com.battery.device.api.discovery.InMemoryDeviceDiscoveryCandidateStore(), adapter, null, null);
|
var plan = service.current(10L).testPlan();
|
assertEquals(2, plan.currentCycleNo());
|
assertEquals(2, plan.currentStepNo());
|
assertEquals(3, plan.plannedCycleCount());
|
assertEquals(List.of(1, 2), plan.rows().stream().map(row -> row.cyclePosition()).toList());
|
assertEquals("放电", plan.rows().get(1).stepName());
|
assertEquals(90d, plan.rows().get(1).durationSeconds());
|
assertEquals("进行中", plan.rows().get(1).executionStatus());
|
when(adapter.testPlan(any(ChannelIdentity.class))).thenReturn(null);
|
org.junit.jupiter.api.Assertions.assertNull(service.current(10L).testPlan());
|
}
|
|
@Test
|
void offlineHintIsIndependentOfHistoricalHandlingStatusAndSurvivesRefresh()
|
{
|
TTest test = test("RECEIVING");
|
var mapper = mock(TTestMapper.class);
|
TTestWarningException warning = new TTestWarningException();
|
warning.setExceptionId(21L);
|
warning.setExceptionType("DEVICE_OFFLINE");
|
warning.setSourceType("CABINET_CHANNEL");
|
warning.setSourceIdentity("dev-sim:1:1:0:1");
|
warning.setMetricCode("DEVICE_ONLINE");
|
warning.setExceptionStatus("OPEN");
|
when(mapper.selectWarningExceptions(10L, "OPEN")).thenReturn(List.of(warning));
|
var service = liveService(test, Instant.now(), false, DeviceWorkStatus.WORKING, mapper);
|
|
for (int attempt = 0; attempt < 2; attempt++) {
|
var alarms = service.current(10L).exceptions();
|
assertEquals(1, alarms.size());
|
assertFalse(alarms.get(0).containsKey("exceptionId"));
|
assertEquals("DEVICE_OFFLINE", alarms.get(0).get("exceptionType"));
|
assertFalse(alarms.get(0).get("message").toString().contains("处理"));
|
}
|
warning.setExceptionStatus("CLOSED");
|
when(mapper.selectWarningExceptions(10L, "OPEN")).thenReturn(List.of());
|
assertEquals(1, service.current(10L).exceptions().size());
|
test.setReceptionStatus("PAUSED");
|
assertTrue(service.current(10L).exceptions().isEmpty());
|
test.setReceptionStatus("ENDED");
|
assertTrue(service.current(10L).exceptions().isEmpty());
|
}
|
|
@Test
|
void recoveredSourcesHideOfflineHintsEvenWhenHistoryRemainsOpen()
|
{
|
TTest test = test("RECEIVING");
|
test.getBatteries().get(0).setCollectorId(7L);
|
var mapper = mock(TTestMapper.class);
|
var warnings = List.of("CABINET_CHANNEL", "COLLECTOR").stream().map(sourceType -> {
|
var warning = new TTestWarningException();
|
warning.setExceptionType("DEVICE_OFFLINE");
|
warning.setSourceType(sourceType);
|
warning.setSourceIdentity("COLLECTOR".equals(sourceType) ? "7" : "dev-sim:1:1:0:1");
|
warning.setExceptionStatus("OPEN");
|
return warning;
|
}).toList();
|
when(mapper.selectWarningExceptions(10L, "OPEN")).thenReturn(warnings);
|
|
for (boolean online : List.of(false, true, true, false)) {
|
var service = liveService(test, Instant.now(), online, DeviceWorkStatus.WORKING, mapper);
|
var state = service.current(10L);
|
assertTrue(state.sources().stream().allMatch(source -> (online ? "ONLINE" : "OFFLINE").equals(source.deviceStatus())));
|
assertEquals(online ? 0 : 2, state.exceptions().size());
|
assertTrue(warnings.stream().allMatch(warning -> "OPEN".equals(warning.getExceptionStatus())));
|
}
|
}
|
|
@Test
|
void receivingOfflineCabinetAndCollectorAreVisibleAsOpenAlarms()
|
{
|
TTest test = test("RECEIVING");
|
test.getBatteries().get(0).setCollectorId(7L);
|
var service = liveService(test, Instant.now(), false);
|
|
var alarms = service.current(10L).exceptions();
|
|
assertEquals(2, alarms.size());
|
assertTrue(alarms.stream().allMatch(item -> "OPEN".equals(item.get("status"))));
|
assertTrue(alarms.stream().anyMatch(item -> "COLLECTOR".equals(item.get("sourceType"))));
|
assertTrue(alarms.stream().anyMatch(item -> "CABINET_CHANNEL".equals(item.get("sourceType"))));
|
}
|
|
/**
|
* 非接收状态只读来源可达性：状态取自所链接来源实例的连接结果，且**不请求**任何测量，
|
* 否则暂停期间仍会产生新观测。
|
*/
|
@Test
|
void nonReceivingConsoleReadsCollectorStatusWithoutRequestingMeasurements() throws Exception
|
{
|
TTest test = test("RECEIVING");
|
test.getBatteries().get(0).setCollectorId(7L);
|
var tests = mock(ITTestService.class);
|
when(tests.selectTestById(10L)).thenReturn(test);
|
var mapper = mock(TTestMapper.class);
|
when(mapper.selectCollectorCode(7L)).thenReturn("dev-sim-collector-1");
|
var adapter = mock(TestDataObservationAdapter.class);
|
var adapters = mock(com.battery.business.mapper.TDeviceAdapterMapper.class);
|
var source = new com.battery.business.domain.TDeviceAdapter();
|
source.setConnectionStatus("ONLINE");
|
when(adapters.selectDeviceAdapterByCode("dev-sim")).thenReturn(source);
|
var service = new TTestLiveServiceImpl(tests,
|
new com.battery.device.api.discovery.InMemoryDeviceDiscoveryCandidateStore(), adapter, mapper, adapters);
|
test.setReceptionStatus("RECEIVING");
|
var receiving = service.current(10L).sources().stream()
|
.filter(item -> item.sourceType().equals("COLLECTOR")).findFirst().orElseThrow();
|
assertEquals("ONLINE", receiving.deviceStatus());
|
verify(adapter).latestCollector(org.mockito.ArgumentMatchers.eq("dev-sim"),
|
org.mockito.ArgumentMatchers.eq("dev-sim-collector-1"));
|
org.mockito.Mockito.clearInvocations(adapter);
|
for (String status : List.of("PAUSED", "ENDED", "NOT_STARTED"))
|
{
|
test.setReceptionStatus(status);
|
var paused = service.current(10L).sources().stream()
|
.filter(item -> item.sourceType().equals("COLLECTOR")).findFirst().orElseThrow();
|
assertEquals("ONLINE", paused.deviceStatus());
|
assertTrue(paused.latestSnapshot().isEmpty());
|
}
|
verify(adapter, never()).latestCollector(org.mockito.ArgumentMatchers.anyString(),
|
org.mockito.ArgumentMatchers.anyString());
|
}
|
|
/** 来源不提供辅助测量（新威实测无辅助数据）时如实标注，并且不因此生成离线异常。 */
|
@Test
|
void collectorWithoutAuxiliaryMeasurementsStaysOnlineAndIsMarkedUnavailable() throws Exception
|
{
|
TTest test = test("RECEIVING");
|
test.getBatteries().get(0).setCollectorId(7L);
|
var tests = mock(ITTestService.class);
|
when(tests.selectTestById(10L)).thenReturn(test);
|
var mapper = mock(TTestMapper.class);
|
when(mapper.selectCollectorCode(7L)).thenReturn("neware76-collector");
|
var adapter = mock(TestDataObservationAdapter.class);
|
when(adapter.latestCollector(org.mockito.ArgumentMatchers.anyString(),
|
org.mockito.ArgumentMatchers.anyString())).thenReturn(null);
|
var adapters = mock(com.battery.business.mapper.TDeviceAdapterMapper.class);
|
var source = new com.battery.business.domain.TDeviceAdapter();
|
source.setConnectionStatus("ONLINE");
|
when(adapters.selectDeviceAdapterByCode("dev-sim")).thenReturn(source);
|
var service = new TTestLiveServiceImpl(tests,
|
new com.battery.device.api.discovery.InMemoryDeviceDiscoveryCandidateStore(), adapter, mapper, adapters);
|
|
var collector = service.current(10L).sources().stream()
|
.filter(item -> item.sourceType().equals("COLLECTOR")).findFirst().orElseThrow();
|
|
assertEquals("ONLINE", collector.deviceStatus());
|
assertEquals(Boolean.FALSE, collector.latestSnapshot().get("auxiliaryDataAvailable"));
|
assertFalse(collector.latestSnapshot().containsKey("measurements"));
|
assertTrue(service.current(10L).exceptions().stream()
|
.noneMatch(item -> "COLLECTOR".equals(item.get("sourceType"))));
|
}
|
|
@Test
|
void notStartedShouldNotExposePreTestDeviceSnapshot()
|
{
|
Instant now = Instant.parse("2026-09-02T06:00:00Z");
|
TTest test = test("NOT_STARTED");
|
TTestLiveServiceImpl service = liveService(test, now);
|
|
LiveState state = service.current(10L);
|
|
assertTrue(state.sources().stream().allMatch(source -> source.latestSnapshot().isEmpty()));
|
}
|
|
@Test
|
void receivingShouldExposeBtsCoreMeasurements()
|
{
|
Instant now = Instant.parse("2026-09-02T06:00:00Z");
|
TTest test = test("RECEIVING");
|
TTestLiveServiceImpl service = liveService(test, now);
|
|
LiveState state = service.current(10L);
|
var snapshot = state.sources().stream().filter(source -> "CABINET_CHANNEL".equals(source.sourceType()))
|
.findFirst().orElseThrow().latestSnapshot();
|
|
assertEquals(new BigDecimal("3.700"), snapshot.get("voltageV"));
|
assertEquals(new BigDecimal("1.200"), snapshot.get("currentA"));
|
assertEquals(new BigDecimal("2.500"), snapshot.get("capacityAh"));
|
assertEquals(now.toEpochMilli(), snapshot.get("observedAt"));
|
}
|
|
@Test
|
void onlineCabinetWithStoppedWorkStatusShouldRemainOnlineAndNotRaiseSourceAlarm()
|
{
|
Instant now = Instant.parse("2026-09-02T06:00:00Z");
|
TTestLiveServiceImpl service = liveService(test("RECEIVING"), now, true, DeviceWorkStatus.STOPPED);
|
|
LiveState state = service.current(10L);
|
var cabinet = state.sources().stream().filter(source -> "CABINET_CHANNEL".equals(source.sourceType()))
|
.findFirst().orElseThrow();
|
|
assertEquals("ONLINE", cabinet.deviceStatus());
|
assertTrue(state.exceptions().isEmpty());
|
}
|
|
@Test
|
void pausedShouldNotExposeUnavailableSourceAsAlarm()
|
{
|
Instant now = Instant.parse("2026-09-02T06:00:00Z");
|
TTest test = test("PAUSED");
|
TTestLiveServiceImpl service = liveService(test, now, false);
|
|
LiveState state = service.current(10L);
|
|
assertTrue(state.exceptions().isEmpty());
|
}
|
|
@Test
|
void completedTestShouldNotExposeOpenWarningInLiveAlarm()
|
{
|
Instant now = Instant.parse("2026-09-02T06:00:00Z");
|
TTestMapper mapper = mock(TTestMapper.class);
|
TTestWarningException warning = new TTestWarningException();
|
warning.setExceptionId(1L);
|
warning.setSourceType("COLLECTOR");
|
warning.setMetricCode("TEMPERATURE");
|
warning.setAbnormalValue(new BigDecimal("25.3"));
|
warning.setUnit("℃");
|
warning.setExceptionStatus("OPEN");
|
when(mapper.selectWarningExceptions(10L, "OPEN")).thenReturn(List.of(warning));
|
|
TTestLiveServiceImpl service = liveService(test("ENDED"), now, true, DeviceWorkStatus.STOPPED, mapper);
|
|
LiveState state = service.current(10L);
|
|
assertTrue(state.exceptions().isEmpty());
|
verify(mapper, never()).selectWarningExceptions(10L, "OPEN");
|
}
|
|
private static TTest test(String receptionStatus)
|
{
|
TTest test = new TTest();
|
test.setTestId(10L);
|
test.setTestStatus("RECEIVING".equals(receptionStatus) ? "RUNNING" : "WAITING");
|
test.setReceptionStatus(receptionStatus);
|
TTestBattery battery = new TTestBattery();
|
battery.setBatteryNo("B-01");
|
TTestChannelBinding binding = new TTestChannelBinding();
|
binding.setChannelId(1L);
|
binding.setAdapterCode("dev-sim");
|
binding.setDevtype(1);
|
binding.setDevid(1);
|
binding.setSubdevid(0);
|
binding.setChlid(1);
|
battery.setBinding(binding);
|
test.setBatteries(List.of(battery));
|
return test;
|
}
|
|
private static TTestLiveServiceImpl liveService(TTest test, Instant now)
|
{
|
return liveService(test, now, true);
|
}
|
|
private static TTestLiveServiceImpl liveService(TTest test, Instant now, boolean online)
|
{
|
return liveService(test, now, online, DeviceWorkStatus.WORKING);
|
}
|
|
private static TTestLiveServiceImpl liveService(TTest test, Instant now, boolean online,
|
DeviceWorkStatus workStatus)
|
{
|
return liveService(test, now, online, workStatus, null);
|
}
|
|
private static TTestLiveServiceImpl liveService(TTest test, Instant now, boolean online,
|
DeviceWorkStatus workStatus, TTestMapper mapper)
|
{
|
return liveService(test, now, online, workStatus, mapper, now);
|
}
|
|
private static TTestLiveServiceImpl liveService(TTest test, Instant now, Instant deviceTime)
|
{
|
return liveService(test, now, true, DeviceWorkStatus.WORKING, null, deviceTime);
|
}
|
|
private static TTestLiveServiceImpl liveService(TTest test, Instant now, boolean online,
|
DeviceWorkStatus workStatus, TTestMapper mapper, Instant deviceTime)
|
{
|
DeviceDiscoveryCandidateStore candidates = mock(DeviceDiscoveryCandidateStore.class);
|
TestDataObservationAdapter simulator = mock(TestDataObservationAdapter.class);
|
ITTestService tests = mock(ITTestService.class);
|
when(tests.selectTestById(10L)).thenReturn(test);
|
ChannelIdentity identity = new ChannelIdentity("dev-sim", 1, 1, 0, 1);
|
RealtimeSnapshot snapshot = new RealtimeSnapshot(identity, workStatus, 1, 1, "cc", null,
|
new BigDecimal("1.200"), new BigDecimal("3.700"), new BigDecimal("2.500"),
|
new BigDecimal("9.250"), 12L, 12L, "0", now,
|
java.util.Map.of(), com.battery.device.api.model.DeviceConnectionStatus.ONLINE, deviceTime, List.of(), null);
|
DeviceDiscoveryCandidateStore.ChannelCandidate channel = new DeviceDiscoveryCandidateStore.ChannelCandidate(
|
identity, online, workStatus, now, snapshot);
|
DeviceDiscoveryCandidateStore.CabinetCandidate cabinet = new DeviceDiscoveryCandidateStore.CabinetCandidate(
|
"dev-sim", 1L, "dev-sim", 1, 1, "127.0.0.2", "mac", "cabinet", "vendor", "model",
|
List.of(channel), now, now.plusSeconds(60));
|
when(candidates.snapshot(any(Instant.class))).thenReturn(
|
new DeviceDiscoveryCandidateStore.CandidateSnapshot(List.of(cabinet)));
|
var plan = new TestDataObservationAdapter.TestPlan(1, 1, 1, List.of());
|
when(simulator.testPlan(any(ChannelIdentity.class))).thenReturn(plan);
|
if (mapper != null && test.getBatteries().get(0).getCollectorId() != null) {
|
when(mapper.selectCollectorCode(7L)).thenReturn("collector-7");
|
when(simulator.latestCollector(org.mockito.ArgumentMatchers.nullable(String.class), org.mockito.ArgumentMatchers.eq("collector-7"))).thenReturn(
|
new TestDataObservationAdapter.CollectorRealtime("collector-7", online, 1000, List.of(), null));
|
}
|
|
return new CurrentLiveService(tests, candidates, simulator, mapper);
|
}
|
|
@Test
|
void initialWriteFailureShouldCompleteStreamWithoutErrorDispatch()
|
{
|
RecordingEmitter emitter = new RecordingEmitter(0);
|
TestLiveStream service = new TestableLiveService(emitter);
|
|
try
|
{
|
service.stream(4L);
|
}
|
finally
|
{
|
service.close();
|
}
|
|
assertTrue(emitter.completed);
|
assertFalse(emitter.completedWithError);
|
}
|
|
@Test
|
void scheduledWriteFailureShouldCompleteStreamWithoutErrorDispatch()
|
{
|
RecordingEmitter emitter = new RecordingEmitter(6);
|
TestLiveStream service = new TestableLiveService(emitter);
|
|
try
|
{
|
service.stream(4L);
|
Awaitility.await().atMost(Duration.ofSeconds(5)).until(() -> emitter.completed);
|
}
|
finally
|
{
|
service.close();
|
}
|
|
assertFalse(emitter.completedWithError);
|
}
|
|
private static final class TestableLiveService extends TestLiveStream
|
{
|
private final RecordingEmitter emitter;
|
|
private TestableLiveService(RecordingEmitter emitter)
|
{
|
super(testLive());
|
this.emitter = emitter;
|
}
|
|
@Override
|
protected SseEmitter createEmitter()
|
{
|
return emitter;
|
}
|
|
private static ITestLiveService testLive() {
|
ITestLiveService live = mock(ITestLiveService.class);
|
when(live.current(any())).thenReturn(new LiveState(4L, "RUNNING", "RECEIVING", Instant.now(), List.of(), List.of()));
|
return live;
|
}
|
}
|
|
private static final class CurrentLiveService extends TTestLiveServiceImpl
|
{
|
private CurrentLiveService(ITTestService tests, DeviceDiscoveryCandidateStore candidates,
|
TestDataObservationAdapter simulator, TTestMapper mapper)
|
{
|
super(tests, candidates, simulator, mapper, null);
|
}
|
}
|
|
private static final class RecordingEmitter extends SseEmitter
|
{
|
private final int successfulWrites;
|
private int writes;
|
private volatile boolean completed;
|
private volatile boolean completedWithError;
|
|
private RecordingEmitter(int successfulWrites)
|
{
|
this.successfulWrites = successfulWrites;
|
}
|
|
@Override
|
public void send(SseEventBuilder builder) throws IOException
|
{
|
if (writes++ >= successfulWrites)
|
throw new AsyncRequestNotUsableException("client disconnected", new IOException("connection closed"));
|
}
|
|
@Override
|
public synchronized void complete()
|
{
|
completed = true;
|
}
|
|
@Override
|
public synchronized void completeWithError(Throwable error)
|
{
|
completedWithError = true;
|
}
|
}
|
}
|
