fpc-comet/tests/cmtestutil.pas

{ cmtestutil -- shared helpers for the fpc-comet test suite.

  Mirrors fpc-binkp's testutil but pinned to IComTransport
  so two TComSession instances can be wired up over a single
  in-process pipe.  This unit deliberately stays tiny:

  - TByteQueue: thread-safe FIFO of bytes
  - TMemPipeEnd: IComTransport that pushes to a "remote read"
    queue and pops from a "local read" queue
  - TMemPipePair: spawns two pipe ends sharing two queues
    (A->B and B->A)
  - TTest: assertion runner with the same Check/CheckEqual
    surface fpc-binkp uses, so test programs read the same.

  An in-memory IComFileProvider lands in a follow-up batch
  when cm.xfer is in place. }

unit cmtestutil;

{$mode objfpc}{$H+}
{$modeswitch advancedrecords}

interface

uses
  Classes, SysUtils, SyncObjs,
  cm.types, cm.transport, cm.provider, cr.sha;

type
  TByteQueue = class
  private
    FLock: TCriticalSection;
    FData: TBytes;
    FLen:  Integer;
    FClosed: Boolean;
  public
    constructor Create;
    destructor  Destroy; override;
    procedure   Push(const Buf; Len: Integer);
    function    Pop(var Buf; MaxLen: Integer): Integer;
    function    Available: Integer;
    procedure   Close;
    function    IsClosed: Boolean;
  end;

  TMemPipeEnd = class(TInterfacedObject, IComTransport)
  private
    FSendQ, FRecvQ: TByteQueue;
    FDesc: string;
  public
    constructor Create(ASend, ARecv: TByteQueue; const ADesc: string);
    function Send(const Buf; Len: Integer): Integer;
    function Recv(var Buf; Len: Integer): Integer;
    function WaitReady(WantRead, WantWrite: Boolean;
                       TimeoutMs: Integer): Integer;
    procedure ShutdownWrite;
    procedure Close;
    function IsConnected: Boolean;
    function PeerDescription: string;
  end;

  TMemPipePair = class
  private
    FQueueAB, FQueueBA: TByteQueue;
    FEndA, FEndB: IComTransport;
  public
    constructor Create;
    destructor  Destroy; override;
    property EndA: IComTransport read FEndA;
    property EndB: IComTransport read FEndB;
  end;

  TTest = class
  private
    FRun, FFailed: Integer;
    FName: string;
  public
    constructor Create(const AName: string);
    procedure Group(const Msg: string);
    procedure Check(Cond: Boolean; const Msg: string);
    procedure CheckEqual(A, B: Integer; const Msg: string);
    procedure CheckEqualStr(const A, B: string; const Msg: string);
    function  ExitCode: Integer;
    procedure Summary;
  end;

  { An outbound file the memory provider will offer. }
  TMemSendFile = record
    Name: string;
    Data: TBytes;
    Time: LongInt;
  end;

  { A received file captured by the memory provider. }
  TMemRecvFile = record
    Name: string;
    Data: TBytes;
    Time: LongInt;
  end;

  TMemProvider = class(TInterfacedObject, IComFileProvider)
  private
    FOutbound: array of TMemSendFile;
    FOutIdx:   Integer;
    FOutStreams: array of TMemoryStream;
    FReceived: array of TMemRecvFile;
    FStagingMap: TStringList;       { TempPath -> idx into FStagingStreams }
    FStagingStreams: array of TMemoryStream;
    FStagingNames:   array of string;
    FStagingTimes:   array of LongInt;
    FAcked:    array of string;
    FFailed:   array of string;
    { Pre-staged partials -- a previous session's partial
      receive that we want to resume.  Indexed by filename. }
    FPartialNames: TStringList;     { lowercased filename -> bytes }
    FPartialData:  array of TBytes;
  public
    constructor Create;
    destructor  Destroy; override;
    procedure   QueueSend(const Name: string;
                          const Data: TBytes; Time_: LongInt);
    { Pre-stage a partial under Name; the next OpenForReceive
      for that name returns a stream pre-populated with these
      bytes from offset 0..Length(Data), positioned at the
      end (Position = Length(Data)).  GetPartialSize returns
      Length(Data) for the same name. }
    procedure   QueuePartial(const Name: string; const Data: TBytes);

    function  NextOutbound(out Item: TCometSendItem): TCometSendDecision;
    procedure AcknowledgeSent(const Item: TCometSendItem);
    procedure AcknowledgeFailed(const Item: TCometSendItem;
                                Reason: TCometSendFailReason);
    function  OpenForReceive(const Name: string; Size: Int64;
                              Time: LongInt; ResumeOffset: Int64;
                              const OverrideDir: string;
                              out TempPath: string): TStream;
    function  GetPartialSize(const Name: string;
                              Time: LongInt): Int64;
    function  FinalizeReceive(const TempPath, Name: string;
                              Time: LongInt): string;
    procedure CleanupReceive(const TempPath: string;
                              KeepForResume: Boolean);

    function ReceivedCount: Integer;
    function ReceivedAt(I: Integer): TMemRecvFile;
    function AckedCount:    Integer;
    function AckedAt(I: Integer): string;
    function FailedCount:   Integer;
    function FailedAt(I: Integer): string;
  end;

