package eu.dnetlib.enabling.resultset.push;

import static org.junit.Assert.*; // NOPMD

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;

import com.google.common.collect.Lists;

import eu.dnetlib.miscutils.cache.Cache;

/**
 * subclass this unit test class and provide a concrete transient push resultset dao implementation.
 *
 * @author marko
 *
 */
@Ignore
public abstract class AbstractTransientPushResultSetDaoTest {
	/**
	 * logger.
	 */
	private static final Log log = LogFactory.getLog(AbstractTransientPushResultSetDaoTest.class); // NOPMD by marko on 11/24/08 5:02 PM

	/**
	 * test first element.
	 */
	private static final String TWO = "two";

	/**
	 * test second element.
	 */
	private static final String ONE = "one";

	/**
	 * some key.
	 */
	private static final String RS_ID = "123";

	/**
	 * instance under test.
	 */
	private TransientPushResultSetDao dao;

	/**
	 * prepare the instance under test.
	 *
	 */
	@Before
	public void setUp() {
		setDao(newInstance());
	}

	/**
	 * specific unit tests will override this method and provide a concrete instance to test.
	 *
	 * @return new transient push resultset dao instance
	 */
	protected abstract TransientPushResultSetDao newInstance();

	/**
	 * test add elements.
	 */
	@Test
	public void testAddElements() {
		final List<String> freshGet = getDao().getElements(RS_ID, 1, 0);
		assertNotNull("fresh get", freshGet);
		assertEquals("fresh get", 0, freshGet.size());

		final List<String> list = new ArrayList<String>();
		list.add(ONE);

		getDao().addElements(RS_ID, list);

		final List<String> firstGet = getDao().getElements(RS_ID, 1, 1);
		assertEquals("first get", 1, firstGet.size());
		assertEquals("first get", ONE, firstGet.get(0));

		final List<String> listAppend = new ArrayList<String>();
		listAppend.add(TWO);

		getDao().addElements(RS_ID, listAppend);

		final List<String> secondGet = getDao().getElements(RS_ID, 1, 2);
		assertEquals("second get", 2, secondGet.size());
		assertEquals("second get", ONE, secondGet.get(0));
		assertEquals("second get", TWO, secondGet.get(1));

		final List<String> listBig = new ArrayList<String>();
		for (int i = 0; i < 90; i++)
			listBig.add(Integer.toString(i));

		getDao().addElements(RS_ID, listBig);

		final List<String> thirdGet = getDao().getElements(RS_ID, 3, 90);
		assertEquals("third get", 88, thirdGet.size());
		assertEquals("third get", "0", thirdGet.get(0));
		assertEquals("third get", "1", thirdGet.get(1));
	}

	/**
	 * check get elements.
	 */
	@Test
	public void testGetElements() {
		final List<String> list = new ArrayList<String>();
		list.add(ONE);

		getDao().addElements(RS_ID, list);

		final List<String> res = getDao().getElements(RS_ID, 1, 1);
		assertEquals("check size", 1, res.size());
		assertEquals("check element", ONE, res.get(0));
	}

	/**
	 * introduced for reproducing of #904. with correct size (limited by 100)
	 */
	@Test
	public void testAddElementsDifferentSizeCorrectSize() {
		List<Integer> sizes = Lists.newArrayList(51, 51, 51, 51);

		int total = 0;
		int run = 0;
		String lastName = null;

		List<String> records = new LinkedList<String>();
		for (int size : sizes) {
			for (int i = 0; i < size; i++) {
				lastName = "bla-" + run + "-" + i;
				records.add(lastName);
			}

			dao.addElements(RS_ID, records);
			records.clear();

			total += size;
			run++;
		}

		inspect();

		assertEquals("check size", total, dao.getSize(RS_ID));
		assertEquals("check last", lastName, dao.getElements(RS_ID, total, total).get(0));

		assertEquals("check first page", "bla-0-11", dao.getElements(RS_ID, 12, 12).get(0));
		assertEquals("check last page", "bla-1-11", dao.getElements(RS_ID, 51 + 12, 51 + 12).get(0));

		assertEquals("check range", 10, dao.getElements(RS_ID, 1, 10).size());
		assertEquals("check range", 10, dao.getElements(RS_ID, 51 + 1, 51 + 10).size());
		assertEquals("check range", 51, dao.getElements(RS_ID, 1, 51).size());
		assertEquals("check range", 52, dao.getElements(RS_ID, 1, 52).size());
		assertEquals("check range", 51, dao.getElements(RS_ID, 2, 52).size());

		assertEquals("check range", 100, dao.getElements(RS_ID, 1, 100).size());
		assertEquals("check range", 99, dao.getElements(RS_ID, 2, 100).size());
		assertEquals("check range", 100, dao.getElements(RS_ID, 2, 101).size());

		assertEquals("check range", 200, dao.getElements(RS_ID, 2, 201).size());
	}

	@Test
	public void testAddElementsSplitted() {
		
		List<Integer> sizes = Lists.newArrayList(860,50);

		int total = 0;
		int run = 0;
		String lastName = null;

		List<String> records = new LinkedList<String>();
		for (int size : sizes) {
			for (int i = 0; i < size; i++) {
				lastName = "bla-" + run + "-" + i;
				records.add(lastName);
			}

			dao.addElements(RS_ID, records);
			records.clear();

			total += size;
			run++;
		}
		
		inspect();

		assertEquals("check size", total, dao.getSize(RS_ID));
	}
	
	private void inspect() {
		CacheTransientResultSetDaoImpl dao = (CacheTransientResultSetDaoImpl) this.dao;
		ResultSetDescriptor desc = dao.getResultSetCache().get(RS_ID);
		Cache<String, List<String>> cache = dao.getCache();

		log.debug("----------------");
		log.debug("ranges: " + desc.getRanges());
		log.debug("last range index: " + desc.getLastRange());
		for (int i = 0; i < desc.getRanges(); i++) {
			final List<String> range = cache.get(RS_ID + "-" + i);
			log.debug("range: " + i + " has " + range.size() + " elements");
			log.debug(" " + range);
		}

		log.debug("----------------");
	}

	@Test
	public void testSizeOfEmpty() {
		assertEquals("empty size", 0, dao.getSize(RS_ID));
	}

	public TransientPushResultSetDao getDao() {
		return dao;
	}

	public void setDao(final TransientPushResultSetDao dao) {
		this.dao = dao;
	}

}