Skip to content

Commit a2dab4b

Browse files
somdoronsomdoron
committed
tests
1 parent b14b92e commit a2dab4b

File tree

6 files changed

+484
-33
lines changed

6 files changed

+484
-33
lines changed

‎session-state-machine/src/test/scala/zio/raft/sessionstatemachine/IdempotencySpec.scala‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
package zio.raft.sessionstatemachine
2+
3+
import zio.test.*
4+
import zio.test.Assertion.*
5+
import zio.{UIO, ZIO}
6+
import zio.raft.{Command, HMap, Index}
7+
import zio.raft.protocol.{SessionId, RequestId}
8+
import zio.stream.{Stream, ZStream}
9+
import java.time.Instant
10+
11+
/**
12+
* Contract test for idempotency checking (PC-1).
13+
*
14+
* Tests that duplicate requests return cached responses without calling applyCommand.
15+
*/
16+
object IdempotencySpec extends ZIOSpecDefault:
17+
18+
sealed trait TestCommand extends Command
19+
object TestCommand:
20+
case class Increment(by: Int) extends TestCommand:
21+
type Response = Int
22+
23+
import zio.prelude.Newtype
24+
object CounterKey extends Newtype[String]
25+
type CounterKey = CounterKey.Type
26+
given HMap.KeyLike[CounterKey] = HMap.KeyLike.forNewtype(CounterKey)
27+
28+
type TestSchema = ("counter", CounterKey, Int) *: EmptyTuple
29+
type CombinedSchema = Tuple.Concat[SessionSchema, TestSchema]
30+
31+
val counterKey = CounterKey("value")
32+
33+
// SR = String (the actual server request payload type)
34+
class TestStateMachine extends SessionStateMachine[TestCommand, String, TestSchema]:
35+
var callCount = 0 // Track how many times applyCommand is called
36+
37+
protected def applyCommand(cmd: TestCommand, createdAt: Instant): StateWriter[HMap[CombinedSchema], ServerRequestForSession[String], cmd.Response] =
38+
callCount += 1
39+
cmd match
40+
case TestCommand.Increment(by) =>
41+
for {
42+
state <- StateWriter.get[HMap[CombinedSchema]]
43+
current = state.get["counter"](counterKey).getOrElse(0)
44+
newValue = current + by
45+
newState = state.updated["counter"](counterKey, newValue)
46+
_ <- StateWriter.set(newState)
47+
} yield newValue.asInstanceOf[cmd.Response]
48+
49+
protected def handleSessionCreated(sid: SessionId, caps: Map[String, String], createdAt: Instant): StateWriter[HMap[CombinedSchema], ServerRequestForSession[String], Unit] =
50+
StateWriter.succeed(())
51+
52+
protected def handleSessionExpired(sid: SessionId, capabilities: Map[String, String], createdAt: Instant): StateWriter[HMap[CombinedSchema], ServerRequestForSession[String], Unit] =
53+
StateWriter.succeed(())
54+
55+
def takeSnapshot(state: HMap[CombinedSchema]): Stream[Nothing, Byte] =
56+
ZStream.empty
57+
58+
def restoreFromSnapshot(stream: Stream[Nothing, Byte]): UIO[HMap[CombinedSchema]] =
59+
ZIO.succeed(HMap.empty)
60+
61+
def shouldTakeSnapshot(lastSnapshotIndex: Index, lastSnapshotSize: Long, commitIndex: Index): Boolean =
62+
false
63+
64+
def spec = suite("Idempotency with Composite Keys")(
65+
66+
test("PC-1: First request calls applyCommand, second request returns cached without calling") {
67+
val sm = new TestStateMachine()
68+
val state0 = HMap.empty[sm.Schema]
69+
val now = Instant.now()
70+
val sessionId = SessionId("s1")
71+
72+
// Create session first (cast to match state machine type)
73+
val createCmd: SessionCommand[TestCommand, String] =
74+
SessionCommand.CreateSession[String](now, sessionId, Map.empty)
75+
.asInstanceOf[SessionCommand[TestCommand, String]]
76+
val (state1, _) = sm.apply(createCmd).run(state0)
77+
78+
// First request - should call applyCommand
79+
val cmd1: SessionCommand[TestCommand, String] =
80+
SessionCommand.ClientRequest(now, sessionId, RequestId(1), RequestId(1), TestCommand.Increment(10))
81+
val (state2, result1) = sm.apply(cmd1).run(state1)
82+
val response1 = result1.asInstanceOf[(Int, List[Any])]._1
83+
84+
assertTrue(sm.callCount == 1) &&
85+
assertTrue(response1 == 10)
86+
87+
// Second request with same ID - should NOT call applyCommand
88+
val cmd2: SessionCommand[TestCommand, String] =
89+
SessionCommand.ClientRequest(now, sessionId, RequestId(1), RequestId(1), TestCommand.Increment(999))
90+
val (state3, result2) = sm.apply(cmd2).run(state2)
91+
val response2 = result2.asInstanceOf[(Int, List[Any])]._1
92+
93+
assertTrue(
94+
sm.callCount == 1 && // Still 1, not called again!
95+
response2 == 10 // Cached response, not 999!
96+
)
97+
},
98+
99+
test("Different requestIds call applyCommand separately") {
100+
val sm = new TestStateMachine()
101+
val state0 = HMap.empty[sm.Schema]
102+
val now = Instant.now()
103+
val sessionId = SessionId("s1")
104+
105+
val createCmd: SessionCommand[TestCommand, String] =
106+
SessionCommand.CreateSession[String](now, sessionId, Map.empty)
107+
.asInstanceOf[SessionCommand[TestCommand, String]]
108+
val (state1, _) = sm.apply(createCmd).run(state0)
109+
110+
// First request
111+
val cmd1: SessionCommand[TestCommand, String] =
112+
SessionCommand.ClientRequest(now, sessionId, RequestId(1), RequestId(1), TestCommand.Increment(5))
113+
val (state2, result1) = sm.apply(cmd1).run(state1)
114+
val response1 = result1.asInstanceOf[(Int, List[Any])]._1
115+
116+
// Second request with DIFFERENT ID
117+
val cmd2: SessionCommand[TestCommand, String] =
118+
SessionCommand.ClientRequest(now, sessionId, RequestId(2), RequestId(1), TestCommand.Increment(3))
119+
val (state3, result2) = sm.apply(cmd2).run(state2)
120+
val response2 = result2.asInstanceOf[(Int, List[Any])]._1
121+
122+
assertTrue(
123+
sm.callCount == 2 && // Both requests processed
124+
response1 == 5 &&
125+
response2 == 8 // 5 + 3
126+
)
127+
}
128+
)

