frostfs-sdk-csharp/src/FrostFS.SDK.Tests/SmokeTests.cs

using System.Security.Cryptography;
using FrostFS.SDK.ClientV2;
using FrostFS.SDK.ClientV2.Interfaces;
using FrostFS.SDK.ModelsV2;
using FrostFS.SDK.ModelsV2.Enums;
using FrostFS.SDK.ModelsV2.Netmap;
using FrostFS.SDK.Cryptography;

using Grpc.Core;
using Microsoft.Extensions.Options;
using Grpc.Core.Interceptors;
using System.Diagnostics;

using static FrostFS.Session.SessionToken.Types.Body;
using FrostFS.SDK.ClientV2.Parameters;
using FrostFS.SDK.Tests;

namespace FrostFS.SDK.SmokeTests;

public abstract class SmokeTestsBase
{
    protected readonly string key = "KwHDAJ66o8FoLBjVbjP2sWBmgBMGjt7Vv4boA7xQrBoAYBE397Aq";
    protected readonly string url = "http://172.23.32.4:8080";

    protected ECDsa Key { get; }

    protected OwnerId OwnerId { get; }

    protected ModelsV2.Version Version { get; }

    protected Context Ctx { get; }

    protected SmokeTestsBase()
    {
        Key = key.LoadWif();
        OwnerId = OwnerId.FromKey(Key);
        Version = new ModelsV2.Version(2, 13);

        Ctx = new Context { Key = Key, OwnerId = OwnerId, Version = Version };
    }
}

public class SmokeTests : SmokeTestsBase
{
    private static readonly PrmWait lightWait = new (100, 1);

    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public async void NetworkMapTest(bool isSingleOnwerClient)
    {
        using var client = isSingleOnwerClient ? Client.GetSingleOwnerInstance(GetSingleOwnerOptions(this.key, this.url)) : Client.GetInstance(GetOptions(this.url));
        
        PrmNetmapSnapshot? prm = isSingleOnwerClient ? default : new () { Context = Ctx };
        var result = await client.GetNetmapSnapshotAsync(prm);

        Assert.True(result.Epoch > 0);
        Assert.Single(result.NodeInfoCollection);

        var item = result.NodeInfoCollection[0];
        Assert.Equal(2, item.Version.Major);
        Assert.Equal(13, item.Version.Minor);
        Assert.Equal(NodeState.Online, item.State);
        Assert.True(item.PublicKey.Length > 0);
        Assert.Single(item.Addresses);
        Assert.Equal(9, item.Attributes.Count);
    }


    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public async void NodeInfoTest(bool isSingleOnwerClient)
    {
        using var client = isSingleOnwerClient ? Client.GetSingleOwnerInstance(GetSingleOwnerOptions(this.key, this.url)) : Client.GetInstance(GetOptions(this.url));

        PrmNodeInfo? prm = isSingleOnwerClient ? default : new() { Context = Ctx };

        var result = await client.GetNodeInfoAsync(prm);

        Assert.Equal(2, result.Version.Major);
        Assert.Equal(13, result.Version.Minor);
        Assert.Equal(NodeState.Online, result.State);
        Assert.Equal(33, result.PublicKey.Length);
        Assert.Single(result.Addresses);
        Assert.Equal(9, result.Attributes.Count);
    }

    [Fact]
    public async void NodeInfo_Statistics_Test()
    {
        var ctx = new Context
        {
            Callback = (cs) => Console.WriteLine($"{cs.MethodName} took {cs.ElapsedMicroSeconds} microseconds")
        };

        using var client = Client.GetSingleOwnerInstance(GetSingleOwnerOptions(this.key, this.url));

        var result = await client.GetNodeInfoAsync();       
    }

    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public async void GetSessionTest(bool isSingleOnwerClient)
    {
        using var client = isSingleOnwerClient ? Client.GetSingleOwnerInstance(GetSingleOwnerOptions(this.key, this.url)) : Client.GetInstance(GetOptions(this.url));

        PrmSessionCreate? prm = isSingleOnwerClient ? new PrmSessionCreate(100) : new PrmSessionCreate(100) { Context = Ctx };

        var token = await client.CreateSessionAsync(prm);

        var session = new Session.SessionToken().Deserialize(token.Token);

        var ownerHash = Base58.Decode(OwnerId.Value);
       
        Assert.NotNull(session);
        Assert.Null(session.Body.Container);
        Assert.Null(session.Body.Object);
        Assert.Equal(16, session.Body.Id.Length);
        Assert.Equal(100ul, session.Body.Lifetime.Exp);
        Assert.Equal(ownerHash, session.Body.OwnerId.Value);
        Assert.Equal(33, session.Body.SessionKey.Length);
        Assert.Equal(ContextOneofCase.None, session.Body.ContextCase);
    }

