CompositeOperator extends Operator<Self> { @Override public PactModule asPactModule(final EvaluationContext context) { if (LOG.isTraceEnabled()) LOG.trace("Transforming\n" + this); final ElementarySopremoModule elementaryPlan = this.asElementaryOperators(context); if (LOG.isTraceEnabled()) LOG.trace(" to elementary plan\n" + elementaryPlan); context.pushOperator(this); final PactModule pactModule = elementaryPlan.asPactModule(context); context.popOperator(); return pactModule; } CompositeOperator(final int numberOfInputs, final int numberOfOutputs);  CompositeOperator(final int minInputs, final int maxInputs, final int minOutputs, final int maxOutputs);  CompositeOperator(); @Override abstract ElementarySopremoModule asElementaryOperators(EvaluationContext context); @Override PactModule asPactModule(final EvaluationContext context);  }
@Test public void testAsPactModule() throws Exception { final Operator<?> input1 = new Source("file: final Operator<?> input2 = new Source("file: final Operator<?> input3 = new Source("file: final CompositeOperator<?> fixture = new CompositeOperatorImpl(1); fixture.setInputs(input1, input2, input3); final EvaluationContext context = new EvaluationContext(); final PactModule result = fixture.asPactModule(context); assertNotNull(result); final List<Level<Contract>> reachableNodes = GraphLevelPartitioner.getLevels( result.getAllOutputs(), ContractNavigator.INSTANCE); assertEquals(3, reachableNodes.get(0).getLevelNodes().size()); assertEquals(1, reachableNodes.get(1).getLevelNodes().size()); assertEquals(1, reachableNodes.get(2).getLevelNodes().size()); assertEquals(1, reachableNodes.get(3).getLevelNodes().size()); for (int index = 0; index < 3; index++) assertTrue(FileDataSource.class.isInstance(reachableNodes.get(0).getLevelNodes() .get(index))); assertSame(ElementaryOperatorImpl.Implementation.class, reachableNodes.get(1) .getLevelNodes().get(0).getUserCodeClass()); assertSame(ElementaryOperatorImpl.Implementation.class, reachableNodes.get(2) .getLevelNodes().get(0).getUserCodeClass()); assertTrue(FileDataSink.class.isInstance(reachableNodes.get(3).getLevelNodes().get(0))); }