fix: publish audits that share a name as separate snapshot nodes

benchmarks/duckshop/edge_cases/orchestration/shared_inline_audits.json

@@ -0,0 +1,15 @@
+{
+  "description": "Two models attach the same inline audit to a column of the same name, and the same explicitly named audit. Each attachment is its own snapshot node.",
+  "files": {
+    "sqlbuild_project.toml": "name = \"orders\"\nadapter = \"duckdb\"\n[connections.local]\ndatabase = \"warehouse.duckdb\"\n[targets.development]\nschema = \"main\"\n",
+    "models/orders.sql": "MODEL (\n  materialized table,\n  description \"Synthetic rows.\",\n  columns (id (audits [not_null, unique, accepted_values (name \"id_known\", values [1, 2])])),\n);\n\nSELECT 1 AS id\n",
+    "models/customers.sql": "MODEL (\n  materialized table,\n  description \"Synthetic rows.\",\n  columns (id (audits [not_null, accepted_values (name \"id_known\", values [1, 2])])),\n);\n\nSELECT 2 AS id\n"
+  },
+  "expected_audits": [
+    "audit:id_known:model:customers:id",
+    "audit:id_known:model:orders:id",
+    "audit:not_null:model:customers:id",
+    "audit:not_null:model:orders:id",
+    "audit:unique:model:orders:id"
+  ]
+}

crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot.rs