    [Fact]
    public async void CreateObjectWithSessionToken()
    {
        using var client = Client.GetSingleOwnerInstance(GetSingleOwnerOptions(this.key, this.url));
       
        await Cleanup(client);

        var token = await client.CreateSessionAsync(new PrmSessionCreate(int.MaxValue));

        var createContainerParam = new PrmContainerCreate(
            new ModelsV2.Container(BasicAcl.PublicRW, new PlacementPolicy(true, new Replica(1))));

        createContainerParam.XHeaders.Add("key1", "value1");

        var containerId = await client.CreateContainerAsync(createContainerParam);

        var bytes = GetRandomBytes(1024);

        var param = new PrmObjectPut
        {
            Header = new ObjectHeader(
                containerId: containerId,
                type: ModelsV2.Enums.ObjectType.Regular,
                new ObjectAttribute("fileName", "test")),
            Payload = new MemoryStream(bytes),
            ClientCut = false,
            SessionToken = token        
        };
 
        var objectId = await client.PutObjectAsync(param);

        var @object = await client.GetObjectAsync(new PrmObjectGet(containerId, objectId));

        var downloadedBytes = new byte[@object.Header.PayloadLength];
        MemoryStream ms = new(downloadedBytes);

        byte[]? chunk = null;
        while ((chunk = await @object.ObjectReader!.ReadChunk()) != null)
        {
            ms.Write(chunk);
        }

        Assert.Equal(MD5.HashData(bytes), MD5.HashData(downloadedBytes));

        await Cleanup(client);
    }

    [Fact]
    public async void FilterTest()
    {
        using var client = Client.GetSingleOwnerInstance(GetSingleOwnerOptions(this.key, this.url));

        await Cleanup(client);

        var createContainerParam = new PrmContainerCreate(
           new ModelsV2.Container(BasicAcl.PublicRW, new PlacementPolicy(true, new Replica(1))))
        {
            WaitParams = lightWait
        };

        var containerId = await client.CreateContainerAsync(createContainerParam);

        var bytes = new byte[] { 1, 2, 3 };

        var ParentHeader = new ObjectHeader(
                containerId: containerId,
                type: ModelsV2.Enums.ObjectType.Regular)
        {
            PayloadLength = 3
        };

        var param = new PrmObjectPut
        {
            Header = new ObjectHeader(
                containerId: containerId,
                type: ModelsV2.Enums.ObjectType.Regular,
                new ObjectAttribute("fileName", "test"))
            {
                Split = new Split(),
            },
            Payload = new MemoryStream(bytes),
            ClientCut = false
        };

        var objectId = await client.PutObjectAsync(param);

        var head = await client.GetObjectHeadAsync(new PrmObjectHeadGet(containerId, objectId));

        var ecdsaKey = this.key.LoadWif();

        var networkInfo = await client.GetNetmapSnapshotAsync();

        await CheckFilter(client, containerId, new FilterByContainerId(ObjectMatchType.Equals, containerId));

        await CheckFilter(client, containerId, new FilterByOwnerId(ObjectMatchType.Equals, OwnerId.FromKey(ecdsaKey)));

        await CheckFilter(client, containerId, new FilterBySplitId(ObjectMatchType.Equals, param.Header.Split.SplitId));

        await CheckFilter(client, containerId, new FilterByAttribute(ObjectMatchType.Equals, "fileName", "test"));

        await CheckFilter(client, containerId, new FilterByObjectId(ObjectMatchType.Equals, objectId));

        await CheckFilter(client, containerId, new FilterByVersion(ObjectMatchType.Equals, networkInfo.NodeInfoCollection[0].Version));

        await CheckFilter(client, containerId, new FilterByEpoch(ObjectMatchType.Equals, networkInfo.Epoch));

        await CheckFilter(client, containerId, new FilterByPayloadLength(ObjectMatchType.Equals, 3));

        var checkSum = CheckSum.CreateCheckSum(bytes);

        await CheckFilter(client, containerId, new FilterByPayloadHash(ObjectMatchType.Equals, checkSum));

        await CheckFilter(client, containerId, new FilterByPhysicallyStored());
    }

