<commit_before>﻿namespace MBrace.Core.Tests

#nowarn "0444" // Disable mbrace warnings
open System
open System.Linq
open System.Collections.Generic
open System.Net
open System.IO
open FsCheck
open NUnit.Framework
open MBrace.Flow
open MBrace.Flow.Internals
open MBrace.Core
open System.Text

// Helper type
type Separator = N | R | RN

[<ReferenceEquality; NoComparison>]
type private ReferenceEqualityContainer<'T> = { Payload : 'T }

// General-purpose, runtime independent CloudFlow tests
[<TestFixture>]
module ``CloudFlow Core property tests`` =

    [<Test>]
    let ``ResizeArray concatenator`` () =
        let check (inputs : int[][]) =
            let tas = inputs |> Array.map (fun ts -> new ResizeArray<_>(ts))
            let expected = inputs |> Array.concat
            ResizeArray.concat tas |> Seq.toArray |> shouldEqual expected

        Check.QuickThrowOnFail check

    [<Test>]
    let ``Partition by size`` () =
        let check (inputs : int64 [], maxSize : int64) =
            let maxSize = 1L + abs maxSize
            let partitions =
                inputs
                |> Partition.partitionBySize (fun i -> async { return abs i }) maxSize
                |> Async.RunSynchronously
        
            Array.concat partitions |> shouldEqual inputs

            for p in partitions do
                if Array.sum p > maxSize then
                    Array.sum p.[0 .. p.Length - 2] |> shouldBe (fun s -> s <= maxSize)

        Check.QuickThrowOnFail check


