feat(05-01): wire diff generation into snapshot subscriber
- Add RETURNING id to snapshot INSERT for new_snapshot_id capture - Call generate_and_store_diff after successful commit (best-effort) - Outer try/except safety net ensures snapshot ack never blocked by diff - Update subscriber tests to mock diff service Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -20,6 +20,7 @@ from sqlalchemy.exc import IntegrityError, OperationalError
|
|||||||
|
|
||||||
from app.config import settings
|
from app.config import settings
|
||||||
from app.database import AdminAsyncSessionLocal
|
from app.database import AdminAsyncSessionLocal
|
||||||
|
from app.services.config_diff_service import generate_and_store_diff
|
||||||
from app.services.openbao_service import OpenBaoTransitService
|
from app.services.openbao_service import OpenBaoTransitService
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -134,12 +135,13 @@ async def handle_config_snapshot(msg) -> None:
|
|||||||
|
|
||||||
# --- INSERT new snapshot ---
|
# --- INSERT new snapshot ---
|
||||||
try:
|
try:
|
||||||
await session.execute(
|
insert_result = await session.execute(
|
||||||
text(
|
text(
|
||||||
"INSERT INTO router_config_snapshots "
|
"INSERT INTO router_config_snapshots "
|
||||||
"(device_id, tenant_id, config_text, sha256_hash, collected_at) "
|
"(device_id, tenant_id, config_text, sha256_hash, collected_at) "
|
||||||
"VALUES (CAST(:device_id AS uuid), CAST(:tenant_id AS uuid), "
|
"VALUES (CAST(:device_id AS uuid), CAST(:tenant_id AS uuid), "
|
||||||
":config_text, :sha256_hash, :collected_at)"
|
":config_text, :sha256_hash, :collected_at) "
|
||||||
|
"RETURNING id"
|
||||||
),
|
),
|
||||||
{
|
{
|
||||||
"device_id": device_id,
|
"device_id": device_id,
|
||||||
@@ -149,6 +151,7 @@ async def handle_config_snapshot(msg) -> None:
|
|||||||
"collected_at": collected_at,
|
"collected_at": collected_at,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
new_snapshot_id = insert_result.scalar_one()
|
||||||
await session.commit()
|
await session.commit()
|
||||||
except IntegrityError:
|
except IntegrityError:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
@@ -170,6 +173,15 @@ async def handle_config_snapshot(msg) -> None:
|
|||||||
await msg.nak()
|
await msg.nak()
|
||||||
return
|
return
|
||||||
|
|
||||||
|
# --- Diff generation (best-effort) ---
|
||||||
|
try:
|
||||||
|
await generate_and_store_diff(device_id, tenant_id, str(new_snapshot_id), session)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.warning(
|
||||||
|
"Diff generation failed for device %s (non-fatal): %s",
|
||||||
|
device_id, exc,
|
||||||
|
)
|
||||||
|
|
||||||
logger.info(
|
logger.info(
|
||||||
"Config snapshot stored for device %s tenant %s",
|
"Config snapshot stored for device %s tenant %s",
|
||||||
device_id,
|
device_id,
|
||||||
|
|||||||
@@ -62,6 +62,9 @@ async def test_new_snapshot_encrypted_and_stored():
|
|||||||
), patch(
|
), patch(
|
||||||
"app.services.config_snapshot_subscriber.OpenBaoTransitService",
|
"app.services.config_snapshot_subscriber.OpenBaoTransitService",
|
||||||
return_value=mock_openbao,
|
return_value=mock_openbao,
|
||||||
|
), patch(
|
||||||
|
"app.services.config_snapshot_subscriber.generate_and_store_diff",
|
||||||
|
new_callable=AsyncMock,
|
||||||
):
|
):
|
||||||
await handle_config_snapshot(msg)
|
await handle_config_snapshot(msg)
|
||||||
|
|
||||||
@@ -248,6 +251,9 @@ async def test_first_snapshot_for_device_always_stored():
|
|||||||
), patch(
|
), patch(
|
||||||
"app.services.config_snapshot_subscriber.OpenBaoTransitService",
|
"app.services.config_snapshot_subscriber.OpenBaoTransitService",
|
||||||
return_value=mock_openbao,
|
return_value=mock_openbao,
|
||||||
|
), patch(
|
||||||
|
"app.services.config_snapshot_subscriber.generate_and_store_diff",
|
||||||
|
new_callable=AsyncMock,
|
||||||
):
|
):
|
||||||
await handle_config_snapshot(msg)
|
await handle_config_snapshot(msg)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user