path: root/tests/FundLab.Api.Tests/ProcessCollectorTests.fs

namespace FundLab.Api.Tests

module ProcessCollectorTests =

    open System
    open System.Diagnostics
    open System.IO
    open System.Threading
    open System.Threading.Tasks
    open Xunit
    open FundLab.Api

    let private secretSentinel = "fund-lab-SECRET-SENTINEL"

    let private fixtureDir () =
        let dir = Path.Combine(Path.GetTempPath(), "fund-lab-collector-" + Guid.NewGuid().ToString("N"))

        Directory.CreateDirectory(dir) |> ignore

        dir

    let private writeScript (dir: string) (name: string) (body: string) =
        let path = Path.Combine(dir, name)
        File.WriteAllText(path, body)
        path

    let private searchEnvelope =
        "{\"schema_version\":\"fund-lab.akshare.v1\",\"operation\":\"search\",\"source\":\"fixture\",\"source_revision\":\"fixture/1\",\"collected_at\":\"2026-09-21T00:00:00Z\",\"_synthetic\":true,\"instruments\":[{\"code\":\"000001\",\"name\":\"fixture fund\",\"fund_type\":null}]}"

    let private hangScript dir =
        writeScript dir "hang.sh" ("sleep 60 & child=$!\necho $child > pidfile\nwait $child\n")

    let private orphanScript dir =
        writeScript
            dir
            "orphan.sh"
            ("sleep 30 & child=$!\necho $child > pidfile\nprintf '%s' '" + searchEnvelope + "'\nexit 0\n")

    let private recordedPid (dir: string) =
        Int32.Parse(File.ReadAllText(Path.Combine(dir, "pidfile")).Trim())

    let private pidAlive (pid: int) = Directory.Exists(sprintf "/proc/%d" pid)

    let private waitForPidDeath (pid: int) =
        let mutable dead = not (pidAlive pid)
        let mutable waited = 0

        while not dead && waited < 10000 do
            Thread.Sleep(100)
            waited <- waited + 100
            dead <- not (pidAlive pid)

        dead

    [<Fact>]
    let ``collector captures stdout from a successful subprocess`` () =
        let dir = fixtureDir ()

        try
            let script = writeScript dir "success.sh" ("printf '%s' '" + searchEnvelope + "'")
            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 30) :> IMarketDataCollector

            match collector.Search("000001", CancellationToken.None) with
            | Ok raw ->
                Assert.Contains("000001", raw)
                Assert.Contains("\"_synthetic\":true", raw)
            | Error message -> Assert.Fail(sprintf "expected success, got %s" message)
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``collector drains concurrent stderr flood without deadlocking`` () =
        let dir = fixtureDir ()

        try
            let body = "yes 'flood padding padding padding padding' | head -n 5000 >&2\nprintf '%s' '" + searchEnvelope + "'"

            let script = writeScript dir "flood.sh" body

            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 30) :> IMarketDataCollector

            match collector.Search("000001", CancellationToken.None) with
            | Ok raw -> Assert.Contains("000001", raw)
            | Error message -> Assert.Fail(sprintf "expected success, got %s" message)
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``collector enforces deadline, kills owned processes and sanitizes error`` () =
        let dir = fixtureDir ()

        try
            let script = hangScript dir
            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 1) :> IMarketDataCollector
            let stopwatch = Stopwatch.StartNew()

            match collector.Search("000001", CancellationToken.None) with
            | Error message ->
                stopwatch.Stop()
                Assert.Contains("did not finish within 1 seconds", message)
                Assert.DoesNotContain(dir, message)
                Assert.DoesNotContain(secretSentinel, message)
            | Ok raw -> Assert.Fail(sprintf "expected timeout error, got %s" raw)

            Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(20.0))
            Assert.True(waitForPidDeath (recordedPid dir), "backgrounded subprocess survived the deadline kill")
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``collector reports cancellation and kills owned processes`` () =
        let dir = fixtureDir ()

        try
            let script = hangScript dir
            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 30) :> IMarketDataCollector
            use cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(500.0))

            match collector.Search("000001", cts.Token) with
            | Error message ->
                Assert.Equal("collector run was cancelled", message)
                Assert.DoesNotContain(dir, message)
                Assert.DoesNotContain(secretSentinel, message)
            | Ok raw -> Assert.Fail(sprintf "expected cancellation, got %s" raw)

            Assert.True(waitForPidDeath (recordedPid dir), "backgrounded subprocess survived cancellation")
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``collector rejects orphan-held pipes at the total exit and EOF deadline`` () =
        let dir = fixtureDir ()

        try
            let script = orphanScript dir
            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 2) :> IMarketDataCollector
            let stopwatch = Stopwatch.StartNew()

            let result = collector.Search("000001", CancellationToken.None)
            stopwatch.Stop()

            match result with
            | Ok raw -> Assert.Fail(sprintf "recovered payload after expired deadline: %s" raw)
            | Error message ->
                Assert.Equal("collector did not finish within 2 seconds", message)
                Assert.DoesNotContain(dir, message)
                Assert.DoesNotContain(secretSentinel, message)

            Assert.True(stopwatch.Elapsed >= TimeSpan.FromSeconds(1.8))
            Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(15.0))
            Assert.True(waitForPidDeath (recordedPid dir), "pipe-holding orphan survived the EOF deadline kill")
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``cancellation during EOF wait interrupts promptly and kills descendants`` () =
        let dir = fixtureDir ()

        try
            let script = orphanScript dir
            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 30) :> IMarketDataCollector
            use cts = new CancellationTokenSource()

            let searchTask = Task.Run(fun () -> collector.Search("000001", cts.Token))
            Thread.Sleep(700)

            let stopwatch = Stopwatch.StartNew()
            cts.Cancel()
            let completed = searchTask.Wait(10000)
            stopwatch.Stop()

            Assert.True(completed, "search did not return promptly after cancellation")

            match searchTask.Result with
            | Error message -> Assert.Equal("collector run was cancelled", message)
            | Ok raw -> Assert.Fail(sprintf "expected cancellation, got %s" raw)

            Assert.True(stopwatch.Elapsed < TimeSpan.FromSeconds(5.0), "cancellation during EOF wait was not prompt")
            Assert.True(waitForPidDeath (recordedPid dir), "pipe-holding orphan survived EOF-wait cancellation")
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``pre-cancelled request does not launch the collector`` () =
        let dir = fixtureDir ()

        try
            let script =
                writeScript dir "recording.sh" ("echo $$ > pidfile\nprintf '%s' '" + searchEnvelope + "'\n")

            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 30) :> IMarketDataCollector
            use cts = new CancellationTokenSource()
            cts.Cancel()

            match collector.Search("000001", cts.Token) with
            | Error message ->
                Assert.Equal("collector run was cancelled", message)
                Assert.DoesNotContain(dir, message)
            | Ok raw -> Assert.Fail(sprintf "expected pre-cancel, got %s" raw)

            Assert.False(File.Exists(Path.Combine(dir, "pidfile")), "collector process was launched despite pre-cancelled request")
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``collector surfaces malformed payload for parser to reject`` () =
        let dir = fixtureDir ()

        try
            let script = writeScript dir "malformed.sh" "printf '%s' '{ not json at all'"
            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 30) :> IMarketDataCollector

            match collector.Search("000001", CancellationToken.None) with
            | Ok raw ->
                Assert.Equal("{ not json at all", raw)

                match MarketData.parseSearchPayload raw with
                | Error _ -> ()
                | Ok _ -> Assert.Fail("expected malformed payload to fail parsing")
            | Error message -> Assert.Fail(sprintf "expected raw output, got error %s" message)
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``nonzero exit keeps stderr sentinel and paths out of the public error`` () =
        let dir = fixtureDir ()

        try
            let body =
                "printf 'secret=%s\\n' '"
                + secretSentinel
                + "' >&2\nprintf 'script lives under "
                + dir
                + "\\n' >&2\nexit 3\n"

            let script = writeScript dir "failing.sh" body
            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 30) :> IMarketDataCollector

            match collector.Search("000001", CancellationToken.None) with
            | Error message ->
                Assert.Equal("collector exited with code 3", message)
                Assert.DoesNotContain(secretSentinel, message)
                Assert.DoesNotContain(dir, message)
            | Ok raw -> Assert.Fail(sprintf "expected failure, got %s" raw)
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``start failure keeps executable path out of the public error`` () =
        let dir = fixtureDir ()

        try
            let script = writeScript dir "success.sh" ("printf '%s' '" + searchEnvelope + "'")
            let missingExecutable = "/nonexistent-" + secretSentinel + "/python3"

            let collector = ProcessMarketDataCollector(missingExecutable, script, None, 30) :> IMarketDataCollector

            match collector.Search("000001", CancellationToken.None) with
            | Error message ->
                let sanitized =
                    message = "could not run the AKShare collector"
                    || message.StartsWith("collector exited with code")

                Assert.True(sanitized, sprintf "unexpected error message: %s" message)
                Assert.DoesNotContain(secretSentinel, message)
                Assert.DoesNotContain(dir, message)
            | Ok raw -> Assert.Fail(sprintf "expected failure, got %s" raw)
        finally
            Directory.Delete(dir, true)

    [<Fact>]
    let ``missing collector script yields sanitized error without paths`` () =
        let dir = fixtureDir ()

        try
            let script = Path.Combine(dir, "missing.py")
            let collector = ProcessMarketDataCollector("/bin/sh", script, None, 30) :> IMarketDataCollector

            match collector.Search("000001", CancellationToken.None) with
            | Error message ->
                Assert.Equal("collector script is not deployed", message)
                Assert.DoesNotContain(dir, message)
            | Ok raw -> Assert.Fail(sprintf "expected error, got %s" raw)
        finally
            Directory.Delete(dir, true)