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

import static nz.org.riskscape.engine.Assert.*;
import static nz.org.riskscape.engine.output.ShapeFileNullMapper.*;
import static org.hamcrest.Matchers.*;
import static org.junit.Assert.*;

import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.file.Path;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.geotools.api.data.DataStore;
import org.geotools.api.data.DataStoreFinder;
import org.geotools.api.data.FeatureSource;
import org.geotools.feature.FeatureIterator;
import org.geotools.referencing.CRS;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.locationtech.jts.geom.Coordinate;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.geom.GeometryFactory;
import org.locationtech.jts.geom.Point;
import org.locationtech.jts.io.WKTReader;
import org.locationtech.jts.io.WKTWriter;
import org.geotools.api.feature.simple.SimpleFeature;
import org.geotools.api.feature.simple.SimpleFeatureType;
import org.geotools.api.referencing.crs.CoordinateReferenceSystem;
import org.riskscape.cli.FeatureBackedStructMap;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;

import nz.org.riskscape.engine.CrsHelper;
import nz.org.riskscape.engine.GeoHelper;
import nz.org.riskscape.engine.RiskscapeException;
import nz.org.riskscape.engine.SRIDSet;
import nz.org.riskscape.engine.TemporaryDirectoryTestHelper;
import nz.org.riskscape.engine.Tuple;
import nz.org.riskscape.engine.types.Nullable;
import nz.org.riskscape.engine.types.Struct;
import nz.org.riskscape.engine.types.Struct.StructMember;
import nz.org.riskscape.engine.types.TypeRegistry;
import nz.org.riskscape.engine.types.Types;
import nz.org.riskscape.engine.types.WithinRange;
import nz.org.riskscape.engine.types.WithinSet;
import nz.org.riskscape.engine.typeset.TypeSet;
import nz.org.riskscape.engine.util.Pair;
import nz.org.riskscape.problem.ProblemSink;

@SuppressWarnings("unchecked")
public class ShapefileWriterTest implements CrsHelper, TemporaryDirectoryTestHelper {

  Path tmpDir;
  File tmpfile;
  private CoordinateReferenceSystem crs;
  private Point point;
  private GeometryFactory geometryFactory;
  SRIDSet sridSet = new SRIDSet();
  TypeSet typeSet = new TypeSet(TypeRegistry.withDefaults());


  @Before
  public void setup() throws IOException {
    tmpDir = createTempDirectory("ShapefileWriterTest");
    tmpfile = File.createTempFile(getClass().getSimpleName(), ".shp", tmpDir.toFile());
    crs = nzMapGrid();
    geometryFactory = new GeometryFactory();
    point = geometryFactory.createPoint(new Coordinate(0, 0));
    point.setSRID(sridSet.get(crs));
  }

  @After
  public void teardown() throws Exception {
    remove(tmpDir);
  }

