MycLib/Test/AST/Test.Myc.Ast.Pipes.pas

unit Test.Myc.Ast.Pipes;

interface

uses
    System.SysUtils,
    System.Classes,
    System.Generics.Collections,
    System.SyncObjs,
    DUnitX.TestFramework,
    // Core Data & Types
    Myc.Data.Value,
    Myc.Data.Scalar,
    Myc.Data.Keyword,
    Myc.Data.Stream,
    Myc.Data.Series,
    // AST & Compiler
    Myc.Ast,
    Myc.Ast.Nodes,
    Myc.Ast.Types,
    Myc.Ast.Environment,
    Myc.Ast.Script;

type
    // Mock Stream that allows manual signaling
    TTestStream = class(TInterfacedObject, IStream)
    private
        FSeries: IScalarRecordSeries;
        FObservers: TList<IStreamObserver>;
        FLock: TSpinLock;
    public
        constructor Create(const ASeries: IScalarRecordSeries);
        destructor Destroy; override;

        // Manual trigger
        procedure Emit(CycleID: Integer);

        // IStream implementation
        function Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
        procedure Unsubscribe(Tag: TSubscriptionTag);
        function GetSeries: IScalarRecordSeries;
        property Series: IScalarRecordSeries read GetSeries;
    end;

    [TestFixture]
    [IgnoreMemoryLeaks]
    TTestPipes = class
    private
        FEnv: TAstEnvironment;

        function CreateSourceSeries(
            const FieldName: string;
            Kind: TScalar.TKind;
            const Values: TArray<Int64>
        ): IWriteableScalarRecordSeries;
    public
        [Setup]
        procedure Setup;

        [Test]
        procedure Test_SimplePipe_Arithmetic;

        [Test]
        procedure Test_TypeChecker_Fail_On_Invalid_Field;

        [Test]
        procedure Test_MultiSource_Pipe;

    end;

implementation

//------------------------------------------------------------------------------
// TTestStream Implementation
//------------------------------------------------------------------------------

constructor TTestStream.Create(const ASeries: IScalarRecordSeries);
begin
    inherited Create;
    FSeries := ASeries;
    FObservers := TList<IStreamObserver>.Create;
    FLock := Default(TSpinLock);
end;

destructor TTestStream.Destroy;
begin
    FObservers.Free;
    inherited;
end;

function TTestStream.GetSeries: IScalarRecordSeries;
begin
    Result := FSeries;
end;

function TTestStream.Subscribe(const Observer: IStreamObserver): TSubscriptionTag;
begin
    FLock.Enter;
    try
        FObservers.Add(Observer);
        // Use the interface pointer as tag.
        Result := TSubscriptionTag(Observer);
    finally
        FLock.Exit;
    end;
end;

procedure TTestStream.Unsubscribe(Tag: TSubscriptionTag);
var
    obs: IStreamObserver;
begin
    // We treat the tag as the interface pointer.
    // Note: This relies on the caller ensuring the object is still alive
    // if they want to unsubscribe, which is standard behavior.
    obs := IStreamObserver(Tag);

    FLock.Enter;
    try
        FObservers.Remove(obs);
    finally
        FLock.Exit;
    end;
end;

procedure TTestStream.Emit(CycleID: Integer);
var
    obs: IStreamObserver;
    observersCopy: TArray<IStreamObserver>;
begin
    FLock.Enter;
    try
        observersCopy := FObservers.ToArray;
    finally
        FLock.Exit;
    end;

    // Simulate new data arrival
    for obs in observersCopy do
    begin
        obs.OnSignal(TStreamSignal.Create(skData, CycleID));
    end;
end;

//------------------------------------------------------------------------------
// TTestPipes Implementation
//------------------------------------------------------------------------------

procedure TTestPipes.Setup;
begin
    FEnv := TAstEnvironment.Construct(nil);
end;

function TTestPipes.CreateSourceSeries(
    const FieldName: string;
    Kind: TScalar.TKind;
    const Values: TArray<Int64>
): IWriteableScalarRecordSeries;
var
    fields: TArray<TScalarRecordField>;
    def: IScalarRecordDefinition;
    i: Integer;
    val: TScalar.TValue;
    recValues: TArray<TScalar.TValue>;
begin
    SetLength(fields, 1);
    fields[0] := TScalarRecordField.Create(TKeywordRegistry.Intern(FieldName), Kind);
    def := TKeywordMappingRegistry<TScalar.TKind>.Intern(fields);

    Result := TScalarRecordSeries.Create(def);

    SetLength(recValues, 1);
    for i := 0 to High(Values) do
    begin
        val.AsInt64 := Values[i];
        recValues[0] := val;
        Result.Add(recValues);
    end;
