package org.talend.dataprep.transformation.api.action.metadata.delete;

import static com.google.common.collect.Sets.newHashSet;
import static java.util.Arrays.asList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.talend.dataprep.transformation.api.action.metadata.ActionMetadataTestUtils.getColumn;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import org.junit.Test;
import org.talend.dataprep.api.dataset.ColumnMetadata;
import org.talend.dataprep.api.dataset.DataSetRow;
import org.talend.dataprep.api.dataset.RowMetadata;
import org.talend.dataprep.api.type.Type;
import org.talend.dataprep.transformation.api.action.context.TransformationContext;
import org.talend.dataprep.transformation.api.action.metadata.ActionMetadataTestUtils;

/**
 * Test class for DeleteInvalid action. Creates one consumer, and test it.
 *
 * @see DeleteInvalid
 */
public class DeleteInvalidTest {

    /** The action to test. */
    private DeleteInvalid deleteInvalid;

    private Map<String, String> parameters;

    /**
     * Default constructor.
     */
    public DeleteInvalidTest() throws IOException {
        deleteInvalid = new DeleteInvalid();
        parameters = ActionMetadataTestUtils.parseParameters(deleteInvalid, //
                DeleteInvalidTest.class.getResourceAsStream("deleteInvalidAction.json"));
    }

    @Test
    public void should_delete_because_non_valid() {
        //given
        final Map<String, String> values = new HashMap<>();
        values.put("0001", "David Bowie");
        values.put("0002", "N");
        values.put("0003", "Something");

        final RowMetadata rowMetadata = new RowMetadata();
        rowMetadata.setColumns(asList(ColumnMetadata.Builder.column() //
                .type(Type.STRING) //
                .computedId("0002") //
                .invalidValues(newHashSet("N")) //
                .build()));

        final DataSetRow row = new DataSetRow(values);
        row.setRowMetadata(rowMetadata);

        //when
        deleteInvalid.applyOnColumn( row, new TransformationContext(), parameters, "0002");

        //then
        assertTrue(row.isDeleted());
        assertEquals("David Bowie", row.get("0001"));
    }

    @Test
    public void should_delete_invalid_values_not_in_metadata() {
        // given
        final Map<String, String> values = new HashMap<>();
        values.put("0001", "1");
        values.put("0002", "N"); // invalid value
        values.put("0003", "2");

        final RowMetadata rowMetadata = new RowMetadata();
        rowMetadata.setColumns(asList(ColumnMetadata.Builder.column() //
                .type(Type.INTEGER) //
                .computedId("0002") //
                .invalidValues(Collections.emptySet()) // no registered invalid values
                .build()));

        final DataSetRow row = new DataSetRow(values);
        row.setRowMetadata(rowMetadata);

        // when
        deleteInvalid.applyOnColumn(row, new TransformationContext(), parameters, "0002");

        // then row is deleted...
        assertTrue(row.isDeleted());

        // ... and column metadata invalid values are also updated
        final Set<String> invalidValues = row.getRowMetadata().getById("0002").getQuality().getInvalidValues();
        assertEquals(1, invalidValues.size());
        assertTrue(invalidValues.contains("N"));
    }

    @Test
    public void should_accept_column() {
        for (Type type : Type.values()) {
            assertTrue(deleteInvalid.acceptColumn(getColumn(type)));
        }
    }

}

@@ -2310,32 +2310,103 @@
 0001%22));%0A    %7D%0A%0A
+    /**%0A     * see https://jira.talendforge.org/browse/TDP-457%0A     */%0A
     @Test%0A    pu

update unit test comment