@@ -75,9 +75,12 @@ pub(crate) fn insert_node(
     key: String,
     node: SnapshotNode,
 ) -> PyResult<BTreeMap<String, SnapshotNode>> {
-    if nodes.insert(key, node).is_some() {
-        return Err(PyValueError::new_err("duplicate snapshot node identity"));
+    if nodes.contains_key(&key) {
+        return Err(PyValueError::new_err(format!(
+            "duplicate snapshot node identity {key}"
+        )));
     }
+    nodes.insert(key, node);
     Ok(nodes)
 }
 

crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_resources.rs

@@ -4,6 +4,7 @@ use std::collections::BTreeMap;
 
 use pyo3::prelude::{PyAnyMethods, PyResult, Python};
 use serde_json::{Value, json};
+use sqlbuild_core::nested_progress::main::audit_resource_id::audit_resource_id;
 use sqlbuild_orchestration::snapshots::models::SnapshotNode;
 
 use crate::bindings::_helpers::compiled_project::project::NativeCompiledProject;
@@ -101,9 +102,17 @@ pub(crate) fn resource_nodes(
         let target: Option<Value> = attached.target.as_ref().map(
             |target| json!({"kind": target.kind, "name": target.name, "column": target.column}),
         );
+        let key = match &attached.target {
+            Some(target) => audit_resource_id(
+                attached.compiled_name(),
+                &target.kind,
+                (Some(&target.name), target.column.as_deref()),
+            ),
+            None => format!("audit:{}", attached.compiled_name()),
+        };
         nodes = insert_node(
             nodes,
-            format!("audit:{}", attached.compiled_name()),
+            key,
             SnapshotNode {
                 kind: "audit".into(),
                 definition: json!({

tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py

@@ -313,6 +313,15 @@ class SchedulerCrashTestCase:
     expected_minimum_kills: int
 
 
+@dataclass(frozen=True)
+class SharedAuditPublicationTestCase:
+    description: str
+    create_project: Callable[[Path], None]
+    target: str
+    expected_audits: frozenset[str]
+    expected_count: int
+
+
 @dataclass(frozen=True)
 class ProjectRolesTestCase:
     description: str

tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py

@@ -1221,7 +1221,7 @@ def scheduler_crash_history(directory: Path, fixture: dict[str, Any]) -> dict[st
 
 
 def publish_generation(
-    project: Path, api: SnapshotApi, generation: int
+    project: Path, api: SnapshotApi, generation: int, target: str = "development"
 ) -> subprocess.CompletedProcess[str]:
     """Publish one generation of a project through the installed location command."""
     return subprocess.run(
@@ -1234,7 +1234,7 @@ def publish_generation(
             "--location-id",
             api.location_id,
             "--target",
-            "development",
+            target,
             "--server",
             api.url,
             "--generation",
@@ -1326,3 +1326,29 @@ def request_job_run(worker: ScheduledWorker, job: str) -> dict[str, Any]:
     )
     assert requested.returncode == 0, requested.stderr
     return wait_for_run(worker.api, worker.process, worker.log, run_id=run_id)
+
+
+SHARED_AUDITS: Path = Path("benchmarks/duckshop/edge_cases/orchestration/shared_inline_audits.json")
+
+
+def create_playground(project: Path) -> None:
+    """Create the default playground project with the installed CLI."""
+    created: subprocess.CompletedProcess[str] = subprocess.run(
+        [str(Path(sys.executable).with_name("sqb")), "playground", str(project)],
+        capture_output=True,
+        text=True,
+        timeout=60,
+        check=False,
+    )
+    assert created.returncode == 0, created.stderr
+
+
+def write_shared_audits(project: Path) -> None:
+    """Write the DuckShop project whose models share inline and named audits."""
+    write_files(project_dir=project, files=json.loads(SHARED_AUDITS.read_text())["files"])
+
+
+def audit_node_keys(api: SnapshotApi, target: str) -> frozenset[str]:
+    """Keys of the current snapshot's audit nodes."""
+    nodes: dict[str, Any] = current_snapshot(api, target)["nodes"]
+    return frozenset(filter(lambda key: nodes[key]["kind"] == "audit", nodes))

tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_shared_audits.py

@@ -0,0 +1,69 @@
+"""Audits sharing a name across attachments publish as one snapshot node per attachment."""
+
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from tests.integration.src.sqlbuild.compiler.compiled_project.publication._test_types import (
+    SharedAuditPublicationTestCase,
+)
+from tests.integration.src.sqlbuild.compiler.compiled_project.publication.helpers import (
+    SHARED_AUDITS,
+    audit_node_keys,
+    create_playground,
+    current_snapshot,
+    publish_generation,
+    snapshot_api,
+    write_shared_audits,
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SharedAuditPublicationTestCase(
+            description="the default playground publishes every inline audit attachment",
+            create_project=create_playground,
+            target="dev",
+            expected_audits=frozenset(
+                {
+                    "audit:not_null:model:stg_customers:customer_id",
+                    "audit:not_null:model:stg_orders:order_id",
+                    "audit:unique:model:stg_customers:customer_id",
+                    "audit:unique:seed:waffle_types:waffle_type_id",
+                    "audit:revenue_is_non_negative:model:daily_revenue",
+                }
+            ),
+            expected_count=30,
+        ),
+        SharedAuditPublicationTestCase(
+            description="DuckShop shared inline and named audits publish once per attachment",
+            create_project=write_shared_audits,
+            target="development",
+            expected_audits=frozenset(json.loads(SHARED_AUDITS.read_text())["expected_audits"]),
+            expected_count=5,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_audits_sharing_a_name_when_published_then_each_attachment_is_a_stable_node(
+    test_case: SharedAuditPublicationTestCase, tmp_path: Path
+) -> None:
+    project: Path = tmp_path / "project"
+    test_case.create_project(project)
+    with snapshot_api(tmp_path / "control") as api:
+        first: subprocess.CompletedProcess[str] = publish_generation(
+            project, api, 1, test_case.target
+        )
+        assert first.returncode == 0, first.stderr
+        audits: frozenset[str] = audit_node_keys(api, test_case.target)
+        assert test_case.expected_audits <= audits
+        assert len(audits) == test_case.expected_count
+        second: subprocess.CompletedProcess[str] = publish_generation(
+            project, api, 2, test_case.target
+        )
+        assert second.returncode == 0, second.stderr
+        assert audit_node_keys(api, test_case.target) == audits
+        assert current_snapshot(api, test_case.target)["changes"] == []