implementation

{ ---- TByteQueue ---- }

constructor TByteQueue.Create;
begin
  inherited Create;
  FLock := TCriticalSection.Create;
end;

destructor TByteQueue.Destroy;
begin
  FLock.Free;
  inherited;
end;

procedure TByteQueue.Push(const Buf; Len: Integer);
begin
  if Len <= 0 then Exit;
  FLock.Enter;
  try
    if FLen + Len > Length(FData) then
      SetLength(FData, FLen + Len + 4096);
    Move(Buf, FData[FLen], Len);
    Inc(FLen, Len);
  finally
    FLock.Leave;
  end;
end;

function TByteQueue.Pop(var Buf; MaxLen: Integer): Integer;
begin
  Result := 0;
  if MaxLen <= 0 then Exit;
  FLock.Enter;
  try
    if FLen = 0 then Exit;
    if MaxLen >= FLen then Result := FLen else Result := MaxLen;
    Move(FData[0], Buf, Result);
    if Result < FLen then
      Move(FData[Result], FData[0], FLen - Result);
    Dec(FLen, Result);
  finally
    FLock.Leave;
  end;
end;

function TByteQueue.Available: Integer;
begin
  FLock.Enter;
  try
    Result := FLen;
  finally
    FLock.Leave;
  end;
end;

procedure TByteQueue.Close;
begin
  FLock.Enter;
  try
    FClosed := True;
  finally
    FLock.Leave;
  end;
end;

function TByteQueue.IsClosed: Boolean;
begin
  FLock.Enter;
  try
    Result := FClosed;
  finally
    FLock.Leave;
  end;
end;

{ ---- TMemPipeEnd ---- }

constructor TMemPipeEnd.Create(ASend, ARecv: TByteQueue; const ADesc: string);
begin
  inherited Create;
  FSendQ := ASend;
  FRecvQ := ARecv;
  FDesc  := ADesc;
end;

function TMemPipeEnd.Send(const Buf; Len: Integer): Integer;
begin
  if FSendQ.IsClosed then Exit(-1);
  FSendQ.Push(Buf, Len);
  Result := Len;
end;

function TMemPipeEnd.Recv(var Buf; Len: Integer): Integer;
begin
  Result := FRecvQ.Pop(Buf, Len);
  if (Result = 0) and FRecvQ.IsClosed then Result := -1;
end;

function TMemPipeEnd.WaitReady(WantRead, WantWrite: Boolean;
                                TimeoutMs: Integer): Integer;
begin
  Result := 0;
  if WantRead and (FRecvQ.Available > 0) then
    Result := Result or 1;
  if WantWrite then
    Result := Result or 2;
end;

procedure TMemPipeEnd.ShutdownWrite;
begin
  FSendQ.Close;
end;

procedure TMemPipeEnd.Close;
begin
  FSendQ.Close;
  FRecvQ.Close;
end;

function TMemPipeEnd.IsConnected: Boolean;
begin
  Result := not (FSendQ.IsClosed and FRecvQ.IsClosed);
end;

function TMemPipeEnd.PeerDescription: string;
begin
  Result := FDesc;
end;

{ ---- TMemPipePair ---- }

constructor TMemPipePair.Create;
begin
  inherited Create;
  FQueueAB := TByteQueue.Create;
  FQueueBA := TByteQueue.Create;
  FEndA := TMemPipeEnd.Create(FQueueAB, FQueueBA, 'mempipe:B');
  FEndB := TMemPipeEnd.Create(FQueueBA, FQueueAB, 'mempipe:A');
end;

destructor TMemPipePair.Destroy;
begin
  FEndA := nil;
  FEndB := nil;
  FQueueAB.Free;
  FQueueBA.Free;
  inherited;
