package com.battery.device.neware;
|
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
|
import java.sql.Connection;
|
import java.sql.PreparedStatement;
|
import java.sql.Statement;
|
import java.time.Clock;
|
import java.time.Instant;
|
import java.time.ZoneOffset;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
import javax.sql.DataSource;
|
import org.junit.jupiter.api.Test;
|
import com.battery.device.api.model.AdapterCapabilities;
|
import com.battery.device.api.model.ChannelIdentity;
|
import com.battery.device.api.registry.AdapterInstanceConfig;
|
|
/**
|
* 只读防线测试：非查询语句在到达驱动之前被拒绝，且该数据源在其生命周期内从未发出过非查询语句。
|
*/
|
class ReadOnlyVendorDataSourceTest
|
{
|
private static final String ADAPTER_CODE = "neware76";
|
private static final ChannelIdentity CHANNEL = new ChannelIdentity(ADAPTER_CODE, 21, 36, 1, 2);
|
|
private ReadOnlyVendorDataSource dataSourceOver(FakeVendorDatabase database)
|
{
|
return new ReadOnlyVendorDataSource(database.dataSource(), ADAPTER_CODE);
|
}
|
|
@Test
|
void queryStatementsAreForwardedAndRecorded() throws Exception
|
{
|
ReadOnlyVendorDataSource dataSource = new ReadOnlyVendorDataSource(new FakeVendorDatabase().dataSource(),
|
ADAPTER_CODE);
|
try (Connection connection = dataSource.getConnection();
|
PreparedStatement statement = connection.prepareStatement("select 1 from channel"))
|
{
|
statement.executeQuery();
|
}
|
try (Connection connection = dataSource.getConnection();
|
Statement statement = connection.createStatement())
|
{
|
statement.executeQuery("show tables");
|
statement.executeQuery("describe `channel`");
|
}
|
assertTrue(dataSource.audit().rejectedStatements().isEmpty());
|
assertEquals(3, dataSource.audit().forwardedStatements().size());
|
}
|
|
@Test
|
void nonQueryStatementsAreRejectedBeforeTheyReachTheDriver() throws Exception
|
{
|
FakeVendorDatabase database = new FakeVendorDatabase();
|
ReadOnlyVendorDataSource dataSource = dataSourceOver(database);
|
List<String> nonQueries = List.of("update channel set chl_status = 1",
|
"insert into channel(dev_uid) values(210036)", "delete from channel",
|
"drop table data_210036_2026_09_18", "create index idx on channel(chl_status)",
|
"truncate table channel", "set global read_only = 0", "grant select on *.* to x",
|
"alter table channel add column x int", "call some_procedure()");
|
for (String sql : nonQueries)
|
{
|
assertThrows(NonQueryStatementException.class, () -> {
|
try (Connection connection = dataSource.getConnection();
|
PreparedStatement statement = connection.prepareStatement(sql))
|
{
|
statement.execute();
|
}
|
}, "非查询语句必须被拒绝：" + sql);
|
}
|
assertEquals(nonQueries.size(), dataSource.audit().rejectedStatements().size());
|
assertTrue(database.statements().isEmpty(), "被拒绝的语句不得到达驱动");
|
// 语句级执行同样被拦下，并且没有任何非查询语句到达驱动。
|
assertThrows(NonQueryStatementException.class, () -> {
|
try (Connection connection = dataSource.getConnection(); Statement statement = connection.createStatement())
|
{
|
statement.executeUpdate("update channel set chl_status = 1");
|
}
|
});
|
assertThrows(NonQueryStatementException.class,
|
() -> dataSource.getConnection().prepareCall("{call anything()}"));
|
assertTrue(database.statements().isEmpty(), "被拒绝的语句不得到达驱动");
|
assertEquals(nonQueries.size() + 1, dataSource.audit().rejectedStatements().size());
|
}
|
|
/**
|
* 这条断言对应验收要求：整个适配器生命周期内，该数据源没有发出过任何非查询语句。
|
*/
|
@Test
|
void adapterLifecycleNeverSendsANonQueryStatement()
|
{
|
FakeVendorDatabase database = new FakeVendorDatabase()
|
.registeredChannel(210036L, 1, 2)
|
.test(new FakeVendorDatabase.TestRow(210036L, 1, 2, 440L, "2026-09-18 09:00:00",
|
"data_210036_2026_09_18", null))
|
.sample(new FakeVendorDatabase.SampleRow("data_210036_2026_09_18", 440L, 1, 2, 7435L, 3, 12, "2",
|
300000L, 13733, -11889, 0, 0, 50350024, 1103331, "2026-09-28 09:54:56"));
|
ReadOnlyVendorDataSource guarded = dataSourceOver(database);
|
AdapterInstanceConfig config = new AdapterInstanceConfig(ADAPTER_CODE, NewareMysqlAdapter.ADAPTER_TYPE,
|
"26.1.2.3", 3306, 60000, "Asia/Hong_Kong", true, AdapterCapabilities.observeOnly(),
|
Map.of("database", "bts63"));
|
NewareMysqlAdapter adapter = new NewareMysqlAdapter(config, guarded, () -> { },
|
Clock.fixed(Instant.parse("2026-09-28T02:00:00Z"), ZoneOffset.UTC), 0);
|
|
adapter.discoverChannels();
|
adapter.inquire(Set.of(CHANNEL));
|
adapter.latestCabinet(CHANNEL);
|
adapter.currentWatermark(CHANNEL);
|
adapter.samplesSince(CHANNEL,
|
new com.battery.device.api.TestDataObservationAdapter.SampleWatermark(
|
Instant.parse("2026-09-28T01:00:00Z"), 1L));
|
|
assertTrue(guarded.audit().rejectedStatements().isEmpty(),
|
"适配器不应尝试任何非查询语句：" + guarded.audit().rejectedStatements());
|
assertTrue(!database.statements().isEmpty(), "适配器应当实际读取来源");
|
for (String statement : database.statements())
|
{
|
assertTrue(ReadOnlyVendorDataSource.isQuery(statement), "到达驱动的语句必须是查询：" + statement);
|
}
|
}
|
|
@Test
|
void statementClassificationKeepsQueriesAndRejectsEverythingElse()
|
{
|
assertTrue(ReadOnlyVendorDataSource.isQuery("select dev_uid from channel"));
|
assertTrue(ReadOnlyVendorDataSource.isQuery("\n SELECT 1"));
|
assertTrue(ReadOnlyVendorDataSource.isQuery("/* probe */ show grants"));
|
assertTrue(ReadOnlyVendorDataSource.isQuery("-- probe\n describe channel"));
|
assertTrue(ReadOnlyVendorDataSource.isQuery("desc `channel`"));
|
assertTrue(!ReadOnlyVendorDataSource.isQuery("update channel set chl_status = 0"));
|
assertTrue(!ReadOnlyVendorDataSource.isQuery(""));
|
// 以 select 开头但紧跟字母的标识符不是查询语句。
|
assertTrue(!ReadOnlyVendorDataSource.isQuery("selection_helper()"));
|
}
|
|
@Test
|
void rejectedStatementsNeverReachTheDriverEvenWhenTheApplicationKeepsTrying()
|
{
|
// 连续尝试写操作后，来源侧一条语句都没收到，说明拦截发生在驱动之前。
|
FakeVendorDatabase database = new FakeVendorDatabase();
|
ReadOnlyVendorDataSource dataSource = dataSourceOver(database);
|
for (String sql : List.of("update channel set chl_status = 1", "delete from channel",
|
"insert into channel(dev_uid) values(210036)"))
|
{
|
try
|
{
|
dataSource.getConnection().prepareStatement(sql);
|
}
|
catch (Exception ignored)
|
{
|
// 每条都被白名单挡下。
|
}
|
}
|
assertTrue(database.statements().isEmpty());
|
assertEquals(3, dataSource.audit().rejectedStatements().size());
|
}
|
|
/** 加锁读以 select 开头却是真的会阻塞厂商写入，所以必须在到达驱动之前被单独拒绝。 */
|
@Test
|
void lockingReadsAreRejectedBeforeTheyReachTheDriver()
|
{
|
FakeVendorDatabase database = new FakeVendorDatabase();
|
ReadOnlyVendorDataSource dataSource = dataSourceOver(database);
|
for (String sql : List.of("select dev_uid from channel for update",
|
"select dev_uid from channel FOR SHARE",
|
"select dev_uid from channel lock in share mode",
|
"select dev_uid from channel with (updlock)",
|
"/* probe */ select dev_uid from channel with (tablock, holdlock)"))
|
{
|
assertTrue(ReadOnlyVendorDataSource.isQuery(sql), "加锁读仍然以查询开头：" + sql);
|
assertTrue(ReadOnlyVendorDataSource.containsLockingClause(sql), "应当识别为加锁读：" + sql);
|
assertThrows(NonQueryStatementException.class,
|
() -> dataSource.getConnection().prepareStatement(sql), sql);
|
}
|
assertTrue(database.statements().isEmpty(), "加锁读不得到达驱动");
|
assertTrue(!ReadOnlyVendorDataSource.containsLockingClause("select dev_uid from channel"));
|
assertTrue(!ReadOnlyVendorDataSource.containsLockingClause(""));
|
}
|
|
/** 连接超时与读取超时都必须由我们显式给出，且不得写入只读事务参数（5.5 不支持）。 */
|
@Test
|
void vendorJdbcUrlCarriesBothTimeoutsAndNoReadOnlySessionVariable()
|
{
|
String url = NewareMysqlAdapterFactory.jdbcUrl("26.1.2.3", 3306, "bts63");
|
|
assertTrue(url.contains("connectTimeout="), url);
|
assertTrue(url.contains("socketTimeout="), url);
|
assertTrue(!url.contains("transaction_read_only"), url);
|
}
|
}
|
