/*
 * 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.steps;

import static nz.org.riskscape.engine.Matchers.*;

import static org.hamcrest.Matchers.*;
import static org.junit.Assert.*;

import java.util.Arrays;
import java.util.List;

import org.junit.Test;

import com.google.common.collect.Lists;

import nz.org.riskscape.dsl.ProblemCodes;
import nz.org.riskscape.engine.Tuple;
import nz.org.riskscape.engine.sort.SortBy.Direction;
import nz.org.riskscape.engine.steps.SortStep.InMemorySortCollector;
import nz.org.riskscape.engine.steps.SortStep.LocalProblems;
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.problem.ResultOrProblems;
import nz.org.riskscape.rl.ast.ExpressionProblems;

@SuppressWarnings("unchecked")
public class SortStepTest extends BaseStepTest<SortStep> {

  @Override
  protected SortStep createStep() {
    return new SortStep(engine);
  }

  @Test
  public void canSortByAnInteger() throws Exception {
    // inputType includes a delta attribute to show that it doesn't get in the way when a delta
    // is not being calculated.
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING, "delta", Types.TEXT);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, 10L, 5L, "foo"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 1L, 5L, "bar"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 5L, "baz"));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(inputType, 1L, 5L, "bar"),
        Tuple.ofValues(inputType, 5L, 5L, "baz"),
        Tuple.ofValues(inputType, 10L, 5L, "foo")
    ));
  }

  @Test
  public void canSortByExpression() throws Exception {
    Struct inputType = Struct.of("foo", Types.TEXT);
    addStubInput(inputType);
    addParam("by", toExpression("str_length(foo)"));

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, "five"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, "three"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, "two"));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(inputType, "two"),
        Tuple.ofValues(inputType, "five"),
        Tuple.ofValues(inputType, "three")
    ));
  }

  @Test
  public void canSortByTwoIntegers() throws Exception {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING);
    addStubInput(inputType);
    addParam("by", toExpression("[v1, v2]"));
    addParam("direction", Direction.ASC);
    addParam("direction", Direction.DESC);

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, 10L, 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 3L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 8L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 10L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 1L, 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 5L));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(inputType, 1L, 5L),
        Tuple.ofValues(inputType, 5L, 10L),
        Tuple.ofValues(inputType, 5L, 8L),
        Tuple.ofValues(inputType, 5L, 5L),
        Tuple.ofValues(inputType, 5L, 3L),
        Tuple.ofValues(inputType, 10L, 5L)
    ));
  }

  @Test
  public void canSortByTwoIntegersAndOnlySpecifyOneDirection() throws Exception {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING);

    addStubInput(inputType);
    addParam("by", toExpression("[v1, v2]"));
    addParam("direction", Direction.DESC);
    //addParam("direction", Direction.DESC);  // the second sort will default to asc

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, 10L, 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 3L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 8L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 10L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 1L, 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 5L));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(inputType, 10L, 5L),
        Tuple.ofValues(inputType, 5L, 3L),
        Tuple.ofValues(inputType, 5L, 5L),
        Tuple.ofValues(inputType, 5L, 8L),
        Tuple.ofValues(inputType, 5L, 10L),
        Tuple.ofValues(inputType, 1L, 5L)
    ));
  }

  @Test
  public void canSortByTwoStrings() throws Exception {
    Struct inputType = Struct.of("make", Types.TEXT, "model", Types.TEXT);

    addStubInput(inputType);
    addParam("by", toExpression("[make, model]"));
    addParam("direction", Direction.ASC);
    addParam("direction", Direction.DESC);

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, "VW", "Golf RS"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, "VW", "Polo"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, "VW", "Golf"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, "Subaru", "Legacy"));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(inputType, "Subaru", "Legacy"),
        Tuple.ofValues(inputType, "VW", "Polo"),
        Tuple.ofValues(inputType, "VW", "Golf RS"),
        Tuple.ofValues(inputType, "VW", "Golf")
    ));
  }

  @Test
  public void canSortByAText() throws Exception {
    Struct inputType = Struct.of("v1", Types.TEXT, "v2", Types.FLOATING);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, "dog", 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, "bird", 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, "cat", 5L));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(inputType, "bird", 5L),
        Tuple.ofValues(inputType, "cat", 5L),
        Tuple.ofValues(inputType, "dog", 5L)
    ));
  }

  @Test
  public void canCombineTwoAccumulators() throws Exception {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();

    List<Tuple> acc1 = sorter.newAccumulator();
    sorter.accumulate(acc1, Tuple.ofValues(inputType, 10L, 5L));
    sorter.accumulate(acc1, Tuple.ofValues(inputType, 1L, 5L));
    sorter.accumulate(acc1, Tuple.ofValues(inputType, 5L, 5L));

    List<Tuple> acc2 = sorter.newAccumulator();
    sorter.accumulate(acc2, Tuple.ofValues(inputType, 10L, 5L));
    sorter.accumulate(acc2, Tuple.ofValues(inputType, 1L, 5L));
    sorter.accumulate(acc2, Tuple.ofValues(inputType, 5L, 5L));

    List<Tuple> combined = sorter.combine(acc1, acc2);
    assertThat(sorter.process(combined), iteratorWithTuples(
        Tuple.ofValues(inputType, 1L, 5L),
        Tuple.ofValues(inputType, 1L, 5L),
        Tuple.ofValues(inputType, 5L, 5L),
        Tuple.ofValues(inputType, 5L, 5L),
        Tuple.ofValues(inputType, 10L, 5L),
        Tuple.ofValues(inputType, 10L, 5L)
    ));
  }

  @Test
  public void canSortByAnIntegerAndProduceDelta() throws Exception {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));
    addParam("delta", expressionParser.parse("(prev, current) -> current.v1 - prev.v1"));

    Struct expected = inputType.and("delta", Nullable.INTEGER);

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();
    assertThat(sorter.getTargetType(), is(expected));

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, 10L, 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 1L, 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 5L));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(expected, 1L, 5L, null),
        Tuple.ofValues(expected, 5L, 5L, 4L),
        Tuple.ofValues(expected, 10L, 5L, 5L)
    ));
  }

  @Test
  public void canSortByAnIntegerAndProduceDeltaAsName() throws Exception {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING, "delta", Types.TEXT);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));
    addParam("delta", expressionParser.parse("(prev, current) -> current.v1 - prev.v1"));
    addParam("delta-attribute", "myDelta");

    Struct expected = inputType.and("myDelta", Nullable.INTEGER);

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();
    assertThat(sorter.getTargetType(), is(expected));

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, 10L, 5L, "foo"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 1L, 5L, "bar"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 5L, "baz"));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(expected, 1L, 5L, "bar"),
        Tuple.ofValues(expected, 5L, 5L, "baz", 4L),
        Tuple.ofValues(expected, 10L, 5L, "foo", 5L)
    ));
  }

  @Test
  public void canCascadeDelta() throws Exception {
    Struct inputType = Struct.of(
        "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    // Probably the most simple cascading hazard - keep a running total of the values
    String expression = "(prev, current) -> if_null(prev.delta, 0) + current.v1";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "integer");

    Struct expected = inputType.and("delta", Types.INTEGER);

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();
    assertThat(sorter.getTargetType(), is(expected));

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, 10L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 1L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(expected, 1L, 1L),
        Tuple.ofValues(expected, 5L, 6L),
        Tuple.ofValues(expected, 10L, 16L)
      ));
  }

  @Test
  public void deltaCanBeWhateverYouWant() throws Exception {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING, "name", Types.TEXT);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    // this delta calculation doesn't really make any sense for a delta. But the point is that the user
    // can make it do whatever they want/need given the two tuples.
    addParam("delta", expressionParser.parse("(prev, current) -> prev.name + current.name"));

    Struct expected = inputType.and("delta", Nullable.TEXT);

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();
    assertThat(sorter.getTargetType(), is(expected));

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, 10L, 5L, "foo"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 1L, 5L, "bar"));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 5L, "baz"));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(expected, 1L, 5L, "bar"),
        Tuple.ofValues(expected, 5L, 5L, "baz", "barbaz"),
        Tuple.ofValues(expected, 10L, 5L, "foo", "bazfoo")
    ));
  }

  @Test
  public void canSortByAnIntegerAndProduceDeltaWithPsuedoIdentity() throws Exception {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));
    addParam("delta", expressionParser.parse(
        "(prev, current) -> if_then_else(is_null(prev), 0 , current.v1 - prev.v1)"));

    // ideally the delta result type would be Types.INTEGER but sadly if_then_else isn't that smart.
    Struct expected = inputType.and("delta", Nullable.INTEGER);

    ResultOrProblems<?> collector = step.realizeSimple(input());
    assertThat(collector.hasProblems(), is(false));
    InMemorySortCollector sorter = (InMemorySortCollector) collector.get();
    assertThat(sorter.getTargetType(), is(expected));

    List<Tuple> acc = sorter.newAccumulator();

    sorter.accumulate(acc, Tuple.ofValues(inputType, 10L, 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 1L, 5L));
    sorter.accumulate(acc, Tuple.ofValues(inputType, 5L, 5L));

    assertThat(sorter.process(acc), iteratorWithTuples(
        Tuple.ofValues(expected, 1L, 5L, 0L),
        Tuple.ofValues(expected, 5L, 5L, 4L),
        Tuple.ofValues(expected, 10L, 5L, 5L)
    ));
  }

  @Test
  public void errorIfDeltaAttributeAlreadyExists() {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING, "delta", Types.TEXT);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));
    addParam("delta", expressionParser.parse("(prev, current) -> current.v1 - prev.v1"));

    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(is(LocalProblems.get().deltaAttributeAlreadyExists("delta", inputType)))
    ));

    addParam("delta-attribute", "v1");
    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(is(LocalProblems.get().deltaAttributeAlreadyExists("v1", inputType)))
    ));
  }

  @Test
  public void errorIfSpecifiedDifferentToActual() throws Exception {
    Struct inputType = Struct.of(
        "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> 'foo'";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "integer");

    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(is(LocalProblems.get().deltaTypeDifferentToActual(Types.INTEGER, Types.TEXT))))
      );
  }

  @Test
  public void errorIfSpecifiedNonNullableButActualNullable() throws Exception {
    Struct inputType = Struct.of(
        "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> prev.delta";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "integer");

    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(
            is(LocalProblems.get().deltaTypeDifferentToActual(Types.INTEGER, Nullable.of(Types.INTEGER))))
      ));
  }

  @Test
  public void worksIfSpecifiedNullableButActuallyNonNullable() throws Exception {
    Struct inputType = Struct.of(
        "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> if_null(prev.delta, 0)";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "nullable(integer)");

    ResultOrProblems<InMemorySortCollector> collector =
        (ResultOrProblems<InMemorySortCollector>) step.realizeSimple(input());

    assertThat(collector, result(any(Object.class)));

    InMemorySortCollector result = collector.get();
    assertThat(result.deltaExpression.get().getResultType(), is(Types.INTEGER));
  }

  @Test
  public void worksIfAnythingSpecified() throws Exception {
    Struct inputType = Struct.of(
      "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> {foo: prev, bar: 'baz'}";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "anything");

    ResultOrProblems<InMemorySortCollector> collector =
        (ResultOrProblems<InMemorySortCollector>) step.realizeSimple(input());

    assertThat(collector, result(any(Object.class)));

    InMemorySortCollector result = collector.get();
    assertThat((Struct)result.deltaExpression.get().getResultType(), any(Struct.class));
  }

  @Test
  public void worksIfStructSpecified() throws Exception {
    Struct inputType = Struct.of(
      "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> { " +
        "foo: if_null(prev.delta.foo, 0) + 1," +
        "bar: str(if_null(prev.delta.foo, 1)) + ', ' " +
        "}";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "struct(foo: integer, bar: text)");

    ResultOrProblems<InMemorySortCollector> collector =
        (ResultOrProblems<InMemorySortCollector>) step.realizeSimple(input());

    assertThat(collector, result(any(Object.class)));

    InMemorySortCollector result = collector.get();
    assertThat(result.deltaExpression.get().getResultType(), is(Struct.of("foo", Types.INTEGER, "bar", Types.TEXT)));
  }

  @Test
  public void worksIfNullableStructSpecified() throws Exception {
    Struct inputType = Struct.of(
      "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> { foo: 2 } * prev.delta";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "nullable(struct(foo: integer))");

    ResultOrProblems<InMemorySortCollector> collector =
        (ResultOrProblems<InMemorySortCollector>) step.realizeSimple(input());

    assertThat(collector, result(any(Object.class)));

    InMemorySortCollector result = collector.get();
    assertThat(result.deltaExpression.get().getResultType(), is(Nullable.of(Struct.of("foo", Types.INTEGER))));
  }

  @Test
  public void errorIfSpecifiedIntergerButActualFloat() throws Exception {
    Struct inputType = Struct.of(
        "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> 5.0";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "integer");
    System.out.println(step.realizeSimple(input()));

    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(is(LocalProblems.get()
            .deltaTypeDifferentToActual(Types.INTEGER, Types.FLOATING))))
      );
  }

  @Test
  public void errorIfSpecifiedFloatButActualInteger() throws Exception {
    Struct inputType = Struct.of(
      "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> 5";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "floating");

    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(is(LocalProblems.get()
            .deltaTypeDifferentToActual(Types.FLOATING, Types.INTEGER))
        )
    ));
  }

  @Test
  public void errorIfDeltaLambdaNoGood() {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));
    addParam("delta", expressionParser.parse("(prev, current) -> current.v1 - prev.v"));

    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(
            is(ExpressionProblems.get().noSuchStructMember("prev.v",
                Lists.newArrayList("prev.v1", "prev.v2", "current.v1", "current.v2"))))
    ));
  }

  @Test
  public void errorIfDeltaLambdaToFewArgs() {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));
    addParam("delta", expressionParser.parse("(prev) -> prev.v"));

    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(
            is(ExpressionProblems.get().lambdaArityError(
                expressionParser.parse("(prev) -> prev.v"),
                1,
                2
            ))
        )
    ));
  }

  @Test
  public void errorIfDeltaLambdaToManyArgs() {
    Struct inputType = Struct.of("v1", Types.INTEGER, "v2", Types.FLOATING);

    addStubInput(inputType);
    addParam("by", toExpression("v1"));
    addParam("delta", expressionParser.parse("(prev, current, next) -> current.v1 - prev.v1"));

    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(
            is(ExpressionProblems.get().lambdaArityError(
                expressionParser.parse("(prev, current, next) -> current.v1 - prev.v1"),
                3,
                2
            ))
        )
    ));
  }

  @Test
  public void errorIfSortFieldNotPresent() throws Exception {
    Struct inputType = Struct.of("v1", Types.TEXT, "v2", Types.FLOATING);

    addStubInput(inputType);
    addParam("by", toExpression("bad"));

    assertThat(step.realizeSimple(input()), failedResult(
        equalTo(ExpressionProblems.get().noSuchStructMember("bad", Arrays.asList("v1", "v2")))
    ));
  }

  @Test
  public void errorIfInvalidDelta() throws Exception {
    Struct inputType = Struct.of(
      "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> prev.delta";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "struct(incomplete: bogus");

    assertThat(step.realizeSimple(input()), failedResult(
      hasAncestorProblem(isError(ProblemCodes.UNEXPECTED_TOKEN)))
    );
  }

  @Test
  public void errorIfTryToAccessCurrDelta() throws Exception {
    Struct inputType = Struct.of(
      "v1", Types.INTEGER
    );

    addStubInput(inputType);
    addParam("by", toExpression("v1"));

    String expression = "(prev, current) -> prev.delta + current.delta";

    addParam("delta", expressionParser.parse(expression));
    addParam("delta-type", "nullable(integer)");

    assertThat(step.realizeSimple(input()), failedResult(
        hasAncestorProblem(
            is(ExpressionProblems.get()
                .noSuchStructMember("current.delta", List.of("prev.v1", "prev.delta", "current.v1")))
        )
      ));
  }

}
