/**
 * 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
 *
 *     https://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.
 */

#include "boost/filesystem.hpp"
#include "Stream.hh"
#include "Exception.hh"
#include <boost/test/included/unit_test_framework.hpp>
#include <boost/test/parameterized_test.hpp>

namespace avro {
namespace stream {

struct CheckEmpty1 {
    void operator()(InputStream& is) {
        const uint8_t* d;
        size_t n;
        BOOST_CHECK(! is.next(&d, &n));
    }
};

struct CheckEmpty2 {
    void operator()(InputStream& is) {
        StreamReader r;
        r.reset(is);
        BOOST_CHECK_THROW(r.read(), Exception);
    }
};

struct TestData {
    size_t chunkSize;
    size_t dataSize;
};

struct Fill1 {
    void operator()(OutputStream& os, size_t len) {
        StreamWriter w;
        w.reset(os);
        for (size_t i = 0; i < len; ++i) {
            w.write(i % 10 + '0');
        }
        w.flush();
    }
};

struct Fill2 {
    void operator()(OutputStream& os, size_t len) {
        for (size_t i = 0; i < len;) {
            uint8_t *b;
            size_t n;
            os.next(&b, &n);
            size_t j = 0;
            for (; i < len && j < n; ++j, ++i, ++b) {
                *b = i % 10 + '0';
            }
            if (i == len) {
                os.backup(n - j);
            }
        }
        os.flush();
    }
};

struct Verify1 {
    void operator()(InputStream& is, size_t dataSize) {
        StreamReader r;
        r.reset(is);
        for (size_t i = 0; i < dataSize; ++i) {
            BOOST_CHECK_EQUAL(i % 10 + '0', r.read());
        }
        BOOST_CHECK_THROW(r.read(), Exception);
    }
};

struct Verify2 {
    void operator()(InputStream& is, size_t len) {
        const uint8_t *b;
        size_t n;

        for (size_t i = 0; i < len;) {
            BOOST_REQUIRE(is.next(&b, &n));
            size_t j = 0;
            for (; i < len && j < n; ++j, ++i, ++b) {
                BOOST_CHECK_EQUAL(*b, i % 10 + '0');
            }
            BOOST_CHECK_EQUAL(j, n);
        }
        BOOST_CHECK(! is.next(&b, &n));
    }
};

template <typename V>
void testEmpty_memoryStream() {
    std::unique_ptr<OutputStream> os = memoryOutputStream();
    std::unique_ptr<InputStream> is = memoryInputStream(*os);
    V()(*is);
}

template <typename F, typename V>
void testNonEmpty_memoryStream(const TestData& td)
{
    std::unique_ptr<OutputStream> os = memoryOutputStream(td.chunkSize);
    F()(*os, td.dataSize);

    std::unique_ptr<InputStream> is = memoryInputStream(*os);
    V()(*is, td.dataSize);
}

void testNonEmpty2(const TestData& td) {
    std::vector<uint8_t> v;
    for (size_t i = 0; i < td.dataSize; ++i) {
        v.push_back(i % 10 + '0');
    }

    uint8_t v2 = 0;
    std::unique_ptr<InputStream> is = memoryInputStream(v.empty() ? &v2 : &v[0], v.size());
    Verify1()(*is, td.dataSize);
}

static const char filename[] = "test_str.bin";

struct FileRemover {
    const boost::filesystem::path file;
    FileRemover(const char* filename) : file(filename) { }
    ~FileRemover() { boost::filesystem::remove(file); }
};

template <typename V>
void testEmpty_fileStream() {
    FileRemover fr(filename);
    {
        std::unique_ptr<OutputStream> os = fileOutputStream(filename);
    }
    std::unique_ptr<InputStream> is = fileInputStream(filename);
    V()(*is);
}

template <typename F, typename V>
void testNonEmpty_fileStream(const TestData& td)
{
    FileRemover fr(filename);
    {
        std::unique_ptr<OutputStream> os = fileOutputStream(filename,
            td.chunkSize);
        F()(*os, td.dataSize);
    }

    std::unique_ptr<InputStream> is = fileInputStream(filename, td.chunkSize);
    V()(*is, td.dataSize);
}

TestData data[] = {
    { 100, 0 },
    { 100, 1 },
    { 100, 10 },
    { 100, 100 },
    { 100, 101 },
    { 100, 1000 },
    { 100, 1024 }
};

}   // namespace stream

}   // namespace

boost::unit_test::test_suite*
init_unit_test_suite( int argc, char* argv[] )
{
    boost::unit_test::test_suite* ts =
        BOOST_TEST_SUITE("Avro C++ unit test suite for streams");

    ts->add(BOOST_TEST_CASE(
        &avro::stream::testEmpty_memoryStream<avro::stream::CheckEmpty1>));
    ts->add(BOOST_TEST_CASE(
        &avro::stream::testEmpty_memoryStream<avro::stream::CheckEmpty2>));

    ts->add(BOOST_PARAM_TEST_CASE(
        (&avro::stream::testNonEmpty_memoryStream<avro::stream::Fill1,
            avro::stream::Verify1>),
        avro::stream::data,
        avro::stream::data +
        sizeof(avro::stream::data) / sizeof(avro::stream::data[0])));
    ts->add(BOOST_PARAM_TEST_CASE(
        (&avro::stream::testNonEmpty_memoryStream<avro::stream::Fill2,
            avro::stream::Verify1>),
        avro::stream::data,
        avro::stream::data +
        sizeof(avro::stream::data) / sizeof(avro::stream::data[0])));
    ts->add(BOOST_PARAM_TEST_CASE(
        (&avro::stream::testNonEmpty_memoryStream<avro::stream::Fill2,
            avro::stream::Verify2>),
        avro::stream::data,
        avro::stream::data +
        sizeof(avro::stream::data) / sizeof(avro::stream::data[0])));

    ts->add(BOOST_PARAM_TEST_CASE(&avro::stream::testNonEmpty2,
        avro::stream::data,
        avro::stream::data +
        sizeof(avro::stream::data) / sizeof(avro::stream::data[0])));

    ts->add(BOOST_TEST_CASE(
        &avro::stream::testEmpty_fileStream<avro::stream::CheckEmpty1>));
    ts->add(BOOST_TEST_CASE(
        &avro::stream::testEmpty_fileStream<avro::stream::CheckEmpty2>));

    ts->add(BOOST_PARAM_TEST_CASE(
        (&avro::stream::testNonEmpty_fileStream<avro::stream::Fill1,
            avro::stream::Verify1>),
        avro::stream::data,
        avro::stream::data +
        sizeof(avro::stream::data) / sizeof(avro::stream::data[0])));
    ts->add(BOOST_PARAM_TEST_CASE(
        (&avro::stream::testNonEmpty_fileStream<avro::stream::Fill2,
            avro::stream::Verify1>),
        avro::stream::data,
        avro::stream::data +
        sizeof(avro::stream::data) / sizeof(avro::stream::data[0])));
    ts->add(BOOST_PARAM_TEST_CASE(
        (&avro::stream::testNonEmpty_fileStream<avro::stream::Fill2,
            avro::stream::Verify2>),
        avro::stream::data,
        avro::stream::data +
        sizeof(avro::stream::data) / sizeof(avro::stream::data[0])));
    return ts;
}