/*
 * RiskScape™ Copyright New Zealand Institute for Earth Science Limited
 * (Earth Sciences New Zealand) is distributed for research purposes only
 * under the terms of AGPLv3.
 *
 * RiskScape™ Copyright 2026 New Zealand Institute for Earth Science
 * Limited (Earth Sciences New Zealand). All rights reserved. Source code
 * available under the AGPLv3.
 * 
 * This program is free software: you can redistribute it and/or modify it under
 *  the terms of the GNU Affero General Public License as published by the Free
 *  Software Foundation, either version 3 of the License, or (at your option) any
 *  later version.
 * 
 * This program is distributed for RESEARCH PURPOSES ONLY, in the hope that it will
 * be useful for research and education initiatives.
 * 
 * If you are not a researcher, or you are a researcher who wishes to use this
 * program on terms other than AGPLv3 (including those who wish to restrict the
 * distribution of any source code created using this program), please contact:
 * https://riskscape.org.nz
 * 
 * This program is distributed WITHOUT ANY WARRANTY; without even the implied
 * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Affero General Public License for more details.  You should have received a copy
 * of the GNU Affero General Public License along with this program.  If not, see
 * <http://www.gnu.org/licenses/>.
 * 
 * By way of summary only, under the AGPLv3:
 *     • Permissions of this strongest copyleft license are conditioned
 *       on making available complete source code of licensed works and
 *       modifications, which include larger works using a licensed work,
 *       under the same license.
 *     • Copyright and license notices must be preserved.
 *     • Contributors provide an express grant of patent rights.
 *     • When a modified version is used to provide a service over a
 *       network, the complete source code of the modified version must be made
 *       available.
 */
package nz.org.riskscape.engine.pipeline;

import static nz.org.riskscape.engine.Matchers.*;
import static nz.org.riskscape.engine.problem.ProblemMatchers.isProblem;
import static org.hamcrest.Matchers.*;
import static org.junit.Assert.*;

import java.util.List;
import java.util.Map;

import org.hamcrest.Matcher;
import org.junit.Before;
import org.junit.Test;

import com.google.common.collect.Range;

import nz.org.riskscape.engine.ArgsProblems;
import nz.org.riskscape.engine.DefaultEngine;
import nz.org.riskscape.engine.DefaultProject;
import nz.org.riskscape.engine.bind.Parameter;
import nz.org.riskscape.engine.bind.ParameterBinder;
import nz.org.riskscape.engine.bind.ParameterSet;
import nz.org.riskscape.engine.bind.impl.EnumBinder;
import nz.org.riskscape.engine.bind.impl.NumberBinder;
import nz.org.riskscape.engine.i18n.RiskscapeMessage;
import nz.org.riskscape.engine.problem.GeneralProblems;
import nz.org.riskscape.engine.relation.ListRelation;
import nz.org.riskscape.engine.steps.Input;
import nz.org.riskscape.engine.steps.JoinStep;
import nz.org.riskscape.engine.steps.SelectStep;
import nz.org.riskscape.engine.steps.UnionStep;
import nz.org.riskscape.engine.types.Nullable;
import nz.org.riskscape.engine.types.Struct;
import nz.org.riskscape.engine.types.Types;
import nz.org.riskscape.rl.ExpressionParser;
import nz.org.riskscape.rl.ast.Expression;
import nz.org.riskscape.rl.ast.ExpressionProblems;
import nz.org.riskscape.rl.ast.PipelineDeclaration;
import nz.org.riskscape.rl.ast.PropertyAccess;

public class DefaultPipelineRealizerTest {

  DefaultEngine engine = new DefaultEngine();
  DefaultProject project = (DefaultProject) engine.emptyProject();
  ExecutionContext executionContext = project.newExecutionContext();

  ParameterSet inputStepParameters = ParameterSet.EMPTY;

