aos/events/event_loop_param_test.h - RealtimeRoboticsGroup/test - Gitiles

 #ifndef _AOS_EVENTS_EVENT_LOOP_PARAM_TEST_H_
 #define _AOS_EVENTS_EVENT_LOOP_PARAM_TEST_H_

 #include <initializer_list>
 #include <string_view>
 #include <vector>

 #include "gtest/gtest.h"

 #include "aos/events/event_loop.h"
 #include "aos/events/test_message_generated.h"
 #include "aos/events/test_message_schema.h"
 #include "aos/events/timing_report_schema.h"
 #include "aos/flatbuffers.h"
 #include "aos/json_to_flatbuffer.h"
 #include "aos/logging/log_message_schema.h"
 #include "aos/network/message_bridge_client_schema.h"
 #include "aos/network/message_bridge_server_schema.h"
 #include "aos/network/timestamp_schema.h"

 namespace aos::testing {

 class EventLoopTestFactory {
  public:
   EventLoopTestFactory()
       : flatbuffer_(configuration::AddSchema(
             R"config({
   "channels": [
     {
       "name": "/aos",
       "type": "aos.logging.LogMessageFbs"
     },
     {
       "name": "/aos",
       "type": "aos.timing.Report"
     },
     {
       "name": "/test",
       "type": "aos.TestMessage",
       "frequency": 800
     },
     {
       "name": "/test1",
       "type": "aos.TestMessage"
     },
     {
       "name": "/test2",
       "type": "aos.TestMessage"
     },
     {
       "name": "/test3",
       "type": "aos.TestMessage",
       "channel_storage_duration": 10000000000
     }
   ]
 })config",
             {aos::FlatbufferSpan<reflection::Schema>(
                  logging::LogMessageFbsSchema()),
              aos::FlatbufferSpan<reflection::Schema>(timing::ReportSchema()),
              aos::FlatbufferSpan<reflection::Schema>(TestMessageSchema())})) {}

   virtual ~EventLoopTestFactory() {}

   // Makes a connected event loop.
   virtual std::unique_ptr<EventLoop> Make(std::string_view name) = 0;
   // Makes a primary event loop.  This is the one the tests will try to use for
   // anything blocking.
   virtual std::unique_ptr<EventLoop> MakePrimary(std::string_view name) = 0;

   // Runs the loops until they quit.
   virtual Result<void> Run() = 0;

   virtual std::unique_ptr<ExitHandle> MakeExitHandle() = 0;

   // Quits the loops.
   virtual void Exit() = 0;

   // Advances time by sleeping.  Can't be called from inside a loop.
   virtual void SleepFor(::std::chrono::nanoseconds duration) = 0;

   // Sets the config to a config with a max size with an invalid alignment.
   void InvalidChannelAlignment() {
     flatbuffer_ = configuration::AddSchema(
         R"config({
   "channels": [
     {
       "name": "/aos",
       "type": "aos.logging.LogMessageFbs"
     },
     {
       "name": "/aos",
       "type": "aos.timing.Report"
     },
     {
       "name": "/test",
       "type": "aos.TestMessage",
       "max_size": 13
     },
     {
       "name": "/test1",
       "type": "aos.TestMessage"
     },
     {
       "name": "/test2",
       "type": "aos.TestMessage"
     },
     {
       "name": "/test3",
       "type": "aos.TestMessage",
       "channel_storage_duration": 10000000000
     }
   ]
 })config",
         {aos::FlatbufferSpan<reflection::Schema>(
              logging::LogMessageFbsSchema()),
          aos::FlatbufferSpan<reflection::Schema>(timing::ReportSchema()),
          aos::FlatbufferSpan<reflection::Schema>(TestMessageSchema())});
   }

   void PinReads() {
     static const std::string kJson = R"config({
   "channels": [
     {
       "name": "/aos",
       "type": "aos.logging.LogMessageFbs",
       "read_method": "PIN",
       "num_readers": 10
     },
     {
       "name": "/aos",
       "type": "aos.timing.Report",
       "read_method": "PIN",
       "num_readers": 10
     },
     {
       "name": "/test",
       "type": "aos.TestMessage",
       "read_method": "PIN",
       "num_readers": 10,
       "frequency": 800
     },
     {
       "name": "/test1",
       "type": "aos.TestMessage",
       "read_method": "PIN",
       "num_readers": 10
     },
     {
       "name": "/test2",
       "type": "aos.TestMessage",
       "read_method": "PIN",
       "num_readers": 10
     },
     {
       "name": "/test3",
       "type": "aos.TestMessage",
       "read_method": "PIN",
       "num_readers": 10,
       "channel_storage_duration": 10000000000
     }
   ]
 })config";

     flatbuffer_ = configuration::AddSchema(
         kJson, {aos::FlatbufferSpan<reflection::Schema>(
                     logging::LogMessageFbsSchema()),
                 aos::FlatbufferSpan<reflection::Schema>(timing::ReportSchema()),
                 aos::FlatbufferSpan<reflection::Schema>(TestMessageSchema())});
   }

   void EnableNodes(std::string_view my_node) {
     static const std::string kJson = R"config({
   "channels": [
     {
       "name": "/me/aos",
       "type": "aos.logging.LogMessageFbs",
       "source_node": "me"
     },
     {
       "name": "/them/aos",
       "type": "aos.logging.LogMessageFbs",
       "source_node": "them"
     },
     {
       "name": "/me/aos",
       "type": "aos.message_bridge.Timestamp",
       "source_node": "me",
       "destination_nodes": [
         {
           "name": "them"
         }
       ]
     },
     {
       "name": "/them/aos",
       "type": "aos.message_bridge.Timestamp",
       "source_node": "them",
       "destination_nodes": [
         {
           "name": "me"
         }
       ]
     },
     {
       "name": "/me/aos",
       "type": "aos.message_bridge.ServerStatistics",
       "source_node": "me",
       "frequency": 2
     },
     {
       "name": "/them/aos",
       "type": "aos.message_bridge.ServerStatistics",
       "source_node": "them",
       "frequency": 2
     },
     {
       "name": "/me/aos",
       "type": "aos.message_bridge.ClientStatistics",
       "source_node": "me",
       "frequency": 2
     },
     {
       "name": "/them/aos",
       "type": "aos.message_bridge.ClientStatistics",
       "source_node": "them",
       "frequency": 2
     },
     {
       "name": "/aos",
       "type": "aos.timing.Report",
       "source_node": "me"
     },
     {
       "name": "/test",
       "type": "aos.TestMessage",
       "source_node": "me"
     },
     {
       "name": "/test1",
       "type": "aos.TestMessage",
       "source_node": "me"
     },
     {
       "name": "/test2",
       "type": "aos.TestMessage",
       "source_node": "me"
     },
     {
       "name": "/test_forward",
       "type": "aos.TestMessage",
       "source_node": "them",
       "destination_nodes": [
         {
           "name": "me"
         }
       ]
     },
     {
       "name": "/test_noforward",
       "type": "aos.TestMessage",
       "source_node": "them"
     }
   ],
   "nodes": [
     {
       "name": "me",
       "hostname": "myhostname"
     },
     {
       "name": "them",
       "hostname": "themhostname"
     }
   ],
   "maps": [
     {
       "match": {
         "name": "/aos*",
         "source_node": "me"
       },
       "rename": {
         "name": "/me/aos"
       }
     },
     {
       "match": {
         "name": "/aos*",
         "source_node": "them"
       },
       "rename": {
         "name": "/them/aos"
       }
     }
   ]
 })config";

     flatbuffer_ = configuration::MergeConfiguration(
         configuration::MergeConfiguration(
             aos::FlatbufferDetachedBuffer<Configuration>(
                 JsonToFlatbuffer<Configuration>(kJson))),
         {aos::FlatbufferSpan<reflection::Schema>(
              logging::LogMessageFbsSchema()),
          aos::FlatbufferSpan<reflection::Schema>(timing::ReportSchema()),
          aos::FlatbufferSpan<reflection::Schema>(TestMessageSchema()),
          aos::FlatbufferSpan<reflection::Schema>(
              message_bridge::ClientStatisticsSchema()),
          aos::FlatbufferSpan<reflection::Schema>(
              message_bridge::ServerStatisticsSchema()),
          aos::FlatbufferSpan<reflection::Schema>(
              message_bridge::TimestampSchema())});

     my_node_ = configuration::GetNode(&flatbuffer_.message(), my_node);
   }

   const Node *my_node() const { return my_node_; }

   const Configuration *configuration() { return &flatbuffer_.message(); }

  private:
   FlatbufferDetachedBuffer<Configuration> flatbuffer_;

   const Node *my_node_ = nullptr;
 };

 enum class DoTimingReports { kYes, kNo };

 class AbstractEventLoopTest
     : public ::testing::TestWithParam<
           std::tuple<std::function<EventLoopTestFactory *()>, ReadMethod,
                      DoTimingReports>> {
  public:
   AbstractEventLoopTest() : factory_(std::get<0>(GetParam())()) {
     if (read_method() == ReadMethod::PIN) {
       factory_->PinReads();
     }
   }

   ReadMethod read_method() const { return std::get<1>(GetParam()); }
   DoTimingReports do_timing_reports() const { return std::get<2>(GetParam()); }

   ::std::unique_ptr<EventLoop> Make(std::string_view name = "");

   ::std::unique_ptr<EventLoop> MakePrimary(std::string_view name = "primary") {
     ++event_loop_count_;
     auto result = factory_->MakePrimary(name);
     if (do_timing_reports() == DoTimingReports::kNo) {
       result->SkipTimingReport();
     }
     return result;
   }

   void InvalidChannelAlignment() { factory_->InvalidChannelAlignment(); }

   void EnableNodes(std::string_view my_node) { factory_->EnableNodes(my_node); }

   Result<void> Run() { return factory_->Run(); }

   std::unique_ptr<ExitHandle> MakeExitHandle() {
     return factory_->MakeExitHandle();
   }

   void Exit() { return factory_->Exit(); }

   void SleepFor(::std::chrono::nanoseconds duration) {
     return factory_->SleepFor(duration);
   }

   const Configuration *configuration() { return factory_->configuration(); }

   const Node *my_node() const { return factory_->my_node(); }

   // Ends the given event loop at the given time from now.
   void EndEventLoop(EventLoop *loop, ::std::chrono::milliseconds duration) {
     auto end_timer = loop->AddTimer([this]() { this->Exit(); });
     end_timer->Schedule(loop->monotonic_now() + duration);
     end_timer->set_name("end");
   }

   // Verifies that the buffer_index values for all of the given objects are
   // consistent.
   void VerifyBuffers(
       int number_buffers,
       std::vector<std::reference_wrapper<const Fetcher<TestMessage>>> fetchers,
       std::vector<std::reference_wrapper<const Sender<TestMessage>>> senders);

   // Helper function for testing the sent too fast check using a PhasedLoop with
   // an interval that sends exactly at the frequency of the channel
   void TestSentTooFastCheckEdgeCase(
       const std::function<RawSender::Error(int, int)> expected_err,
       const bool send_twice_at_end);

  private:
   const ::std::unique_ptr<EventLoopTestFactory> factory_;

   int event_loop_count_ = 0;
 };

 using AbstractEventLoopDeathTest = AbstractEventLoopTest;

 // Returns the frequency of the /test TestMessage channel
 int TestChannelFrequency(EventLoop *event_loop);
 // Returns the queue size of the /test TestMessage channel
 int TestChannelQueueSize(EventLoop *event_loop);
 // Sends a test message with value 0 with the given sender
 RawSender::Error SendTestMessage(aos::Sender<TestMessage> &sender);

 }  // namespace aos::testing

 #endif  // _AOS_EVENTS_EVENT_LOOP_PARAM_TEST_H_
	#ifndef _AOS_EVENTS_EVENT_LOOP_PARAM_TEST_H_
	#define _AOS_EVENTS_EVENT_LOOP_PARAM_TEST_H_

	#include <initializer_list>
	#include <string_view>
	#include <vector>

	#include "gtest/gtest.h"

	#include "aos/events/event_loop.h"
	#include "aos/events/test_message_generated.h"
	#include "aos/events/test_message_schema.h"
	#include "aos/events/timing_report_schema.h"
	#include "aos/flatbuffers.h"
	#include "aos/json_to_flatbuffer.h"
	#include "aos/logging/log_message_schema.h"
	#include "aos/network/message_bridge_client_schema.h"
	#include "aos/network/message_bridge_server_schema.h"
	#include "aos/network/timestamp_schema.h"

	namespace aos::testing {

	class EventLoopTestFactory {
	public:
	EventLoopTestFactory()
	: flatbuffer_(configuration::AddSchema(
	R"config({
	"channels": [
	{
	"name": "/aos",
	"type": "aos.logging.LogMessageFbs"
	},
	{
	"name": "/aos",
	"type": "aos.timing.Report"
	},
	{
	"name": "/test",
	"type": "aos.TestMessage",
	"frequency": 800
	},
	{
	"name": "/test1",
	"type": "aos.TestMessage"
	},
	{
	"name": "/test2",
	"type": "aos.TestMessage"
	},
	{
	"name": "/test3",
	"type": "aos.TestMessage",
	"channel_storage_duration": 10000000000
	}
	]
	})config",
	{aos::FlatbufferSpan<reflection::Schema>(
	logging::LogMessageFbsSchema()),
	aos::FlatbufferSpan<reflection::Schema>(timing::ReportSchema()),
	aos::FlatbufferSpan<reflection::Schema>(TestMessageSchema())})) {}

	virtual ~EventLoopTestFactory() {}

	// Makes a connected event loop.
	virtual std::unique_ptr<EventLoop> Make(std::string_view name) = 0;
	// Makes a primary event loop. This is the one the tests will try to use for
	// anything blocking.
	virtual std::unique_ptr<EventLoop> MakePrimary(std::string_view name) = 0;

	// Runs the loops until they quit.
	virtual Result<void> Run() = 0;

	virtual std::unique_ptr<ExitHandle> MakeExitHandle() = 0;

	// Quits the loops.
	virtual void Exit() = 0;

	// Advances time by sleeping. Can't be called from inside a loop.
	virtual void SleepFor(::std::chrono::nanoseconds duration) = 0;

	// Sets the config to a config with a max size with an invalid alignment.
	void InvalidChannelAlignment() {
	flatbuffer_ = configuration::AddSchema(
	R"config({
	"channels": [
	{
	"name": "/aos",
	"type": "aos.logging.LogMessageFbs"
	},
	{
	"name": "/aos",
	"type": "aos.timing.Report"
	},
	{
	"name": "/test",
	"type": "aos.TestMessage",
	"max_size": 13
	},
	{
	"name": "/test1",
	"type": "aos.TestMessage"
	},
	{
	"name": "/test2",
	"type": "aos.TestMessage"
	},
	{
	"name": "/test3",
	"type": "aos.TestMessage",
	"channel_storage_duration": 10000000000
	}
	]
	})config",
	{aos::FlatbufferSpan<reflection::Schema>(
	logging::LogMessageFbsSchema()),
	aos::FlatbufferSpan<reflection::Schema>(timing::ReportSchema()),
	aos::FlatbufferSpan<reflection::Schema>(TestMessageSchema())});
	}

	void PinReads() {
	static const std::string kJson = R"config({
	"channels": [
	{
	"name": "/aos",
	"type": "aos.logging.LogMessageFbs",
	"read_method": "PIN",
	"num_readers": 10
	},
	{
	"name": "/aos",
	"type": "aos.timing.Report",
	"read_method": "PIN",
	"num_readers": 10
	},
	{
	"name": "/test",
	"type": "aos.TestMessage",
	"read_method": "PIN",
	"num_readers": 10,
	"frequency": 800
	},
	{
	"name": "/test1",
	"type": "aos.TestMessage",
	"read_method": "PIN",
	"num_readers": 10
	},
	{
	"name": "/test2",
	"type": "aos.TestMessage",
	"read_method": "PIN",
	"num_readers": 10
	},
	{
	"name": "/test3",
	"type": "aos.TestMessage",
	"read_method": "PIN",
	"num_readers": 10,
	"channel_storage_duration": 10000000000
	}
	]
	})config";

	flatbuffer_ = configuration::AddSchema(
	kJson, {aos::FlatbufferSpan<reflection::Schema>(
	logging::LogMessageFbsSchema()),
	aos::FlatbufferSpan<reflection::Schema>(timing::ReportSchema()),
	aos::FlatbufferSpan<reflection::Schema>(TestMessageSchema())});
	}

	void EnableNodes(std::string_view my_node) {
	static const std::string kJson = R"config({
	"channels": [
	{
	"name": "/me/aos",
	"type": "aos.logging.LogMessageFbs",
	"source_node": "me"
	},
	{
	"name": "/them/aos",
	"type": "aos.logging.LogMessageFbs",
	"source_node": "them"
	},
	{
	"name": "/me/aos",
	"type": "aos.message_bridge.Timestamp",
	"source_node": "me",
	"destination_nodes": [
	{
	"name": "them"
	}
	]
	},
	{
	"name": "/them/aos",
	"type": "aos.message_bridge.Timestamp",
	"source_node": "them",
	"destination_nodes": [
	{
	"name": "me"
	}
	]
	},
	{
	"name": "/me/aos",
	"type": "aos.message_bridge.ServerStatistics",
	"source_node": "me",
	"frequency": 2
	},
	{
	"name": "/them/aos",
	"type": "aos.message_bridge.ServerStatistics",
	"source_node": "them",
	"frequency": 2
	},
	{
	"name": "/me/aos",
	"type": "aos.message_bridge.ClientStatistics",
	"source_node": "me",
	"frequency": 2
	},
	{
	"name": "/them/aos",
	"type": "aos.message_bridge.ClientStatistics",
	"source_node": "them",
	"frequency": 2
	},
	{
	"name": "/aos",
	"type": "aos.timing.Report",
	"source_node": "me"
	},
	{
	"name": "/test",
	"type": "aos.TestMessage",
	"source_node": "me"
	},
	{
	"name": "/test1",
	"type": "aos.TestMessage",
	"source_node": "me"
	},
	{
	"name": "/test2",
	"type": "aos.TestMessage",
	"source_node": "me"
	},
	{
	"name": "/test_forward",
	"type": "aos.TestMessage",
	"source_node": "them",
	"destination_nodes": [
	{
	"name": "me"
	}
	]
	},
	{
	"name": "/test_noforward",
	"type": "aos.TestMessage",
	"source_node": "them"
	}
	],
	"nodes": [
	{
	"name": "me",
	"hostname": "myhostname"
	},
	{
	"name": "them",
	"hostname": "themhostname"
	}
	],
	"maps": [
	{
	"match": {
	"name": "/aos*",
	"source_node": "me"
	},
	"rename": {
	"name": "/me/aos"
	}
	},
	{
	"match": {
	"name": "/aos*",
	"source_node": "them"
	},
	"rename": {
	"name": "/them/aos"
	}
	}
	]
	})config";

	flatbuffer_ = configuration::MergeConfiguration(
	configuration::MergeConfiguration(
	aos::FlatbufferDetachedBuffer<Configuration>(
	JsonToFlatbuffer<Configuration>(kJson))),
	{aos::FlatbufferSpan<reflection::Schema>(
	logging::LogMessageFbsSchema()),
	aos::FlatbufferSpan<reflection::Schema>(timing::ReportSchema()),
	aos::FlatbufferSpan<reflection::Schema>(TestMessageSchema()),
	aos::FlatbufferSpan<reflection::Schema>(
	message_bridge::ClientStatisticsSchema()),
	aos::FlatbufferSpan<reflection::Schema>(
	message_bridge::ServerStatisticsSchema()),
	aos::FlatbufferSpan<reflection::Schema>(
	message_bridge::TimestampSchema())});

	my_node_ = configuration::GetNode(&flatbuffer_.message(), my_node);
	}

	const Node *my_node() const { return my_node_; }

	const Configuration *configuration() { return &flatbuffer_.message(); }

	private:
	FlatbufferDetachedBuffer<Configuration> flatbuffer_;

	const Node *my_node_ = nullptr;
	};

	enum class DoTimingReports { kYes, kNo };

	class AbstractEventLoopTest
	: public ::testing::TestWithParam<
	std::tuple<std::function<EventLoopTestFactory *()>, ReadMethod,
	DoTimingReports>> {
	public:
	AbstractEventLoopTest() : factory_(std::get<0>(GetParam())()) {
	if (read_method() == ReadMethod::PIN) {
	factory_->PinReads();
	}
	}

	ReadMethod read_method() const { return std::get<1>(GetParam()); }
	DoTimingReports do_timing_reports() const { return std::get<2>(GetParam()); }

	::std::unique_ptr<EventLoop> Make(std::string_view name = "");

	::std::unique_ptr<EventLoop> MakePrimary(std::string_view name = "primary") {
	++event_loop_count_;
	auto result = factory_->MakePrimary(name);
	if (do_timing_reports() == DoTimingReports::kNo) {
	result->SkipTimingReport();
	}
	return result;
	}

	void InvalidChannelAlignment() { factory_->InvalidChannelAlignment(); }

	void EnableNodes(std::string_view my_node) { factory_->EnableNodes(my_node); }

	Result<void> Run() { return factory_->Run(); }

	std::unique_ptr<ExitHandle> MakeExitHandle() {
	return factory_->MakeExitHandle();
	}

	void Exit() { return factory_->Exit(); }

	void SleepFor(::std::chrono::nanoseconds duration) {
	return factory_->SleepFor(duration);
	}

	const Configuration *configuration() { return factory_->configuration(); }

	const Node *my_node() const { return factory_->my_node(); }

	// Ends the given event loop at the given time from now.
	void EndEventLoop(EventLoop *loop, ::std::chrono::milliseconds duration) {
	auto end_timer = loop->AddTimer([this]() { this->Exit(); });
	end_timer->Schedule(loop->monotonic_now() + duration);
	end_timer->set_name("end");
	}

	// Verifies that the buffer_index values for all of the given objects are
	// consistent.
	void VerifyBuffers(
	int number_buffers,
	std::vector<std::reference_wrapper<const Fetcher<TestMessage>>> fetchers,
	std::vector<std::reference_wrapper<const Sender<TestMessage>>> senders);

	// Helper function for testing the sent too fast check using a PhasedLoop with
	// an interval that sends exactly at the frequency of the channel
	void TestSentTooFastCheckEdgeCase(
	const std::function<RawSender::Error(int, int)> expected_err,
	const bool send_twice_at_end);

	private:
	const ::std::unique_ptr<EventLoopTestFactory> factory_;

	int event_loop_count_ = 0;
	};

	using AbstractEventLoopDeathTest = AbstractEventLoopTest;

	// Returns the frequency of the /test TestMessage channel
	int TestChannelFrequency(EventLoop *event_loop);
	// Returns the queue size of the /test TestMessage channel
	int TestChannelQueueSize(EventLoop *event_loop);
	// Sends a test message with value 0 with the given sender
	RawSender::Error SendTestMessage(aos::Sender<TestMessage> &sender);

	} // namespace aos::testing

	#endif // _AOS_EVENTS_EVENT_LOOP_PARAM_TEST_H_