@Test(expected = RuntimeException.class) public void addQuery_insertQueryNotCorrectType() throws Exception { final RyaStreamsClient mockClient = mock(RyaStreamsClient.class); final AddQuery addQuery = mock(AddQuery.class); when(mockClient.getAddQuery()).thenReturn(addQuery); final String sparql = "SELECT * WHERE { ?a ?b ?c }"; final StreamsQuery addedQuery = new StreamsQuery(UUID.randomUUID(), sparql, true, true); when(addQuery.addQuery(eq(sparql), eq(false), eq(true))).thenReturn(addedQuery); final SparqlPrompt prompt = mock(SparqlPrompt.class); when(prompt.getSparql()).thenReturn(Optional.of(sparql)); final SharedShellState state = new SharedShellState(); state.connectedToAccumulo(mock(AccumuloConnectionDetails.class), mock(RyaClient.class)); state.connectedToInstance("unitTest"); state.connectedToRyaStreams(mockClient); final RyaStreamsCommands commands = new RyaStreamsCommands(state, prompt, mock(ConsolePrinter.class)); commands.addQuery(true, true); }
@CliCommand(value = STREAM_QUERIES_ADD_CMD, help = "Add a SPARQL query to the Rya Streams subsystem.") public String addQuery( @CliOption(key = {"inactive"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will add the query, but not run it. (default: false)") final boolean inactive, @CliOption(key = {"insert"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will insert the query's results back into Rya. (default: false)") final boolean isInsert) { final RyaStreamsClient streamsClient = state.getShellState().getRyaStreamsCommands().get(); try { final Optional<String> sparql = sparqlPrompt.getSparql(); if(!sparql.isPresent()) { return ""; } final boolean isConstructQuery = QueryInvestigator.isConstruct(sparql.get()); final boolean isInsertQuery = QueryInvestigator.isInsertWhere(sparql.get()); if(isInsert && isConstructQuery) { consolePrinter.println("WARNING: CONSTRUCT is part of the SPARQL Query API, so they do not normally"); consolePrinter.println("get written back to the triple store. Consider using an INSERT, which is"); consolePrinter.println("part of the SPARQL Update API, in the future."); } if(!isInsert && isInsertQuery) { consolePrinter.println("WARNING: INSERT is part of the SPARQL Update API, so they normally get written"); consolePrinter.println("back to the triple store. Consider using a CONSTRUCT, which is part of the"); consolePrinter.println("SPARQL Query API, in the future."); } if(isInsert && !(isConstructQuery || isInsertQuery)) { throw new RuntimeException("Only CONSTRUCT queries and INSERT updates may be inserted back to the triple store."); } final StreamsQuery streamsQuery = streamsClient.getAddQuery().addQuery(sparql.get(), !inactive, isInsert); return "The added query's ID is " + streamsQuery.getQueryId(); } catch (final MalformedQueryException | IOException | RyaStreamsException e) { throw new RuntimeException("Unable to add the SPARQL query to the Rya Streams subsystem.", e); } }
RyaStreamsCommands implements CommandMarker { @CliCommand(value = STREAM_QUERIES_ADD_CMD, help = "Add a SPARQL query to the Rya Streams subsystem.") public String addQuery( @CliOption(key = {"inactive"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will add the query, but not run it. (default: false)") final boolean inactive, @CliOption(key = {"insert"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will insert the query's results back into Rya. (default: false)") final boolean isInsert) { final RyaStreamsClient streamsClient = state.getShellState().getRyaStreamsCommands().get(); try { final Optional<String> sparql = sparqlPrompt.getSparql(); if(!sparql.isPresent()) { return ""; } final boolean isConstructQuery = QueryInvestigator.isConstruct(sparql.get()); final boolean isInsertQuery = QueryInvestigator.isInsertWhere(sparql.get()); if(isInsert && isConstructQuery) { consolePrinter.println("WARNING: CONSTRUCT is part of the SPARQL Query API, so they do not normally"); consolePrinter.println("get written back to the triple store. Consider using an INSERT, which is"); consolePrinter.println("part of the SPARQL Update API, in the future."); } if(!isInsert && isInsertQuery) { consolePrinter.println("WARNING: INSERT is part of the SPARQL Update API, so they normally get written"); consolePrinter.println("back to the triple store. Consider using a CONSTRUCT, which is part of the"); consolePrinter.println("SPARQL Query API, in the future."); } if(isInsert && !(isConstructQuery || isInsertQuery)) { throw new RuntimeException("Only CONSTRUCT queries and INSERT updates may be inserted back to the triple store."); } final StreamsQuery streamsQuery = streamsClient.getAddQuery().addQuery(sparql.get(), !inactive, isInsert); return "The added query's ID is " + streamsQuery.getQueryId(); } catch (final MalformedQueryException | IOException | RyaStreamsException e) { throw new RuntimeException("Unable to add the SPARQL query to the Rya Streams subsystem.", e); } } }
RyaStreamsCommands implements CommandMarker { @CliCommand(value = STREAM_QUERIES_ADD_CMD, help = "Add a SPARQL query to the Rya Streams subsystem.") public String addQuery( @CliOption(key = {"inactive"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will add the query, but not run it. (default: false)") final boolean inactive, @CliOption(key = {"insert"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will insert the query's results back into Rya. (default: false)") final boolean isInsert) { final RyaStreamsClient streamsClient = state.getShellState().getRyaStreamsCommands().get(); try { final Optional<String> sparql = sparqlPrompt.getSparql(); if(!sparql.isPresent()) { return ""; } final boolean isConstructQuery = QueryInvestigator.isConstruct(sparql.get()); final boolean isInsertQuery = QueryInvestigator.isInsertWhere(sparql.get()); if(isInsert && isConstructQuery) { consolePrinter.println("WARNING: CONSTRUCT is part of the SPARQL Query API, so they do not normally"); consolePrinter.println("get written back to the triple store. Consider using an INSERT, which is"); consolePrinter.println("part of the SPARQL Update API, in the future."); } if(!isInsert && isInsertQuery) { consolePrinter.println("WARNING: INSERT is part of the SPARQL Update API, so they normally get written"); consolePrinter.println("back to the triple store. Consider using a CONSTRUCT, which is part of the"); consolePrinter.println("SPARQL Query API, in the future."); } if(isInsert && !(isConstructQuery || isInsertQuery)) { throw new RuntimeException("Only CONSTRUCT queries and INSERT updates may be inserted back to the triple store."); } final StreamsQuery streamsQuery = streamsClient.getAddQuery().addQuery(sparql.get(), !inactive, isInsert); return "The added query's ID is " + streamsQuery.getQueryId(); } catch (final MalformedQueryException | IOException | RyaStreamsException e) { throw new RuntimeException("Unable to add the SPARQL query to the Rya Streams subsystem.", e); } } @Autowired  RyaStreamsCommands(
            final SharedShellState state,
            final SparqlPrompt sparqlPrompt,
            final ConsolePrinter consolePrinter); }
RyaStreamsCommands implements CommandMarker { @CliCommand(value = STREAM_QUERIES_ADD_CMD, help = "Add a SPARQL query to the Rya Streams subsystem.") public String addQuery( @CliOption(key = {"inactive"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will add the query, but not run it. (default: false)") final boolean inactive, @CliOption(key = {"insert"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will insert the query's results back into Rya. (default: false)") final boolean isInsert) { final RyaStreamsClient streamsClient = state.getShellState().getRyaStreamsCommands().get(); try { final Optional<String> sparql = sparqlPrompt.getSparql(); if(!sparql.isPresent()) { return ""; } final boolean isConstructQuery = QueryInvestigator.isConstruct(sparql.get()); final boolean isInsertQuery = QueryInvestigator.isInsertWhere(sparql.get()); if(isInsert && isConstructQuery) { consolePrinter.println("WARNING: CONSTRUCT is part of the SPARQL Query API, so they do not normally"); consolePrinter.println("get written back to the triple store. Consider using an INSERT, which is"); consolePrinter.println("part of the SPARQL Update API, in the future."); } if(!isInsert && isInsertQuery) { consolePrinter.println("WARNING: INSERT is part of the SPARQL Update API, so they normally get written"); consolePrinter.println("back to the triple store. Consider using a CONSTRUCT, which is part of the"); consolePrinter.println("SPARQL Query API, in the future."); } if(isInsert && !(isConstructQuery || isInsertQuery)) { throw new RuntimeException("Only CONSTRUCT queries and INSERT updates may be inserted back to the triple store."); } final StreamsQuery streamsQuery = streamsClient.getAddQuery().addQuery(sparql.get(), !inactive, isInsert); return "The added query's ID is " + streamsQuery.getQueryId(); } catch (final MalformedQueryException | IOException | RyaStreamsException e) { throw new RuntimeException("Unable to add the SPARQL query to the Rya Streams subsystem.", e); } } @Autowired  RyaStreamsCommands(
            final SharedShellState state,
            final SparqlPrompt sparqlPrompt,
            final ConsolePrinter consolePrinter); @CliAvailabilityIndicator({ STREAMS_CONFIGURE_CMD, STREAMS_DETAILS_CMD}) boolean areConfigCommandsAvailable(); @CliAvailabilityIndicator({ STREAM_QUERIES_ADD_CMD, STREAM_QUERIES_DELETE_CMD, STREAM_QUERIES_START_CMD, STREAM_QUERIES_STOP_CMD, STREAM_QUERIES_LIST_CMD, STREAM_QUERIES_DETAILS_CMD}) boolean areQueriesCommandsAvailable(); @CliCommand(value = STREAMS_CONFIGURE_CMD, help = "Connect a Rya Streams subsystem to a Rya Instance.") String configureRyaStreams(
            @CliOption(key = {"kafkaHostname"}, mandatory = true, help = "The hostname of the Kafka Broker.")
            final String kafkaHostname,
            @CliOption(key = {"kafkaPort"}, mandatory = true, help = "The port of the Kafka Broker.")
            final int kafkaPort); @CliCommand(value = STREAMS_DETAILS_CMD, help = "Print information about which Rya Streams subsystem the Rya instance is connected to.") String printRyaStreamsDetails(); @CliCommand(value = STREAM_QUERIES_ADD_CMD, help = "Add a SPARQL query to the Rya Streams subsystem.") String addQuery(
            @CliOption(key = {"inactive"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true",
                       help = "Setting this flag will add the query, but not run it. (default: false)")
            final boolean inactive,
            @CliOption(key = {"insert"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true",
                       help = "Setting this flag will insert the query's results back into Rya. (default: false)")
            final boolean isInsert); @CliCommand(value = STREAM_QUERIES_DELETE_CMD, help = "Delete a SPARQL query from the Rya Streams subsystem.") String deleteQuery(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query to remove.")
            final String queryId); @CliCommand(value = STREAM_QUERIES_START_CMD, help = "Start processing a SPARQL query using the Rya Streams subsystem.") String startQuery(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query to start processing.")
            final String queryId); @CliCommand(value = STREAM_QUERIES_STOP_CMD, help = "Stop processing a SPARQL query using the Rya Streams subsystem.") String stopQuery(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query to stop processing.")
            final String queryId); @CliCommand(value = STREAM_QUERIES_LIST_CMD, help = "List the queries that are being managed by the configured Rya Streams subsystem.") String listQueries(); @CliCommand(value = STREAM_QUERIES_DETAILS_CMD, help = "Print detailed information about a specific query managed by the Rya Streams subsystem.") String printQueryDetails(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query whose details will be printed.")
            final String queryId); }
RyaStreamsCommands implements CommandMarker { @CliCommand(value = STREAM_QUERIES_ADD_CMD, help = "Add a SPARQL query to the Rya Streams subsystem.") public String addQuery( @CliOption(key = {"inactive"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will add the query, but not run it. (default: false)") final boolean inactive, @CliOption(key = {"insert"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true", help = "Setting this flag will insert the query's results back into Rya. (default: false)") final boolean isInsert) { final RyaStreamsClient streamsClient = state.getShellState().getRyaStreamsCommands().get(); try { final Optional<String> sparql = sparqlPrompt.getSparql(); if(!sparql.isPresent()) { return ""; } final boolean isConstructQuery = QueryInvestigator.isConstruct(sparql.get()); final boolean isInsertQuery = QueryInvestigator.isInsertWhere(sparql.get()); if(isInsert && isConstructQuery) { consolePrinter.println("WARNING: CONSTRUCT is part of the SPARQL Query API, so they do not normally"); consolePrinter.println("get written back to the triple store. Consider using an INSERT, which is"); consolePrinter.println("part of the SPARQL Update API, in the future."); } if(!isInsert && isInsertQuery) { consolePrinter.println("WARNING: INSERT is part of the SPARQL Update API, so they normally get written"); consolePrinter.println("back to the triple store. Consider using a CONSTRUCT, which is part of the"); consolePrinter.println("SPARQL Query API, in the future."); } if(isInsert && !(isConstructQuery || isInsertQuery)) { throw new RuntimeException("Only CONSTRUCT queries and INSERT updates may be inserted back to the triple store."); } final StreamsQuery streamsQuery = streamsClient.getAddQuery().addQuery(sparql.get(), !inactive, isInsert); return "The added query's ID is " + streamsQuery.getQueryId(); } catch (final MalformedQueryException | IOException | RyaStreamsException e) { throw new RuntimeException("Unable to add the SPARQL query to the Rya Streams subsystem.", e); } } @Autowired  RyaStreamsCommands(
            final SharedShellState state,
            final SparqlPrompt sparqlPrompt,
            final ConsolePrinter consolePrinter); @CliAvailabilityIndicator({ STREAMS_CONFIGURE_CMD, STREAMS_DETAILS_CMD}) boolean areConfigCommandsAvailable(); @CliAvailabilityIndicator({ STREAM_QUERIES_ADD_CMD, STREAM_QUERIES_DELETE_CMD, STREAM_QUERIES_START_CMD, STREAM_QUERIES_STOP_CMD, STREAM_QUERIES_LIST_CMD, STREAM_QUERIES_DETAILS_CMD}) boolean areQueriesCommandsAvailable(); @CliCommand(value = STREAMS_CONFIGURE_CMD, help = "Connect a Rya Streams subsystem to a Rya Instance.") String configureRyaStreams(
            @CliOption(key = {"kafkaHostname"}, mandatory = true, help = "The hostname of the Kafka Broker.")
            final String kafkaHostname,
            @CliOption(key = {"kafkaPort"}, mandatory = true, help = "The port of the Kafka Broker.")
            final int kafkaPort); @CliCommand(value = STREAMS_DETAILS_CMD, help = "Print information about which Rya Streams subsystem the Rya instance is connected to.") String printRyaStreamsDetails(); @CliCommand(value = STREAM_QUERIES_ADD_CMD, help = "Add a SPARQL query to the Rya Streams subsystem.") String addQuery(
            @CliOption(key = {"inactive"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true",
                       help = "Setting this flag will add the query, but not run it. (default: false)")
            final boolean inactive,
            @CliOption(key = {"insert"}, mandatory = false, unspecifiedDefaultValue = "false", specifiedDefaultValue = "true",
                       help = "Setting this flag will insert the query's results back into Rya. (default: false)")
            final boolean isInsert); @CliCommand(value = STREAM_QUERIES_DELETE_CMD, help = "Delete a SPARQL query from the Rya Streams subsystem.") String deleteQuery(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query to remove.")
            final String queryId); @CliCommand(value = STREAM_QUERIES_START_CMD, help = "Start processing a SPARQL query using the Rya Streams subsystem.") String startQuery(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query to start processing.")
            final String queryId); @CliCommand(value = STREAM_QUERIES_STOP_CMD, help = "Stop processing a SPARQL query using the Rya Streams subsystem.") String stopQuery(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query to stop processing.")
            final String queryId); @CliCommand(value = STREAM_QUERIES_LIST_CMD, help = "List the queries that are being managed by the configured Rya Streams subsystem.") String listQueries(); @CliCommand(value = STREAM_QUERIES_DETAILS_CMD, help = "Print detailed information about a specific query managed by the Rya Streams subsystem.") String printQueryDetails(
            @CliOption(key= {"queryId"}, mandatory = true, help = "The ID of the query whose details will be printed.")
            final String queryId); static final String STREAMS_CONFIGURE_CMD; static final String STREAMS_DETAILS_CMD; static final String STREAM_QUERIES_ADD_CMD; static final String STREAM_QUERIES_DELETE_CMD; static final String STREAM_QUERIES_START_CMD; static final String STREAM_QUERIES_STOP_CMD; static final String STREAM_QUERIES_LIST_CMD; static final String STREAM_QUERIES_DETAILS_CMD; }