﻿using HiveMapReduce.Exceptions;
using HiveMapReduce.Prototype;
using HiveMapReduce.Streaming;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace HiveMapReduce.UnitTest
{
    [TestClass]
    public class StreamingIOTest
    {
        [TestMethod]
        public void TestStreamingOutputToFileThenStreamingInput()
        {
            var fileName = TestHelpers.GetFileFromWorkingDirectory("TestStreamingOutputToFileThenStreamingInput." + Guid.NewGuid().ToString() + ".txt");

            Row row = new Row(new Schema(@"colInt:int,colNullableInt:int?,
                                            colLong:long,colNullableLong:long?,
                                            colString:string,
                                            colBoolean:bool,colNullableBoolean:bool?,
                                            colDouble:double,colNullableDouble:double?,
                                            colFloat:float,colNullableFloat:float?,
                                            colBinary:binary,colUdt:udt(array<int>)"), new StreamingColumnData(new ColumnSchema("fakeColumn:int")));

            using (var streamingOutput = new StreamingOutput())
            {

                var fsOut = File.OpenWrite(fileName);
                streamingOutput.SetOutputStream(fsOut);

                for (var i = 0; i < 1000; i++)
                {
                    row[0].SetValue(i);
                    row[1].SetValue(i % 2 == 0 ? (int?)null : i);
                    row[2].SetValue(((long)i) << 32);
                    row[3].SetValue(i % 2 == 1 ? (long?)null : ((long)i) << 32);
                    row[4].SetValue(i.ToString() + i.ToString().GetHashCode().ToString("X8"));
                    row[5].SetValue(i % 2 == 0);
                    row[6].SetValue(i % 4 == 0 ? (bool?)null : i % 2 == 0);
                    row[7].SetValue(Math.PI * i);
                    row[8].SetValue(i % 3 == 0 ? (double?)null : Math.PI / i);
                    row[9].SetValue((float)Math.PI * i);
                    row[10].SetValue(i % 3 == 1 ? (float?)null : (float)Math.PI / i);

                    byte[] bytes = i % 7 == 0 ? null : new byte[i];
                    if (bytes != null)
                    {
                        for (var j = 0; j < i; j++)
                        {
                            bytes[j] = (byte)(j % 256);
                        }
                    }

                    row[11].SetValue(bytes);

                    row[12].SetValue(null);

                    streamingOutput.Commit(row);
                }

                streamingOutput.Flush();

            }

            using (var streamingInput = new StreamingInput())
            {
                var fsIn = File.OpenRead(fileName);
                streamingInput.SetInputStream(fsIn);

                var iter = 0;
                foreach (var r in streamingInput.GetRows(row))
                {
                    Assert.IsTrue(r[0].GetInt() == iter);

                    if (iter % 2 == 0)
                    {
                        Assert.IsTrue(r[1].IsNull());
                    }
                    else
                    {
                        Assert.AreEqual(r[1].GetInt(), iter);
                    }

                    Assert.AreEqual(r[2].GetLong(), ((long)iter) << 32);

                    if (iter % 2 == 1)
                    {
                        Assert.IsTrue(r[3].IsNull());
                    }
                    else
                    {
                        Assert.AreEqual(r[3].GetLong(), ((long)iter) << 32);
                    }

                    Assert.AreEqual(r[4].GetString(), (iter.ToString() + iter.ToString().GetHashCode().ToString("X8")));

                    Assert.AreEqual(r[5].GetBoolean(), (iter % 2 == 0));

                    if (iter % 4 == 0)
                    {
                        Assert.IsTrue(r[6].IsNull());
                    }
                    else
                    {
                        Assert.AreEqual(r[6].GetBoolean(), (iter % 2 == 0));
                    }

                    Assert.IsTrue(r[7].GetDouble() == Math.PI * iter);

                    if (iter % 3 == 0)
                    {
                        Assert.IsTrue(r[8].IsNull());
                    }
                    else
                    {
                        Assert.AreEqual(r[8].GetDouble(), Math.PI / iter);
                    }

                    Assert.AreEqual(r[9].GetFloat(), (float)Math.PI * iter);

                    if (iter % 3 == 1)
                    {
                        Assert.IsTrue(r[10].IsNull());
                    }
                    else
                    {
                        Assert.AreEqual(r[10].GetFloat(), (float)Math.PI / iter);
                    }

                    if (iter % 7 == 0)
                    {
                        Assert.IsTrue(r[11].IsNull());
                    }
                    else
                    {
                        Assert.AreEqual(r[11].GetBytes().Length, iter);
                    }

                    Assert.IsTrue(r[12].IsNull());

                    iter++;
                }
            }
        }

        [TestMethod]
        public void TestNullStreamingOutput()
        {
            var fileName = TestHelpers.GetFileFromWorkingDirectory("TestNullStreamingOutput." + Guid.NewGuid().ToString() + ".txt");

            using (var streamingOutput = new StreamingOutput())
            {

                var fsOut = File.OpenWrite(fileName);
                streamingOutput.SetOutputStream(fsOut);

                Row row = new Row(new Schema(@"colInt:int"), new StreamingColumnData(new ColumnSchema("fakeColumn:int")));

                Assert.AreEqual(typeof(ArgumentNullException), TestHelpers.GetException(() => { streamingOutput.Commit(row); }).GetType());
            }
        }

        [TestMethod]
        public void TestSchemaMismatchStreamingInput()
        {
            var fileName = TestHelpers.GetFileFromWorkingDirectory("TestNullStreamingOutput." + Guid.NewGuid().ToString() + ".txt");

            using (var streamingOutput = new StreamingOutput())
            {
                var fsOut = File.OpenWrite(fileName);
                streamingOutput.SetOutputStream(fsOut);

                Row row = new Row(new Schema(@"colInt:int"), new StreamingColumnData(new ColumnSchema("fakeColumn:int")));

                row[0].SetValue(1);

                streamingOutput.Commit(row);

                streamingOutput.Flush();
            }

            using (var streamingInput = new StreamingInput())
            {

                var fsIn = File.OpenRead(fileName);
                streamingInput.SetInputStream(fsIn);

                Row rowMismatch = new Row(new Schema(@"colInt:int,colLong:long"), new StreamingColumnData(new ColumnSchema("fakeColumn:int")));

                Assert.AreEqual(typeof(ArgumentOutOfRangeException), TestHelpers.GetException(() => { foreach (var r in streamingInput.GetRows(rowMismatch)) { } }).GetType());
            }
        }

        [TestMethod]
        public void TestStreamingOutputNotInitialized()
        {
            using (var streamingOutput = new StreamingOutput())
            {

                Assert.AreEqual(typeof(StreamNotInitializedException), TestHelpers.GetException(() => streamingOutput.Commit(null)).GetType());

                Row row = new Row(new Schema(@"colInt:int"), new StreamingColumnData(new ColumnSchema("fakeColumn:int")));
                Assert.AreEqual(typeof(StreamNotInitializedException), TestHelpers.GetException(() => streamingOutput.Commit(row)).GetType());
                Assert.AreEqual(typeof(StreamNotInitializedException), TestHelpers.GetException(() => streamingOutput.Flush()).GetType());

                Assert.AreEqual(typeof(ArgumentNullException), TestHelpers.GetException(() => streamingOutput.SetOutputStream(null)).GetType());

                var fileNameRead = TestHelpers.GetFileFromWorkingDirectory("TestStreamingNotInitialized.Read." + Guid.NewGuid().ToString() + ".txt");

                var fsInOut = File.OpenWrite(fileNameRead);
                fsInOut.Close();

                var fsIn = File.OpenRead(fileNameRead);
                Assert.AreEqual(typeof(ArgumentOutOfRangeException), TestHelpers.GetException(() => streamingOutput.SetOutputStream(fsIn)).GetType());

                var fileName = TestHelpers.GetFileFromWorkingDirectory("TestStreamingNotInitialized." + Guid.NewGuid().ToString() + ".txt");
                var fsOut = File.OpenWrite(fileName);
                streamingOutput.SetOutputStream(fsOut);

                Assert.AreEqual(typeof(ArgumentNullException), TestHelpers.GetException(() => streamingOutput.Commit(null)).GetType());
            }
        }
    }
}