Andrii Landiak
5 months ago
51 changed files with 1114 additions and 342 deletions
@ -0,0 +1,155 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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. |
||||
|
*/ |
||||
|
package org.thingsboard.server.edge; |
||||
|
|
||||
|
import org.junit.After; |
||||
|
import org.junit.Assert; |
||||
|
import org.junit.Test; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.test.context.TestPropertySource; |
||||
|
import org.thingsboard.edge.exception.EdgeFeatureDisabledException; |
||||
|
import org.thingsboard.server.common.data.ApiUsageRecordKey; |
||||
|
import org.thingsboard.server.common.data.ApiUsageStateValue; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.stats.TbApiUsageReportClient; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
import org.thingsboard.server.dao.usagerecord.ApiUsageStateService; |
||||
|
import org.thingsboard.server.edge.imitator.EdgeImitator; |
||||
|
import org.thingsboard.server.service.edge.rpc.session.EdgeSessionsHolder; |
||||
|
|
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
@TestPropertySource(properties = { |
||||
|
"usage.stats.report.enabled=true", |
||||
|
"usage.stats.report.interval=2", |
||||
|
"usage.stats.report.urgent_interval=1" |
||||
|
}) |
||||
|
public class EdgeApiUsageDisabledTest extends AbstractEdgeTest { |
||||
|
|
||||
|
private static final int MAX_EDGE_EVENTS = 1; |
||||
|
|
||||
|
@Autowired |
||||
|
private ApiUsageStateService apiUsageStateService; |
||||
|
|
||||
|
@Autowired |
||||
|
private TbApiUsageReportClient apiUsageReportClient; |
||||
|
|
||||
|
@Autowired |
||||
|
private EdgeSessionsHolder edgeSessionsHolder; |
||||
|
|
||||
|
@After |
||||
|
public void restoreEdgeLimit() { |
||||
|
try { |
||||
|
loginSysAdmin(); |
||||
|
updateDefaultTenantProfileConfig(cfg -> cfg.setMaxEdgeEvents(0)); |
||||
|
} catch (Exception ignored) {} |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testLiveSessionForceClosedWhenEdgeStateDisabled() throws Exception { |
||||
|
await().atMost(10, TimeUnit.SECONDS).until(() -> sessionConnected(tenantId)); |
||||
|
|
||||
|
loginSysAdmin(); |
||||
|
updateDefaultTenantProfileConfig(cfg -> cfg.setMaxEdgeEvents(MAX_EDGE_EVENTS)); |
||||
|
loginTenantAdmin(); |
||||
|
|
||||
|
for (int i = 0; i < MAX_EDGE_EVENTS + 5; i++) { |
||||
|
apiUsageReportClient.report(tenantId, null, ApiUsageRecordKey.EDGE_EVENT_COUNT); |
||||
|
} |
||||
|
|
||||
|
await().atMost(15, TimeUnit.SECONDS).until(() -> apiUsageStateService.findTenantApiUsageState(tenantId).getEdgeState() == ApiUsageStateValue.DISABLED); |
||||
|
|
||||
|
await().atMost(10, TimeUnit.SECONDS).until(() -> !sessionConnected(tenantId)); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testReconnectRejectedWhenEdgeStateDisabled() throws Exception { |
||||
|
await().atMost(10, TimeUnit.SECONDS).until(() -> sessionConnected(tenantId)); |
||||
|
|
||||
|
loginSysAdmin(); |
||||
|
updateDefaultTenantProfileConfig(cfg -> cfg.setMaxEdgeEvents(MAX_EDGE_EVENTS)); |
||||
|
loginTenantAdmin(); |
||||
|
|
||||
|
for (int i = 0; i < MAX_EDGE_EVENTS + 5; i++) { |
||||
|
apiUsageReportClient.report(tenantId, null, ApiUsageRecordKey.EDGE_EVENT_COUNT); |
||||
|
} |
||||
|
|
||||
|
await().atMost(15, TimeUnit.SECONDS).until(() -> !sessionConnected(tenantId)); |
||||
|
|
||||
|
EdgeImitator rejectedImitator = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); |
||||
|
rejectedImitator.connect(); |
||||
|
|
||||
|
await().atMost(10, TimeUnit.SECONDS) |
||||
|
.untilAsserted(() -> { |
||||
|
Exception closeException = rejectedImitator.getCloseException(); |
||||
|
Assert.assertNotNull("Imitator must receive a connection-rejected callback", closeException); |
||||
|
Assert.assertTrue("Expected EdgeFeatureDisabledException, got: " + closeException, closeException instanceof EdgeFeatureDisabledException); |
||||
|
}); |
||||
|
Assert.assertFalse("Edge session must not be admitted when edge feature is disabled", |
||||
|
sessionConnected(tenantId)); |
||||
|
|
||||
|
try { |
||||
|
rejectedImitator.disconnect(); |
||||
|
} catch (Exception ignored) {} |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testReconnectAdmittedAfterEdgeStateReEnabled() throws Exception { |
||||
|
await().atMost(10, TimeUnit.SECONDS).until(() -> sessionConnected(tenantId)); |
||||
|
|
||||
|
loginSysAdmin(); |
||||
|
updateDefaultTenantProfileConfig(cfg -> cfg.setMaxEdgeEvents(MAX_EDGE_EVENTS)); |
||||
|
loginTenantAdmin(); |
||||
|
|
||||
|
for (int i = 0; i < MAX_EDGE_EVENTS + 5; i++) { |
||||
|
apiUsageReportClient.report(tenantId, null, ApiUsageRecordKey.EDGE_EVENT_COUNT); |
||||
|
} |
||||
|
|
||||
|
await().atMost(15, TimeUnit.SECONDS) |
||||
|
.until(() -> apiUsageStateService.findTenantApiUsageState(tenantId).getEdgeState() == ApiUsageStateValue.DISABLED); |
||||
|
await().atMost(10, TimeUnit.SECONDS) |
||||
|
.until(() -> !sessionConnected(tenantId)); |
||||
|
|
||||
|
loginSysAdmin(); |
||||
|
updateDefaultTenantProfileConfig(cfg -> cfg.setMaxEdgeEvents(0)); |
||||
|
loginTenantAdmin(); |
||||
|
|
||||
|
await().atMost(15, TimeUnit.SECONDS) |
||||
|
.until(() -> apiUsageStateService.findTenantApiUsageState(tenantId).getEdgeState() == ApiUsageStateValue.ENABLED); |
||||
|
|
||||
|
EdgeImitator reconnectImitator = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); |
||||
|
reconnectImitator.connect(); |
||||
|
|
||||
|
try { |
||||
|
await().atMost(15, TimeUnit.SECONDS) |
||||
|
.until(() -> sessionConnected(tenantId)); |
||||
|
Assert.assertNull("Imitator must not receive a close callback after re-enable", |
||||
|
reconnectImitator.getCloseException()); |
||||
|
} finally { |
||||
|
try { |
||||
|
reconnectImitator.disconnect(); |
||||
|
} catch (Exception ignored) {} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private boolean sessionConnected(TenantId tenantId) { |
||||
|
return edgeSessionsHolder.getByTenantId(tenantId).stream().anyMatch(s -> s.getState().isConnected()); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,126 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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. |
||||
|
*/ |
||||
|
package org.thingsboard.server.transport.coap; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.awaitility.Awaitility; |
||||
|
import org.eclipse.californium.core.CoapObserveRelation; |
||||
|
import org.eclipse.californium.core.coap.CoAP; |
||||
|
import org.junit.After; |
||||
|
import org.junit.Assert; |
||||
|
import org.junit.Before; |
||||
|
import org.junit.Test; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.test.context.TestPropertySource; |
||||
|
import org.thingsboard.server.common.data.ApiUsageRecordKey; |
||||
|
import org.thingsboard.server.common.data.ApiUsageStateValue; |
||||
|
import org.thingsboard.server.common.data.CoapDeviceType; |
||||
|
import org.thingsboard.server.common.data.TenantProfile; |
||||
|
import org.thingsboard.server.common.data.TransportPayloadType; |
||||
|
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; |
||||
|
import org.thingsboard.server.common.msg.session.FeatureType; |
||||
|
import org.thingsboard.server.common.stats.TbApiUsageReportClient; |
||||
|
import org.thingsboard.server.common.transport.service.DefaultTransportService; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
import org.thingsboard.server.dao.usagerecord.ApiUsageStateService; |
||||
|
|
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
@TestPropertySource(properties = { |
||||
|
"usage.stats.report.enabled=true", |
||||
|
"usage.stats.report.interval=2", |
||||
|
"usage.stats.report.urgent_interval=1", |
||||
|
}) |
||||
|
@Slf4j |
||||
|
public class CoapTransportFeatureDisabledTest extends AbstractCoapIntegrationTest { |
||||
|
|
||||
|
private static final int MAX_TRANSPORT_MESSAGES = 10; |
||||
|
private static final double WARN_THRESHOLD = 0.5; |
||||
|
|
||||
|
@Autowired |
||||
|
private ApiUsageStateService apiUsageStateService; |
||||
|
|
||||
|
@Autowired |
||||
|
private TbApiUsageReportClient apiUsageReportClient; |
||||
|
|
||||
|
@Autowired |
||||
|
private DefaultTransportService defaultTransportService; |
||||
|
|
||||
|
@Before |
||||
|
public void beforeTest() throws Exception { |
||||
|
loginSysAdmin(); |
||||
|
TenantProfile tenantProfile = doGet("/api/tenantProfile/" + tenantProfileId, TenantProfile.class); |
||||
|
DefaultTenantProfileConfiguration config = |
||||
|
(DefaultTenantProfileConfiguration) tenantProfile.getProfileData().getConfiguration(); |
||||
|
config.setMaxTransportMessages(MAX_TRANSPORT_MESSAGES); |
||||
|
config.setWarnThreshold(WARN_THRESHOLD); |
||||
|
doPost("/api/tenantProfile", tenantProfile); |
||||
|
|
||||
|
CoapTestConfigProperties configProperties = CoapTestConfigProperties.builder() |
||||
|
.deviceName("Coap transport disable test device") |
||||
|
.coapDeviceType(CoapDeviceType.DEFAULT) |
||||
|
.transportPayloadType(TransportPayloadType.JSON) |
||||
|
.build(); |
||||
|
processBeforeTest(configProperties); |
||||
|
} |
||||
|
|
||||
|
@After |
||||
|
public void afterTest() throws Exception { |
||||
|
try { |
||||
|
processAfterTest(); |
||||
|
} finally { |
||||
|
try { |
||||
|
loginSysAdmin(); |
||||
|
TenantProfile tenantProfile = doGet("/api/tenantProfile/" + tenantProfileId, TenantProfile.class); |
||||
|
DefaultTenantProfileConfiguration config = |
||||
|
(DefaultTenantProfileConfiguration) tenantProfile.getProfileData().getConfiguration(); |
||||
|
config.setMaxTransportMessages(0); |
||||
|
doPost("/api/tenantProfile", tenantProfile); |
||||
|
} catch (Exception ignored) {} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testCoapObserveSessionClosedWhenTransportDisabled() { |
||||
|
client = new CoapTestClient(accessToken, FeatureType.ATTRIBUTES); |
||||
|
CoapTestCallback callback = new CoapTestCallback(); |
||||
|
CoapObserveRelation observeRelation = client.getObserveRelation(callback); |
||||
|
|
||||
|
Awaitility.await("await initial observe response") |
||||
|
.atMost(10, TimeUnit.SECONDS) |
||||
|
.until(() -> CoAP.ResponseCode.CONTENT.equals(callback.getResponseCode()) |
||||
|
&& callback.getObserve() != null); |
||||
|
|
||||
|
Assert.assertFalse("CoAP transport must hold at least one registered session after observe", |
||||
|
defaultTransportService.sessions.isEmpty()); |
||||
|
|
||||
|
for (int i = 0; i < MAX_TRANSPORT_MESSAGES + 5; i++) { |
||||
|
apiUsageReportClient.report(tenantId, null, ApiUsageRecordKey.TRANSPORT_MSG_COUNT); |
||||
|
} |
||||
|
|
||||
|
Awaitility.await("transport state flips to DISABLED") |
||||
|
.atMost(15, TimeUnit.SECONDS) |
||||
|
.until(() -> apiUsageStateService.findTenantApiUsageState(tenantId).getTransportState() == ApiUsageStateValue.DISABLED); |
||||
|
|
||||
|
Awaitility.await("CoAP session is removed from DefaultTransportService") |
||||
|
.atMost(10, TimeUnit.SECONDS) |
||||
|
.until(() -> defaultTransportService.sessions.isEmpty()); |
||||
|
|
||||
|
observeRelation.proactiveCancel(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,125 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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. |
||||
|
*/ |
||||
|
package org.thingsboard.server.transport.mqtt; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.awaitility.Awaitility; |
||||
|
import org.junit.After; |
||||
|
import org.junit.Assert; |
||||
|
import org.junit.Before; |
||||
|
import org.junit.Test; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.test.context.DynamicPropertyRegistry; |
||||
|
import org.springframework.test.context.DynamicPropertySource; |
||||
|
import org.springframework.test.context.TestPropertySource; |
||||
|
import org.thingsboard.server.common.data.ApiUsageRecordKey; |
||||
|
import org.thingsboard.server.common.data.ApiUsageStateValue; |
||||
|
import org.thingsboard.server.common.data.Device; |
||||
|
import org.thingsboard.server.common.data.TenantProfile; |
||||
|
import org.thingsboard.server.common.data.security.DeviceCredentials; |
||||
|
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; |
||||
|
import org.thingsboard.server.common.stats.TbApiUsageReportClient; |
||||
|
import org.thingsboard.server.controller.AbstractControllerTest; |
||||
|
import org.thingsboard.server.dao.service.DaoSqlTest; |
||||
|
import org.thingsboard.server.dao.usagerecord.ApiUsageStateService; |
||||
|
import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; |
||||
|
|
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest.MQTT_PORT; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
@TestPropertySource(properties = { |
||||
|
"service.integrations.supported=ALL", |
||||
|
"transport.mqtt.enabled=true", |
||||
|
"usage.stats.report.enabled=true", |
||||
|
"usage.stats.report.interval=2", |
||||
|
"usage.stats.report.urgent_interval=1", |
||||
|
}) |
||||
|
@Slf4j |
||||
|
public class MqttTransportFeatureDisabledTest extends AbstractControllerTest { |
||||
|
|
||||
|
private static final int MAX_TRANSPORT_MESSAGES = 10; |
||||
|
private static final double WARN_THRESHOLD = 0.5; |
||||
|
|
||||
|
@DynamicPropertySource |
||||
|
static void props(DynamicPropertyRegistry registry) { |
||||
|
log.warn("transport.mqtt.bind_port = {}", MQTT_PORT); |
||||
|
registry.add("transport.mqtt.bind_port", () -> MQTT_PORT); |
||||
|
} |
||||
|
|
||||
|
@Autowired |
||||
|
private ApiUsageStateService apiUsageStateService; |
||||
|
|
||||
|
@Autowired |
||||
|
private TbApiUsageReportClient apiUsageReportClient; |
||||
|
|
||||
|
private String deviceAccessToken; |
||||
|
|
||||
|
@Before |
||||
|
public void beforeTest() throws Exception { |
||||
|
loginSysAdmin(); |
||||
|
TenantProfile tenantProfile = doGet("/api/tenantProfile/" + tenantProfileId, TenantProfile.class); |
||||
|
DefaultTenantProfileConfiguration config = |
||||
|
(DefaultTenantProfileConfiguration) tenantProfile.getProfileData().getConfiguration(); |
||||
|
config.setMaxTransportMessages(MAX_TRANSPORT_MESSAGES); |
||||
|
config.setWarnThreshold(WARN_THRESHOLD); |
||||
|
doPost("/api/tenantProfile", tenantProfile); |
||||
|
|
||||
|
loginTenantAdmin(); |
||||
|
Device device = new Device(); |
||||
|
device.setName("Transport disable test device"); |
||||
|
device.setType("default"); |
||||
|
device = doPost("/api/device", device, Device.class); |
||||
|
DeviceCredentials credentials = doGet("/api/device/" + device.getId().getId() + "/credentials", DeviceCredentials.class); |
||||
|
deviceAccessToken = credentials.getCredentialsId(); |
||||
|
} |
||||
|
|
||||
|
@After |
||||
|
public void afterTest() throws Exception { |
||||
|
try { |
||||
|
loginSysAdmin(); |
||||
|
TenantProfile tenantProfile = doGet("/api/tenantProfile/" + tenantProfileId, TenantProfile.class); |
||||
|
DefaultTenantProfileConfiguration config = (DefaultTenantProfileConfiguration) tenantProfile.getProfileData().getConfiguration(); |
||||
|
config.setMaxTransportMessages(0); |
||||
|
doPost("/api/tenantProfile", tenantProfile); |
||||
|
} catch (Exception ignored) {} |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testLiveMqttSessionClosedWhenTransportDisabled() throws Exception { |
||||
|
MqttTestClient client = new MqttTestClient(); |
||||
|
client.connectAndWait(deviceAccessToken); |
||||
|
Assert.assertTrue("MQTT client must be connected before flipping transport state", client.isConnected()); |
||||
|
|
||||
|
for (int i = 0; i < MAX_TRANSPORT_MESSAGES + 5; i++) { |
||||
|
apiUsageReportClient.report(tenantId, null, ApiUsageRecordKey.TRANSPORT_MSG_COUNT); |
||||
|
} |
||||
|
|
||||
|
Awaitility.await("transport state flips to DISABLED") |
||||
|
.atMost(15, TimeUnit.SECONDS) |
||||
|
.until(() -> apiUsageStateService.findTenantApiUsageState(tenantId).getTransportState() == ApiUsageStateValue.DISABLED); |
||||
|
|
||||
|
Awaitility.await("MQTT client receives server-side disconnect") |
||||
|
.atMost(10, TimeUnit.SECONDS) |
||||
|
.until(() -> !client.isConnected()); |
||||
|
|
||||
|
try { |
||||
|
client.disconnectForcibly(); |
||||
|
} catch (Exception ignored) {} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,29 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2026 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed 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. |
||||
|
*/ |
||||
|
package org.thingsboard.edge.exception; |
||||
|
|
||||
|
import java.io.Serial; |
||||
|
|
||||
|
public class EdgeFeatureDisabledException extends RuntimeException { |
||||
|
|
||||
|
@Serial |
||||
|
private static final long serialVersionUID = -4719918663724404639L; |
||||
|
|
||||
|
public EdgeFeatureDisabledException(String message) { |
||||
|
super(message); |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue