Enable noop source support in controller

Additional telemetry apis for testing.

Change-Id: I60a61ffbbe86bbb0474777905de922434deabbe2
Issue-ID: POLICY-1716
Signed-off-by: jhh <jorge.hernandez-herrero@att.com>

3 files changed, 286 insertions, 179 deletions

diff --git a/policy-management/src/main/java/org/onap/policy/drools/controller/DroolsControllerFactory.java b/policy-management/src/main/java/org/onap/policy/drools/controller/DroolsControllerFactory.java
index 16daa947..f91e6dc4 100644
--- a/policy-management/src/main/java/org/onap/policy/drools/controller/DroolsControllerFactory.java
+++ b/policy-management/src/main/java/org/onap/policy/drools/controller/DroolsControllerFactory.java
@@ -279,10 +279,10 @@ class IndexedDroolsControllerFactory implements DroolsControllerFactory {
                     propertyTopicEntityPrefix = PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + ".";
                 }
             } else if (commInfra == CommInfrastructure.NOOP) {
-                if (!isSource) {
-                    propertyTopicEntityPrefix = PolicyEndPointProperties.PROPERTY_NOOP_SINK_TOPICS + ".";
+                if (isSource) {
+                    propertyTopicEntityPrefix = PolicyEndPointProperties.PROPERTY_NOOP_SOURCE_TOPICS + ".";
                 } else {
-                    continue;
+                    propertyTopicEntityPrefix = PolicyEndPointProperties.PROPERTY_NOOP_SINK_TOPICS + ".";
                 }
             } else {
                 throw new IllegalArgumentException("Invalid Communication Infrastructure: " + commInfra);
diff --git a/policy-management/src/main/java/org/onap/policy/drools/server/restful/RestManager.java b/policy-management/src/main/java/org/onap/policy/drools/server/restful/RestManager.java
index 30392438..aac92505 100644
--- a/policy-management/src/main/java/org/onap/policy/drools/server/restful/RestManager.java
+++ b/policy-management/src/main/java/org/onap/policy/drools/server/restful/RestManager.java
@@ -49,15 +49,11 @@ import javax.ws.rs.QueryParam;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Response.Status;
+import org.onap.policy.common.endpoints.event.comm.Topic;
 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
 import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
 import org.onap.policy.common.endpoints.event.comm.TopicSink;
 import org.onap.policy.common.endpoints.event.comm.TopicSource;
-import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicSink;
-import org.onap.policy.common.endpoints.event.comm.bus.DmaapTopicSource;
-import org.onap.policy.common.endpoints.event.comm.bus.NoopTopicSink;
-import org.onap.policy.common.endpoints.event.comm.bus.UebTopicSink;
-import org.onap.policy.common.endpoints.event.comm.bus.UebTopicSource;
 import org.onap.policy.drools.controller.DroolsController;
 import org.onap.policy.drools.features.PolicyControllerFeatureAPI;
 import org.onap.policy.drools.features.PolicyEngineFeatureAPI;
@@ -1693,56 +1689,61 @@ public class RestManager {
     }
 
     /**
-     * GET.
-     *
-     * @return response object
+     * GET sources of a communication type.
      */
     @GET
-    @Path("engine/topics/sources/ueb")
-    @ApiOperation(value = "Retrieves the UEB managed topic sources", notes = "UEB Topic Sources Agregation",
-            responseContainer = "List", response = UebTopicSource.class)
-    public Response uebSources() {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getUebTopicSources()).build();
-    }
-
-    /**
-     * GET.
-     *
-     * @return response object
+    @Path("engine/topics/sources/{comm: ueb|dmaap|noop}")
+    @ApiOperation(value = "Retrieves managed topic sources", notes = "Sources for a communication infrastructure",
+            responseContainer = "List", response = TopicSource.class)
+    public Response commSources(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm
+    ) {
+        List<TopicSource> sources = new ArrayList<>();
+        switch (CommInfrastructure.valueOf(comm.toUpperCase())) {
+            case UEB:
+                sources.addAll(TopicEndpoint.manager.getUebTopicSources());
+                break;
+            case DMAAP:
+                sources.addAll(TopicEndpoint.manager.getDmaapTopicSources());
+                break;
+            case NOOP:
+                sources.addAll(TopicEndpoint.manager.getNoopTopicSources());
+                break;
+            default:
+                logger.debug("Invalid communication mechanism parameter: {}", comm);
+                break;
+        }
+        return Response.status(Response.Status.OK).entity(sources).build();
+    }
+
+    /**
+     * GET sinks of a communication type.
      */
     @GET
-    @Path("engine/topics/sinks/ueb")
-    @ApiOperation(value = "Retrieves the UEB managed topic sinks", notes = "UEB Topic Sinks Agregation",
-            responseContainer = "List", response = UebTopicSink.class)
-    public Response uebSinks() {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getUebTopicSinks()).build();
+    @Path("engine/topics/sinks/{comm: ueb|dmaap|noop}")
+    @ApiOperation(value = "Retrieves managed topic sinks", notes = "Communication Infrastructure Sinks",
+            responseContainer = "List", response = TopicSink.class)
+    public Response commSinks(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm
+    ) {
+        List<TopicSink> sinks = new ArrayList<>();
+        switch (CommInfrastructure.valueOf(comm.toUpperCase())) {
+            case UEB:
+                sinks.addAll(TopicEndpoint.manager.getUebTopicSinks());
+                break;
+            case DMAAP:
+                sinks.addAll(TopicEndpoint.manager.getDmaapTopicSinks());
+                break;
+            case NOOP:
+                sinks.addAll(TopicEndpoint.manager.getNoopTopicSinks());
+                break;
+            default:
+                logger.debug("Invalid communication mechanism parameter: {}", comm);
+                break;
+        }
+        return Response.status(Response.Status.OK).entity(sinks).build();
     }
 
-    /**
-     * GET.
-     *
-     * @return response object
-     */
-    @GET
-    @Path("engine/topics/sources/dmaap")
-    @ApiOperation(value = "Retrieves the DMaaP managed topic sources", notes = "DMaaP Topic Sources Agregation",
-            responseContainer = "List", response = DmaapTopicSource.class)
-    public Response dmaapSources() {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getDmaapTopicSources()).build();
-    }
-
-    /**
-     * GET.
-     *
-     * @return response object
-     */
-    @GET
-    @Path("engine/topics/sinks/dmaap")
-    @ApiOperation(value = "Retrieves the DMaaP managed topic sinks", notes = "DMaaP Topic Sinks Agregation",
-            responseContainer = "List", response = DmaapTopicSink.class)
-    public Response dmaapSinks() {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getDmaapTopicSinks()).build();
-    }
 
     /**
      * GET a source.
@@ -1817,127 +1818,179 @@ public class RestManager {
     }
 
     /**
-     * GET noop sinks.
+     * GET source topic switches.
      */
     @GET
-    @Path("engine/topics/sinks/noop")
-    @ApiOperation(value = "Retrieves the NOOP managed topic sinks", notes = "NOOP Topic Sinks Agregation",
-            responseContainer = "List", response = NoopTopicSink.class)
-    public Response noopSinks() {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getNoopTopicSinks()).build();
+    @Path("engine/topics/sources/{comm: ueb|dmaap|noop}/{topic}/switches")
+    @ApiOperation(value = "Topic Control Switches", notes = "List of the Topic Control Switches",
+            responseContainer = "List")
+    public Response commSourceTopicSwitches(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm
+    ) {
+        return Response.status(Response.Status.OK).entity(Arrays.asList(Switches.values())).build();
     }
 
     /**
-     * GET.
-     *
-     * @return response object
+     * GET sink topic switches.
      */
     @GET
-    @Path("engine/topics/sources/ueb/{topic}/switches")
-    @ApiOperation(value = "UEB Topic Control Switches", notes = "List of the UEB Topic Control Switches",
-            responseContainer = "List")
-    public Response uebTopicSwitches() {
+    @Path("engine/topics/sinks/{comm: ueb|dmaap|noop}/{topic}/switches")
+    @ApiOperation(value = "Topic Control Switches", notes = "List of the Topic Control Switches",
+        responseContainer = "List")
+    public Response commSinkTopicSwitches(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm
+    ) {
         return Response.status(Response.Status.OK).entity(Arrays.asList(Switches.values())).build();
     }
 
     /**
-     * PUT.
-     *
-     * @return response object
+     * PUTs a lock on a topic.
      */
     @PUT
-    @Path("engine/topics/sources/ueb/{topic}/switches/lock")
-    @ApiOperation(value = "Locks an UEB Source topic", response = UebTopicSource.class)
+    @Path("engine/topics/sources/{comm: ueb|dmaap|noop}/{topic}/switches/lock")
+    @ApiOperation(value = "Locks a topic", response = TopicSource.class)
     @ApiResponses(value = {@ApiResponse(code = 406,
             message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
-    public Response uebTopicLock(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        final UebTopicSource source = TopicEndpoint.manager.getUebTopicSource(topic);
-        final boolean success = source.lock();
-        if (success) {
-            return Response.status(Status.OK).entity(source).build();
-        } else {
-            return Response.status(Status.NOT_ACCEPTABLE).entity(makeTopicOperError(topic)).build();
-        }
+    public Response commSourceTopicLock(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic
+    ) {
+        TopicSource source =
+            TopicEndpoint.manager.getTopicSource(CommInfrastructure.valueOf(comm.toUpperCase()), topic);
+        return getResponse(topic, source.lock(), source);
     }
 
     /**
-     * DELETE.
-     *
-     * @return response object
+     * DELETEs the lock on a topic.
      */
     @DELETE
-    @Path("engine/topics/sources/ueb/{topic}/switches/lock")
-    @ApiOperation(value = "Unlocks an UEB Source topic", response = UebTopicSource.class)
+    @Path("engine/topics/sources/{comm: ueb|dmaap|noop}/{topic}/switches/lock")
+    @ApiOperation(value = "Unlocks topic", response = TopicSource.class)
     @ApiResponses(value = {@ApiResponse(code = 406,
             message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
-    public Response uebTopicUnlock(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        final UebTopicSource source = TopicEndpoint.manager.getUebTopicSource(topic);
-        final boolean success = source.unlock();
-        if (success) {
-            return Response.status(Status.OK).entity(source).build();
-        } else {
-            return Response.status(Status.NOT_ACCEPTABLE).entity(makeTopicOperError(topic)).build();
-        }
+    public Response commSourceTopicUnlock(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic
+    ) {
+        TopicSource source =
+            TopicEndpoint.manager.getTopicSource(CommInfrastructure.valueOf(comm.toUpperCase()), topic);
+        return getResponse(topic, source.unlock(), source);
     }
 
-    private Error makeTopicOperError(String topic) {
-        return new Error("cannot perform operation on " + topic);
+    /**
+     * Starts a topic source.
+     */
+    @PUT
+    @Path("engine/topics/sources/{comm: ueb|dmaap|noop}/{topic}/switches/activation")
+    @ApiOperation(value = "Starts a topic", response = TopicSource.class)
+    @ApiResponses(value = {@ApiResponse(code = 406,
+        message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
+    public Response commSourceTopicActivation(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic
+    ) {
+        TopicSource source =
+            TopicEndpoint.manager.getTopicSource(CommInfrastructure.valueOf(comm.toUpperCase()), topic);
+        return getResponse(topic, source.start(), source);
     }
 
     /**
-     * GET.
-     *
-     * @return response object
+     * Stops a topic source.
      */
-    @GET
-    @Path("engine/topics/sources/dmaap/{topic}/switches")
-    @ApiOperation(value = "DMaaP Topic Control Switches", notes = "List of the DMaaP Topic Control Switches",
-            responseContainer = "List")
-    public Response dmaapTopicSwitches() {
-        return Response.status(Response.Status.OK).entity(Arrays.asList(Switches.values())).build();
+    @DELETE
+    @Path("engine/topics/sources/{comm: ueb|dmaap|noop}/{topic}/switches/activation")
+    @ApiOperation(value = "Stops a topic", response = TopicSource.class)
+    @ApiResponses(value = {@ApiResponse(code = 406,
+        message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
+    public Response commSourceTopicDeactivation(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic
+    ) {
+        TopicSource source =
+            TopicEndpoint.manager.getTopicSource(CommInfrastructure.valueOf(comm.toUpperCase()), topic);
+        return getResponse(topic, source.stop(), source);
     }
 
     /**
-     * PUT.
-     *
-     * @return response object
+     * PUTs a lock on a topic.
      */
     @PUT
-    @Path("engine/topics/sources/dmaap/{topic}/switches/lock")
-    @ApiOperation(value = "Locks an DMaaP Source topic", response = DmaapTopicSource.class)
+    @Path("engine/topics/sinks/{comm: ueb|dmaap|noop}/{topic}/switches/lock")
+    @ApiOperation(value = "Locks a topic sink", response = TopicSink.class)
     @ApiResponses(value = {@ApiResponse(code = 406,
-            message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
-    public Response dmmapTopicLock(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        final DmaapTopicSource source = TopicEndpoint.manager.getDmaapTopicSource(topic);
-        final boolean success = source.lock();
-        if (success) {
-            return Response.status(Status.OK).entity(source).build();
-        } else {
-            return Response.status(Status.NOT_ACCEPTABLE).entity(makeTopicOperError(topic)).build();
-        }
+        message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
+    public Response commSinkTopicLock(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic
+    ) {
+        TopicSink sink =
+            TopicEndpoint.manager.getTopicSink(CommInfrastructure.valueOf(comm.toUpperCase()), topic);
+        return getResponse(topic, sink.lock(), sink);
     }
 
     /**
-     * DELETE.
-     *
-     * @return response object
+     * DELETEs the lock on a topic.
      */
     @DELETE
-    @Path("engine/topics/sources/dmaap/{topic}/switches/lock")
-    @ApiOperation(value = "Unlocks an DMaaP Source topic", response = DmaapTopicSource.class)
+    @Path("engine/topics/sinks/{comm: ueb|dmaap|noop}/{topic}/switches/lock")
+    @ApiOperation(value = "Unlocks a topic sink", response = TopicSink.class)
     @ApiResponses(value = {@ApiResponse(code = 406,
-            message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
-    public Response dmaapTopicUnlock(
-            @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        final DmaapTopicSource source = TopicEndpoint.manager.getDmaapTopicSource(topic);
-        final boolean success = source.unlock();
+        message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
+    public Response commSinkTopicUnlock(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic
+    ) {
+        TopicSink sink =
+            TopicEndpoint.manager.getTopicSink(CommInfrastructure.valueOf(comm.toUpperCase()), topic);
+        return getResponse(topic, sink.unlock(), sink);
+    }
+
+    /**
+     * Starts a topic sink.
+     */
+    @PUT
+    @Path("engine/topics/sinks/{comm: ueb|dmaap|noop}/{topic}/switches/activation")
+    @ApiOperation(value = "Starts a topic sink", response = TopicSink.class)
+    @ApiResponses(value = {@ApiResponse(code = 406,
+        message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
+    public Response commSinkTopicActivation(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic
+    ) {
+        TopicSink sink =
+            TopicEndpoint.manager.getTopicSink(CommInfrastructure.valueOf(comm.toUpperCase()), topic);
+        return getResponse(topic, sink.start(), sink);
+    }
+
+    /**
+     * Stops a topic sink.
+     */
+    @DELETE
+    @Path("engine/topics/sinks/{comm: ueb|dmaap|noop}/{topic}/switches/activation")
+    @ApiOperation(value = "Stops a topic", response = TopicSource.class)
+    @ApiResponses(value = {@ApiResponse(code = 406,
+        message = "The system is an administrative state that prevents " + "this request to be fulfilled")})
+    public Response commSinkTopicDeactivation(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic
+    ) {
+        TopicSink sink =
+            TopicEndpoint.manager.getTopicSink(CommInfrastructure.valueOf(comm.toUpperCase()), topic);
+        return getResponse(topic, sink.stop(), sink);
+    }
+
+    private Response getResponse(String topicName, boolean success, Topic topic) {
         if (success) {
-            return Response.status(Status.OK).entity(source).build();
+            return Response.status(Status.OK).entity(topic).build();
         } else {
-            return Response.status(Status.SERVICE_UNAVAILABLE).entity(makeTopicOperError(topic)).build();
+            return Response.status(Status.NOT_ACCEPTABLE).entity(makeTopicOperError(topicName)).build();
         }
     }
 
+    private Error makeTopicOperError(String topic) {
+        return new Error("cannot perform operation on " + topic);
+    }
+
     /**
      * Offers and event to a topic in a communication infrastructure.
      *
diff --git a/policy-management/src/test/java/org/onap/policy/drools/server/restful/test/RestManagerTest.java b/policy-management/src/test/java/org/onap/policy/drools/server/restful/test/RestManagerTest.java
index c484497c..3df9fac2 100644
--- a/policy-management/src/test/java/org/onap/policy/drools/server/restful/test/RestManagerTest.java
+++ b/policy-management/src/test/java/org/onap/policy/drools/server/restful/test/RestManagerTest.java
@@ -340,63 +340,56 @@ public class RestManagerTest {
         assertEquals(200, response.getStatusLine().getStatusCode());
         httpDelete.releaseConnection();
 
-        /*
-         * PUT: /engine/topics/sources/ueb/topic/switches/lock
-         * /engine/topics/sources/dmaap/topic/switches/lock DELETE:
-         * /engine/topics/sources/ueb/topic/switches/lock
-         * /engine/topics/sources/dmaap/topic/switches/lock
-         */
-        httpPut = new HttpPut(HOST_URL + "/engine/topics/sources/ueb/" + UEB_TOPIC + "/switches/lock");
+        putDeleteSwitch("/engine/topics/sources/ueb/", UEB_TOPIC, "lock");
+        putDeleteSwitch("/engine/topics/sources/dmaap/", DMAAP_TOPIC, "lock");
+        putDeleteSwitch("/engine/topics/sources/noop/", NOOP_TOPIC, "lock");
+        putDeleteSwitch("/engine/topics/sinks/ueb/", UEB_TOPIC, "lock");
+        putDeleteSwitch("/engine/topics/sinks/dmaap/", DMAAP_TOPIC, "lock");
+        putDeleteSwitch("/engine/topics/sinks/noop/", NOOP_TOPIC, "lock");
+
+        putDeleteSwitch("/engine/topics/sources/ueb/", UEB_TOPIC, "activation");
+        putDeleteSwitch("/engine/topics/sources/dmaap/", DMAAP_TOPIC, "activation");
+        putDeleteSwitch("/engine/topics/sources/noop/", NOOP_TOPIC, "activation");
+        putDeleteSwitch("/engine/topics/sinks/ueb/", UEB_TOPIC, "activation");
+        putDeleteSwitch("/engine/topics/sinks/dmaap/", DMAAP_TOPIC, "activation");
+        putDeleteSwitch("/engine/topics/sinks/noop/", NOOP_TOPIC, "activation");
+
+        putSwitch("/engine/topics/sources/ueb/", UEB_TOPIC, "activation");
+        putSwitch("/engine/topics/sources/dmaap/", DMAAP_TOPIC, "activation");
+        putSwitch("/engine/topics/sources/noop/", NOOP_TOPIC, "activation");
+        putSwitch("/engine/topics/sinks/ueb/", UEB_TOPIC, "activation");
+        putSwitch("/engine/topics/sinks/dmaap/", DMAAP_TOPIC, "activation");
+        putSwitch("/engine/topics/sinks/noop/", NOOP_TOPIC, "activation");
+
+        httpPut = new HttpPut(HOST_URL + "/engine/tools/loggers/ROOT/debug");
         response = client.execute(httpPut);
-        logger.info(httpPut.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        logger.info(httpPut.getRequestLine() + "response code: {}", response.getStatusLine().getStatusCode());
         assertEquals(200, response.getStatusLine().getStatusCode());
         httpPut.releaseConnection();
 
-        httpDelete = new HttpDelete(HOST_URL + "/engine/topics/sources/ueb/" + UEB_TOPIC + "/switches/lock");
-        response = client.execute(httpDelete);
-        logger.info(httpDelete.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
-        assertEquals(200, response.getStatusLine().getStatusCode());
-        httpDelete.releaseConnection();
+    }
 
-        httpPut = new HttpPut(HOST_URL + "/engine/topics/sources/dmaap/" + DMAAP_TOPIC + "/switches/lock");
-        response = client.execute(httpPut);
-        logger.info(httpPut.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
-        assertEquals(200, response.getStatusLine().getStatusCode());
-        httpPut.releaseConnection();
+    private void putDeleteSwitch(String urlPrefix, String topic, String control) throws IOException {
+        putSwitch(urlPrefix, topic, control);
+        deleteSwitch(urlPrefix, topic, control);
+    }
 
-        httpDelete = new HttpDelete(HOST_URL + "/engine/topics/sources/dmaap/" + DMAAP_TOPIC + "/switches/lock");
-        response = client.execute(httpDelete);
+    private void deleteSwitch(String urlPrefix, String topic, String control) throws IOException {
+        HttpDelete httpDelete = new HttpDelete(HOST_URL + urlPrefix + topic + "/switches/" + control);
+        CloseableHttpResponse response = client.execute(httpDelete);
         logger.info(httpDelete.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
         assertEquals(200, response.getStatusLine().getStatusCode());
         httpDelete.releaseConnection();
+    }
 
-        /*
-         * PUT: /engine/switches/activation DELETE: /engine/switches/activation
-         */
-        /*
-         * httpPut = new HttpPut(HOST_URL + "/engine/switches/activation"); response =
-         * client.execute(httpPut); logger.info(httpPut.getRequestLine() + " response code: {}",
-         * response.getStatusLine().getStatusCode()); assertEquals(200,
-         * response.getStatusLine().getStatusCode()); httpPut.releaseConnection();
-         * 
-         * httpDelete = new HttpDelete(HOST_URL + "/engine/switches/activation"); response =
-         * client.execute(httpDelete); logger.info(httpDelete.getRequestLine() +
-         * " response code: {}", response.getStatusLine().getStatusCode()); assertEquals(200,
-         * response.getStatusLine().getStatusCode()); httpDelete.releaseConnection();
-         */
-
-        /*
-         * PUT: /engine/tools/loggers/logger/level
-         */
-        httpPut = new HttpPut(HOST_URL + "/engine/tools/loggers/ROOT/debug");
-        response = client.execute(httpPut);
-        logger.info(httpPut.getRequestLine() + "response code: {}", response.getStatusLine().getStatusCode());
+    private void putSwitch(String urlPrefix, String topic, String control) throws IOException {
+        HttpPut httpPut = new HttpPut(HOST_URL + urlPrefix + topic + "/switches/" + control);
+        CloseableHttpResponse response = client.execute(httpPut);
+        logger.info(httpPut.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
         assertEquals(200, response.getStatusLine().getStatusCode());
         httpPut.releaseConnection();
-
     }
 
-
     @Test
     public void getTest() throws IOException {
         HttpGet httpGet;
@@ -735,6 +728,18 @@ public class RestManagerTest {
         assertEquals(200, response.getStatusLine().getStatusCode());
         httpGet.releaseConnection();
 
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sources/noop");
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(200, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/noop");
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(200, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
         httpGet = new HttpGet(HOST_URL + "/engine/topics/sources/ueb/" + UEB_TOPIC);
         response = client.execute(httpGet);
         logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
@@ -783,6 +788,30 @@ public class RestManagerTest {
         assertEquals(500, response.getStatusLine().getStatusCode());
         httpGet.releaseConnection();
 
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sources/noop/" + NOOP_TOPIC);
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(200, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sources/noop/foobar");
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(500, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/noop/" + NOOP_TOPIC);
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(200, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/noop/foobar");
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(500, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
         httpGet = new HttpGet(HOST_URL + "/engine/topics/sources/ueb/" + UEB_TOPIC + "/events");
         response = client.execute(httpGet);
         logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
@@ -831,6 +860,30 @@ public class RestManagerTest {
         assertEquals(500, response.getStatusLine().getStatusCode());
         httpGet.releaseConnection();
 
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sources/noop/" + NOOP_TOPIC + "/events");
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(200, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sources/noop/foobar/events");
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(500, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/noop/" + NOOP_TOPIC + "/events");
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(200, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/noop/foobar/events");
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(500, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
         httpGet = new HttpGet(HOST_URL + "/engine/topics/sources/ueb/" + UEB_TOPIC + "/switches");
         response = client.execute(httpGet);
         logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
@@ -843,23 +896,25 @@ public class RestManagerTest {
         assertEquals(200, response.getStatusLine().getStatusCode());
         httpGet.releaseConnection();
 
-        /*
-         * GET: /engine/topics/sinks/noop /engine/topics/sinks/noop/topic
-         * /engine/topics/sinks/noop/topic/events
-         */
-        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/noop");
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sources/noop/" + NOOP_TOPIC + "/switches");
         response = client.execute(httpGet);
         logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
         assertEquals(200, response.getStatusLine().getStatusCode());
         httpGet.releaseConnection();
 
-        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/noop/" + NOOP_TOPIC);
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/ueb/" + UEB_TOPIC + "/switches");
         response = client.execute(httpGet);
         logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
         assertEquals(200, response.getStatusLine().getStatusCode());
         httpGet.releaseConnection();
 
-        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/noop/" + NOOP_TOPIC + "/events");
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/dmaap/" + DMAAP_TOPIC + "/switches");
+        response = client.execute(httpGet);
+        logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(200, response.getStatusLine().getStatusCode());
+        httpGet.releaseConnection();
+
+        httpGet = new HttpGet(HOST_URL + "/engine/topics/sinks/noop/" + NOOP_TOPIC + "/switches");
         response = client.execute(httpGet);
         logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
         assertEquals(200, response.getStatusLine().getStatusCode());
@@ -885,7 +940,6 @@ public class RestManagerTest {
         logger.info(httpGet.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
         assertEquals(200, response.getStatusLine().getStatusCode());
         httpGet.releaseConnection();
-
     }
 
     /**