/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.beam.sdk.io.redis;

import java.io.IOException;
import java.net.ServerSocket;
import java.util.ArrayList;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import redis.embedded.RedisServer;

/**
 * Test on the Redis IO.
 */
public class RedisIOTest {

  @Rule public TestPipeline writePipeline = TestPipeline.create();
  @Rule public TestPipeline readPipeline = TestPipeline.create();

  private EmbeddedRedis embeddedRedis;

  @Before
  public void before() throws Exception {
    embeddedRedis = new EmbeddedRedis();
  }

  @After
  public void after() throws Exception {
    embeddedRedis.close();
  }

  @Test
  public void testWriteRead() throws Exception {
    ArrayList<KV<String, String>> data = new ArrayList<>();
    for (int i = 0; i < 100; i++) {
      KV<String, String> kv = KV.of("key " + i, "value " + i);
      data.add(kv);
    }
    PCollection<KV<String, String>> write = writePipeline.apply(Create.of(data));
    write.apply(RedisIO.write().withEndpoint("::1", embeddedRedis.getPort()));

    writePipeline.run();

    PCollection<KV<String, String>> read = readPipeline.apply("Read",
        RedisIO.read().withEndpoint("::1", embeddedRedis.getPort())
            .withKeyPattern("key*"));
    PAssert.that(read).containsInAnyOrder(data);

    PCollection<KV<String,  String>> readNotMatch = readPipeline.apply("ReadNotMatch",
        RedisIO.read().withEndpoint("::1", embeddedRedis.getPort())
            .withKeyPattern("foobar*"));
    PAssert.thatSingleton(readNotMatch.apply(Count.globally())).isEqualTo(0L);

    readPipeline.run();
  }

  @Test
  public void testWriteBuildsCorrectly() {
    RedisIO.Write writeOp = RedisIO.write().withEndpoint("test", 111);
    Assert.assertEquals(writeOp.connectionConfiguration().port(), 111);
    Assert.assertEquals(writeOp.connectionConfiguration().host(), "test");
  }

  /**
   * Simple embedded Redis instance wrapper to control Redis server.
   */
  private static class EmbeddedRedis implements AutoCloseable {

    private final int port;
    private final RedisServer redisServer;

    public EmbeddedRedis() throws IOException {
      try (ServerSocket serverSocket = new ServerSocket(0)) {
        port = serverSocket.getLocalPort();
      }
      redisServer = new RedisServer(port);
      redisServer.start();
    }

    public int getPort() {
      return this.port;
    }

    @Override
    public void close() {
      redisServer.stop();
    }

  }

}

@@ -2717,32 +2717,468 @@
 ine.run();%0A  %7D%0A%0A
+  @Test%0A  public void testReadBuildsCorrectly() %7B%0A    RedisIO.Read read = RedisIO.read().withEndpoint(%22test%22, 111).withAuth(%22pass%22).withTimeout(5);%0A    Assert.assertEquals(%22test%22, read.connectionConfiguration().host());%0A    Assert.assertEquals(111, read.connectionConfiguration().port());%0A    Assert.assertEquals(%22pass%22, read.connectionConfiguration().auth());%0A    Assert.assertEquals(5, read.connectionConfiguration().timeout());%0A  %7D%0A%0A
   @Test%0A  public
@@ -3235,18 +3235,16 @@
 te write
-Op
  = Redis
@@ -3275,24 +3275,56 @@
 %22test%22, 111)
+.withAuth(%22pass%22).withTimeout(5)
 ;%0A    Assert
@@ -3329,39 +3329,45 @@
 rt.assertEquals(
+%22test%22, 
 write
-Op
 .connectionConfi
@@ -3381,19 +3381,84 @@
 n().
+host());%0A    Assert.assertEquals(111, write.connectionConfiguration().
 port()
-, 111
 );%0A 
@@ -3484,15 +3484,21 @@
 als(
+%22pass%22, 
 write
-Op
 .con
@@ -3524,22 +3524,85 @@
 n().
-host(), %22test%22
+auth());%0A    Assert.assertEquals(5, write.connectionConfiguration().timeout()
 );%0A 

Fix error-prone warnings and add extra test for Read construction