### Question:
Engine implements ReactiveStreamsEngine { @Override public <T, R> SubscriberWithCompletionStage<T, R> buildSubscriber(Graph graph) { Processor<T, T> processor = new ConnectableProcessor<>(); Multi<T> flowable = Multi.createFrom().publisher(processor); for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (operator instanceof ProcessorOperator) { flowable = applyProcessors(flowable, stage, (ProcessorOperator) operator); } else if (operator instanceof TerminalOperator) { CompletionStage<R> result = applySubscriber(Transformer.apply(flowable), stage, (TerminalOperator) operator); return new DefaultSubscriberWithCompletionStage<>(processor, result); } else { throw new UnsupportedStageException(stage); } } throw new IllegalArgumentException("The graph does not have a valid final stage"); }  @Override Publisher<T> buildPublisher(Graph graph); @Override SubscriberWithCompletionStage<T, R> buildSubscriber(Graph graph); @Override Processor<T, R> buildProcessor(Graph graph); @Override CompletionStage<T> buildCompletion(Graph graph);  }### Answer:
@Test public void testUnknownTerminalStage() { assertThrows(UnsupportedStageException.class, () -> { engine = new Engine(); List<Stage> stages = new ArrayList<>(); stages.add((Stage.Map) () -> i -> (int) i + 1); stages.add(new Stage() { }); Graph graph = () -> stages; engine.buildSubscriber(graph); }); }
@Test public void testInvalidSubscriber() { assertThrows(IllegalArgumentException.class, () -> { engine = new Engine(); List<Stage> stages = new ArrayList<>(); stages.add((Stage.Map) () -> i -> (int) i + 1); Graph graph = () -> stages; engine.buildSubscriber(graph); }); }