Rules Service Integration Tests

Data Services
1

I have a simple Rules data service with a few simple stages it executes. An issue I have found with rules services is that you can easily unit test each individual action you implement, but testing the pipeline as a whole is troublesome.
The reason is that you need to start the data service in order to initialize and start the Rules service and most data services have dependencies on running instances of kafka/mongo.

To try and alleviate this, I have attempted to setup a unit test that configures the service to run with the transient data type and local event type:

System.setProperty("DATA_SERVICE_TYPE", "transient");
System.setProperty("EVENT_SERVICE_TYPE", "local");

I set the above as part of the static @BeforeAll annotated method in my unit test.
Beyond that, the Rules service Constructor accepts a IStreamDataReader actionInputStream. I have set that to use a StreamDataManagerNonPersisted.newBuilder().build(); as a test.

When I run my test, the data service starts without error, but when I attempt to publish StreamData to the “actionInputStream” I cannot seem to get that to work.
I am attempting to publish data to the “actionInputStream” in my unit test as follows:

StreamData data = StreamData.newBuilder()
                .withDiscriminator("<my_discriminator>")
                .putData(FalconPersistRule.DATA_KEY, <my_dataobject_ref>)
                .build();

Flowable.just(data).concatWith(Flowable.fromPublisher(<actionInputStream>.getPublisher(Context.makeSystemContext())));

Seeing that the service starts up without error using the “transient” data type, “local” event type, as well as the “StreamDataManagerNonPersisted actionInputStream” leads me to believe that my issue is with how I am publishing the data in my test. But please let me know if what I am trying to do is even possible.

My main questions:

  • Can you configure a Rules data service to run without an external mongo/kafka instance as I have configured?
  • How can I publish data to the actionInputStream such that the Rules service will process that data?
  • Is there a different way to execute these types of tests that I am missing? (I want to avoid the requirement of running external dependencies. I have already thought about using testcontainers, but that requires docker-in-docker with our CI pipelines, which I also want to avoid.)
2

There’s no reason you shouldn’t be able to test with transient (DataService) and local (EventService), and I completely agree that that’s what you should do.

Any chance you could post a full error? Also could you double check that you’ve started the actionInputStream, calling its start(...) method? I don’t think that should lead to an error message, but it would lead to events not getting consumed in the RulesService.

3

Also assuming those system props are to populate an OWNER Config, I’d consider running ConfigCache.clear(); before you set the properties, and maybe again in an @AfterAll.

That will guarantee if your test suite shares a JVM for multiple tests, which I think Gradle may do, that your variables are actually considered.

4

Hi Mark.

I am never receiving an error, the server starts just fine with the transient and local service types. My problem is that the actionInputStream is never seeming to receive the data I am trying to send to it. I checked and I am calling the stream’s start(...) method.

Im wondering if I am failing to correctly use the stream’s Publisher to send the data. This is how I am attempting to send the data to the stream:

StreamData data = StreamData.newBuilder()
                .withDiscriminator("<my_discriminator>")
                .putData(FalconPersistRule.DATA_KEY, <my_dataobject_ref>)
                .build();
        
        Flowable.just(data).concatWith(Flowable.fromPublisher(TaskingDsApplication.actionInputStream.getPublisher(Context.makeSystemContext()))).subscribe();

My understanding is that Flowable.just(...) will emit whatever it is given, and concatWith will concatenate the first Flowable with the second. The second Flowable is the Publisher from the actionInputStream, so I’d think it would receive the data from Flowable.just(...).

If you know of any place in the Rules service code I can pop a breakpoint to test, let me know. I’m also free to huddle on slack if you want to have a closer look.

Any thoughts and help is appreciated :slight_smile:

5

To write to a Stream, you should use the StreamDataManager (actionInputStream)'s write(...) method. Are you saying that instead of calling write(...), you are just trying to push things into its Flowable? I don’t think that will actually work as the Flowable commands you showed actually make a new Publisher. They don’t modify the actionInputStream’s Publisher.

6

Perfect. That was the issue. My test is passing now. Thanks Mark!

1 Like
7

Awesome! And not a problem!

1 Like