    private static async Task CheckFilter(IFrostFSClient client, ContainerId containerId, IObjectFilter filter)
    {
        var resultObjectsCount = 0;

        PrmObjectSearch searchParam = new(containerId) { Filters = [filter] };

        await foreach (var objId in client.SearchObjectsAsync(searchParam))
        {
            resultObjectsCount++;
            var objHeader = await client.GetObjectHeadAsync(new PrmObjectHeadGet(containerId, objId));
        }

        Assert.True(0 < resultObjectsCount, $"Filter for {filter.Key} doesn't work");
    }

    [Theory]
    [InlineData(1)]
    [InlineData(3 * 1024 * 1024)]            // exactly one chunk size - 3MB
    [InlineData(6 * 1024 * 1024 + 100)]
    public async void SimpleScenarioTest(int objectSize)
    {
        using var client = Client.GetSingleOwnerInstance(GetSingleOwnerOptions(this.key, this.url));
       
        await Cleanup(client);

        bool callbackInvoked = false;
        var ctx = new Context
        {
           // Timeout = TimeSpan.FromSeconds(20),
            Callback = new((CallStatistics cs) =>
            {
                callbackInvoked = true;
                Assert.True(cs.ElapsedMicroSeconds > 0);
            })
        };

        var createContainerParam = new PrmContainerCreate(
            new ModelsV2.Container(BasicAcl.PublicRW, new PlacementPolicy(true, new Replica(1))))
        {
            Context = ctx
        };

        var containerId = await client.CreateContainerAsync(createContainerParam);

        var container = await client.GetContainerAsync(new PrmContainerGet(containerId) { Context = ctx });
        Assert.NotNull(container);
        Assert.True(callbackInvoked);

        var bytes = GetRandomBytes(objectSize);

        var param = new PrmObjectPut
        {
            Header = new ObjectHeader(
                containerId: containerId,
                type: ModelsV2.Enums.ObjectType.Regular,
                new ObjectAttribute("fileName", "test")),
            Payload = new MemoryStream(bytes),
            ClientCut = false,
            Context = new Context
            {
                Callback = new((CallStatistics cs) => Assert.True(cs.ElapsedMicroSeconds > 0))
            }
        };

        var objectId = await client.PutObjectAsync(param);

        var filter = new FilterByAttribute(ObjectMatchType.Equals, "fileName", "test");

        bool hasObject = false;
        await foreach (var objId in client.SearchObjectsAsync(new PrmObjectSearch(containerId) { Filters = [filter] }))
        {
            hasObject = true;

            var objHeader = await client.GetObjectHeadAsync(new PrmObjectHeadGet(containerId, objectId));
            Assert.Equal((ulong)bytes.Length, objHeader.PayloadLength);
            Assert.Single(objHeader.Attributes);
            Assert.Equal("fileName", objHeader.Attributes.First().Key);
            Assert.Equal("test", objHeader.Attributes.First().Value);
        }

        Assert.True(hasObject);

        var @object = await client.GetObjectAsync(new PrmObjectGet(containerId, objectId));

        var downloadedBytes = new byte[@object.Header.PayloadLength];
        MemoryStream ms = new(downloadedBytes);

        byte[]? chunk = null;
        while ((chunk = await @object.ObjectReader!.ReadChunk()) != null)
        {
            ms.Write(chunk);
        }

        Assert.Equal(MD5.HashData(bytes), MD5.HashData(downloadedBytes));

        await Cleanup(client);

        await foreach (var _ in client.ListContainersAsync())
        {
            Assert.Fail("Containers exist");
        }
    }

