"""Synapse s2s gateway tests — signature verification and target CRUD."""
import json
from unittest.mock import AsyncMock, MagicMock
import pytest
from fastapi.testclient import TestClient
from navi.config import Settings
S2S_SECRET = "s2s-target-secret-0123456789"
@pytest.fixture
def synapse_client(monkeypatch):
import navi.api.routes.synapse as synapse_mod
import navi.auth.deps as auth_deps
import navi.synapse.reactions as reactions
# Anonymous admin (auth disabled) — CRUD routes resolve without a cookie.
monkeypatch.setattr(
auth_deps, "settings",
Settings(_env_file=None, navi_persona_file="", navi_auth_enabled=False),
)
monkeypatch.setattr(synapse_mod, "_pool", AsyncMock(return_value=MagicMock()))
# Don't spawn real reaction runs from tests; collect the calls instead.
reactions_scheduled: list = []
monkeypatch.setattr(
reactions, "schedule_reaction",
lambda envelope, user_id, event_type: reactions_scheduled.append(
(envelope, user_id, event_type)
),
)
from navi.main import app
client = TestClient(app)
client.reactions_scheduled = reactions_scheduled
return client
def _target(**overrides):
target = {"id": 1, "user_id": "u1", "token_ref": "tgt-1", "secret": S2S_SECRET}
target.update(overrides)
return target
def _sign(body: bytes, secret: str = S2S_SECRET):
from gnexus_synapse import make_signature
return {"X-Gnexus-Signature": make_signature(body, secret),
"X-Gnexus-Event-Type": "gntodo.task.created"}
def _post(tc: TestClient, url: str, body: dict, secret: str = S2S_SECRET):
raw = json.dumps(body).encode()
return tc.post(url, content=raw, headers=_sign(raw, secret))
def test_synapse_delivery_503_without_targets(synapse_client, monkeypatch):
import navi.api.routes.synapse as synapse_mod
monkeypatch.setattr(
synapse_mod.SynapseTargetStore, "active_secrets",
AsyncMock(return_value=[]),
)
envelope = {"event_id": "e1", "subject": "topic", "action": "done"}
resp = _post(synapse_client, "/webhooks/synapse", envelope)
assert resp.status_code == 503
def test_synapse_delivery_401_on_foreign_secret(synapse_client, monkeypatch):
import navi.api.routes.synapse as synapse_mod
monkeypatch.setattr(
synapse_mod.SynapseTargetStore, "active_secrets",
AsyncMock(return_value=[_target()]),
)
envelope = {"event_id": "e2", "subject": "topic", "action": "done"}
resp = _post(synapse_client, "/webhooks/synapse", envelope, secret="wrong-secret-0123")
assert resp.status_code == 401
def test_synapse_delivery_401_on_broken_json(synapse_client, monkeypatch):
"""A correctly signed non-JSON body must not leak a 500 — it is a 401."""
import navi.api.routes.synapse as synapse_mod
monkeypatch.setattr(
synapse_mod.SynapseTargetStore, "active_secrets",
AsyncMock(return_value=[_target()]),
)
resp = synapse_client.post(
"/webhooks/synapse", content=b"not-json{", headers=_sign(b"not-json{"),
)
assert resp.status_code == 401
def test_synapse_delivery_200_and_both_spellings(synapse_client, monkeypatch):
import navi.api.routes.synapse as synapse_mod
monkeypatch.setattr(
synapse_mod.SynapseTargetStore, "active_secrets",
AsyncMock(return_value=[_target()]),
)
envelope = {
"event_id": "e3", "subject": "task", "action": "completed",
"payload": {"user_id": "u1"}, "source": "gntodo",
}
client = synapse_client
resp = _post(client, "/webhooks/synapse", envelope)
assert resp.status_code == 200
assert resp.json() == {"received": True, "event_id": "e3"}
resp_slash = _post(client, "/webhooks/synapse/", envelope)
assert resp_slash.status_code == 200
def test_synapse_duplicate_delivery_acknowledged_once(synapse_client, monkeypatch):
"""A replayed event id is acked again but only marked once."""
import navi.api.routes.synapse as synapse_mod
monkeypatch.setattr(
synapse_mod.SynapseTargetStore, "active_secrets",
AsyncMock(return_value=[_target()]),
)
synapse_mod._seen_events.clear()
envelope = {"event_id": "e4", "subject": "t", "action": "a"}
client = synapse_client
first = _post(client, "/webhooks/synapse", envelope)
assert first.status_code == 200
second = _post(client, "/webhooks/synapse", envelope)
assert second.status_code == 200
assert synapse_mod._seen_events["e4"] is True
def test_synapse_delivery_schedules_reaction(synapse_client, monkeypatch):
"""A fresh, verified delivery hands the envelope to the reaction layer."""
import navi.api.routes.synapse as synapse_mod
monkeypatch.setattr(
synapse_mod.SynapseTargetStore, "active_secrets",
AsyncMock(return_value=[_target()]),
)
synapse_mod._seen_events.clear()
envelope = {
"event_id": "e9", "subject": "task", "action": "created",
"payload": {"user_id": "u1"}, "source": "gntodo",
}
resp = _post(synapse_client, "/webhooks/synapse", envelope)
assert resp.status_code == 200
assert synapse_client.reactions_scheduled == [(
envelope, "u1", "gntodo.task.created",
)]
def test_synapse_replayed_delivery_skips_reaction(synapse_client, monkeypatch):
"""A replayed event id is acked again but does not react twice."""
import navi.api.routes.synapse as synapse_mod
monkeypatch.setattr(
synapse_mod.SynapseTargetStore, "active_secrets",
AsyncMock(return_value=[_target()]),
)
synapse_mod._seen_events.clear()
envelope = {"event_id": "e10", "subject": "t", "action": "a"}
first = _post(synapse_client, "/webhooks/synapse", envelope)
second = _post(synapse_client, "/webhooks/synapse", envelope)
assert first.status_code == second.status_code == 200
assert len(synapse_client.reactions_scheduled) == 1
def test_synapse_target_crud(synapse_client, monkeypatch):
import navi.api.routes.synapse as synapse_mod
import navi.synapse.store as store_mod
created = MagicMock(
id=7, token_ref="tgt-9",
created_at=__import__("datetime").datetime(2026, 10, 7, 12, 0, 0),
)
monkeypatch.setattr(store_mod.SynapseTargetStore, "create", AsyncMock(return_value=created))
monkeypatch.setattr(store_mod.SynapseTargetStore, "list_for_user", AsyncMock(return_value=[created]))
monkeypatch.setattr(store_mod.SynapseTargetStore, "revoke", AsyncMock(return_value=True))
client = synapse_client
created_resp = client.post(
"/synapse-targets",
json={"token_ref": "tgt-9", "secret": S2S_SECRET},
)
assert created_resp.status_code == 200
assert created_resp.json()["token_ref"] == "tgt-9"
listed = client.get("/synapse-targets")
assert listed.status_code == 200
assert listed.json()["items"][0]["token_ref"] == "tgt-9"
revoked = client.delete("/synapse-targets/7")
assert revoked.status_code == 204
def test_synapse_settings_get_defaults(synapse_client, monkeypatch):
"""No row yet — the endpoint returns the defaults, not a 500."""
import navi.api.routes.synapse as synapse_mod
from navi.synapse.settings_store import SynapseSettings
async def fake_get(self, user_id):
return SynapseSettings(user_id=user_id)
monkeypatch.setattr(synapse_mod.SynapseSettingsStore, "get", fake_get)
resp = synapse_client.get("/synapse-settings")
assert resp.status_code == 200
body = resp.json()
assert body["reactions_enabled"] is False
assert body["push_target"] == "app"
assert body["completion_notify"] == "important"
assert body["instructions"] == ""
# tells the UI whether Synapse-linked options are usable (no keys in tests)
assert body["source_ready"] is False
def test_synapse_settings_put_records_user_edit(synapse_client, monkeypatch):
"""Changing instructions via PUT stores an edited_by='user' version."""
import navi.api.routes.synapse as synapse_mod
from navi.synapse.settings_store import SynapseSettings
current = SynapseSettings(user_id="u1", instructions="")
saved_instructions: list = []
async def fake_get(self, user_id):
return current
async def fake_save_instructions(self, user_id, content, edited_by, reason=None):
saved_instructions.append((content, edited_by, reason))
current.instructions = content
async def fake_save(self, settings_obj):
pass
monkeypatch.setattr(synapse_mod.SynapseSettingsStore, "get", fake_get)
monkeypatch.setattr(synapse_mod.SynapseSettingsStore, "save_instructions", fake_save_instructions)
monkeypatch.setattr(synapse_mod.SynapseSettingsStore, "save", fake_save)
resp = synapse_client.put(
"/synapse-settings",
json={
"reactions_enabled": True,
"push_target": "app_synapse",
"completion_notify": "always",
"instructions": "Тот, кто приходит от gntodo, отвечает.",
},
)
assert resp.status_code == 200
assert resp.json()["reactions_enabled"] is True
assert saved_instructions == [(
"Тот, кто приходит от gntodo, отвечает.", "user", None
)]
def test_synapse_settings_put_same_instructions_skips_version(synapse_client, monkeypatch):
import navi.api.routes.synapse as synapse_mod
from navi.synapse.settings_store import SynapseSettings
current = SynapseSettings(user_id="u1")
async def fake_get(self, user_id):
return current
async def fake_save_instructions(self, *args, **kwargs):
raise AssertionError("unchanged instructions must not record a version")
async def fake_save(self, settings_obj):
pass
monkeypatch.setattr(synapse_mod.SynapseSettingsStore, "get", fake_get)
monkeypatch.setattr(synapse_mod.SynapseSettingsStore, "save_instructions", fake_save_instructions)
monkeypatch.setattr(synapse_mod.SynapseSettingsStore, "save", fake_save)
resp = synapse_client.put(
"/synapse-settings",
json={
"reactions_enabled": False,
"push_target": "app",
"completion_notify": "never",
"instructions": "",
},
)
assert resp.status_code == 200
def test_synapse_settings_validation(synapse_client, monkeypatch):
import navi.api.routes.synapse as synapse_mod
monkeypatch.setattr(
synapse_mod, "_pool", AsyncMock(return_value=MagicMock())
)
resp = synapse_client.put(
"/synapse-settings",
json={
"reactions_enabled": True,
"push_target": "carrier_pigeon",
"completion_notify": "always",
"instructions": "",
},
)
assert resp.status_code == 422
