refactor(#1375): type connector operation responses (#1388)

This commit is contained in:
m57 authored and GitHub committed 2026-09-29 21:28:40 +03:00
1 parent 8ae2ddbbee
commit c8edf3665f
12 files changed
+1168 -125

No files matched your search

@@ -7,6 +7,7 @@ import com.m57.hermescontrol.data.model.ConnectorCatalogEntry
import com.m57.hermescontrol.data.model.ConnectorError
import com.m57.hermescontrol.data.model.ConnectorPolicy
import com.m57.hermescontrol.data.model.ConnectorTool
import com.m57.hermescontrol.data.ws.contract.ConnectionRespondParams
import com.m57.hermescontrol.data.ws.contract.ConnectorOwner
import com.m57.hermescontrol.data.ws.contract.ConnectorsAccountsParams
import com.m57.hermescontrol.data.ws.contract.ConnectorsAccountsRemoveParams
@@ -61,10 +62,9 @@ interface AccountConnectorRepository {
expectedRevision: String,
): AccountConnectorResult<ConnectorPolicy>
suspend fun operationRequest(
method: String,
params: Map<String, Any>,
): Any?
suspend fun operationRespond(params: ConnectionRespondParams): Any?
suspend fun operationWake(params: ConnectorsOperationStatusParams): Any?
suspend fun operationStatus(
opId: String,
@@ -74,26 +74,8 @@ interface AccountConnectorRepository {
}
class HermesAccountConnectorRepository(
private val rpc: suspend (String, Map<String, Any>) -> Any? = { method, params ->
val deferred = HermesWsClient.request(method, params)
try {
deferred.await()
} catch (
e: CancellationException,
) {
deferred.cancel(e)
throw e
} finally {
if (!deferred.isCompleted) deferred.cancel()
}
},
private val caller: TypedRpcCaller = HermesRpcCaller,
) : AccountConnectorRepository {
private suspend fun request(
method: String,
params: Map<String, Any> = emptyMap(),
) = rpc(method, params)
private suspend fun <T> guarded(block: suspend () -> T): AccountConnectorResult<T> =
try {
AccountConnectorResult.Success(block())
@@ -111,12 +93,6 @@ class HermesAccountConnectorRepository(
AccountConnectorResult.Failure(ConnectorError.NetworkError("Failed to communicate with gateway."))
}
private fun accountParams(params: Map<String, Any> = emptyMap()): Map<String, Any> =
buildMap {
putAll(params)
AuthManager.activeProfileId.value?.let { put("profile", it) }
}
private fun obj(value: Any?): JsonObject = ConnectorParser.toJsonElement(value).jsonObject
private fun JsonObject.str(k: String) = this[k]?.jsonPrimitive?.contentOrNull.orEmpty()
@@ -173,10 +149,23 @@ class HermesAccountConnectorRepository(
)
}
override suspend fun operationRequest(
method: String,
params: Map<String, Any>,
): Any? = request(method, accountParams(params + ("owner" to mapOf("type" to "account"))))
override suspend fun operationRespond(params: ConnectionRespondParams): Any? =
caller.call(
RpcMethods.CONNECTION_RESPOND,
params.copy(
owner = ConnectorOwner.account(),
profile = AuthManager.activeProfileId.value,
),
)
override suspend fun operationWake(params: ConnectorsOperationStatusParams): Any? =
caller.call(
RpcMethods.CONNECTORS_OPERATION_WAKE,
params.copy(
owner = ConnectorOwner.account(),
profile = AuthManager.activeProfileId.value,
),
)
override suspend fun operationStatus(opId: String) =
guarded {
@@ -0,0 +1,26 @@
package com.m57.hermescontrol.data.ws.contract
import kotlinx.serialization.SerialName
import kotlinx.serialization.Serializable
@Serializable
data class ConnectionAnswerTarget(
@SerialName("name") val name: String,
@SerialName("status") val status: String,
@SerialName("detail") val detail: String? = null,
@SerialName("env") val env: Map<String, String>? = null,
)
@Serializable
data class ConnectionAnswer(
@SerialName("targets") val targets: List<ConnectionAnswerTarget>? = null,
@SerialName("settled_by") val settledBy: String? = null,
)
@Serializable
data class ConnectionRespondParams(
@SerialName("owner") val owner: ConnectorOwner,
@SerialName("op_id") val opId: String,
@SerialName("result") val result: ConnectionAnswer,
@SerialName("profile") val profile: String? = null,
)
@@ -315,6 +315,22 @@ object RpcMethods {
JsonElement.serializer(),
)
// Passthrough JsonElement results: callers keep their existing legacy parsers.
val CONNECTION_RESPOND: RpcMethod<ConnectionRespondParams, JsonElement> =
RpcMethod(
WsMethods.CONNECTION_RESPOND,
ConnectionRespondParams.serializer(),
JsonElement.serializer(),
)
// Passthrough JsonElement results: callers keep their existing legacy parsers.
val CONNECTORS_OPERATION_WAKE: RpcMethod<ConnectorsOperationStatusParams, JsonElement> =
RpcMethod(
WsMethods.CONNECTORS_OPERATION_WAKE,
ConnectorsOperationStatusParams.serializer(),
JsonElement.serializer(),
)
/** Every typed method. GatewayContractTest iterates this; future migrations append here. */
val all: List<RpcMethod<*, *>> =
listOf(
@@ -356,5 +372,7 @@ object RpcMethods {
CONNECTORS_POLICY_GET,
CONNECTORS_POLICY_SET,
CONNECTORS_TOOLS,
CONNECTION_RESPOND,
CONNECTORS_OPERATION_WAKE,
)
}
@@ -3,14 +3,27 @@ package com.m57.hermescontrol.ui.chat
import com.m57.hermescontrol.data.model.ConnectionOperationSnapshot
import com.m57.hermescontrol.data.model.ConnectorError
import com.m57.hermescontrol.data.ws.ConnectorParser
import com.m57.hermescontrol.data.ws.ConnectorRepository
import com.m57.hermescontrol.data.ws.HermesWsClient
import com.m57.hermescontrol.data.ws.WsMethods
import com.m57.hermescontrol.data.ws.contract.ConnectionAnswer
import com.m57.hermescontrol.data.ws.contract.ConnectionAnswerTarget
import com.m57.hermescontrol.data.ws.contract.ConnectionRespondParams
import com.m57.hermescontrol.data.ws.contract.ConnectorOwner
import com.m57.hermescontrol.data.ws.contract.ConnectorsOperationStatusParams
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
sealed interface ConnectionOperationRequest {
data class Respond(
val params: ConnectionRespondParams,
) : ConnectionOperationRequest
data class Wake(
val params: ConnectorsOperationStatusParams,
) : ConnectionOperationRequest
}
sealed class ConnectionPendingAction {
abstract val opId: String
abstract val observedSeq: Long
@@ -45,10 +58,7 @@ data class ConnectionOperationUiState(
)
fun interface ConnectionOperationRequester {
suspend fun request(
method: String,
params: Map<String, Any>,
): Any?
suspend fun request(action: ConnectionOperationRequest): Any?
}
internal class ConnectionResumeCheckpoint(
@@ -163,19 +173,20 @@ class ChatConnectionOperationDelegate(
observedSeq = current.seq,
),
) ?: return
val answer =
buildMap<String, Any> {
put("name", target)
put("status", if (approved) "approved" else "skipped")
if (safeEnv.isNotEmpty()) put("env", safeEnv)
}
val answerTarget =
ConnectionAnswerTarget(
name = target,
status = if (approved) "approved" else "skipped",
env = safeEnv.takeIf { it.isNotEmpty() },
)
dispatch(
method = WsMethods.CONNECTION_RESPOND,
params =
mapOf(
"owner" to ownerParams(),
"op_id" to snapshot.opId,
"result" to mapOf("targets" to listOf(answer)),
action =
ConnectionOperationRequest.Respond(
ConnectionRespondParams(
owner = owner(),
opId = snapshot.opId,
result = ConnectionAnswer(targets = listOf(answerTarget)),
),
),
)
}
@@ -183,12 +194,13 @@ class ChatConnectionOperationDelegate(
suspend fun continueOperation() {
val snapshot = begin(ConnectionPendingAction.Continue(currentOp(), currentSeq())) ?: return
dispatch(
method = WsMethods.CONNECTION_RESPOND,
params =
mapOf(
"owner" to ownerParams(),
"op_id" to snapshot.opId,
"result" to mapOf("settled_by" to "continue"),
action =
ConnectionOperationRequest.Respond(
ConnectionRespondParams(
owner = owner(),
opId = snapshot.opId,
result = ConnectionAnswer(settledBy = "continue"),
),
),
)
}
@@ -198,36 +210,40 @@ class ChatConnectionOperationDelegate(
if (current.opId != expectedOpId) return
val snapshot = begin(ConnectionPendingAction.Wake(current.opId, current.seq)) ?: return
dispatch(
method = WsMethods.CONNECTORS_OPERATION_WAKE,
params = mapOf("owner" to ownerParams(), "op_id" to snapshot.opId),
action =
ConnectionOperationRequest.Wake(
ConnectorsOperationStatusParams(
owner = owner(),
opId = snapshot.opId,
),
),
clearOnSuccess = true,
unknownOperationSettles = true,
)
}
private suspend fun dispatch(
method: String,
params: Map<String, Any>,
action: ConnectionOperationRequest,
clearOnSuccess: Boolean = false,
unknownOperationSettles: Boolean = false,
) {
val action = _state.value.pendingAction
val pending = _state.value.pendingAction
val actionGeneration = generation
try {
requester.request(method, params)
requester.request(action)
// Success only acknowledges receipt. Keep the exactly-once lock until
// a newer authoritative snapshot advances the operation sequence.
if (clearOnSuccess && generation == actionGeneration && _state.value.pendingAction == action) {
if (clearOnSuccess && generation == actionGeneration && _state.value.pendingAction == pending) {
_state.value = _state.value.copy(pendingAction = null)
}
} catch (error: CancellationException) {
throw error
} catch (error: Exception) {
if (unknownOperationSettles && error is HermesWsClient.HermesRpcException && error.code == 4004) {
if (generation == actionGeneration && _state.value.pendingAction == action) {
action?.opId?.let(::markSettled)
if (generation == actionGeneration && _state.value.pendingAction == pending) {
pending?.opId?.let(::markSettled)
}
} else if (generation == actionGeneration && _state.value.pendingAction == action) {
} else if (generation == actionGeneration && _state.value.pendingAction == pending) {
_state.value =
_state.value.copy(
pendingAction = null,
@@ -269,8 +285,8 @@ class ChatConnectionOperationDelegate(
snapshot.sessionId == sessionId
}
private fun ownerParams(): Map<String, String> =
if (accountOwned) mapOf("type" to "account") else ConnectorRepository.sessionOwner(checkNotNull(sessionId))
private fun owner(): ConnectorOwner =
if (accountOwned) ConnectorOwner.account() else ConnectorOwner.session(checkNotNull(sessionId))
private fun currentOp(): String =
_state.value.operation
@@ -461,8 +461,16 @@ class ChatViewModel(
constructor(application: Application) : this(application, startCleanup = true)
private val connectionOperationDelegate =
ChatConnectionOperationDelegate { method, params ->
HermesWsClient.request(method, params).await()
ChatConnectionOperationDelegate { action ->
when (action) {
is ConnectionOperationRequest.Respond -> {
HermesWsClient.call(RpcMethods.CONNECTION_RESPOND, action.params)
}
is ConnectionOperationRequest.Wake -> {
HermesWsClient.call(RpcMethods.CONNECTORS_OPERATION_WAKE, action.params)
}
}
}
// ── Internal state ───────────────────────────────────────────────────
@@ -14,6 +14,7 @@ import com.m57.hermescontrol.data.ws.AccountConnectorRepository
import com.m57.hermescontrol.data.ws.HermesWsClient
import com.m57.hermescontrol.data.ws.WsEvent
import com.m57.hermescontrol.ui.chat.ChatConnectionOperationDelegate
import com.m57.hermescontrol.ui.chat.ConnectionOperationRequest
import com.m57.hermescontrol.ui.chat.ConnectionOperationRequester
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Job
@@ -51,7 +52,13 @@ class AccountConnectorsViewModel(
private val scopedJobs = mutableListOf<Job>()
private val operation =
ChatConnectionOperationDelegate(
requester = ConnectionOperationRequester(repository::operationRequest),
requester =
ConnectionOperationRequester { action ->
when (action) {
is ConnectionOperationRequest.Respond -> repository.operationRespond(action.params)
is ConnectionOperationRequest.Wake -> repository.operationWake(action.params)
}
},
accountOwned = true,
)
val connectionState = operation.state
@@ -3,6 +3,10 @@ package com.m57.hermescontrol.data.ws
import com.m57.hermescontrol.data.local.AuthManager
import com.m57.hermescontrol.data.model.AccountConnectorResult
import com.m57.hermescontrol.data.model.ConnectorTool
import com.m57.hermescontrol.data.ws.contract.ConnectionAnswer
import com.m57.hermescontrol.data.ws.contract.ConnectionRespondParams
import com.m57.hermescontrol.data.ws.contract.ConnectorOwner
import com.m57.hermescontrol.data.ws.contract.ConnectorsOperationStatusParams
import io.mockk.every
import io.mockk.mockkObject
import io.mockk.unmockkAll
@@ -62,14 +66,19 @@ class AccountConnectorRepositoryTest {
operationFixture()
}
repo.operationStatus("op-a")
repo.operationRequest(
WsMethods.CONNECTION_RESPOND,
mapOf(
"op_id" to "op-a",
"result" to mapOf("settled_by" to "continue"),
repo.operationRespond(
ConnectionRespondParams(
owner = ConnectorOwner.account(),
opId = "op-a",
result = ConnectionAnswer(settledBy = "continue"),
),
)
repo.operationWake(
ConnectorsOperationStatusParams(
owner = ConnectorOwner.account(),
opId = "op-a",
),
)
repo.operationRequest(WsMethods.CONNECTORS_OPERATION_WAKE, mapOf("op_id" to "op-a"))
assertEquals(
listOf(
WsMethods.CONNECTORS_OPERATION_STATUS,
@@ -86,6 +95,23 @@ class AccountConnectorRepositoryTest {
assertEquals("op-a", params["op_id"])
assertFalse("session_id" in params)
}
assertEquals(
mapOf(
"owner" to mapOf("type" to "account"),
"op_id" to "op-a",
"result" to mapOf("settled_by" to "continue"),
"profile" to "work",
),
calls[1].second,
)
assertEquals(
mapOf(
"owner" to mapOf("type" to "account"),
"op_id" to "op-a",
"profile" to "work",
),
calls[2].second,
)
}
@Test fun `catalog accounts tools and removal use native account endpoints`() =
@@ -213,7 +239,7 @@ class AccountConnectorRepositoryTest {
}
private fun accountRepo(handler: suspend (String, Map<String, Any>) -> Any?) =
HermesAccountConnectorRepository(rpc = handler, caller = fakeCaller(handler))
HermesAccountConnectorRepository(caller = fakeCaller(handler))
private fun operationFixture() =
mapOf(
@@ -0,0 +1,562 @@
package com.m57.hermescontrol.data.ws
import android.util.Log
import com.m57.hermescontrol.data.local.AuthManager
import com.m57.hermescontrol.data.remote.CookieManager
import com.m57.hermescontrol.data.remote.DashboardSessionTokenRefresher
import com.m57.hermescontrol.data.remote.OkHttpProvider
import com.m57.hermescontrol.data.remote.buildFakePersistentCookieJar
import com.m57.hermescontrol.data.ws.contract.ConnectionAnswer
import com.m57.hermescontrol.data.ws.contract.ConnectionAnswerTarget
import com.m57.hermescontrol.data.ws.contract.ConnectionRespondParams
import com.m57.hermescontrol.data.ws.contract.ConnectorOwner
import com.m57.hermescontrol.data.ws.contract.ConnectorsOperationStatusParams
import com.m57.hermescontrol.data.ws.contract.HermesRpcCaller
import com.m57.hermescontrol.data.ws.contract.RpcMethods
import io.mockk.coEvery
import io.mockk.every
import io.mockk.mockk
import io.mockk.mockkObject
import io.mockk.mockkStatic
import io.mockk.unmockkAll
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import kotlinx.coroutines.withTimeoutOrNull
import kotlinx.serialization.json.JsonElement
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.buildJsonArray
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import okhttp3.Response
import okhttp3.WebSocket
import okhttp3.WebSocketListener
import okhttp3.mockwebserver.MockResponse
import okhttp3.mockwebserver.MockWebServer
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicReference
class HermesWsClientConnectionOperationTest {
private lateinit var mockWebServer: MockWebServer
@Before
fun setUp() {
mockkStatic(Log::class)
every { Log.d(any<String>(), any<String>()) } returns 0
every { Log.i(any<String>(), any<String>()) } returns 0
every { Log.w(any<String>(), any<String>()) } returns 0
every { Log.w(any<String>(), any<String>(), any<Throwable>()) } returns 0
every { Log.e(any<String>(), any<String>(), any()) } returns 0
mockWebServer = MockWebServer()
mockWebServer.start()
mockkObject(AuthManager)
every { AuthManager.wsUrl() } returns mockWebServer.url("/").toString().replace("http://", "ws://")
every { AuthManager.isAutoReconnect() } returns false
every { AuthManager.getSessionCookie() } returns null
every { AuthManager.serverStore } returns
mockk<com.m57.hermescontrol.data.config.ServerStore>().also {
every { it.getLatestState() } returns
com.m57.hermescontrol.data.config
.ServerStoreState()
}
CookieManager.setJarForTest(buildFakePersistentCookieJar())
mockkObject(DashboardSessionTokenRefresher)
coEvery { DashboardSessionTokenRefresher.refreshAsync() } returns null
every { DashboardSessionTokenRefresher.refresh() } returns null
HermesWsClient.disconnect(clearPendingMessages = true)
HermesWsClient.releaseExternalActivityConnectionLease()
HermesWsClient.releaseBackgroundConnectionLease()
HermesWsClient.setAppForeground(true)
AuthManager.resetAuthStateForTest()
}
@After
fun tearDown() {
AuthManager.resetAuthStateForTest()
HermesWsClient.releaseExternalActivityConnectionLease()
HermesWsClient.releaseBackgroundConnectionLease()
HermesWsClient.disconnect(clearPendingMessages = true)
runBlocking {
withTimeoutOrNull(5000) {
HermesWsClient.connectionStatus.first { it == ConnectionStatus.DISCONNECTED }
}
}
runCatching { mockWebServer.shutdown() }
unmockkAll()
}
private fun connectClient() {
HermesWsClient.connect()
runBlocking {
withTimeout(5000) { HermesWsClient.connectionStatus.first { it == ConnectionStatus.CONNECTED } }
}
}
@Test
fun typedCallConnectionRespondWithApprovedEnvAndAccountOwnerEncodesFrameAndReturnsRawServerJsonUntouched() {
val opened = CountDownLatch(1)
val captured = CountDownLatch(1)
val frameRef = AtomicReference<JsonObject?>(null)
val serverResultJson =
buildJsonObject {
put("status", "ok")
put("settled", true)
}
mockWebServer.enqueue(
MockResponse().withWebSocketUpgrade(
object : WebSocketListener() {
override fun onOpen(
webSocket: WebSocket,
response: Response,
) = opened.countDown()
override fun onMessage(
webSocket: WebSocket,
text: String,
) {
val json = OkHttpProvider.json.parseToJsonElement(text) as? JsonObject ?: return
val name = (json["method"] as? JsonPrimitive)?.content
if (name == "connection.respond") {
frameRef.set(json)
captured.countDown()
val id = (json["id"] as? JsonPrimitive)?.content ?: "1"
webSocket.send(
"""{"jsonrpc":"2.0","id":"$id","result":""" + serverResultJson + """}""",
)
}
}
},
),
)
connectClient()
assertTrue(opened.await(5, TimeUnit.SECONDS))
val result: JsonElement =
runBlocking {
HermesWsClient.call(
RpcMethods.CONNECTION_RESPOND,
ConnectionRespondParams(
owner = ConnectorOwner.account(),
opId = "op-approved-123",
result =
ConnectionAnswer(
targets =
listOf(
ConnectionAnswerTarget(
name = "github",
status = "approved",
detail = "OAuth granted",
env = mapOf("TOKEN" to "dummy_secret_token_abc"),
),
),
settledBy = null,
),
profile = "prod",
),
)
}
assertTrue("Frame must reach the server", captured.await(5, TimeUnit.SECONDS))
val frame = requireNotNull(frameRef.get())
assertFalse("Outgoing frames have NO jsonrpc field", frame.containsKey("jsonrpc"))
assertEquals("connection.respond", (frame["method"] as? JsonPrimitive)?.content)
val params = frame["params"] as? JsonObject
val expectedParams =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "account")
},
)
put("op_id", "op-approved-123")
put(
"result",
buildJsonObject {
put(
"targets",
buildJsonArray {
add(
buildJsonObject {
put("name", "github")
put("status", "approved")
put("detail", "OAuth granted")
put(
"env",
buildJsonObject {
put("TOKEN", "dummy_secret_token_abc")
},
)
},
)
},
)
},
)
put("profile", "prod")
}
assertEquals(expectedParams, params)
assertEquals(serverResultJson, result)
}
@Test
fun typedCallConnectionRespondWithSkippedOmittedEnvAndSessionOwnerEncodesFrameAndReturnsRawServerJsonUntouched() {
val opened = CountDownLatch(1)
val captured = CountDownLatch(1)
val frameRef = AtomicReference<JsonObject?>(null)
val serverResultJson =
buildJsonObject {
put("status", "ok")
put("settled", false)
}
mockWebServer.enqueue(
MockResponse().withWebSocketUpgrade(
object : WebSocketListener() {
override fun onOpen(
webSocket: WebSocket,
response: Response,
) = opened.countDown()
override fun onMessage(
webSocket: WebSocket,
text: String,
) {
val json = OkHttpProvider.json.parseToJsonElement(text) as? JsonObject ?: return
val name = (json["method"] as? JsonPrimitive)?.content
if (name == "connection.respond") {
frameRef.set(json)
captured.countDown()
val id = (json["id"] as? JsonPrimitive)?.content ?: "1"
webSocket.send(
"""{"jsonrpc":"2.0","id":"$id","result":""" + serverResultJson + """}""",
)
}
}
},
),
)
connectClient()
assertTrue(opened.await(5, TimeUnit.SECONDS))
val result: JsonElement =
runBlocking {
HermesWsClient.call(
RpcMethods.CONNECTION_RESPOND,
ConnectionRespondParams(
owner = ConnectorOwner.session("sess-skip-456"),
opId = "op-skip-456",
result =
ConnectionAnswer(
targets =
listOf(
ConnectionAnswerTarget(
name = "slack",
status = "skipped",
detail = null,
env = null,
),
),
settledBy = null,
),
profile = null,
),
)
}
assertTrue("Frame must reach the server", captured.await(5, TimeUnit.SECONDS))
val frame = requireNotNull(frameRef.get())
assertFalse("Outgoing frames have NO jsonrpc field", frame.containsKey("jsonrpc"))
assertEquals("connection.respond", (frame["method"] as? JsonPrimitive)?.content)
val params = frame["params"] as? JsonObject
val expectedParams =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "sess-skip-456")
},
)
put("op_id", "op-skip-456")
put(
"result",
buildJsonObject {
put(
"targets",
buildJsonArray {
add(
buildJsonObject {
put("name", "slack")
put("status", "skipped")
},
)
},
)
},
)
}
assertEquals(expectedParams, params)
assertFalse(params?.containsKey("profile") ?: true)
assertEquals(serverResultJson, result)
}
@Test
fun typedCallConnectionRespondWithContinueAndOmittedTargetsEncodesFrameAndReturnsRawServerJsonUntouched() {
val opened = CountDownLatch(1)
val captured = CountDownLatch(1)
val frameRef = AtomicReference<JsonObject?>(null)
val serverResultJson =
buildJsonObject {
put("status", "ok")
put("settled", true)
}
mockWebServer.enqueue(
MockResponse().withWebSocketUpgrade(
object : WebSocketListener() {
override fun onOpen(
webSocket: WebSocket,
response: Response,
) = opened.countDown()
override fun onMessage(
webSocket: WebSocket,
text: String,
) {
val json = OkHttpProvider.json.parseToJsonElement(text) as? JsonObject ?: return
val name = (json["method"] as? JsonPrimitive)?.content
if (name == "connection.respond") {
frameRef.set(json)
captured.countDown()
val id = (json["id"] as? JsonPrimitive)?.content ?: "1"
webSocket.send(
"""{"jsonrpc":"2.0","id":"$id","result":""" + serverResultJson + """}""",
)
}
}
},
),
)
connectClient()
assertTrue(opened.await(5, TimeUnit.SECONDS))
val result: JsonElement =
runBlocking {
HermesWsClient.call(
RpcMethods.CONNECTION_RESPOND,
ConnectionRespondParams(
owner = ConnectorOwner.session("sess-continue-789"),
opId = "op-continue-789",
result =
ConnectionAnswer(
targets = null,
settledBy = "continue",
),
profile = null,
),
)
}
assertTrue("Frame must reach the server", captured.await(5, TimeUnit.SECONDS))
val frame = requireNotNull(frameRef.get())
assertFalse("Outgoing frames have NO jsonrpc field", frame.containsKey("jsonrpc"))
assertEquals("connection.respond", (frame["method"] as? JsonPrimitive)?.content)
val params = frame["params"] as? JsonObject
val expectedParams =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "sess-continue-789")
},
)
put("op_id", "op-continue-789")
put(
"result",
buildJsonObject {
put("settled_by", "continue")
},
)
}
assertEquals(expectedParams, params)
assertFalse(params?.containsKey("profile") ?: true)
assertEquals(serverResultJson, result)
}
@Test
fun typedCallConnectorsOperationWakeEncodesOwnerSessionOpIdAndProfileAndReturnsRawServerJsonUntouched() {
val opened = CountDownLatch(1)
val captured = CountDownLatch(1)
val frameRef = AtomicReference<JsonObject?>(null)
val serverResultJson =
buildJsonObject {
put("op_id", "op-wake-777")
put("status", "running")
}
mockWebServer.enqueue(
MockResponse().withWebSocketUpgrade(
object : WebSocketListener() {
override fun onOpen(
webSocket: WebSocket,
response: Response,
) = opened.countDown()
override fun onMessage(
webSocket: WebSocket,
text: String,
) {
val json = OkHttpProvider.json.parseToJsonElement(text) as? JsonObject ?: return
val name = (json["method"] as? JsonPrimitive)?.content
if (name == "connectors.operation.wake") {
frameRef.set(json)
captured.countDown()
val id = (json["id"] as? JsonPrimitive)?.content ?: "1"
webSocket.send(
"""{"jsonrpc":"2.0","id":"$id","result":""" + serverResultJson + """}""",
)
}
}
},
),
)
connectClient()
assertTrue(opened.await(5, TimeUnit.SECONDS))
val result: JsonElement =
runBlocking {
HermesWsClient.call(
RpcMethods.CONNECTORS_OPERATION_WAKE,
ConnectorsOperationStatusParams(
owner = ConnectorOwner.session("sess-wake-99"),
opId = "op-wake-777",
profile = "custom-profile",
),
)
}
assertTrue("Frame must reach the server", captured.await(5, TimeUnit.SECONDS))
val frame = requireNotNull(frameRef.get())
assertFalse("Outgoing frames have NO jsonrpc field", frame.containsKey("jsonrpc"))
assertEquals("connectors.operation.wake", (frame["method"] as? JsonPrimitive)?.content)
val params = frame["params"] as? JsonObject
val expectedParams =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "sess-wake-99")
},
)
put("op_id", "op-wake-777")
put("profile", "custom-profile")
}
assertEquals(expectedParams, params)
assertEquals(serverResultJson, result)
}
@Test
fun hermesRpcCallerCallDelegatesForWakeWithAccountOwnerAndNullProfile() {
val opened = CountDownLatch(1)
val captured = CountDownLatch(1)
val frameRef = AtomicReference<JsonObject?>(null)
val serverResultJson =
buildJsonObject {
put("op_id", "op-wake-888")
put("status", "done")
}
mockWebServer.enqueue(
MockResponse().withWebSocketUpgrade(
object : WebSocketListener() {
override fun onOpen(
webSocket: WebSocket,
response: Response,
) = opened.countDown()
override fun onMessage(
webSocket: WebSocket,
text: String,
) {
val json = OkHttpProvider.json.parseToJsonElement(text) as? JsonObject ?: return
val name = (json["method"] as? JsonPrimitive)?.content
if (name == "connectors.operation.wake") {
frameRef.set(json)
captured.countDown()
val id = (json["id"] as? JsonPrimitive)?.content ?: "1"
webSocket.send(
"""{"jsonrpc":"2.0","id":"$id","result":""" + serverResultJson + """}""",
)
}
}
},
),
)
connectClient()
assertTrue(opened.await(5, TimeUnit.SECONDS))
val result: JsonElement =
runBlocking {
HermesRpcCaller.call(
RpcMethods.CONNECTORS_OPERATION_WAKE,
ConnectorsOperationStatusParams(
owner = ConnectorOwner.account(),
opId = "op-wake-888",
profile = null,
),
)
}
assertTrue("Frame must reach the server", captured.await(5, TimeUnit.SECONDS))
val frame = requireNotNull(frameRef.get())
assertEquals("connectors.operation.wake", (frame["method"] as? JsonPrimitive)?.content)
val params = frame["params"] as? JsonObject
val expectedParams =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "account")
},
)
put("op_id", "op-wake-888")
}
assertEquals(expectedParams, params)
assertFalse(params?.containsKey("profile") ?: true)
assertEquals(serverResultJson, result)
}
}
@@ -332,9 +332,6 @@ class RawRpcParamsRatchetTest {
"ui/bots/group/GroupChatViewModel.kt::PROFILES_CONFIGURE",
"ui/bots/group/GroupChatViewModel.kt::PROFILES_CONFIGURE",
"ui/bots/group/GroupChatViewModel.kt::PROFILES_LIST",
"ui/chat/ChatConnectionOperationDelegate.kt::CONNECTION_RESPOND",
"ui/chat/ChatConnectionOperationDelegate.kt::CONNECTION_RESPOND",
"ui/chat/ChatConnectionOperationDelegate.kt::CONNECTORS_OPERATION_WAKE",
"ui/chat/ChatModelSwitchDelegate.kt::CONFIG_SET",
"ui/chat/ChatModelSwitchDelegate.kt::CONFIG_SET",
"ui/chat/ChatModelSwitchDelegate.kt::CONFIG_SET",
@@ -0,0 +1,257 @@
package com.m57.hermescontrol.data.ws.contract
import com.m57.hermescontrol.data.remote.OkHttpProvider
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.buildJsonArray
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertTrue
import org.junit.Test
class ConnectionOperationTypesTest {
@Test
fun connectionRespondParamsWithApprovedEnvAndAccountOwnerEncodesExactWireJson() {
val params =
ConnectionRespondParams(
owner = ConnectorOwner.account(),
opId = "op-approved-1",
result =
ConnectionAnswer(
targets =
listOf(
ConnectionAnswerTarget(
name = "linear",
status = "approved",
detail = "User accepted linear integration",
env = mapOf("API_KEY" to "dummy_secret_key_123"),
),
),
settledBy = null,
),
profile = "work",
)
val encoded = OkHttpProvider.json.encodeToJsonElement(ConnectionRespondParams.serializer(), params)
val expected =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "account")
},
)
put("op_id", "op-approved-1")
put(
"result",
buildJsonObject {
put(
"targets",
buildJsonArray {
add(
buildJsonObject {
put("name", "linear")
put("status", "approved")
put("detail", "User accepted linear integration")
put(
"env",
buildJsonObject {
put("API_KEY", "dummy_secret_key_123")
},
)
},
)
},
)
},
)
put("profile", "work")
}
assertEquals(expected, encoded)
assertTrue(encoded is JsonObject)
assertEquals(setOf("owner", "op_id", "result", "profile"), (encoded as JsonObject).keys)
val resultObj = (encoded as JsonObject)["result"] as JsonObject
assertEquals(setOf("targets"), resultObj.keys)
assertFalse(resultObj.containsKey("settled_by"))
}
@Test
fun connectionRespondParamsWithSkippedOmittedEnvAndSessionOwnerEncodesExactWireJson() {
val params =
ConnectionRespondParams(
owner = ConnectorOwner.session("sess-xyz"),
opId = "op-skip-2",
result =
ConnectionAnswer(
targets =
listOf(
ConnectionAnswerTarget(
name = "github",
status = "skipped",
detail = null,
env = null,
),
),
settledBy = null,
),
profile = null,
)
val encoded = OkHttpProvider.json.encodeToJsonElement(ConnectionRespondParams.serializer(), params)
val expected =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "sess-xyz")
},
)
put("op_id", "op-skip-2")
put(
"result",
buildJsonObject {
put(
"targets",
buildJsonArray {
add(
buildJsonObject {
put("name", "github")
put("status", "skipped")
},
)
},
)
},
)
}
assertEquals(expected, encoded)
assertTrue(encoded is JsonObject)
assertEquals(setOf("owner", "op_id", "result"), (encoded as JsonObject).keys)
assertFalse((encoded as JsonObject).containsKey("profile"))
val resultObj = (encoded as JsonObject)["result"] as JsonObject
assertEquals(setOf("targets"), resultObj.keys)
assertFalse(resultObj.containsKey("settled_by"))
val targetObj = (resultObj["targets"] as kotlinx.serialization.json.JsonArray)[0] as JsonObject
assertEquals(setOf("name", "status"), targetObj.keys)
assertFalse(targetObj.containsKey("env"))
assertFalse(targetObj.containsKey("detail"))
}
@Test
fun connectionRespondParamsWithContinueAndOmittedTargetsEncodesExactWireJson() {
val params =
ConnectionRespondParams(
owner = ConnectorOwner.session("sess-cont-3"),
opId = "op-cont-3",
result =
ConnectionAnswer(
targets = null,
settledBy = "continue",
),
profile = null,
)
val encoded = OkHttpProvider.json.encodeToJsonElement(ConnectionRespondParams.serializer(), params)
val expected =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "sess-cont-3")
},
)
put("op_id", "op-cont-3")
put(
"result",
buildJsonObject {
put("settled_by", "continue")
},
)
}
assertEquals(expected, encoded)
assertTrue(encoded is JsonObject)
assertEquals(setOf("owner", "op_id", "result"), (encoded as JsonObject).keys)
assertFalse((encoded as JsonObject).containsKey("profile"))
val resultObj = (encoded as JsonObject)["result"] as JsonObject
assertEquals(setOf("settled_by"), resultObj.keys)
assertFalse(resultObj.containsKey("targets"))
}
@Test
fun connectorsOperationWakeUsesConnectorsOperationStatusParamsAndEncodesAllFieldsWhenProfilePresent() {
val params =
ConnectorsOperationStatusParams(
owner = ConnectorOwner.session("sess-wake-1"),
opId = "op-wake-1",
profile = "staging",
)
val encoded = OkHttpProvider.json.encodeToJsonElement(ConnectorsOperationStatusParams.serializer(), params)
val expected =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "sess-wake-1")
},
)
put("op_id", "op-wake-1")
put("profile", "staging")
}
assertEquals(expected, encoded)
assertTrue(encoded is JsonObject)
assertEquals(setOf("owner", "op_id", "profile"), (encoded as JsonObject).keys)
}
@Test
fun connectorsOperationWakeUsesConnectorsOperationStatusParamsAndOmitsProfileWhenNull() {
val params =
ConnectorsOperationStatusParams(
owner = ConnectorOwner.account(),
opId = "op-wake-2",
profile = null,
)
val encoded = OkHttpProvider.json.encodeToJsonElement(ConnectorsOperationStatusParams.serializer(), params)
val expected =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "account")
},
)
put("op_id", "op-wake-2")
}
assertEquals(expected, encoded)
assertTrue(encoded is JsonObject)
assertEquals(setOf("owner", "op_id"), (encoded as JsonObject).keys)
assertFalse((encoded as JsonObject).containsKey("profile"))
}
@Test
fun connectionRespondAndWakeAreRegisteredInGatewayContractRpcMethodsAll() {
val allMethods = RpcMethods.all
val respondMethod = allMethods.firstOrNull { it.name == "connection.respond" }
val wakeMethod = allMethods.firstOrNull { it.name == "connectors.operation.wake" }
org.junit.Assert.assertNotNull("connection.respond must be registered in RpcMethods.all", respondMethod)
org.junit.Assert.assertNotNull("connectors.operation.wake must be registered in RpcMethods.all", wakeMethod)
assertEquals(RpcMethods.CONNECTION_RESPOND, respondMethod)
assertEquals(RpcMethods.CONNECTORS_OPERATION_WAKE, wakeMethod)
}
}
@@ -6,8 +6,13 @@ import com.m57.hermescontrol.data.model.ConnectionOperationTarget
import com.m57.hermescontrol.data.model.ConnectionTargetAction
import com.m57.hermescontrol.data.model.ConnectionTargetKind
import com.m57.hermescontrol.data.model.ConnectionTargetState
import com.m57.hermescontrol.data.remote.OkHttpProvider
import com.m57.hermescontrol.data.ws.HermesWsClient
import com.m57.hermescontrol.data.ws.WsMethods
import com.m57.hermescontrol.data.ws.contract.ConnectionAnswer
import com.m57.hermescontrol.data.ws.contract.ConnectionAnswerTarget
import com.m57.hermescontrol.data.ws.contract.ConnectionRespondParams
import com.m57.hermescontrol.data.ws.contract.ConnectorOwner
import com.m57.hermescontrol.data.ws.contract.ConnectorsOperationStatusParams
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.launch
@@ -15,6 +20,9 @@ import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
import kotlinx.serialization.json.buildJsonArray
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNull
@@ -109,19 +117,65 @@ class ChatConnectionOperationDelegateTest {
)
assertEquals(1, requester.calls.size)
val call = requester.calls.single()
assertEquals(WsMethods.CONNECTION_RESPOND, call.first)
assertEquals(setOf("owner", "op_id", "result"), call.second.keys)
assertEquals(mapOf("type" to "session", "session_id" to "session-a"), call.second["owner"])
assertEquals("op-a", call.second["op_id"])
@Suppress("UNCHECKED_CAST")
val result = call.second["result"] as Map<String, Any>
val action = requester.calls.single()
assertTrue(action is ConnectionOperationRequest.Respond)
val respond = action as ConnectionOperationRequest.Respond
assertEquals(
ConnectionRespondParams(
owner = ConnectorOwner.session("session-a"),
opId = "op-a",
result =
ConnectionAnswer(
targets =
listOf(
ConnectionAnswerTarget(
name = "github",
status = "approved",
env = mapOf("TOKEN" to canary),
),
),
),
),
respond.params,
)
val encoded = OkHttpProvider.json.encodeToJsonElement(ConnectionRespondParams.serializer(), respond.params)
val expectedWire =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "session-a")
},
)
put("op_id", "op-a")
put(
"result",
buildJsonObject {
put(
"targets",
buildJsonArray {
add(
buildJsonObject {
put("name", "github")
put("status", "approved")
put(
"env",
buildJsonObject {
put("TOKEN", canary)
},
)
},
)
},
)
},
)
}
assertEquals(expectedWire, encoded)
@Suppress("UNCHECKED_CAST")
val answer = (result["targets"] as List<Map<String, Any>>).single()
assertEquals("github", answer["name"])
assertEquals("approved", answer["status"])
assertEquals(mapOf("TOKEN" to canary), answer["env"])
assertFalse(
delegate.state.value
.toString()
@@ -166,12 +220,57 @@ class ChatConnectionOperationDelegateTest {
skipDelegate.acceptRequest(snapshot())
skipDelegate.respond("github", mapOf("TOKEN" to "must-drop"), approved = false)
@Suppress("UNCHECKED_CAST")
val skipResult = skipRequester.calls.single().second["result"] as Map<String, Any>
val skipAction = skipRequester.calls.single()
assertTrue(skipAction is ConnectionOperationRequest.Respond)
val skipRespond = skipAction as ConnectionOperationRequest.Respond
assertEquals(
ConnectionRespondParams(
owner = ConnectorOwner.session("session-a"),
opId = "op-a",
result =
ConnectionAnswer(
targets =
listOf(
ConnectionAnswerTarget(
name = "github",
status = "skipped",
),
),
),
),
skipRespond.params,
)
@Suppress("UNCHECKED_CAST")
val skipAnswer = (skipResult["targets"] as List<Map<String, Any>>).single()
assertEquals(mapOf("name" to "github", "status" to "skipped"), skipAnswer)
val encodedSkip =
OkHttpProvider.json.encodeToJsonElement(ConnectionRespondParams.serializer(), skipRespond.params)
val expectedSkipWire =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "session-a")
},
)
put("op_id", "op-a")
put(
"result",
buildJsonObject {
put(
"targets",
buildJsonArray {
add(
buildJsonObject {
put("name", "github")
put("status", "skipped")
},
)
},
)
},
)
}
assertEquals(expectedSkipWire, encodedSkip)
val continueRequester = RecordingRequester()
val continueDelegate = delegate(continueRequester)
@@ -179,14 +278,38 @@ class ChatConnectionOperationDelegateTest {
continueDelegate.acceptRequest(snapshot())
continueDelegate.continueOperation()
val continueAction = continueRequester.calls.single()
assertTrue(continueAction is ConnectionOperationRequest.Respond)
val continueRespond = continueAction as ConnectionOperationRequest.Respond
assertEquals(
mapOf(
"owner" to mapOf("type" to "session", "session_id" to "session-a"),
"op_id" to "op-a",
"result" to mapOf("settled_by" to "continue"),
ConnectionRespondParams(
owner = ConnectorOwner.session("session-a"),
opId = "op-a",
result = ConnectionAnswer(settledBy = "continue"),
),
continueRequester.calls.single().second,
continueRespond.params,
)
val encodedContinue =
OkHttpProvider.json.encodeToJsonElement(ConnectionRespondParams.serializer(), continueRespond.params)
val expectedContinueWire =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "session-a")
},
)
put("op_id", "op-a")
put(
"result",
buildJsonObject {
put("settled_by", "continue")
},
)
}
assertEquals(expectedContinueWire, encodedContinue)
}
@Test
@@ -199,15 +322,31 @@ class ChatConnectionOperationDelegateTest {
delegate.wake("op-a")
val call = requester.calls.single()
assertEquals(WsMethods.CONNECTORS_OPERATION_WAKE, call.first)
val action = requester.calls.single()
assertTrue(action is ConnectionOperationRequest.Wake)
val wake = action as ConnectionOperationRequest.Wake
assertEquals(
mapOf(
"owner" to mapOf("type" to "session", "session_id" to "session-a"),
"op_id" to "op-a",
ConnectorsOperationStatusParams(
owner = ConnectorOwner.session("session-a"),
opId = "op-a",
),
call.second,
wake.params,
)
val encoded =
OkHttpProvider.json.encodeToJsonElement(ConnectorsOperationStatusParams.serializer(), wake.params)
val expectedWire =
buildJsonObject {
put(
"owner",
buildJsonObject {
put("type", "session")
put("session_id", "session-a")
},
)
put("op_id", "op-a")
}
assertEquals(expectedWire, encoded)
}
@Test
@@ -265,7 +404,7 @@ class ChatConnectionOperationDelegateTest {
runTest {
val gate = CompletableDeferred<Unit>()
val requester =
ConnectionOperationRequester { _, _ ->
ConnectionOperationRequester {
gate.await()
throw IllegalStateException("old failure")
}
@@ -355,13 +494,10 @@ class ChatConnectionOperationDelegateTest {
private class RecordingRequester(
private val failure: Throwable? = null,
) : ConnectionOperationRequester {
val calls = mutableListOf<Pair<String, Map<String, Any>>>()
val calls = mutableListOf<ConnectionOperationRequest>()
override suspend fun request(
method: String,
params: Map<String, Any>,
): Any? {
calls += method to params
override suspend fun request(action: ConnectionOperationRequest): Any? {
calls += action
failure?.let { throw it }
return Unit
}
@@ -10,7 +10,9 @@ import com.m57.hermescontrol.data.model.ConnectorPolicy
import com.m57.hermescontrol.data.ws.AccountConnectorRepository
import com.m57.hermescontrol.data.ws.ConnectionOperationParser
import com.m57.hermescontrol.data.ws.WsEvent
import com.m57.hermescontrol.data.ws.WsMethods
import com.m57.hermescontrol.data.ws.contract.ConnectionAnswer
import com.m57.hermescontrol.data.ws.contract.ConnectionRespondParams
import com.m57.hermescontrol.data.ws.contract.ConnectorOwner
import io.mockk.coEvery
import io.mockk.coVerify
import io.mockk.mockk
@@ -143,7 +145,7 @@ class AccountConnectorsViewModelTest {
runTest(dispatcher) {
advanceUntilIdle()
coEvery { repo.connect(any(), any()) } returns AccountConnectorResult.Success(snapshot())
coEvery { repo.operationRequest(any(), any()) } returns mapOf("status" to "ok")
coEvery { repo.operationRespond(any()) } returns mapOf("status" to "ok")
coEvery { repo.operationStatus("op-a") } returns
AccountConnectorResult.Success(snapshot().copy(seq = 2, settled = true))
vm.connect("drive")
@@ -151,12 +153,11 @@ class AccountConnectorsViewModelTest {
vm.continueOperation()
advanceUntilIdle()
coVerify {
repo.operationRequest(
WsMethods.CONNECTION_RESPOND,
mapOf(
"owner" to mapOf("type" to "account"),
"op_id" to "op-a",
"result" to mapOf("settled_by" to "continue"),
repo.operationRespond(
ConnectionRespondParams(
owner = ConnectorOwner.account(),
opId = "op-a",
result = ConnectionAnswer(settledBy = "continue"),
),
)
}