    [Theory]
    [InlineData(1)]
    [InlineData(3 * 1024 * 1024)]            // exactly one chunk size - 3MB
    [InlineData(6 * 1024 * 1024 + 100)]
    public async void SimpleScenarioWithSessionTest(int objectSize)
    {
        using var client = Client.GetSingleOwnerInstance(GetSingleOwnerOptions(this.key, this.url));

        var token = await client.CreateSessionAsync(new PrmSessionCreate(int.MaxValue));

        await Cleanup(client);

        var ctx = new Context
        {
            Timeout = TimeSpan.FromSeconds(20),
            Callback = new((CallStatistics cs) => Assert.True(cs.ElapsedMicroSeconds > 0))
        };

        var createContainerParam = new PrmContainerCreate(
            new ModelsV2.Container(BasicAcl.PublicRW, new PlacementPolicy(true, new Replica(1))))
        {
            Context = ctx
        };

        var containerId = await client.CreateContainerAsync(createContainerParam);

        var container = await client.GetContainerAsync(new PrmContainerGet(containerId) { Context = ctx });
        Assert.NotNull(container);

        var bytes = GetRandomBytes(objectSize);

        var param = new PrmObjectPut
        {
            Header = new ObjectHeader(
                containerId: containerId,
                type: ModelsV2.Enums.ObjectType.Regular,
                new ObjectAttribute("fileName", "test")),
            Payload = new MemoryStream(bytes),
            ClientCut = false,
            Context = new Context
            {
                Callback = new((CallStatistics cs) => Assert.True(cs.ElapsedMicroSeconds > 0))
            },
            SessionToken = token
        };

        var objectId = await client.PutObjectAsync(param);

        var filter = new FilterByAttribute(ObjectMatchType.Equals, "fileName", "test");

        bool hasObject = false;
        await foreach (var objId in client.SearchObjectsAsync(new PrmObjectSearch(containerId) { Filters = [filter], SessionToken = token }))
        {
            hasObject = true;

            var objHeader = await client.GetObjectHeadAsync(new PrmObjectHeadGet(containerId, objectId) { SessionToken = token });
            Assert.Equal((ulong)bytes.Length, objHeader.PayloadLength);
            Assert.Single(objHeader.Attributes);
            Assert.Equal("fileName", objHeader.Attributes.First().Key);
            Assert.Equal("test", objHeader.Attributes.First().Value);
        }

        Assert.True(hasObject);

        var @object = await client.GetObjectAsync(new PrmObjectGet(containerId, objectId) { SessionToken = token });

        var downloadedBytes = new byte[@object.Header.PayloadLength];
        MemoryStream ms = new(downloadedBytes);

        byte[]? chunk = null;
        while ((chunk = await @object.ObjectReader!.ReadChunk()) != null)
        {
            ms.Write(chunk);
        }

        Assert.Equal(MD5.HashData(bytes), MD5.HashData(downloadedBytes));

        await Cleanup(client);

        await foreach (var _ in client.ListContainersAsync())
        {
            Assert.Fail("Containers exist");
        }
    }

