Testkit for DurableState

Akka
1

Hello,

I am using DurableState for my project. Earlier I was using EventSourcedBehavior. But after moving to DurableState all my JUnits are failing. I am using PersistenceTestKit and Plugin. We are also using Sharding.

Can somebody please share which TestKit needs to be used ?

2

Probably missing documentation for this but you can look at the tests in Akka sources: https://github.com/akka/akka/blob/main/akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/state/scaladsl/DurableStateBehaviorSpec.scala

It’s essentially ScalaTestWithActorTestKit with PersistenceTestKitDurableStateStorePlugin.config

3

Thanks for the response. But its still not working. We are using sharding.
This is my actor:
public class InterimActor
extends DurableStateBehavior<InterimActor.Command, InterimActor.State> {

My JUnit:
@ExtendWith(MockitoExtension.class)
class InterimActorTest {
@ClassRule
public static final TestKitJunitResource testKit =
new TestKitJunitResource(
PersistenceTestKitPlugin.getInstance()
.config()
.withFallback(ConfigFactory.defaultApplication().resolve()));

PersistenceTestKit persistenceTestKit = PersistenceTestKit.create(testKit.system());






@BeforeAll
public static void init() {
    if (sharding == null) {
        Cluster cluster = Cluster.get(testKit.system());
        cluster.manager().tell(new Join(cluster.selfMember().address()));
        sharding = ClusterSharding.get(testKit.system());
        sharding.init(
                Entity.of(
                        InterimActor.TYPE_KEY,
                        entityContext ->
                                ProbedInterimActor.create(
                                        myClient,
                                        parentActorProbe,
                                        childActorProbe,
                                        entityContext.getShard(),
                                        cleanup)));
    }
}


@BeforeEach
public void setUp() throws JsonProcessingException {
    persistenceTestKit.clearAll();
    parentActorProbe = testKit.createTestProbe(ParentActor.Command.class);
    childActorProbe = testKit.createTestProbe(ChildActor.Command.class);
}


@Test
void testStartToInterimActor() {
    EntityRef<InterimActor.Command> underTest =
            sharding.entityRefFor(InterimActor.TYPE_KEY, jobId);
    Item Item = ItemObject.getFiniteItem();
    when(myClient.streamContacts(DEFAULT_ACCOUNT_ID, Item.getContactList(), null))
            .thenReturn(responseFlux);
    underTest.tell(new InterimActor.Start(jobId, Item, DEFAULT_ACCOUNT_ID));
    expectedPersisted(InterimActor.Started.class);
    underTest.tell(new InterimActor.Pace(2, 2));
    ChildActorProbe.expectMessageClass(ChildActor.Start.class);
    StepVerifier.create(responseFlux)
            .expectSubscription()
            .thenRequest(TWO)
            .expectNext(jsonNode, jsonNode)
            .thenRequest(TWO)
            .expectNext(jsonNode, jsonNode)
            .expectComplete();
}
4

I updated following:

        new TestKitJunitResource(
                PersistenceTestKitPlugin.getInstance()
                        .config()
                        .withFallback(ConfigFactory.load(TEST_CONF))
                        .withFallback(PersistenceTestKitDurableStateStorePlugin.config())
                        );
5

My test passes, but how do I check what is getting persisted in the State ? I am only able to check what messages are received by my actors. I wanted to see the updates to my state object.

6

You should be able to get the DurableStateStoreQuery as described in Persistence Query • Akka Documentation
Then you can use the getObject(persistenceId) to retrieve the value.