end;

procedure TTestPipes.Test_SimplePipe_Arithmetic;
var
    source: IWriteableScalarRecordSeries;
    mockStream: TTestStream; // Keep reference to fire later
    script: string;
    resVal: TDataValue;
    resStream: IStream;
    resSeries: IScalarRecordSeries;
    item: TScalar;
    key: IKeyword;
begin
    // Arrange
    source := CreateSourceSeries('val', TScalar.TKind.Ordinal, [10, 20, 30]);
    mockStream := TTestStream.Create(source);

    FEnv.RootScope.Define('src', TDataValue.FromStream(mockStream), TTypes.CreateRecordSeries(source.Def));

    // Act
    script := '(pipe [src [:val]] (fn [x] {:res (* x 2)}))';

    // 1. Compile & Link & Run (Constructs the Pipe Object)
    resVal := FEnv.Run(TAstScript.Parse(script));

    Assert.AreEqual(TDataValueKind.vkStream, resVal.Kind);
    resStream := resVal.AsStream;
    resSeries := resStream.Series;

    // 2. Fire Data! (Now that pipe is fully constructed)
    mockStream.Emit(0);

    // Assert
    Assert.AreEqual(Int64(1), resSeries.TotalCount, 'TotalCount mismatch');

    key := TKeywordRegistry.Intern('res');

    item := resSeries.Fields[key].Items[0];
    Assert.AreEqual(Int64(60), item.Value.AsInt64, 'Item[0] incorrect');
end;

procedure TTestPipes.Test_TypeChecker_Fail_On_Invalid_Field;
var
    source: IWriteableScalarRecordSeries;
    mockStream: TTestStream;
    script: string;
begin
    source := CreateSourceSeries('val', TScalar.TKind.Ordinal, [1]);
    mockStream := TTestStream.Create(source);

    FEnv.RootScope.Define('src', TDataValue.FromStream(mockStream), TTypes.CreateRecordSeries(source.Def));

    script := '(pipe [src [:non_existent]] (fn [x] {:res x}))';

    Assert.WillRaise(procedure begin FEnv.Run(TAstScript.Parse(script)); end, ECompilationFailed);
end;

procedure TTestPipes.Test_MultiSource_Pipe;
var
    s1, s2: IWriteableScalarRecordSeries;
    m1, m2: TTestStream;
    script: string;
    key: IKeyword;
    resSeries: IScalarRecordSeries;
begin
    key := TKeywordRegistry.Intern('sum');

    // Arrange
    s1 := CreateSourceSeries('a', TScalar.TKind.Ordinal, [20]);
    s2 := CreateSourceSeries('b', TScalar.TKind.Ordinal, [6]);

    m1 := TTestStream.Create(s1);
    m2 := TTestStream.Create(s2);

    FEnv.RootScope.Define('in1', TDataValue.FromStream(m1), TTypes.CreateRecordSeries(s1.Def));
    FEnv.RootScope.Define('in2', TDataValue.FromStream(m2), TTypes.CreateRecordSeries(s2.Def));

    // Act
    script := '(pipe [in1 [:a] in2 [:b]] (fn [valA valB] {:sum (+ valA valB)}))';

    resSeries := FEnv.Run(TAstScript.Parse(script)).AsStream.Series;

    Assert.AreEqual(Int64(0), resSeries.TotalCount);

    // Partial update - no result expected
    m1.Emit(0);
    Assert.AreEqual(Int64(0), resSeries.TotalCount);

    // Complete update - result expected
    m2.Emit(0);
    Assert.AreEqual(Int64(1), resSeries.TotalCount);

    Assert.AreEqual(Int64(26), resSeries.Fields[key].Items[0].Value.AsInt64);

    // Next cycle
    s1.Add([10]);
    m1.Emit(1);
    Assert.AreEqual(Int64(1), resSeries.TotalCount);

    s2.Add([5]);
    m2.Emit(1);
    Assert.AreEqual(Int64(2), resSeries.TotalCount);

    // Check history (Lookback 0 is newest)
    // Items[0] is the result of Cycle 1 (10 + 5 = 15)
    Assert.AreEqual(Int64(15), resSeries.Fields[key].Items[0].Value.AsInt64);
    // Items[1] is the result of Cycle 0 (20 + 6 = 26)
    Assert.AreEqual(Int64(26), resSeries.Fields[key].Items[1].Value.AsInt64);
end;

end.