‎session-state-machine/src/test/scala/zio/raft/sessionstatemachine/PendingServerRequestSpec.scala‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
package zio.raft.sessionstatemachine
2+
3+
import zio.test.*
4+
import zio.test.Assertion.*
5+
import java.time.Instant
6+
7+
/**
8+
* Contract test for PendingServerRequest.
9+
*
10+
* Tests the PendingServerRequest case class after removing redundant id and sessionId fields.
11+
* Both id and sessionId are now stored only in the HMap composite key, not duplicated in the value.
12+
*/
13+
object PendingServerRequestSpec extends ZIOSpecDefault:
14+
15+
case class TestPayload(data: String, value: Int)
16+
17+
def spec = suite("PendingServerRequest")(
18+
19+
test("create PendingServerRequest with only payload and lastSentAt") {
20+
val now = Instant.now()
21+
val payload = TestPayload("test-data", 42)
22+
23+
val pending = PendingServerRequest(
24+
payload = payload,
25+
lastSentAt = now
26+
)
27+
28+
assertTrue(
29+
pending.payload == payload &&
30+
pending.lastSentAt == now
31+
)
32+
},
33+
34+
test("lastSentAt is not optional - always has a value") {
35+
val payload = TestPayload("data", 123)
36+
val pending = PendingServerRequest(
37+
payload = payload,
38+
lastSentAt = Instant.now()
39+
)
40+
41+
// Verify lastSentAt is not Option[Instant], just Instant
42+
val timestamp: Instant = pending.lastSentAt
43+
44+
assertTrue(timestamp != null)
45+
},
46+
47+
test("PendingServerRequest is immutable") {
48+
val original = PendingServerRequest(
49+
payload = TestPayload("original", 1),
50+
lastSentAt = Instant.now()
51+
)
52+
53+
val newTime = Instant.now().plusSeconds(60)
54+
val updated = original.copy(lastSentAt = newTime)
55+
56+
assertTrue(
57+
original.lastSentAt != updated.lastSentAt &&
58+
original.payload == updated.payload
59+
)
60+
},
61+
62+
test("PendingServerRequest works with different payload types") {
63+
val stringPending = PendingServerRequest(
64+
payload = "string-payload",
65+
lastSentAt = Instant.now()
66+
)
67+
68+
val intPending = PendingServerRequest(
69+
payload = 42,
70+
lastSentAt = Instant.now()
71+
)
72+
73+
assertTrue(
74+
stringPending.payload == "string-payload" &&
75+
intPending.payload == 42
76+
)
77+
}
78+
)

