### Question:
KsqlResource { public KsqlEngine getKsqlEngine() { return ksqlEngine; } KsqlResource(
      KsqlEngine ksqlEngine,
      CommandStore commandStore,
      StatementExecutor statementExecutor,
      long distributedCommandResponseTimeout
  ); @POST Response handleKsqlStatements(KsqlRequest request); List<String> getStatementStrings(String ksqlString); KsqlEngine getKsqlEngine();  }### Answer:
@Test public void testDescribeStatement() throws Exception { KsqlResource testResource = TestKsqlResourceUtil.get(ksqlEngine); final String tableName = "TEST_TABLE"; final String ksqlString = String.format("DESCRIBE %s;", tableName); final ShowColumns ksqlStatement = new ShowColumns(QualifiedName.of(tableName), false, false); SourceDescription testDescription = makeSingleRequest( testResource, ksqlString, Collections.emptyMap(), SourceDescription.class ); SourceDescription expectedDescription = new SourceDescription(testResource.getKsqlEngine().getMetaStore().getSource(tableName), false, "serdes", "topo", "exec-plan", Collections.EMPTY_LIST, Collections.EMPTY_LIST,null); assertEquals(expectedDescription, testDescription); }
@Test public void testListStreamsStatement() throws Exception { KsqlResource testResource = TestKsqlResourceUtil.get(ksqlEngine); final String ksqlString = "LIST STREAMS;"; final ListStreams ksqlStatement = new ListStreams(Optional.empty()); StreamsList streamsList = makeSingleRequest( testResource, ksqlString, Collections.emptyMap(), StreamsList.class ); List<SourceInfo.Stream> testStreams = streamsList.getStreams(); assertEquals(1, testStreams.size()); SourceInfo expectedStream = new SourceInfo.Stream((KsqlStream) testResource.getKsqlEngine().getMetaStore().getSource("TEST_STREAM")); assertEquals(expectedStream, testStreams.get(0)); }
@Test public void testListTablesStatement() throws Exception { KsqlResource testResource = TestKsqlResourceUtil.get(ksqlEngine); final String ksqlString = "LIST TABLES;"; final ListTables ksqlStatement = new ListTables(Optional.empty()); TablesList tablesList = makeSingleRequest( testResource, ksqlString, Collections.emptyMap(), TablesList.class ); List<SourceInfo.Table> testTables = tablesList.getTables(); assertEquals(1, testTables.size()); SourceInfo expectedTable = new SourceInfo.Table((KsqlTable) testResource.getKsqlEngine().getMetaStore().getSource("TEST_TABLE")); assertEquals(expectedTable, testTables.get(0)); }