Revision control
Copy as Markdown
// Copyright (C) 2022 The Qt Company Ltd.
// Copyright (C) 2019 Alexey Edelev <semlanik@gmail.com>
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR GPL-3.0-only
#include <grpcclienttestbase.h>
#include <QtGrpc/QGrpcCallOptions>
#include <QtCore/QCryptographicHash>
#include <QtCore/QFile>
#include <QtCore/QThread>
#include <QtCore/QTimer>
#include <QtTest/QSignalSpy>
#include <QtTest/QTest>
#include <message_latency_defs.h>
#include <testservice_client.grpc.qpb.h>
using namespace Qt::Literals::StringLiterals;
using namespace qtgrpc::tests;
class QtGrpcClientServerStreamTest : public GrpcClientTestBase
{
Q_OBJECT
public:
QtGrpcClientServerStreamTest()
: GrpcClientTestBase(
Channels{ GrpcClientTestBase::Channel::Qt })
{
}
private Q_SLOTS:
void valid();
void cancel();
void deferredCancel();
void hugeBlob();
void multipleStreams();
void multipleStreamsCancel();
void cancelWhileErrorTimeout();
};
void QtGrpcClientServerStreamTest::valid()
{
const int ExpectedMessageCount = 4;
SimpleStringMessage result;
SimpleStringMessage request;
request.setTestFieldString("Stream");
auto stream = client()->testMethodServerStream(request);
auto *streamPtr = stream.get();
QSignalSpy messageReceivedSpy(streamPtr, &QGrpcServerStream::messageReceived);
QVERIFY(messageReceivedSpy.isValid());
QSignalSpy streamFinishedSpy(streamPtr, &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
QObject::connect(streamPtr, &QGrpcServerStream::messageReceived, this,
[&result, stream = std::move(stream)] {
const auto ret = stream->read<SimpleStringMessage>();
QVERIFY(ret.has_value());
result.setTestFieldString(result.testFieldString()
+ ret->testFieldString());
});
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1,
MessageLatencyWithThreshold * ExpectedMessageCount);
auto args = streamFinishedSpy.first();
QCOMPARE(args.count(), 1);
QCOMPARE_EQ(args.first().value<QGrpcStatus>().code(), QtGrpc::StatusCode::Ok);
QCOMPARE(messageReceivedSpy.count(), ExpectedMessageCount);
QCOMPARE_EQ(result.testFieldString(), "Stream1Stream2Stream3Stream4");
}
void QtGrpcClientServerStreamTest::cancel()
{
const int ExpectedMessageCount = 2;
SimpleStringMessage result;
SimpleStringMessage request;
request.setTestFieldString("Stream");
auto stream = client()->testMethodServerStream(request);
QSignalSpy streamFinishedSpy(stream.get(), &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
int i = 0;
QObject::connect(stream.get(), &QGrpcServerStream::messageReceived, this, [&] {
const auto ret = stream->read<SimpleStringMessage>();
QVERIFY(ret.has_value());
result.setTestFieldString(result.testFieldString() + ret->testFieldString());
if (++i == ExpectedMessageCount)
stream->cancel();
});
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1,
MessageLatencyWithThreshold * ExpectedMessageCount);
auto args = streamFinishedSpy.first();
QCOMPARE(args.count(), 1);
QVERIFY(args.first().value<QGrpcStatus>() == QtGrpc::StatusCode::Cancelled);
QCOMPARE(i, 2);
QCOMPARE_EQ(result.testFieldString(), "Stream1Stream2");
}
void QtGrpcClientServerStreamTest::deferredCancel()
{
const int ExpectedMessageCount = 3;
SimpleStringMessage result;
SimpleStringMessage request;
request.setTestFieldString("Stream");
auto stream = client()->testMethodServerStream(request);
QSignalSpy streamFinishedSpy(stream.get(), &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
QSignalSpy messageReceivedSpy(stream.get(), &QGrpcServerStream::messageReceived);
QVERIFY(messageReceivedSpy.isValid());
int i = 0;
QObject::connect(stream.get(), &QGrpcServerStream::messageReceived, this, [&] {
const auto ret = stream->read<SimpleStringMessage>();
QVERIFY(ret.has_value());
result.setTestFieldString(result.testFieldString() + ret->testFieldString());
if (++i == ExpectedMessageCount)
QTimer::singleShot(MessageLatencyThreshold, stream.get(), &QGrpcServerStream::cancel);
});
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1,
MessageLatencyWithThreshold * ExpectedMessageCount);
auto args = streamFinishedSpy.first();
QCOMPARE(args.count(), 1);
QVERIFY(args.first().value<QGrpcStatus>() == QtGrpc::StatusCode::Cancelled);
QCOMPARE(messageReceivedSpy.count(), ExpectedMessageCount);
QCOMPARE_EQ(result.testFieldString(), "Stream1Stream2Stream3");
}
void QtGrpcClientServerStreamTest::hugeBlob()
{
BlobMessage result;
BlobMessage request;
QFile testFile(":/assets/testfile");
QVERIFY(testFile.open(QFile::ReadOnly));
request.setTestBytes(testFile.readAll());
QByteArray dataHash = QCryptographicHash::hash(request.testBytes(), QCryptographicHash::Sha256);
auto stream = client()->testMethodBlobServerStream(request);
auto *streamPtr = stream.get();
QSignalSpy streamFinishedSpy(streamPtr, &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
QObject::connect(streamPtr, &QGrpcServerStream::messageReceived, this,
[&result, stream = streamPtr] {
const auto ret = stream->read<BlobMessage>();
QVERIFY(ret.has_value());
result.setTestBytes(ret->testBytes());
});
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1, MessageLatencyWithThreshold);
auto args = streamFinishedSpy.first();
QCOMPARE(args.count(), 1);
QCOMPARE_EQ(args.first().value<QGrpcStatus>().code(), QtGrpc::StatusCode::Ok);
QVERIFY(!result.testBytes().isEmpty());
QByteArray returnDataHash =
QCryptographicHash::hash(result.testBytes(), QCryptographicHash::Sha256);
QCOMPARE_EQ(returnDataHash, dataHash);
}
void QtGrpcClientServerStreamTest::multipleStreams()
{
const int ExpectedMessageCount = 4;
SimpleStringMessage result;
SimpleStringMessage request;
request.setTestFieldString("Stream");
auto stream = client()->testMethodServerStream(request);
auto *streamPtr = stream.get();
// Ensure we're not reusing streams
QCOMPARE_NE(streamPtr, client()->testMethodServerStream(request).get());
QSignalSpy streamFinishedSpy(streamPtr, &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
QSignalSpy steamMessageRecievedSpy(streamPtr, &QGrpcServerStream::messageReceived);
QVERIFY(steamMessageRecievedSpy.isValid());
QObject::connect(streamPtr, &QGrpcServerStream::messageReceived, this,
[&result, stream = std::move(stream)] {
const auto ret = stream->read<SimpleStringMessage>();
QVERIFY(ret.has_value());
result.setTestFieldString(result.testFieldString()
+ ret->testFieldString());
});
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1,
MessageLatencyWithThreshold * ExpectedMessageCount);
auto args = streamFinishedSpy.first();
QCOMPARE(args.count(), 1);
QCOMPARE_EQ(args.first().value<QGrpcStatus>().code(), QtGrpc::StatusCode::Ok);
QCOMPARE_EQ(steamMessageRecievedSpy.count(), ExpectedMessageCount);
QCOMPARE_EQ(result.testFieldString(), "Stream1Stream2Stream3Stream4");
}
void QtGrpcClientServerStreamTest::multipleStreamsCancel()
{
SimpleStringMessage result;
SimpleStringMessage request;
request.setTestFieldString("Stream");
auto stream = client()->testMethodServerStream(request);
auto streamNext = client()->testMethodServerStream(request);
QCOMPARE_NE(stream, streamNext);
QSignalSpy streamFinishedSpy(stream.get(), &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
QSignalSpy streamNextFinishedSpy(streamNext.get(), &QGrpcServerStream::finished);
QVERIFY(streamNextFinishedSpy.isValid());
streamNext->cancel();
QCOMPARE(streamFinishedSpy.count(), 0);
QCOMPARE(streamNextFinishedSpy.count(), 1);
auto args = streamNextFinishedSpy.first();
QCOMPARE(args.count(), 1);
QVERIFY(args.first().value<QGrpcStatus>() == QtGrpc::StatusCode::Cancelled);
stream = client()->testMethodServerStream(request);
QCOMPARE_NE(stream, streamNext);
streamNext = client()->testMethodServerStream(request);
QCOMPARE_NE(stream, streamNext);
QSignalSpy otherStreamFinishedSpy(stream.get(), &QGrpcServerStream::finished);
QVERIFY(otherStreamFinishedSpy.isValid());
QSignalSpy otherStreamNextFinishedSpy(streamNext.get(), &QGrpcServerStream::finished);
QVERIFY(otherStreamNextFinishedSpy.isValid());
stream->cancel();
QCOMPARE(otherStreamFinishedSpy.count(), 1);
args = otherStreamFinishedSpy.first();
QCOMPARE(args.count(), 1);
QVERIFY(args.first().value<QGrpcStatus>() == QtGrpc::StatusCode::Cancelled);
const int ExpectedMessageCount = 4;
QTRY_COMPARE_EQ_WITH_TIMEOUT(otherStreamNextFinishedSpy.count(), 1,
MessageLatencyWithThreshold * ExpectedMessageCount);
args = otherStreamNextFinishedSpy.first();
QCOMPARE(args.count(), 1);
QCOMPARE_EQ(args.first().value<QGrpcStatus>().code(), QtGrpc::StatusCode::Ok);
}
void QtGrpcClientServerStreamTest::cancelWhileErrorTimeout()
{
SimpleStringMessage result;
SimpleStringMessage request;
request.setTestFieldString("Stream");
auto stream = client()->testMethodServerStream(request);
QSignalSpy streamFinishedSpy(stream.get(), &QGrpcServerStream::finished);
QVERIFY(streamFinishedSpy.isValid());
stream->cancel();
stream.reset();
QTRY_COMPARE_EQ_WITH_TIMEOUT(streamFinishedSpy.count(), 1, MessageLatencyWithThreshold);
auto args = streamFinishedSpy.first();
QCOMPARE(args.count(), 1);
QVERIFY(args.first().value<QGrpcStatus>() == QtGrpc::StatusCode::Cancelled);
}
QTEST_MAIN(QtGrpcClientServerStreamTest)
#include "tst_grpc_client_serverstream.moc"