‎session-state-machine/src/test/scala/zio/raft/sessionstatemachine/SchemaSpec.scala‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
package zio.raft.sessionstatemachine
2+
3+
import zio.test.*
4+
import zio.test.Assertion.*
5+
import zio.raft.HMap
6+
import zio.raft.protocol.{SessionId, RequestId}
7+
import java.time.Instant
8+
9+
/**
10+
* Contract test for SessionSchema with composite keys.
11+
*
12+
* Tests the new architecture:
13+
* - Composite keys (SessionId, RequestId) for cache and serverRequests
14+
* - Byte-based HMap keys with proper numeric ordering
15+
* - Schema concatenation with user schema
16+
*/
17+
object SchemaSpec extends ZIOSpecDefault:
18+
19+
// Test newtypes for user schema
20+
import zio.prelude.Newtype
21+
22+
object CounterKey extends Newtype[String]
23+
type CounterKey = CounterKey.Type
24+
given HMap.KeyLike[CounterKey] = HMap.KeyLike.forNewtype(CounterKey)
25+
26+
type TestUserSchema =
27+
("counter", CounterKey, Int) *:
28+
EmptyTuple
29+
30+
type CombinedSchema = Tuple.Concat[SessionSchema, TestUserSchema]
31+
32+
def spec = suite("Schema with Composite Keys")(
33+
34+
test("SessionSchema has composite key for cache") {
35+
val sessionId = SessionId("session-1")
36+
val state = HMap.empty[SessionSchema]
37+
38+
// Cache uses composite key (SessionId, RequestId)
39+
val withCache = state.updated["cache"](
40+
(sessionId, RequestId(1)),
41+
"cached-response"
42+
)
43+
44+
val retrieved: Option[Any] = withCache.get["cache"]((sessionId, RequestId(1)))
45+
46+
assertTrue(retrieved.contains("cached-response"))
47+
},
48+
49+
test("SessionSchema has composite key for serverRequests") {
50+
val sessionId = SessionId("session-1")
51+
val state = HMap.empty[SessionSchema]
52+
53+
// serverRequests uses composite key (SessionId, RequestId)
54+
val pending = PendingServerRequest(
55+
payload = "test-request",
56+
lastSentAt = Instant.now()
57+
)
58+
59+
val withRequest = state.updated["serverRequests"](
60+
(sessionId, RequestId(5)),
61+
pending
62+
)
63+
64+
val retrieved: Option[PendingServerRequest[String]] =
65+
withRequest.get["serverRequests"]((sessionId, RequestId(5)))
66+
.map(_.asInstanceOf[PendingServerRequest[String]])
67+
68+
assertTrue(
69+
retrieved.isDefined &&
70+
retrieved.get.payload == "test-request"
71+
)
72+
},
73+
74+
test("Combined schema concatenates SessionSchema and UserSchema") {
75+
val state = HMap.empty[CombinedSchema]
76+
77+
val sessionId = SessionId("s1")
78+
79+
// Can use SessionSchema prefixes with composite keys
80+
val withSession = state
81+
.updated["metadata"](sessionId, SessionMetadata(Map.empty, Instant.now()))
82+
.updated["cache"]((sessionId, RequestId(1)), "value1")
83+
84+
// Can use UserSchema prefixes
85+
val withUser = withSession
86+
.updated["counter"](CounterKey("main"), 42)
87+
88+
val metadata: Option[SessionMetadata] = withUser.get["metadata"](sessionId)
89+
val cached: Option[Any] = withUser.get["cache"]((sessionId, RequestId(1)))
90+
val counter: Option[Int] = withUser.get["counter"](CounterKey("main"))
91+
92+
assertTrue(
93+
metadata.isDefined &&
94+
cached.contains("value1") &&
95+
counter.contains(42)
96+
)
97+
},
98+
99+
test("Composite keys enable range queries for session") {
100+
val sessionId = SessionId("session-1")
101+
val state = HMap.empty[SessionSchema]
102+
.updated["cache"]((sessionId, RequestId(1)), "resp1")
103+
.updated["cache"]((sessionId, RequestId(5)), "resp5")
104+
.updated["cache"]((sessionId, RequestId(10)), "resp10")
105+
106+
// Range query: get all cache entries for session with RequestId in [0, 7)
107+
val rangeResults = state.range["cache"](
108+
(sessionId, RequestId.zero),
109+
(sessionId, RequestId(7))
110+
).toList
111+
112+
assertTrue(
113+
rangeResults.length == 2 && // RequestId 1 and 5, not 10
114+
rangeResults.map(_._1._2).toSet == Set(RequestId(1), RequestId(5))
115+
)
116+
},
117+
118+
test("Numeric ordering works correctly for RequestIds") {
119+
val sessionId = SessionId("session-1")
120+
val state = HMap.empty[SessionSchema]
121+
.updated["cache"]((sessionId, RequestId(9)), "nine")
122+
.updated["cache"]((sessionId, RequestId(42)), "forty-two")
123+
.updated["cache"]((sessionId, RequestId(100)), "hundred")
124+
125+
// Range should use numeric ordering, not lexicographic
126+
// RequestId uses big-endian encoding, so 9 < 42 < 100
127+
val all = state.range["cache"](
128+
(sessionId, RequestId.zero),
129+
(sessionId, RequestId.max)
130+
).toList.map(_._1._2)
131+
132+
assertTrue(
133+
all == List(RequestId(9), RequestId(42), RequestId(100)) // Numeric order!
134+
)
135+
}
136+
)

0 commit comments

Comments
 (0)