package com.yahoo.sketches.sampling;

import static com.yahoo.sketches.sampling.PreambleUtil.RESERVOIR_SIZE_INT;
import static com.yahoo.sketches.sampling.PreambleUtil.RESERVOIR_SIZE_SHORT;
import static com.yahoo.sketches.sampling.PreambleUtil.SER_VER_BYTE;
import static org.testng.Assert.assertEquals;

import org.testng.annotations.Test;

import com.yahoo.memory.Memory;
import com.yahoo.memory.NativeMemory;
import com.yahoo.sketches.ArrayOfLongsSerDe;

/**
 * Most parts of the class are tested in the sketch and union tests. This class handles only the
 * edge cases not easily tested elsewhere.
 */
public class VersionConverterTest {
  @Test
  public void checkReadOnlyMemory() {
    int k = 32768;
    short encK = ReservoirSize.computeSize(k);
    ArrayOfLongsSerDe serDe = new ArrayOfLongsSerDe();

    ReservoirItemsSketch<Long> ris = ReservoirItemsSketch.getInstance(k);
    byte[] sketchBytesOrig = ris.toByteArray(serDe);

    // get a new byte[], manually revert to v1, then reconstruct
    byte[] sketchBytes = ris.toByteArray(serDe);
    Memory sketchMem = new NativeMemory(sketchBytes);

    sketchMem.putByte(SER_VER_BYTE, (byte) 1);
    sketchMem.putInt(RESERVOIR_SIZE_INT, 0); // zero out all 4 bytes
    sketchMem.putShort(RESERVOIR_SIZE_SHORT, encK);

    Memory readOnlyMem = sketchMem.asReadOnlyMemory();
    ReservoirItemsSketch<Long> rebuilt = ReservoirItemsSketch.getInstance(readOnlyMem, serDe);
    byte[] rebuiltBytes = rebuilt.toByteArray(serDe);

    assertEquals(sketchBytesOrig.length, rebuiltBytes.length);
    for (int i = 0; i < sketchBytesOrig.length; ++i) {
      assertEquals(sketchBytesOrig[i], rebuiltBytes[i]);
    }

  }
}

@@ -297,16 +297,65 @@
 tEquals;
+%0Aimport static org.testng.Assert.assertNotEquals;
 %0A%0Aimport
@@ -930,61 +930,8 @@
 (k);
-%0A    byte%5B%5D sketchBytesOrig = ris.toByteArray(serDe);
 %0A%0A  
@@ -1100,168 +1100,32 @@
 s);%0A
-%0A
     
-sketchMem.putByte(SER_VER_BYTE, (byte) 1);%0A    sketchMem.putInt(RESERVOIR_SIZE_INT, 0); // zero out all 4 bytes%0A    sketchMem.putShort(RESERVOIR_SIZE_SHORT
+revertToV1(sketchMem
 , en
@@ -1189,158 +1189,318 @@
 ();%0A
+%0A
     
-ReservoirItemsSketch%3CLong%3E rebuilt = ReservoirItemsSketch.getInstance(readOnlyMem, serDe
+// read-only input should generate a new Memory%0A    Memory convertedSketch = VersionConverter.convertSketch1to2(readOnlyMem);%0A    assertNotEquals(readOnlyMem, convertedSketch
 );%0A
+%0A
     
-byte%5B%5D rebuiltBytes = rebuilt.toByteArray(serDe
+// same process on writable Memory should be in-place%0A    convertedSketch = VersionConverter.convertSketch1to2(sketchMem
 );%0A
-%0A
     
@@ -1522,166 +1522,264 @@
 etch
-BytesOrig.length, rebuiltBytes.length);%0A    for (int i = 0; i %3C sketchBytesOrig.length; ++i) %7B%0A      assertEquals(sketchBytesOrig%5Bi%5D, rebuiltBytes%5Bi%5D
+Mem, convertedSketch);%0A  %7D%0A%0A  private Memory revertToV1(Memory mem, short encodedK) %7B%0A    mem.putByte(SER_VER_BYTE, (byte) 1);%0A    mem.putInt(RESERVOIR_SIZE_INT, 0); // zero out all 4 bytes%0A    mem.putShort(RESERVOIR_SIZE_SHORT, encodedK
 );%0A
+%0A
     
-%7D%0A
+return mem;
 %0A  %7D

Improve unit test to compare Memory with and without read-only input