end;

{ ---- TTest ---- }

constructor TTest.Create(const AName: string);
begin
  inherited Create;
  FName := AName;
  Writeln('=== ', FName, ' ===');
end;

procedure TTest.Group(const Msg: string);
begin
  Writeln(Msg);
end;

procedure TTest.Check(Cond: Boolean; const Msg: string);
begin
  Inc(FRun);
  if not Cond then
  begin
    Inc(FFailed);
    Writeln('  FAIL: ', Msg);
  end;
end;

procedure TTest.CheckEqual(A, B: Integer; const Msg: string);
begin
  Check(A = B, Format('%s (got %d, want %d)', [Msg, A, B]));
end;

procedure TTest.CheckEqualStr(const A, B: string; const Msg: string);
begin
  Check(A = B, Format('%s (got %s, want %s)', [Msg, A, B]));
end;

procedure TTest.Summary;
begin
  Writeln;
  Writeln(Format('%d tests run, %d failed', [FRun, FFailed]));
end;

function TTest.ExitCode: Integer;
begin
  if FFailed > 0 then Result := 1 else Result := 0;
end;

{ ---- TMemProvider ---- }

constructor TMemProvider.Create;
begin
  inherited Create;
  FStagingMap := TStringList.Create;
  FStagingMap.Sorted := True;
  FStagingMap.Duplicates := dupError;
  FPartialNames := TStringList.Create;
  FPartialNames.CaseSensitive := False;
end;

destructor TMemProvider.Destroy;
var
  I: Integer;
begin
  for I := 0 to High(FOutStreams) do
    if FOutStreams[I] <> nil then FOutStreams[I].Free;
  for I := 0 to High(FStagingStreams) do
    if FStagingStreams[I] <> nil then FStagingStreams[I].Free;
  FStagingMap.Free;
  FPartialNames.Free;
  inherited;
end;

procedure TMemProvider.QueuePartial(const Name: string; const Data: TBytes);
var
  Idx: Integer;
begin
  Idx := Length(FPartialData);
  SetLength(FPartialData, Idx + 1);
  FPartialData[Idx] := Data;
  FPartialNames.AddObject(Name, TObject(PtrInt(Idx)));
end;

procedure TMemProvider.QueueSend(const Name: string;
  const Data: TBytes; Time_: LongInt);
var
  Idx: Integer;
begin
  Idx := Length(FOutbound);
  SetLength(FOutbound,   Idx + 1);
  SetLength(FOutStreams, Idx + 1);
  FOutbound[Idx].Name := Name;
  FOutbound[Idx].Data := Data;
  FOutbound[Idx].Time := Time_;
end;

function TMemProvider.NextOutbound(out Item: TCometSendItem): TCometSendDecision;
var
  MS: TMemoryStream;
  Digest: TSHA256Digest;
begin
  Item := Default(TCometSendItem);
  if FOutIdx >= Length(FOutbound) then
  begin
    Result := cmSendNone;
    Exit;
  end;

  MS := TMemoryStream.Create;
  if Length(FOutbound[FOutIdx].Data) > 0 then
    MS.WriteBuffer(FOutbound[FOutIdx].Data[0],
                   Length(FOutbound[FOutIdx].Data));
  MS.Position := 0;
  FOutStreams[FOutIdx] := MS;

  if Length(FOutbound[FOutIdx].Data) > 0 then
    Digest := SHA256Buffer(FOutbound[FOutIdx].Data[0],
                           Length(FOutbound[FOutIdx].Data))
  else
    Digest := SHA256Buffer(MS.Memory^, 0);

  Item.Name   := FOutbound[FOutIdx].Name;
  Item.Size   := Length(FOutbound[FOutIdx].Data);
  Item.Time   := FOutbound[FOutIdx].Time;
  Item.Stream := MS;
  Item.Cookie := Pointer(PtrInt(FOutIdx));
  Move(Digest, Item.SHA256, 32);

  Inc(FOutIdx);
  Result := cmSendFile;
end;

procedure TMemProvider.AcknowledgeSent(const Item: TCometSendItem);
var
  Idx: Integer;
begin
  Idx := Integer(PtrInt(Item.Cookie));
  if (Idx >= 0) and (Idx < Length(FOutStreams)) and
     (FOutStreams[Idx] <> nil) then
  begin
    FOutStreams[Idx].Free;
    FOutStreams[Idx] := nil;
  end;
  SetLength(FAcked, Length(FAcked) + 1);
  FAcked[High(FAcked)] := Item.Name;