    [Theory]
    [InlineData(1)]
    [InlineData(64 * 1024 * 1024)]      // exactly 1 block size - 64MB
    [InlineData(64 * 1024 * 1024 - 1)]
    [InlineData(64 * 1024 * 1024 + 1)]
    [InlineData(2 * 64 * 1024 * 1024 + 256)]
    [InlineData(200)]
    public async void ClientCutScenarioTest(int objectSize)
    {        
        using var client = Client.GetSingleOwnerInstance(GetSingleOwnerOptions(this.key, this.url));

        await Cleanup(client);

        var createContainerParam = new PrmContainerCreate(new ModelsV2.Container(BasicAcl.PublicRW, new PlacementPolicy(true, new Replica(1))))
        {
            WaitParams = lightWait
        };

        var containerId = await client.CreateContainerAsync(createContainerParam);

        var ctx = new Context
        {
            Timeout = TimeSpan.FromSeconds(10),
            Interceptors = new([new MetricsInterceptor()])
        };

        var container = await client.GetContainerAsync(new PrmContainerGet(containerId) { Context = ctx });

        Assert.NotNull(container);

        byte[] bytes = GetRandomBytes(objectSize);

        var param = new PrmObjectPut
        {
            Header = new ObjectHeader(
                containerId: containerId,
                type: ModelsV2.Enums.ObjectType.Regular,
                new ObjectAttribute("fileName", "test")),
            Payload = new MemoryStream(bytes),
            ClientCut = true
        };

        var objectId = await client.PutObjectAsync(param);

        var filter = new FilterByAttribute(ObjectMatchType.Equals, "fileName", "test");

        bool hasObject = false;
        await foreach (var objId in client.SearchObjectsAsync(new PrmObjectSearch(containerId, filter)))
        {
            hasObject = true;

            var objHeader = await client.GetObjectHeadAsync(new PrmObjectHeadGet(containerId, objectId));
            Assert.Equal((ulong)bytes.Length, objHeader.PayloadLength);
            Assert.Single(objHeader.Attributes);
            Assert.Equal("fileName", objHeader.Attributes[0].Key);
            Assert.Equal("test", objHeader.Attributes[0].Value);
        }

        Assert.True(hasObject);

        var @object = await client.GetObjectAsync(new PrmObjectGet(containerId, objectId));

        var downloadedBytes = new byte[@object.Header.PayloadLength];
        MemoryStream ms = new(downloadedBytes);

        byte[]? chunk = null;
        while ((chunk = await @object.ObjectReader!.ReadChunk()) != null)
        {
            ms.Write(chunk);
        }

        Assert.Equal(MD5.HashData(bytes), MD5.HashData(downloadedBytes));

        await CheckFilter(client, containerId, new FilterByRootObject());

        await Cleanup(client);

        var deadline = DateTime.UtcNow.Add(TimeSpan.FromSeconds(5));

        IAsyncEnumerator<ContainerId>? enumerator = null;
        do
        {
           if (deadline <= DateTime.UtcNow)
           { 
               Assert.Fail("Containers exist");
               break;
           }

           enumerator = client.ListContainersAsync().GetAsyncEnumerator();
           await Task.Delay(500);        
        }
        while (await enumerator!.MoveNextAsync());           
    }

    private static byte[] GetRandomBytes(int size)
    {
        Random rnd = new();
        var bytes = new byte[size];
        rnd.NextBytes(bytes);
        return bytes;
    }

    private static IOptions<SingleOwnerClientSettings> GetSingleOwnerOptions(string key, string url)
    {
        return Options.Create(new SingleOwnerClientSettings
        {
            Key = key,
            Host = url
        });
    }

    private static IOptions<ClientSettings> GetOptions(string url)
    {
        return Options.Create(new ClientSettings
        {
            Host = url
        });
    }

    static async Task Cleanup(IFrostFSClient client)
    {
        await foreach (var cid in client.ListContainersAsync())
        {
            await client.DeleteContainerAsync(new PrmContainerDelete(cid) { WaitParams = lightWait });
        }
    }
}


public class MetricsInterceptor() : Interceptor
{
    public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
        TRequest request,
        ClientInterceptorContext<TRequest, TResponse> context,
        AsyncUnaryCallContinuation<TRequest, TResponse> continuation)
    {
        var call = continuation(request, context);

        return new AsyncUnaryCall<TResponse>(
            HandleUnaryResponse(call),
            call.ResponseHeadersAsync,
            call.GetStatus,
            call.GetTrailers,
            call.Dispose);
    }
  
    private static async Task<TResponse> HandleUnaryResponse<TResponse>(AsyncUnaryCall<TResponse> call)
    {
        var watch = new Stopwatch();
        watch.Start();
      
        var response = await call.ResponseAsync;

        watch.Stop();

        // Do something with call info
        // var elapsed = watch.ElapsedTicks * 1_000_000/Stopwatch.Frequency;

        return response;    
    }
}