### Question:
Engine implements ReactiveStreamsEngine { @Override public <T> Publisher<T> buildPublisher(Graph graph) { Multi<T> publisher = null; for (Stage stage : graph.getStages()) { Operator operator = Stages.lookup(stage); if (publisher == null) { if (operator instanceof PublisherOperator) { publisher = createPublisher(stage, (PublisherOperator) operator); } else { throw new IllegalArgumentException("Expecting a publisher stage, got a " + stage); } } else { if (operator instanceof ProcessorOperator) { publisher = applyProcessors(publisher, stage, (ProcessorOperator) operator); } else { throw new IllegalArgumentException("Expecting a processor stage, got a " + stage); } } } return publisher; }  @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 testValidPublisher() { engine = new Engine(); List<Stage> stages = new ArrayList<>(); stages.add((Stage.Of) () -> Arrays.asList(1, 2, 3)); stages.add((Stage.Map) () -> i -> (int) i + 1); Graph graph = () -> stages; assertThat(engine.buildPublisher(graph)).isNotNull(); }
@Test public void testInvalidPublisher() { assertThrows(IllegalArgumentException.class, () -> { engine = new Engine(); List<Stage> stages = new ArrayList<>(); stages.add((Stage.Map) () -> i -> (int) i + 1); stages.add(new Stage.FindFirst() { }); Graph graph = () -> stages; engine.buildPublisher(graph); }); }
@Test public void testCreatingPublisherWithUnknownStage() { assertThrows(UnsupportedStageException.class, () -> { engine = new Engine(); List<Stage> stages = new ArrayList<>(); stages.add(new Stage() { }); stages.add((Stage.Map) () -> i -> (int) i + 1); Graph graph = () -> stages; engine.buildPublisher(graph); }); }