[<TestFixture; AbstractClass>]
type ``CloudFlow tests`` () as self =
    let runRemote (workflow : Cloud<'T>) = self.RunRemote(workflow)
    let runLocally (workflow : Cloud<'T>) = self.RunLocally(workflow)

    /// Urls for running HTTP tests
    let testUrls = 
        [| 
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-alls-11.txt";
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-antony-23.txt";
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-as-12.txt";
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-comedy-7.txt";
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-coriolanus-24.txt";
            "http://ocw.mit.edu/ans7870/6/6.006/s08/lecturenotes/files/t8.shakespeare.txt" 
        |]

    let getHttpFileLineCount (url : string) = 
        let mutable lineCount = 0L
        let client = new WebClient()
        use stream = client.OpenRead(url)
        use reader = new StreamReader(stream)

        while reader.ReadLine() <> null do
            lineCount <- lineCount + 1L

        lineCount

    abstract RunRemote : Cloud<'T> -> 'T
    abstract RunLocally : Cloud<'T> -> 'T
    abstract IsSupportedStorageLevel : StorageLevel -> bool
    abstract FsCheckMaxNumberOfTests : int
    abstract FsCheckMaxNumberOfIOBoundTests : int

    // #region Flow persist tests

    [<Test>]
    member __.``1. PersistedCloudFlow : simple persist`` () =
        if __.IsSupportedStorageLevel StorageLevel.Disk then
            let inputs = [|1L .. 1000000L|]
            let persisted = inputs |> CloudFlow.OfArray |> CloudFlow.persist StorageLevel.Disk |> runRemote
            let workers = Cloud.GetWorkerCount() |> runRemote
            persisted.StorageLevel |> shouldEqual StorageLevel.Disk
            persisted.PartitionCount |> shouldEqual workers
            persisted |> CloudFlow.toArray |> runRemote |> shouldEqual inputs
            persisted |> Seq.toArray |> shouldEqual inputs

    [<Test>]
    member __.``1. PersistedCloudFlow : randomized persist`` () =
        let f(xs : int[]) =            
            let x = xs |> CloudFlow.OfArray |> CloudFlow.map ((+)1) |> CloudFlow.persist StorageLevel.Disk |> runRemote
            let y = xs |> Seq.map ((+)1) |> Seq.toArray
            Assert.AreEqual(y, x |> Seq.toArray)
        Check.QuickThrowOnFail(f, self.FsCheckMaxNumberOfTests)

    [<Test>]
    member __.``1. PersistedCloudFlow : caching`` () =
        let inputs = [|1L .. 1000000L|]
        let persisted = inputs |> CloudFlow.OfArray |> CloudFlow.cache |> runRemote
        let workers = Cloud.GetWorkerCount() |> runRemote
        persisted.StorageLevel |> shouldEqual StorageLevel.Memory
        persisted.PartitionCount |> shouldEqual workers
        persisted |> CloudFlow.toArray |> runRemote |> shouldEqual inputs

    [<Test>]
    member __.``1. PersistedCloudFlow : disposal`` () =
        let inputs = [|1 .. 1000000|]
        let persisted = inputs |> CloudFlow.OfArray |> CloudFlow.persist StorageLevel.Disk |> runRemote
        persisted |> Cloud.Dispose |> runRemote
        shouldfail(fun () -> persisted |> CloudFlow.length |> runRemote |> ignore)

    [<Test>]
    member __.``1. PersistedCloudFlow : cached objects`` () =
        // ensure that initial array is large enough for disk caching to kick in,
        // otherwise persisted fragments will be encapsulated in the reference as an optimization.
        // Limit is currently set to 512KiB, so total size must be 512KiB * cluster size.
        let xs = Array.init 10000 (fun _ -> "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.")
        let cv = xs |> CloudFlow.OfArray |> CloudFlow.map (fun x -> { Payload = x }) |> CloudFlow.cache |> runRemote
        let x = cv |> CloudFlow.map (fun ref -> ref.GetHashCode()) |> CloudFlow.toArray |> runRemote
        for i in 1 .. 5 do
            let y = cv |> CloudFlow.map (fun ref -> ref.GetHashCode()) |> CloudFlow.toArray |> runRemote
            Assert.AreEqual(x, y)

    [<Test>]
    member __.``1. PersistedCloudFlow : merging`` () =
        let inputs = [|1 .. 1000000|]
        let N = 10
        let persisted = inputs |> CloudFlow.OfArray |> CloudFlow.persist StorageLevel.MemoryAndDisk |> runRemote
        let merged = PersistedCloudFlow.Concat(Array.init N (fun _ -> persisted))
        merged.PartitionCount |> shouldEqual (N * persisted.PartitionCount)
        
        let partitions = persisted.GetPartitions() |> Array.map snd
        let mergedPartitions = merged.GetPartitions() |> Array.map snd
        
        mergedPartitions
        |> Seq.countBy (fun p -> p.Id)
        |> Seq.forall (fun (_,n) -> n = N)
        |> shouldEqual true
<commit_msg>fix CloudFlow tests<commit_after>﻿namespace MBrace.Core.Tests

#nowarn "0444" // Disable mbrace warnings
open System
open System.Linq
open System.Collections.Generic
open System.Net
open System.IO
open FsCheck
open NUnit.Framework
open MBrace.Flow
open MBrace.Flow.Internals
open MBrace.Core
open System.Text

// Helper type
type Separator = N | R | RN

// General-purpose, runtime independent CloudFlow tests
[<TestFixture>]
module ``CloudFlow Core property tests`` =

    [<Test>]
    let ``ResizeArray concatenator`` () =
        let check (inputs : int[][]) =
            let tas = inputs |> Array.map (fun ts -> new ResizeArray<_>(ts))
            let expected = inputs |> Array.concat
            ResizeArray.concat tas |> Seq.toArray |> shouldEqual expected

        Check.QuickThrowOnFail check

    [<Test>]
    let ``Partition by size`` () =
        let check (inputs : int64 [], maxSize : int64) =
            let maxSize = 1L + abs maxSize
            let partitions =
                inputs
                |> Partition.partitionBySize (fun i -> async { return abs i }) maxSize
                |> Async.RunSynchronously
        
            Array.concat partitions |> shouldEqual inputs

            for p in partitions do
                if Array.sum p > maxSize then
                    Array.sum p.[0 .. p.Length - 2] |> shouldBe (fun s -> s <= maxSize)

        Check.QuickThrowOnFail check


[<TestFixture; AbstractClass>]
type ``CloudFlow tests`` () as self =
    let runRemote (workflow : Cloud<'T>) = self.RunRemote(workflow)
    let runLocally (workflow : Cloud<'T>) = self.RunLocally(workflow)

    /// Urls for running HTTP tests
    let testUrls = 
        [| 
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-alls-11.txt";
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-antony-23.txt";
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-as-12.txt";
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-comedy-7.txt";
            "http://www.textfiles.com/etext/AUTHORS/SHAKESPEARE/shakespeare-coriolanus-24.txt";
            "http://ocw.mit.edu/ans7870/6/6.006/s08/lecturenotes/files/t8.shakespeare.txt" 
        |]

    let getHttpFileLineCount (url : string) = 
        let mutable lineCount = 0L
        let client = new WebClient()
        use stream = client.OpenRead(url)
        use reader = new StreamReader(stream)

        while reader.ReadLine() <> null do
            lineCount <- lineCount + 1L

        lineCount

    abstract RunRemote : Cloud<'T> -> 'T
    abstract RunLocally : Cloud<'T> -> 'T
    abstract IsSupportedStorageLevel : StorageLevel -> bool
    abstract FsCheckMaxNumberOfTests : int
    abstract FsCheckMaxNumberOfIOBoundTests : int

    // #region Flow persist tests

    [<Test>]
    member __.``1. PersistedCloudFlow : Simple StorageLevel.Disk`` () =
        if __.IsSupportedStorageLevel StorageLevel.Disk then
            let inputs = [|1L .. 1000000L|]
            let persisted = inputs |> CloudFlow.OfArray |> CloudFlow.persist StorageLevel.Disk |> runRemote
            let workers = Cloud.GetWorkerCount() |> runRemote
            persisted.StorageLevel |> shouldEqual StorageLevel.Disk
            persisted.PartitionCount |> shouldEqual workers
            persisted |> CloudFlow.toArray |> runRemote |> shouldEqual inputs
            persisted |> Seq.toArray |> shouldEqual inputs

    [<Test>]
    member __.``1. PersistedCloudFlow : Randomized StorageLevel.Disk`` () =
        if __.IsSupportedStorageLevel StorageLevel.Disk then
            let f(xs : int[]) =            
                let x = xs |> CloudFlow.OfArray |> CloudFlow.map ((+)1) |> CloudFlow.persist StorageLevel.Disk |> runRemote
                let y = xs |> Seq.map ((+)1) |> Seq.toArray
                Assert.AreEqual(y, x |> Seq.toArray)
            Check.QuickThrowOnFail(f, self.FsCheckMaxNumberOfTests)

    [<Test>]
    member __.``1. PersistedCloudFlow : StorageLevel.Memory`` () =
        if __.IsSupportedStorageLevel StorageLevel.Memory then
            let inputs = [|1L .. 1000000L|]
            let persisted = inputs |> CloudFlow.OfArray |> CloudFlow.cache |> runRemote
            let workers = Cloud.GetWorkerCount() |> runRemote
            persisted.StorageLevel |> shouldEqual StorageLevel.Memory
            persisted.PartitionCount |> shouldEqual workers
            persisted |> CloudFlow.toArray |> runRemote |> shouldEqual inputs

    [<Test>]
    member __.``1. PersistedCloudFlow : disposal`` () =
        let inputs = [|1 .. 1000000|]
        let persisted = inputs |> CloudFlow.OfArray |> CloudFlow.persist StorageLevel.Disk |> runRemote
        persisted |> Cloud.Dispose |> runRemote
        shouldfail(fun () -> persisted |> CloudFlow.length |> runRemote |> ignore)

    [<Test>]
    member __.``1. PersistedCloudFlow : StorageLevel.Memory caching`` () =
        if __.IsSupportedStorageLevel StorageLevel.Memory then
            // ensure that initial array is large enough for disk caching to kick in,
            // otherwise persisted fragments will be encapsulated in the reference as an optimization.
            // Limit is currently set to 512KiB, so total size must be 512KiB * cluster size.
            // Arrays have reference equality semantics, compare hashcodes to ensure that caching is taking place.
            let cv = [|1 .. 20|] |> CloudFlow.OfArray |> CloudFlow.map (fun _ -> [|1L .. 1000000L|]) |> CloudFlow.cache |> runRemote
            let x = cv |> CloudFlow.map (fun ref -> ref.GetHashCode()) |> CloudFlow.toArray |> runRemote
            for i in 1 .. 5 do
                let y = cv |> CloudFlow.map (fun ref -> ref.GetHashCode()) |> CloudFlow.toArray |> runRemote
                Assert.AreEqual(x, y)

    [<Test>]
    member __.``1. PersistedCloudFlow : Simple StorageLevel.MemorySerialized`` () =
        if __.IsSupportedStorageLevel StorageLevel.MemorySerialized then
            let f(xs : int[]) =            
                let pf = xs |> CloudFlow.OfArray |> CloudFlow.map ((+)1) |> CloudFlow.persist StorageLevel.MemorySerialized |> runRemote
                let ys = pf |> CloudFlow.length |> runRemote
                Assert.AreEqual(ys, xs.LongLength)
            Check.QuickThrowOnFail(f, self.FsCheckMaxNumberOfTests)

    [<Test>]
    member __.``1. PersistedCloudFlow : merging`` () =
        let inputs = [|1 .. 1000000|]
        let N = 10
        let persisted = inputs |> CloudFlow.OfArray |> CloudFlow.persist StorageLevel.MemoryAndDisk |> runRemote
        let merged = PersistedCloudFlow.Concat(Array.init N (fun _ -> persisted))
        merged.PartitionCount |> shouldEqual (N * persisted.PartitionCount)
        
        let partitions = persisted.GetPartitions() |> Array.map snd
        let mergedPartitions = merged.GetPartitions() |> Array.map snd
        
        mergedPartitions
        |> Seq.countBy (fun p -> p.Id)
        |> Seq.forall (fun (_,n) -> n = N)
        |> shouldEqual true

{'fraction_non_alphanumeric': 0.1124251497005988, 'fraction_numerical': 0.013922155688622755, 'mean_word_length': 2.427655207798871, 'pattern_counts': {'":': 0, '<': 42, '<?xml version=': 0, '>': 197, 'https://': 0, 'lorem ipsum': 0, 'www.': 10, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2354723', 'n_tokens_mistral': 4138, 'n_tokens_neox': 3852, 'n_words': 1014}