  Step dummyInputStep = new Step() {

    @Override
    public String getId() {
      return "input";
    }

    @Override
    public RiskscapeMessage getDescription() {
      return null;
    }

    @Override
    public ParameterSet getParameterSet() {
      return inputStepParameters;
    }

    @Override
    public RealizedPipeline realize(RealizationInput input) {
      return input.getRealizedPipeline().add(
          input.newPrototypeStep().realizedBy(this).withResult(ListRelation.ofValues(1L, 2L, 3L))
      );
    }

    @Override
    public Range<Integer> getInputArity() {
      return Range.singleton(0);
    }

  };

  @Before
  public void setup() {
    engine.getPipelineSteps().add(dummyInputStep);
    engine.getPipelineSteps().add(new SelectStep(engine));
    engine.getPipelineSteps().add(new JoinStep(engine));
    engine.getPipelineSteps().add(new UnionStep(engine));
  }

  DefaultPipelineRealizer realizer = new DefaultPipelineRealizer();
  PipelineDeclaration parsed;
  RealizedPipeline realized;

  @Test
  public void canRealizeASingleStepWithNoParameters() {
    parseAndRealize("input()");

    assertStepImplementation("input", sameInstance(dummyInputStep));
  }

  @Test
  public void badStepNameGivesAnError() throws Exception {
    parseAndRealize("output()");

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(
            is(GeneralProblems.get().noSuchObjectExistsDidYouMean("output", Step.class, List.of("input"))
        ))
    ));
  }

  @Test
  public void canRealizeASingleStepWithAParameter() {
    inputStepParameters = ParameterSet.from(Parameter.required("foo", String.class));

    parseAndRealize("input(foo: 'bar')");

    assertStepParameters("input", hasEntry(equalTo("foo"), contains("bar")));
  }

  @Test
  public void badParameterNameGivesAnError() throws Exception {
    inputStepParameters = ParameterSet.from(Parameter.required("foo", String.class));

    parseAndRealize("input(foz: 'bar')");

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        isError(PipelineBuilder.ProblemCodes.STEP_PARAMETER_UNKNOWN)
    )));
  }

  @Test
  public void canRealizeASingleStepAnAnonymousExpressionParameter() {
    inputStepParameters = ParameterSet.from(Parameter.required("foo", Expression.class));

    parseAndRealize("input(foo: 'bar')");

    assertStepParameters("input", hasEntry(equalTo("foo"), contains(instanceOf(Expression.class))));
  }

  @Test
  public void canRealizeTwoStepsTogether() throws Exception {
    parseAndRealize("input() -> select(*)");

    assertFalse(realized.hasFailures());

    RealizedStep inputStep = realized.getStep("input").get();
    RealizedStep selectStep = realized.getStep("select").get();

    assertThat(realized.getDependents(inputStep), contains(selectStep));
    assertThat(selectStep.getDependencies(), contains(inputStep));
  }

  @Test
  public void canRealizeTwoStepsFanningOut() throws Exception {
    parseAndRealize("input() -> select(*) as foo input -> select(*) as bar");

    assertFalse(realized.hasFailures());

    RealizedStep inputStep = realized.getStep("input").get();
    RealizedStep fooStep = realized.getStep("foo").get();
    RealizedStep barStep = realized.getStep("bar").get();

    assertThat(realized.getDependents(inputStep), containsInAnyOrder(fooStep, barStep));
    assertThat(fooStep.getDependencies(), contains(inputStep));
    assertThat(barStep.getDependencies(), contains(inputStep));
  }

  @Test
  public void canRealizeTwoStepsFanningIn() throws Exception {
    parseAndRealize("""
        join(on: true)
        input() as foo -> join.rhs
        input() -> select({ value  as foo }) as bar -> join.lhs
        """);

    assertFalse(realized.hasFailures());

    RealizedStep joinStep = realized.getStep("join").get();
    RealizedStep fooStep = realized.getStep("foo").get();
    RealizedStep barStep = realized.getStep("bar").get();

    // NB: dependencies are in the order that the named inputs are declared in
    // (not the order that the edges appear in the pipeline)
    assertThat(joinStep.getDependencies(), contains(barStep, fooStep));
  }

  @Test
  public void namedInputsJoinedInCorrectOrder() throws Exception {
    // use a left-outer join so the RHS will get a Nullable type assigned to it
    // The join_type parameter requires the enum binder
    requiresBinder(new EnumBinder());
    // NB: we join the RHS first here, but it should be treated as the 2nd input
    parseAndRealize("""
        input()
        -> select({ 'foo' as RHS }) as rhs
        -> join.rhs
        input()
        -> select({ 'bar' as LHS }) as lhs
        -> join(on: true, join_type: 'LEFT_OUTER')
        """);
    assertFalse(realized.hasFailures());
    RealizedStep joinStep = realized.getStep("join").get();
    // check the RHS is assigned the nullable type
    assertThat(joinStep.getProduces(), is(Struct.of("LHS", Types.TEXT, "RHS", Nullable.TEXT)));

    // repeat but declare the steps in the other order
    parseAndRealize("""
        input()
        -> select({ 'bar' as LHS }) as lhs
        -> join(on: true, join_type: 'LEFT_OUTER')
        input()
        -> select({ 'foo' as RHS }) as rhs
        -> join.rhs
        """);
    assertFalse(realized.hasFailures());
    joinStep = realized.getStep("join").get();
    assertThat(joinStep.getProduces(), is(Struct.of("LHS", Types.TEXT, "RHS", Nullable.TEXT)));
  }

  @Test
  public void getErrorWhenStepNeedsInput() throws Exception {
    parseAndRealize("select(*) as foo ");
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        is(GeneralProblems.get().badArity(Input.class, new SelectStep(engine), Range.singleton(1), 0)))
    ));

    parseAndRealize("input() -> join(on: true) ");
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        is(JoinStep.PROBLEMS.noRightHandSide())
    )));
  }

  @Test
  public void canHaveMultipleUnnamedInputs() throws Exception {
    parseAndRealize("""
        input() as foo -> union()
        input() as bar -> union
        input() as baz -> union
        """);

    assertFalse(realized.hasFailures());

    RealizedStep unionStep = realized.getStep("union").get();
    RealizedStep fooStep = realized.getStep("foo").get();
    RealizedStep barStep = realized.getStep("bar").get();
    RealizedStep bazStep = realized.getStep("baz").get();

    assertThat(unionStep.getDependencies(), containsInAnyOrder(fooStep, barStep, bazStep));
  }

  @Test
  public void getErrorForSingleUndefinedStepReference() throws Exception {
    // step referenced but not defined
    parseAndRealize("foo");
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        isError(Pipeline.ProblemCodes.STEP_NAME_UNKNOWN)
    )));

    // we should only get one error when the problem is at the start of the chain
    parseAndRealize("foo -> select(*) as bar -> select(*) as baz");
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        isError(Pipeline.ProblemCodes.STEP_NAME_UNKNOWN)
    )));
  }

  @Test
  public void getErrorForUnchainedStepReference() throws Exception {
    // step defined and referenced but not chained
    parseAndRealize("input() as foo "
        + "foo");
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(PipelineBuilder.ProblemCodes.UNUSED_STEP_REFERENCE))
    ));
  }

  @Test
  public void cannotAddNamedEdgeToStepWithoutNames() {
    parseAndRealize("input() -> select(*).foo ");

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.NAMED_INPUT_NOT_ALLOWED))));

    parseAndRealize("""
        input()
        select(*) as foo
        input -> foo.bar
        """);

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.NAMED_INPUT_NOT_ALLOWED))));
  }

  @Test
  public void cannotAddNamedEdgeToUnknownInput() {
    parseAndRealize("""
        join(on: true)
        input() -> join.bad
        input() -> select({ value  as foo }) -> join.lhs
        """);

    assertTrue(realized.hasFailures());
    Step joinStep = new JoinStep(engine);
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(
            is(PipelineProblems.get().namedInputUnknown(joinStep, "bad", joinStep.getInputNames()))
        )));
  }

  @Test
  public void inputCollisionsDetected() {
    // chaining 2 different steps to the same target
    parseAndRealize("""
        input() -> select(*)
        input() -> select
        """);

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.INPUT_ALREADY_CHAINED)
    )));

    // duplicate edge, i.e. chaining the same 2 steps together twice
    parseAndRealize("""
        input() -> select(*)
        input -> select
        """);

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.EDGE_ALREADY_EXISTS)
    )));
  }

  @Test
  public void inputCollisionsDetectedNamedStep() {
    // chaining 2 different named steps to the same target
    parseAndRealize("""
        input() as foo
        select(*) as bar
        input() as baz
        foo -> bar
        baz -> bar
        """);

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.INPUT_ALREADY_CHAINED)
    )));

    // duplicate edge, i.e. chaining the same 2 named steps together twice
    parseAndRealize("""
        input() as foo
        select(*) as bar
        foo -> bar
        foo -> bar
        """);

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.EDGE_ALREADY_EXISTS)
    )));
  }

  @Test
  public void inputCollisionDetectedOnDefaultInput() {
    // default input name used in both cases
    parseAndRealize("""
        join(on: true)
        input() -> join
        input() -> select({ value  as foo }) -> join
        """);

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.DEFAULT_INPUT_ALREADY_CHAINED)
    )));

    // default input name used in the 2nd/error case
    parseAndRealize("""
        join(on: true)
        input() -> join.lhs
        input() -> select({ value  as foo }) -> join
        """);

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.DEFAULT_INPUT_ALREADY_CHAINED)
    )));

    // input name explicitly used, but it's the default
    parseAndRealize("""
        join(on: true)
        input() -> join.lhs
        input() -> select({ value  as foo }) -> join.lhs
        """);

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.INPUT_ALREADY_CHAINED))
    ));

    // non-default input name used explicitly
    parseAndRealize("""
        join(on: true)
        input() -> join.rhs
        input() -> select({ value  as foo }) -> join.rhs
        """);

    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.INPUT_ALREADY_CHAINED))
    ));
  }

  @Test
  public void inputCollisionDetectedOnNamedStepsDefaultInput() {
    // basically we repeat inputCollisionDetectedOnDefaultInput() except we use
    // named steps this time.
    // joining 2 steps to the same default named input (implicit in both cases)
    parseAndRealize("""
        input() as foo
        join(on: true) as bar
        input() as baz
        foo -> bar
        baz -> bar
        """);
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.DEFAULT_INPUT_ALREADY_CHAINED)
    )));

    // joining 2 steps to the same default named input (mix of implicit and explicit)
    parseAndRealize("""
        input() as foo
        join(on: true) as bar
        input() as baz
        foo -> bar.lhs
        baz -> bar
        """);
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.DEFAULT_INPUT_ALREADY_CHAINED)
    )));

    // joining 2 steps to the same default named input (explicit case)
    parseAndRealize("""
        input() as foo
        join(on: true) as bar
        input() as baz
        foo -> bar.lhs
        baz -> bar.lhs
        """);
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.INPUT_ALREADY_CHAINED)
    )));

    // explicitly joining 2 steps to the same non-default named input
    parseAndRealize("""
        input() as foo
        join(on: true) as bar
        input() as baz
        foo -> bar.rhs
        baz -> bar.rhs
        """);
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.INPUT_ALREADY_CHAINED))));

    // duplicate named edge this time (same source and target steps repeated)
    parseAndRealize("""
        input() as foo
        join(on: true) as bar
        input() as baz
        foo -> bar.rhs
        foo -> bar.rhs
        """);
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(),
        contains(
        hasAncestorProblem(isError(Pipeline.ProblemCodes.EDGE_ALREADY_EXISTS)
    )));
  }

  @Test
  public void chainingToUnknownStepsIsAProblem() throws Exception {
    parseAndRealize("""
        input() -> select(*) as step2
        input() -> stepTwo
        """);

    assertThat(realized.getFailures(), contains(hasAncestorProblem(
              isError(Pipeline.ProblemCodes.STEP_NAME_UNKNOWN)
        )
    ));
  }

  @Test
  public void chainingToUnknownStepWithNameIsAProblem() throws Exception {
    parseAndRealize("""
        input() -> join(on: true) as step2
        input() -> select({ value  as foo }) -> step2.rhs
        input() -> stepTwo.rhs
        """);

    assertThat(realized.getFailures(), contains(hasAncestorProblem(
              isError(Pipeline.ProblemCodes.STEP_NAME_UNKNOWN)
        )
    ));
  }

  @Test
  public void namedInputOnLeadingStepIsAProblem() throws Exception {
    // first line has the problem (other lines are just so the join()
    // step doesn't throw out other errors)
    parseAndRealize("""
        join(on: true).lhs -> select(*)
        input() -> join.rhs
        input() -> select({value as foo}) -> join.lhs
        """);
    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        isError(PipelineBuilder.ProblemCodes.UNUSED_NAMED_INPUT))
    ));

    // repeat with a named step - the last line has the problem this time
    parseAndRealize("""
        input() -> join(on: true) as foo
        input() -> select({value as foo}) -> foo.rhs
        foo.lhs -> select(*)
        """);
    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        isError(PipelineBuilder.ProblemCodes.UNUSED_NAMED_INPUT))
    ));
  }

  @Test
  public void mustKeepUsingKeyworksOnceUsed() throws Exception {
    inputStepParameters = ParameterSet.from(Parameter.required("foo", String.class),
        Parameter.required("bar", String.class));

    // Cannot go back to anon parameters once a keyword arg is given
    parseAndRealize("input(foo: 'bar', 'baz')");

    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        isError(PipelineBuilder.ProblemCodes.KEYWORD_REQUIRED))
    ));
  }

  @Test
  public void wrongNumberOfStepParametersIsAProblem() throws Exception {
    parseAndRealize("input() -> select(*, 'spurious param')");

    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        is(ArgsProblems.get().wrongNumber(1, 2))
    )));
  }

  @Test
  public void notRealizableParameterExpressionsIsAProblem() throws Exception {
    // Java types (string, int, etc) must always be constant expressions
    inputStepParameters = ParameterSet.from(Parameter.required("foo", String.class));

    parseAndRealize("input(foo: bar)");
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        isProblem(ExpressionProblems.class, (r, p) -> p.constantRequired(r.any()))
    )));

    // but Expression-type parameters can be dynamic expressions, that don't have
    // to realize as a constant
    inputStepParameters = ParameterSet.from(Parameter.required("foo", PropertyAccess.class));

    parseAndRealize("input(foo: bar)");
    assertFalse(realized.hasFailures());
    assertThat(realized.getFailures(), empty());
  }

  @Test
  public void bindingToWrongParameterValueIsAProblem() throws Exception {
    // binding to specific types (e.g. 'foo' where we expect a number) is handled by
    // BaseStep.bindParameters(), but check that it doesn't muck up realization.
    // We can use join() as an example here, but it'll require a binder
    requiresBinder(new NumberBinder());
    executionContext = project.newExecutionContext(); // refresh context to get new binder
    parseAndRealize("""
        input() -> join.lhs
        input() -> join.rhs
        join(on: true, initial-index-size: 'foo')
        """);
    assertTrue(realized.hasFailures());
    assertThat(realized.getFailures(), contains(hasAncestorProblem(
        is(NumberBinder.PROBLEMS.numberFormatException("foo", Integer.class))
    )));
  }

  @Test
  public void canIncrementallyChainAStep() throws Exception {
    inputStepParameters = ParameterSet.from(Parameter.required("foo", Expression.class));
    // add a single step
    parseAndRealize("input(foo: bar)");
    RealizedStep inputStep = realized.getStep("input").get();

    // append a new step to the original
    PipelineDeclaration childPipeline = ExpressionParser.INSTANCE.parsePipeline("input -> select({*})");
    realized = executionContext.getPipelineRealizer()
      .realize(realized, childPipeline);

    assertThat(realized.hasFailures(), is(false));

    // original step is there
    assertThat(realized.getStep("input").get(), sameInstance(inputStep));
    // new one is as well, and ...
    assertThat(
        realized.getStep("select").get(),
        allOf(
            // depends on input
            hasProperty("dependencies", contains(sameInstance(inputStep))),
            // the step's ast points to the new pipeline ...
            hasProperty("ast", sameInstance(childPipeline.getFirst().getLast()))
        )
    );

    // ... but the realized pipeline still points to the ast we parsed originally.
    assertThat(realized.getAst(), sameInstance(parsed));
  }

  @Test
  public void canIncrementallyAddTwoIndependentSteps() throws Exception {
    // this is just a sanity check that there's no requirement that an incrementally added step be chained
    inputStepParameters = ParameterSet.from(Parameter.required("foo", Expression.class));
    // add a single step
    parseAndRealize("input(foo: bar) as input");
    RealizedStep inputStep1 = realized.getStep("input").get();

    // append a new step to the original
    realized = executionContext.getPipelineRealizer()
      .realize(realized, ExpressionParser.INSTANCE.parsePipeline("input(foo: bar)"));

    assertThat(realized.hasFailures(), is(false));

    // original step is there
    assertThat(realized.getStep("input").get(), sameInstance(inputStep1));
    // new one is as well, and is an island
    assertThat(
        realized.getStep("input_1").get(),
        hasProperty("dependencies", empty())
    );
  }

  @Test
  public void failsIfIncrementalPipelineTriesToAddTheSameStep() throws Exception {
    inputStepParameters = ParameterSet.from(Parameter.required("foo", Expression.class));
    // add a single step
    parseAndRealize("input(foo: bar) as input");

    // try and add one with the same name
    realized = executionContext.getPipelineRealizer()
      .realize(realized, ExpressionParser.INSTANCE.parsePipeline("input(foo: bar) as input"));

    // the reported problem is a bit useless - it's more likely the duplicate has come from a separate source location,
    // might need a version of stepRedefinition that accepts uris insteads.  Having said that, the planned exec step is
    // the only thing that'll make use of this (so far) and it takes pains to avoid this situation.  So I think for now
    // it's ok that realization fails with a handled error...
    assertThat(
        realized.getFailures(),
        not(empty())
    );
  }

  private void requiresBinder(ParameterBinder binder) {
    engine.getBinders().add(binder);
    // we need to refresh the context so that the new binder is actually used
    executionContext = project.newExecutionContext();
  }

  private void parseAndRealize(String pipelineSource) {
    parsed = ExpressionParser.INSTANCE.parsePipeline(pipelineSource);
    realized = realizer.realize(executionContext, parsed);
  }

  private void assertStepImplementation(String name, Matcher<Step> stepMatcher) {
    RealizedStep realizedStep = realized.getStep(name).orElse(null);
    assertThat(realizedStep, hasProperty("failed", equalTo(false)));
    assertNotNull("foo", realizedStep);
    assertThat(
      realizedStep.getImplementation(),
      stepMatcher
    );
  }

  @SafeVarargs
  private void assertStepParameters(
      String name,
      Matcher<Map<? extends String, ? extends Iterable<? extends Object>>>... entryMatchers
  ) {
    RealizedStep realizedStep = realized.getStep(name).orElse(null);
    assertThat(realizedStep, hasProperty("failed", equalTo(false)));
    assertNotNull("foo", realizedStep);

    assertThat(realizedStep, hasProperty("boundParameters", allOf(entryMatchers)));
  }
}