  @Test
  public void willWrite4326LatLong_WithAxisSwap() throws Exception {
    //Input geometries are 4326, Lat/Long
    crs = CRS.decode("EPSG:4326", false);
    point = geometryFactory.createPoint(new Coordinate(-40, 172));
    point.setSRID(sridSet.get(crs));

    Struct struct = Struct.of("value", Types.INTEGER).and("the_geom", Types.GEOMETRY);

    List<Long> values = Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L);
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(integer -> writer.write(new Tuple(struct).set("value", integer).set("the_geom", point)));
    }

    //Written geometries should to 4326 Long/Lat
    CoordinateReferenceSystem outputCrs = CRS.decode("EPSG:4326", true);
    Point outputPoint = geometryFactory.createPoint(new Coordinate(172, -40));

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Long> collected = forEach(fs, sf -> {
      assertEquals(outputPoint, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));
      return (Long) sf.getAttribute(ShapefileWriter.VALUE_ATTRIBUTE);
    });

    assertEquals(values, collected);
    assertTrue(CRS.equalsIgnoreMetadata(outputCrs, fs.getSchema().getCoordinateReferenceSystem()));
  }

  @Test
  public void canWrite4326LongLat() throws Exception {
    //Input geometries are 4326, Long/Lat
    crs = CRS.decode("EPSG:4326", true);
    point = geometryFactory.createPoint(new Coordinate(172, -50));
    point.setSRID(sridSet.get(crs));

    Struct struct = Struct.of("value", Types.INTEGER).and("the_geom", Types.GEOMETRY);

    List<Long> values = Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L);
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(integer -> writer.write(new Tuple(struct).set("value", integer).set("the_geom", point)));
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Long> collected = forEach(fs, sf -> {
      assertEquals(point, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));
      return (Long) sf.getAttribute(ShapefileWriter.VALUE_ATTRIBUTE);
    });

    assertEquals(values, collected);
    assertTrue(CRS.equalsIgnoreMetadata(crs, fs.getSchema().getCoordinateReferenceSystem()));
  }

  @Test
  public void willWrite2193YX_WithAxisSwap() throws Exception {
    //Input geometries are 2193 YX
    crs = CRS.decode("EPSG:2193", false);
    point = geometryFactory.createPoint(new Coordinate(5501000, 1801000));
    point.setSRID(sridSet.get(crs));

    Struct struct = Struct.of("value", Types.INTEGER).and("the_geom", Types.GEOMETRY);

    List<Long> values = Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L);
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(integer -> writer.write(new Tuple(struct).set("value", integer).set("the_geom", point)));
    }

    //Written geometries should to 2193 XY
    CoordinateReferenceSystem outputCrs = CRS.decode("EPSG:2193", true);
    Point outputPoint = geometryFactory.createPoint(new Coordinate(1801000, 5501000));

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Long> collected = forEach(fs, sf -> {
      assertThat((Geometry)sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE),
          GeoHelper.geometryMatch(outputPoint, GeoHelper.METER_TOLERANCE_NEAREST_MM));
      return (Long) sf.getAttribute(ShapefileWriter.VALUE_ATTRIBUTE);
    });

    assertEquals(values, collected);
    assertTrue(CRS.equalsIgnoreMetadata(outputCrs, fs.getSchema().getCoordinateReferenceSystem()));
  }

  @Test
  public void canWrite2193XY() throws Exception {
    //Input geometries are 2193 XY
    crs = CRS.decode("EPSG:2193", true);
    point = geometryFactory.createPoint(new Coordinate(5501000, 1801000));
    point.setSRID(sridSet.get(crs));

    Struct struct = Struct.of("value", Types.INTEGER).and("the_geom", Types.GEOMETRY);

    List<Long> values = Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L);
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(integer -> writer.write(new Tuple(struct).set("value", integer).set("the_geom", point)));
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Long> collected = forEach(fs, sf -> {
      assertThat((Geometry)sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE),
          GeoHelper.geometryMatch(point, GeoHelper.METER_TOLERANCE_NEAREST_MM));
      return (Long) sf.getAttribute(ShapefileWriter.VALUE_ATTRIBUTE);
    });

    assertEquals(values, collected);
    assertTrue(CRS.equalsIgnoreMetadata(crs, fs.getSchema().getCoordinateReferenceSystem()));
  }

  @Test
  public void canWriteGeometriesWithNoCRS() throws Exception {
    point = geometryFactory.createPoint(new Coordinate(20, 30));
    point.setSRID(0);   // 0 is unknown

    Struct struct = Struct.of("value", Types.INTEGER, "the_geom", Nullable.of(Types.POINT));
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      // we are also testing that first tuple can have a null geom
      writer.write(Tuple.ofValues(struct, 0L));
      writer.write(Tuple.ofValues(struct, 10L, point));
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Pair<Geometry, Long>> collected = forEach(fs, sf -> {
      return Pair.of(
          (Geometry)sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE),
          (Long) sf.getAttribute(ShapefileWriter.VALUE_ATTRIBUTE));
    });

    assertThat(collected, contains(
        allOf(
            hasProperty("left", nullValue()),
            hasProperty("right", is(0L))
        ),
        allOf(
            hasProperty("left", is(point)),
            hasProperty("right", is(10L))
        )
    ));
    assertNull(fs.getSchema().getCoordinateReferenceSystem());
  }


  @Test
  public void canWriteOutATableOfIntegersAndPointGeometrys() throws IOException {
    doTableOfIntegersAndPointsTest(Struct.of("value", Types.INTEGER).and("the_geom", Types.GEOMETRY));
  }

  @Test
  public void canWriteOutATableOfIntegersAndPoints() throws IOException {
    // Using the point geometry subtype should result in same shp file as using geometry type
    doTableOfIntegersAndPointsTest(Struct.of("value", Types.INTEGER).and("the_geom", Types.POINT));
  }

  private void doTableOfIntegersAndPointsTest(Struct struct) throws IOException {

    List<Long> values = Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L);
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(integer -> writer.write(new Tuple(struct).set("value", integer).set("the_geom", point)));
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Long> collected = forEach(fs, sf -> {
      assertEquals(point, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));
      return (Long) sf.getAttribute(ShapefileWriter.VALUE_ATTRIBUTE);
    });

    assertEquals(values, collected);
    assertEquals(crs, fs.getSchema().getCoordinateReferenceSystem());
  }

  @Test
  public void canWriteOutATableOfNullableIntegersAndPoints() throws IOException {

    Struct struct = Struct.of("value", Nullable.INTEGER).and("the_geom", Types.GEOMETRY).build();

    List<Long> values = Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, null);
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(integer -> writer.write(new Tuple(struct).set("value", integer).set("the_geom", point)));
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Long> collected = forEach(fs, sf -> {
      assertEquals(point, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));
      return (Long) sf.getAttribute(ShapefileWriter.VALUE_ATTRIBUTE);
    });

    assertEquals(Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, -9999L), collected);
    assertEquals(crs, fs.getSchema().getCoordinateReferenceSystem());
  }

  @Test
  public void mapsNullValues() throws Exception {
    Struct struct = Struct.of("the_geom", Types.GEOMETRY)
        .and("text", Nullable.TEXT)
        .and("int", Nullable.INTEGER)
        .and("geom2", Nullable.GEOMETRY)
        .and("decimal", Nullable.DECIMAL)
        .and("floating", Nullable.FLOATING)
        .and("boolean", Nullable.BOOLEAN)
        .and("date", Nullable.DATE);


    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      writer.write(Tuple.ofValues(struct, point));
    }

    Struct expectedType = Struct.of("the_geom", Types.GEOMETRY,
        "text", Types.TEXT,
        "int", Types.INTEGER,
        "geom2", Types.GEOMETRY)
        .and("decimal", Types.DECIMAL)
        .and("floating", Types.FLOATING)
        .and("boolean", Types.BOOLEAN)
        .and("date", Types.DATE);

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();

    List<Tuple> collected = forEach(fs, sf -> {
      assertEquals(point, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));

      FeatureBackedStructMap structMap = new FeatureBackedStructMap(sf, expectedType.getMembers().stream()
          .map(StructMember::getKey).collect(Collectors.toSet()));

      return (Tuple) expectedType.coerce(structMap);
    });
    Tuple actual = collected.get(0);


    WKTReader reader = new WKTReader();
    assertEquals(point, actual.fetch("the_geom"));
    assertEquals("", actual.fetch("text"));
    assertEquals(Long.valueOf(-9999L), actual.fetch("int"));
    assertEquals(reader.read("POINT EMPTY"), actual.fetch("geom2"));
    assertEquals(new BigDecimal("-9999"), actual.fetch("decimal"));
    assertEquals(Double.valueOf(-9999D), actual.fetch("floating"));
    assertEquals(false, actual.fetch("boolean"));

    // shapefile only stores the date part, sigh
    LocalDate expectedLocalDate = LocalDate.of(1970, 1, 1);
    LocalDate actualLocalDate = ((Date) actual.fetch("date")).toInstant().atZone(ZoneId.systemDefault()).toLocalDate();

    assertEquals(expectedLocalDate, actualLocalDate);
  }

  @Test
  public void willRenameTheFirstGeometryFieldItFindsToTheGeom() throws IOException {

    Struct struct = Struct.of("value", Types.INTEGER).and("geometry", Types.GEOMETRY).build();

    List<Long> values = Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L);
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(integer -> writer.write(new Tuple(struct).set("value", integer).set("geometry", point)));
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Long> collected = forEach(fs, sf -> {
      assertEquals(point, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));
      return (Long) sf.getAttribute(ShapefileWriter.VALUE_ATTRIBUTE);
    });

    assertEquals(values, collected);
    assertEquals(crs, fs.getSchema().getCoordinateReferenceSystem());
  }

  @SuppressWarnings("resource")
  @Test
  public void willComplainIfNoGeometryFoundInTheStruct() {

    Struct struct = Struct.of("value", Types.INTEGER).and("the_geom", Types.TEXT).build();

    RiskscapeException ex = assertThrows(RiskscapeException.class,
        () ->new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL).write(new Tuple(struct)));
    assertThat(ex.getProblem().getArguments()[0],
        is("output type does not include geometry - output file requires geometry"));
  }

  @Test
  public void canWriteOutATableOfStructs() throws IOException {

    WithinSet simpleSet = new WithinSet(Types.TEXT, "foo", "bar", "baz");
    WithinRange ranged = new WithinRange(Types.INTEGER, 0, 100);

    Struct myStruct = Struct.of("int", Types.INTEGER)
        .and("string", Types.TEXT)
        .and("set", simpleSet)
        .and("ranged", ranged)
        .and("the_geom", Types.GEOMETRY)
        .build();

    List<Tuple> values = Arrays.asList(
        new Tuple(myStruct).set("int", 3L).set("string", "cool").set("set", "bar").set("ranged", 66L)
          .set("the_geom", point),
        new Tuple(myStruct).set("int", 63L).set("string", "story").set("set", "baz").set("ranged", 33L)
          .set("the_geom", point)
    );

    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(value-> writer.write(value));
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Map<String, Object>> collected = forEach(fs, sf -> {
      assertEquals(point, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));

      FeatureBackedStructMap structMap = new FeatureBackedStructMap(sf, myStruct.getMembers().stream()
          .map(StructMember::getKey).collect(Collectors.toSet()));

      return ImmutableMap.copyOf(structMap);
    });

    assertEquals(Lists.transform(values, Tuple::toMap), collected);
  }

  @Test
  public void canWriteOutATableOfNestedStructs() throws IOException {
    Struct nestedStruct = Struct.of("name", Types.TEXT).and("desc", Types.TEXT).build();
    Struct myStruct = Struct.of("the_geom", Types.GEOMETRY)
        .and("a", nestedStruct)
        .and("b", nestedStruct)
        .build();

    List<Tuple> values = Arrays.asList(
        new Tuple(myStruct).set("the_geom", point)
            .set("a", new Tuple(nestedStruct).set("name", "1a-name").set("desc", "1a-desc"))
            .set("b", new Tuple(nestedStruct).set("name", "1b-name").set("desc", "1b-desc")),
        new Tuple(myStruct).set("the_geom", point)
            .set("a", new Tuple(nestedStruct).set("name", "2a-name").set("desc", "2a-desc"))
            .set("b", new Tuple(nestedStruct).set("name", "2b-name").set("desc", "2b-desc"))
    );

    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(value-> writer.write(value));
    }

    Struct expectedStruct = Struct.of("the_geom", Types.GEOMETRY)
        .and("a.name", Types.TEXT).and("a.desc", Types.TEXT)
        .and("b.name", Types.TEXT).and("b.desc", Types.TEXT)
        .build();

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Map<String, Object>> collected = forEach(fs, sf -> {
      assertEquals(point, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));

      FeatureBackedStructMap structMap = new FeatureBackedStructMap(sf, expectedStruct.getMembers().stream()
          .map(StructMember::getKey).collect(Collectors.toSet()));

      return ImmutableMap.copyOf(structMap);
    });

    List<Map<String, Object>> expected = Arrays.asList(
        new Tuple(expectedStruct).set("the_geom", point)
            .set("a.name", "1a-name").set("a.desc", "1a-desc")
            .set("b.name", "1b-name").set("b.desc", "1b-desc").toMap(),
        new Tuple(expectedStruct).set("the_geom", point)
            .set("a.name", "2a-name").set("a.desc", "2a-desc")
            .set("b.name", "2b-name").set("b.desc", "2b-desc").toMap()
    );

    assertEquals(expected, collected);
  }

  @Test
  public void canWriteOutATableOfNestedNullableStructs() throws IOException {
    Struct nestedStruct = Struct.of("name", Types.TEXT).and("desc", Types.TEXT);
    Struct myStruct = Struct.of("the_geom", Types.GEOMETRY)
        .and("a", Nullable.of(nestedStruct))
        .and("b", nestedStruct);

    List<Tuple> values = Arrays.asList(
        new Tuple(myStruct).set("the_geom", point)
            .set("a", new Tuple(nestedStruct).set("name", "1a-name").set("desc", "1a-desc"))
            .set("b", new Tuple(nestedStruct).set("name", "1b-name").set("desc", "1b-desc")),
        new Tuple(myStruct).set("the_geom", point)
            .set("a", null)
            .set("b", new Tuple(nestedStruct).set("name", "2b-name").set("desc", "2b-desc"))
    );

    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      values.forEach(value-> writer.write(value));
    }

    Struct expectedStruct = Struct.of("the_geom", Types.GEOMETRY)
        .and("a.name", Nullable.TEXT).and("a.desc", Nullable.TEXT)
        .and("b.name", Types.TEXT).and("b.desc", Types.TEXT);

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Map<String, Object>> collected = forEach(fs, sf -> {
      assertEquals(point, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));

      FeatureBackedStructMap structMap = new FeatureBackedStructMap(sf, expectedStruct.getMembers().stream()
          .map(StructMember::getKey).collect(Collectors.toSet()));

      return ImmutableMap.copyOf(structMap);
    });
    List<Tuple> expected = Arrays.asList(
        Tuple.ofValues(expectedStruct).set("the_geom", point)
            .set("a.name", "1a-name").set("a.desc", "1a-desc")
            .set("b.name", "1b-name").set("b.desc", "1b-desc"),
        new Tuple(expectedStruct).set("the_geom", point)
            .set("a.name", NULL_VALUE_TEXT).set("a.desc", NULL_VALUE_TEXT)
            .set("b.name", "2b-name").set("b.desc", "2b-desc")
    );

    assertEquals(Lists.transform(expected, Tuple::toMap), collected);
  }

  @Test
  public void willWriteOutExtraGeometryIncludedInTypeAsWKTIfNotSameAsGeometryArgument() throws IOException {
    // the_geom is a magic field name for shapefiles with geotools
    Struct struct = Struct.of("the_geom", Types.GEOMETRY)
        .and("string", Types.TEXT)
        .and("extra_poly", Types.GEOMETRY);
    doWriteExtraGeometryAsWktTest(struct);
  }

  @Test
  public void willWriteOutExtraGeometryIncludedInTypeAsWKTWhenExtraIsGeometrySubType() throws IOException {
    // the_geom is a magic field name for shapefiles with geotools
    Struct struct = Struct.of("the_geom", Types.GEOMETRY)
        .and("string", Types.TEXT)
        .and("extra_poly", Types.POLYGON);
    doWriteExtraGeometryAsWktTest(struct);
  }

  private void doWriteExtraGeometryAsWktTest(Struct struct)  throws IOException {

    Geometry poly = geometryFactory.createPoint(new Coordinate(10, 10));
    WKTWriter wktWriter = new WKTWriter();
    String polyAsWkt = wktWriter.write(poly);

    Tuple value = new Tuple(struct).set("string", "radical").set("the_geom", point).set("extra_poly", poly);
    Map<String, Object> expectedValue =
        ImmutableMap.of("string", "radical", "the_geom", point, "extra_poly", polyAsWkt);

    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      writer.write(value);
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<Map<String, Object>> collected = forEach(fs, sf -> {
      assertEquals(point, sf.getAttribute(ShapefileWriter.GEOMETRY_ATTRIBUTE));
      FeatureBackedStructMap structMap = new FeatureBackedStructMap(sf, struct.getMembers().stream()
          .map(StructMember::getKey).collect(Collectors.toSet()));

      return ImmutableMap.copyOf(structMap);
    });

    assertEquals(expectedValue, collected.get(0));
  }

   @Test(expected=IllegalArgumentException.class)
  public void throwsAnIllegalArgumentExceptionIfGivenAValueThatDoesNotMatchType() throws IOException {
    Struct typeOne = Struct.of("foo", Types.TEXT).and("the_geom", Types.GEOMETRY).build();
    Struct typeTwo = Struct.of("foo", Types.INTEGER).and("the_geom", Types.GEOMETRY).build();

    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      writer.write(new Tuple(typeOne).set("foo", "1").set("the_geom", point));
      writer.write(new Tuple(typeTwo).set("foo", 1).set("the_geom", point));
    }
  }

  @Test
  public void willTruncateAttributeNamesThatAreTooLong() throws IOException {
    Struct type = Struct
        .of("this-name-is-very-long-do-you-know-that", Types.TEXT)
        .and("the_geom", Types.GEOMETRY)
        .build();

    List<String> values = Arrays.asList("1", "2");
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      writer.write(new Tuple(type).set("this-name-is-very-long-do-you-know-that", "1").set("the_geom", point));
      writer.write(new Tuple(type).set("this-name-is-very-long-do-you-know-that", "2").set("the_geom", point));
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<String> collected = forEach(fs, sf -> {
      return (String) sf.getAttribute("this-name-");
    });

    assertEquals(values, collected);
  }

  @Test
  public void willDistinguishBetweenTruncatedNames() throws IOException {
    Struct type = Struct
        .of("this-name-is-very-long-do-you-know-that", Types.TEXT)
        .and("this-name-is-very-long-do-I-know-that", Types.TEXT)
        .and("the_geom", Types.GEOMETRY)
        .build();

    List<String> values = Arrays.asList("foobar");
    try (ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL)) {
      writer.write(new Tuple(type)
          .set("this-name-is-very-long-do-you-know-that", "foo")
          .set("this-name-is-very-long-do-I-know-that", "bar")
          .set("the_geom", point));
    }

    FeatureSource<SimpleFeatureType, SimpleFeature> fs = openWritten();
    List<String> collected = forEach(fs, sf -> {
      return (String) sf.getAttribute("this-name0") + sf.getAttribute("this-name1");
    });

    assertEquals(values, collected);
  }

  @Test
  public void writingNoFeaturesResultsInEmptyFile() throws IOException {
    ShapefileWriter writer = new ShapefileWriter(tmpfile, sridSet, ProblemSink.DEVNULL);
    writer.close();
    assertTrue(tmpfile.exists());
    assertEquals(0, tmpfile.length());
  }

  private <T extends Object> List<T> forEach(FeatureSource<SimpleFeatureType, SimpleFeature> fs,
      Function<SimpleFeature, T> feature) {
    try {
      List<T> collected = new ArrayList<>();
      FeatureIterator<SimpleFeature> features = fs.getFeatures().features();
      while (features.hasNext()) {
        collected.add(feature.apply(features.next()));
      }
      features.close();
      return collected;
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  private FeatureSource<SimpleFeatureType, SimpleFeature> openWritten() {
    return openShapefile(tmpfile);
  }

  private FeatureSource<SimpleFeatureType, SimpleFeature> openShapefile(File shapefile) {

    try {
      DataStore dataStore = DataStoreFinder.getDataStore(ImmutableMap.of("url", shapefile.toURI().toURL()));

      if (dataStore == null) {
        throw new RuntimeException(String.format("Could not read features from %s url - is it in a supported format?",
            shapefile));
      }
      String typeName = dataStore.getTypeNames()[0];

      return dataStore.getFeatureSource(typeName);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

}