end;

procedure TMemProvider.AcknowledgeFailed(const Item: TCometSendItem;
  Reason: TCometSendFailReason);
var
  Idx: Integer;
begin
  Idx := Integer(PtrInt(Item.Cookie));
  if (Idx >= 0) and (Idx < Length(FOutStreams)) and
     (FOutStreams[Idx] <> nil) then
  begin
    FOutStreams[Idx].Free;
    FOutStreams[Idx] := nil;
  end;
  SetLength(FFailed, Length(FFailed) + 1);
  FFailed[High(FFailed)] := Item.Name;
end;

function TMemProvider.OpenForReceive(const Name: string; Size: Int64;
  Time: LongInt; ResumeOffset: Int64; const OverrideDir: string;
  out TempPath: string): TStream;
var
  MS:       TMemoryStream;
  Idx, PI:  Integer;
  Partial:  TBytes;
begin
  MS := TMemoryStream.Create;
  if Size > 0 then
    MS.Size := Size;

  { If we pre-staged a partial for this name, copy those
    bytes into [0..ResumeOffset) so the engine sees a
    valid prefix when it seeds the SHA. }
  if ResumeOffset > 0 then
  begin
    PI := FPartialNames.IndexOf(Name);
    if PI >= 0 then
    begin
      Partial := FPartialData[Integer(PtrInt(FPartialNames.Objects[PI]))];
      if Length(Partial) >= ResumeOffset then
      begin
        MS.Position := 0;
        MS.WriteBuffer(Partial[0], ResumeOffset);
      end;
    end;
  end;
  MS.Position := ResumeOffset;

  Idx := Length(FStagingStreams);
  SetLength(FStagingStreams, Idx + 1);
  SetLength(FStagingNames,   Idx + 1);
  SetLength(FStagingTimes,   Idx + 1);
  FStagingStreams[Idx] := MS;
  FStagingNames[Idx]   := Name;
  FStagingTimes[Idx]   := Time;

  TempPath := Format('mem:%d', [Idx]);
  FStagingMap.AddObject(TempPath, TObject(PtrInt(Idx)));
  Result := MS;
end;

function TMemProvider.GetPartialSize(const Name: string;
  Time: LongInt): Int64;
var
  PI: Integer;
begin
  Result := 0;
  PI := FPartialNames.IndexOf(Name);
  if PI >= 0 then
    Result := Length(FPartialData[Integer(PtrInt(FPartialNames.Objects[PI]))]);
end;

function TMemProvider.FinalizeReceive(const TempPath, Name: string;
  Time: LongInt): string;
var
  I, Idx, R: Integer;
  MS: TMemoryStream;
begin
  Result := '';
  I := FStagingMap.IndexOf(TempPath);
  if I < 0 then Exit;
  Idx := Integer(PtrInt(FStagingMap.Objects[I]));
  MS  := FStagingStreams[Idx];

  R := Length(FReceived);
  SetLength(FReceived, R + 1);
  FReceived[R].Name := Name;
  FReceived[R].Time := Time;
  SetLength(FReceived[R].Data, MS.Size);
  if MS.Size > 0 then
  begin
    MS.Position := 0;
    MS.ReadBuffer(FReceived[R].Data[0], MS.Size);
  end;
  MS.Free;
  FStagingStreams[Idx] := nil;
  FStagingMap.Delete(I);
  Result := 'mem:' + Name;
end;

procedure TMemProvider.CleanupReceive(const TempPath: string;
  KeepForResume: Boolean);
var
  I, Idx: Integer;
begin
  I := FStagingMap.IndexOf(TempPath);
  if I < 0 then Exit;
  Idx := Integer(PtrInt(FStagingMap.Objects[I]));
  if FStagingStreams[Idx] <> nil then
  begin
    FStagingStreams[Idx].Free;
    FStagingStreams[Idx] := nil;
  end;
  FStagingMap.Delete(I);
end;

function TMemProvider.ReceivedCount: Integer;
begin Result := Length(FReceived); end;
function TMemProvider.ReceivedAt(I: Integer): TMemRecvFile;
begin Result := FReceived[I]; end;
function TMemProvider.AckedCount: Integer;
begin Result := Length(FAcked); end;
function TMemProvider.AckedAt(I: Integer): string;
begin Result := FAcked[I]; end;
function TMemProvider.FailedCount: Integer;
begin Result := Length(FFailed); end;
function TMemProvider.FailedAt(I: Integer): string;
begin Result := FFailed[I]; end;

end.