Closes #18159: Expose snapshots to Event Rule condition evaluation (#22637)

Expose an event's prechange and postchange snapshots to event rule
condition evaluation, making snapshots.prechange.<attr> and
snapshots.postchange.<attr> available through the existing dot-path
syntax.

Add changed and unchanged snapshot operators for comparing an attribute
across the two snapshots without requiring a condition value. These
operators support rules such as firing only when a field transitions to a
specific state.

Make condition values optional only for snapshot operators by introducing
a missing-value sentinel, while preserving value requirements for all
other operators. Reject invalid combinations such as using changed or
unchanged with an explicit value or with an attr starting with snapshots.

Fail closed when condition paths traverse invalid snapshot structures,
including raw scalar snapshot values such as status strings, by treating
unresolvable snapshot-operator paths as missing and converting invalid
direct paths to InvalidCondition.

Document the new snapshot path syntax, changed and unchanged operators,
create/delete snapshot behavior, and the serialization differences
between snapshot data and REST API data. Add regression and integration
tests covering validation, transition behavior, null snapshot edge cases,
direct snapshot paths, and event rule evaluation.

docs/reference/conditions.md

@@ -9,7 +9,7 @@ A condition is expressed as a JSON object with the following keys:
 | Key name | Required | Default | Description |
 |----------|----------|---------|-------------|
 | attr     | Yes      | -       | Name of the key within the data being evaluated |
-| value    | Yes      | -       | The reference value to which the given data will be compared |
+| value    | See note | -       | The reference value to which the given data will be compared. Not used by snapshot operators (`changed`, `unchanged`). |
 | op       | No       | `eq`    | The logical operation to be performed |
 | negate   | No       | False   | Negate (invert) the result of the condition's evaluation |
 
@@ -22,6 +22,9 @@ A condition is expressed as a JSON object with the following keys:
 * `lte`: Less than or equal to
 * `in`: Is present within a list of values
 * `contains`: Contains the specified value
+* `regex`: Matches a regular expression
+* `changed`: The attribute's value differs between the pre-change and post-change snapshots (no `value` required)
+* `unchanged`: The attribute's value is the same in both snapshots (no `value` required)
 
 ### Accessing Nested Keys
 
@@ -91,6 +94,59 @@ The following condition will evaluate as true:
 !!! note "Evaluating static choice fields"
     Pay close attention when evaluating static choice fields, such as the `status` field above. These fields typically render as a dictionary specifying both the field's raw value (`value`) and its human-friendly label (`label`). Be sure to specify on which of these you want to match.
 
+## Snapshot Conditions (Event Rules)
+
+When used in an [event rule](../features/event-rules.md), conditions can also inspect the **pre-change and post-change snapshots** captured at the time of the event. This allows rules to fire only when a specific field actually changes value, rather than whenever it has a particular value.
+
+### Snapshot Operators
+
+The `changed` and `unchanged` operators compare an attribute's value across the two snapshots. They do not accept a `value` key.
+
+Fire only when `status` changes (to any value):
+
+```json
+{
+  "attr": "status",
+  "op": "changed"
+}
+```
+
+### Combining with Standard Conditions
+
+The canonical use case — fire only when `status` changes **to** `active` — combines a standard value check with the `changed` operator:
+
+```json
+{
+  "and": [
+    {
+      "attr": "status.value",
+      "value": "active"
+    },
+    {
+      "attr": "status",
+      "op": "changed"
+    }
+  ]
+}
+```
+
+### Direct Snapshot Path Access
+
+You can also read pre- or post-change values directly using the `snapshots.prechange.<attr>` and `snapshots.postchange.<attr>` dot-path syntax with any standard operator:
+
+```json
+{
+  "attr": "snapshots.prechange.status",
+  "value": "planned"
+}
+```
+
+!!! warning "Snapshot serialization format"
+    Snapshot data uses the **model serializer format**, not the REST API format. Choice fields such as `status` are stored as raw strings (e.g. `"active"`) rather than nested objects (e.g. `{"value": "active", "label": "Active"}`). Use `attr: "snapshots.prechange.status"` — not `"snapshots.prechange.status.value"` — when referencing snapshot attributes. The `changed`/`unchanged` operators compare the same format on both sides, so they are not affected by this distinction.
+
+!!! note "Snapshot availability"
+    Snapshots are only populated for update and delete events. For create events, `prechange` is `null` — conditions using the `changed` operator on a create event evaluate to `true` (the field transitioned from non-existent to its initial value), while conditions using `snapshots.prechange.*` paths evaluate to `false`. For delete events, `postchange` is `null` — the `changed` operator evaluates to `true` for any attribute present in the prechange snapshot, and `unchanged` evaluates to `false`.
+
 ## Condition Sets
 
 Multiple conditions can be combined into nested sets using AND or OR logic. This is done by declaring a JSON object with a single key (`and` or `or`) containing a list of condition objects and/or child condition sets.

netbox/extras/conditions.py

@@ -13,6 +13,12 @@ __all__ = (
 AND = 'and'
 OR = 'or'
 
+# Sentinel for a snapshot attribute that could not be resolved (missing key or
+# null snapshot).  Using a unique object ensures that two independently
+# unresolvable values compare equal to each other, which is the correct
+# semantics for the 'unchanged' operator when neither snapshot has the field.
+_MISSING = object()
+
 
 def is_ruleset(data):
     """
@@ -30,8 +36,9 @@ class Condition:
     An individual conditional rule that evaluates a single attribute and its value.
 
     :param attr: The name of the attribute being evaluated
-    :param value: The value being compared
+    :param value: The value being compared (not used by snapshot operators)
     :param op: The logical operation to use when evaluating the value (default: 'eq')
+    :param negate: Invert the result of evaluation
     """
     EQ = 'eq'
     GT = 'gt'
@@ -41,11 +48,16 @@ class Condition:
     IN = 'in'
     CONTAINS = 'contains'
     REGEX = 'regex'
+    CHANGED = 'changed'
+    UNCHANGED = 'unchanged'
 
     OPERATORS = (
-        EQ, GT, GTE, LT, LTE, IN, CONTAINS, REGEX
+        EQ, GT, GTE, LT, LTE, IN, CONTAINS, REGEX, CHANGED, UNCHANGED
     )
 
+    # Operators that compare pre/post snapshots and do not accept a value.
+    SNAPSHOT_OPERATORS = (CHANGED, UNCHANGED)
+
     TYPES = {
         str: (EQ, CONTAINS, REGEX),
         bool: (EQ, CONTAINS),
@@ -55,25 +67,44 @@ class Condition:
         type(None): (EQ,)
     }
 
-    def __init__(self, attr, value, op=EQ, negate=False):
+    def __init__(self, attr, value=_MISSING, op=EQ, negate=False):
         if op not in self.OPERATORS:
             raise ValueError(_("Unknown operator: {op}. Must be one of: {operators}").format(
                 op=op, operators=', '.join(self.OPERATORS)
             ))
-        if type(value) not in self.TYPES:
-            raise ValueError(_("Unsupported value type: {value}").format(value=type(value)))
-        if op not in self.TYPES[type(value)]:
-            raise ValueError(_("Invalid type for {op} operation: {value}").format(op=op, value=type(value)))
+
+        if op in self.SNAPSHOT_OPERATORS:
+            if value is not _MISSING:
+                raise ValueError(_(
+                    "The '{op}' operator compares snapshots and does not accept a value."
+                ).format(op=op))
+            if attr.startswith('snapshots.'):
+                raise ValueError(_(
+                    "The '{op}' operator resolves '{attr}' within each snapshot dict, not the "
+                    "top-level condition context. Use the bare attribute name (e.g. 'status') "
+                    "rather than a snapshot path (e.g. 'snapshots.prechange.status'), which is "
+                    "only valid with standard operators."
+                ).format(op=op, attr=attr))
+            self.value = _MISSING
+        else:
+            if value is _MISSING:
+                raise ValueError(_("A value is required for the '{op}' operator.").format(op=op))
+            if type(value) not in self.TYPES:
+                raise ValueError(_("Unsupported value type: {value}").format(value=type(value)))
+            if op not in self.TYPES[type(value)]:
+                raise ValueError(_("Invalid type for {op} operation: {value}").format(op=op, value=type(value)))
+            self.value = value
 
         self.attr = attr
-        self.value = value
         self.op = op
         self.eval_func = getattr(self, f'eval_{op}')
         self.negate = negate
 
-    def eval(self, data):
+    def _resolve_attr(self, data):
         """
-        Evaluate the provided data to determine whether it matches the condition.
+        Walk self.attr as a dotted key path through data. Raises InvalidCondition on
+        missing keys, or when an intermediate value can't be indexed by key (e.g. a
+        REST API-style path like 'status.value' applied to a raw snapshot value).
         """
         def _get(obj, key):
             if isinstance(obj, list):
@@ -81,9 +112,46 @@ class Condition:
             return operator.getitem(obj or {}, key)
 
         try:
-            value = functools.reduce(_get, self.attr.split('.'), data)
+            return functools.reduce(_get, self.attr.split('.'), data)
         except KeyError:
             raise InvalidCondition(f"Invalid key path: {self.attr}")
+        except TypeError as e:
+            raise InvalidCondition(f"Invalid key path: {self.attr} ({e})")
+
+    def _resolve_snapshot_attr(self, snapshot):
+        """
+        Walk self.attr through a snapshot dict, returning _MISSING on any miss.
+        Snapshots use the model serializer format (raw field values), not the REST
+        API format, so e.g. status is stored as "active" not {"value": "active"}.
+        """
+        if snapshot is None:
+            return _MISSING
+        try:
+            obj = snapshot
+            for key in self.attr.split('.'):
+                if isinstance(obj, list):
+                    obj = [operator.getitem(item or {}, key) for item in obj]
+                else:
+                    obj = operator.getitem(obj or {}, key)
+            return obj
+        except (KeyError, TypeError):
+            return _MISSING
+
+    def eval(self, data):
+        """
+        Evaluate the provided data to determine whether it matches the condition.
+        """
+        if self.op in self.SNAPSHOT_OPERATORS:
+            snapshots = data.get('snapshots') if isinstance(data, dict) else None
+            if snapshots is None:
+                raise InvalidCondition(
+                    f"No snapshot data available for '{self.op}' operator. "
+                    f"Snapshot operators are only meaningful on update and delete events."
+                )
+            result = self.eval_func(snapshots)
+            return not result if self.negate else result
+
+        value = self._resolve_attr(data)
         try:
             result = self.eval_func(value)
         except TypeError as e:
@@ -128,6 +196,27 @@ class Condition:
     def eval_regex(self, value):
         return re.match(self.value, value) is not None
 
+    # Snapshot comparison operators
+    # These resolve self.attr in both the prechange and postchange snapshots and
+    # compare the resulting values.  _MISSING is used when a snapshot is absent
+    # or does not contain the attribute.
+    #
+    # Fail-closed semantics:
+    #   changed:   False when attr is absent from both snapshots (field never existed)
+    #   unchanged: False when attr is absent from both snapshots (avoids silent pass on typos)
+
+    def eval_changed(self, snapshots):
+        pre = self._resolve_snapshot_attr(snapshots.get('prechange'))
+        post = self._resolve_snapshot_attr(snapshots.get('postchange'))
+        return pre != post
+
+    def eval_unchanged(self, snapshots):
+        pre = self._resolve_snapshot_attr(snapshots.get('prechange'))
+        post = self._resolve_snapshot_attr(snapshots.get('postchange'))
+        if pre is _MISSING and post is _MISSING:
+            return False
+        return pre == post
+
 
 class ConditionSet:
     """

netbox/extras/events.py

@@ -176,8 +176,13 @@ def process_event_rules(event_rules, object_type, event):
 
     for event_rule in event_rules:
 
-        # Evaluate event rule conditions (if any)
-        if not event_rule.eval_conditions(event['data']):
+        # Evaluate event rule conditions (if any).
+        # Snapshots are merged into the condition context so conditions can
+        # reference snapshots.prechange.<attr> and snapshots.postchange.<attr>
+        # using the standard dot-path syntax, and so the 'changed'/'unchanged'
+        # operators can access pre/post values.
+        condition_data = {**event['data'], 'snapshots': event.get('snapshots')}
+        if not event_rule.eval_conditions(condition_data):
             continue
 
         # Guard against action_data that is valid JSON but not a dict

netbox/extras/tests/test_conditions.py

@@ -321,3 +321,249 @@ class ConditionSetTestCase(TestCase):
         })
 
         self.assertFalse(form.is_valid())
+
+
+class SnapshotConditionTestCase(TestCase):
+    """
+    Tests for snapshot-aware conditions: the 'changed'/'unchanged' operators and
+    direct snapshot attribute access via the snapshots.prechange.* / snapshots.postchange.*
+    dot-path syntax.
+    """
+
+    def _make_condition_data(self, site, snapshots):
+        """Return a condition evaluation context as produced by process_event_rules()."""
+        return {**serialize_for_event(site), 'snapshots': snapshots}
+
+    #
+    # Validation
+    #
+
+    def test_changed_operator_rejects_value(self):
+        with self.assertRaises(ValueError):
+            Condition('status', value='active', op='changed')
+
+    def test_unchanged_operator_rejects_value(self):
+        with self.assertRaises(ValueError):
+            Condition('status', value='active', op='unchanged')
+
+    def test_snapshot_operator_rejects_snapshot_path_attr(self):
+        """Snapshot operators must not use a snapshots.prechange.* path — that's only for standard operators."""
+        with self.assertRaises(ValueError):
+            Condition('snapshots.prechange.status', op='changed')
+        with self.assertRaises(ValueError):
+            Condition('snapshots.postchange.status', op='unchanged')
+
+    def test_standard_operator_requires_value(self):
+        with self.assertRaises(ValueError):
+            Condition('status', op='eq')
+
+    #
+    # 'changed' operator
+    #
+
+    def test_changed_true_when_attr_differs(self):
+        c = Condition('status', op='changed')
+        snapshots = {
+            'prechange': {'status': 'planned'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertTrue(c.eval({'snapshots': snapshots}))
+
+    def test_changed_false_when_attr_same(self):
+        c = Condition('status', op='changed')
+        snapshots = {
+            'prechange': {'status': 'active'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertFalse(c.eval({'snapshots': snapshots}))
+
+    def test_changed_true_when_prechange_missing_attr(self):
+        # attr present in postchange but absent from prechange snapshot
+        c = Condition('description', op='changed')
+        snapshots = {
+            'prechange': {},
+            'postchange': {'description': 'hello'},
+        }
+        self.assertTrue(c.eval({'snapshots': snapshots}))
+
+    def test_changed_true_when_prechange_is_none(self):
+        # OBJECT_CREATED events have no prechange snapshot
+        c = Condition('status', op='changed')
+        snapshots = {
+            'prechange': None,
+            'postchange': {'status': 'active'},
+        }
+        self.assertTrue(c.eval({'snapshots': snapshots}))
+
+    def test_changed_false_when_both_snapshots_missing_attr(self):
+        # If neither snapshot has the attr, nothing changed
+        c = Condition('nonexistent', op='changed')
+        snapshots = {
+            'prechange': {'status': 'active'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertFalse(c.eval({'snapshots': snapshots}))
+
+    def test_changed_false_when_path_traverses_scalar(self):
+        # Snapshot choice fields are raw strings, not nested dicts. A path like
+        # 'status.value' hits a TypeError when traversing into the string; both
+        # sides resolve to _MISSING and the operator returns False (no change).
+        c = Condition('status.value', op='changed')
+        snapshots = {
+            'prechange': {'status': 'planned'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertFalse(c.eval({'snapshots': snapshots}))
+
+    def test_changed_negated(self):
+        c = Condition('status', op='changed', negate=True)
+        snapshots = {
+            'prechange': {'status': 'planned'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertFalse(c.eval({'snapshots': snapshots}))
+
+    def test_changed_raises_when_no_snapshots(self):
+        c = Condition('status', op='changed')
+        with self.assertRaises(InvalidCondition):
+            c.eval({'status': {'value': 'active'}})
+
+    #
+    # 'unchanged' operator
+    #
+
+    def test_unchanged_true_when_attr_same(self):
+        c = Condition('status', op='unchanged')
+        snapshots = {
+            'prechange': {'status': 'active'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertTrue(c.eval({'snapshots': snapshots}))
+
+    def test_unchanged_false_when_attr_differs(self):
+        c = Condition('status', op='unchanged')
+        snapshots = {
+            'prechange': {'status': 'planned'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertFalse(c.eval({'snapshots': snapshots}))
+
+    def test_unchanged_false_when_both_snapshots_missing_attr(self):
+        # Fail-closed: a typo or non-existent attr resolves to _MISSING on both
+        # sides; unchanged must return False rather than silently passing.
+        c = Condition('statsu', op='unchanged')
+        snapshots = {
+            'prechange': {'status': 'active'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertFalse(c.eval({'snapshots': snapshots}))
+
+    #
+    # Direct snapshot path access (snapshots.prechange.* / snapshots.postchange.*)
+    #
+
+    def test_snapshot_path_access_prechange(self):
+        c = Condition('snapshots.prechange.status', value='planned', op='eq')
+        snapshots = {
+            'prechange': {'status': 'planned'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertTrue(c.eval({'snapshots': snapshots}))
+
+    def test_snapshot_path_access_postchange(self):
+        c = Condition('snapshots.postchange.status', value='active', op='eq')
+        snapshots = {
+            'prechange': {'status': 'planned'},
+            'postchange': {'status': 'active'},
+        }
+        self.assertTrue(c.eval({'snapshots': snapshots}))
+
+    def test_snapshot_path_rest_api_style_attr_raises_invalid_condition(self):
+        """
+        Snapshots store raw values (e.g. status="planned"), not REST API-style nested
+        dicts (status={"value": "planned"}). A '.value' suffix on a snapshot path must
+        fail closed with InvalidCondition rather than raising a raw TypeError.
+        """
+        c = Condition('snapshots.prechange.status.value', value='planned', op='eq')
+        snapshots = {
+            'prechange': {'status': 'planned'},
+            'postchange': {'status': 'active'},
+        }
+        with self.assertRaises(InvalidCondition):
+            c.eval({'snapshots': snapshots})
+
+    #
+    # EventRule.eval_conditions integration
+    #
+
+    def test_event_rule_changed_operator(self):
+        """
+        Verify the canonical use case: fire only when status changes to active.
+        """
+        event_rule = EventRule(
+            name='Notify on activation',
+            event_types=[OBJECT_UPDATED],
+            conditions={
+                'and': [
+                    {'attr': 'status.value', 'value': 'active'},
+                    {'attr': 'status', 'op': 'changed'},
+                ]
+            }
+        )
+        site = Site.objects.create(name='Site 2', slug='site-2', status=SiteStatusChoices.STATUS_ACTIVE)
+
+        # status changed planned → active: should fire
+        data_changed = self._make_condition_data(site, {
+            'prechange': {'status': SiteStatusChoices.STATUS_PLANNED},
+            'postchange': {'status': SiteStatusChoices.STATUS_ACTIVE},
+        })
+        self.assertTrue(event_rule.eval_conditions(data_changed))
+
+        # status already active, description updated: should NOT fire
+        data_unchanged = self._make_condition_data(site, {
+            'prechange': {'status': SiteStatusChoices.STATUS_ACTIVE},
+            'postchange': {'status': SiteStatusChoices.STATUS_ACTIVE},
+        })
+        self.assertFalse(event_rule.eval_conditions(data_unchanged))
+
+    def test_event_rule_snapshot_path_with_existing_operator(self):
+        """
+        Conditions can reference prechange/postchange data using the standard
+        snapshots.prechange.<attr> dot-path and existing operators.
+        Note: snapshot values use model serializer format (raw strings, not nested
+        dicts), so 'status' not 'status.value'.
+        """
+        event_rule = EventRule(
+            name='Was planned',
+            event_types=[OBJECT_UPDATED],
+            conditions={
+                'attr': 'snapshots.prechange.status',
+                'value': SiteStatusChoices.STATUS_PLANNED,
+            }
+        )
+        site = Site.objects.create(name='Site 3', slug='site-3', status=SiteStatusChoices.STATUS_ACTIVE)
+        data = self._make_condition_data(site, {
+            'prechange': {'status': SiteStatusChoices.STATUS_PLANNED},
+            'postchange': {'status': SiteStatusChoices.STATUS_ACTIVE},
+        })
+        self.assertTrue(event_rule.eval_conditions(data))
+
+    def test_event_rule_snapshot_path_rest_api_style_attr_must_return_false(self):
+        """
+        An EventRule condition mistakenly using a REST API-style '.value' suffix on a
+        snapshot path must fail closed (return False) rather than crashing evaluation.
+        """
+        event_rule = EventRule(
+            name='Was planned (REST-style mistake)',
+            event_types=[OBJECT_UPDATED],
+            conditions={
+                'attr': 'snapshots.prechange.status.value',
+                'value': SiteStatusChoices.STATUS_PLANNED,
+            }
+        )
+        site = Site.objects.create(name='Site 4', slug='site-4', status=SiteStatusChoices.STATUS_ACTIVE)
+        data = self._make_condition_data(site, {
+            'prechange': {'status': SiteStatusChoices.STATUS_PLANNED},
+            'postchange': {'status': SiteStatusChoices.STATUS_ACTIVE},
+        })
+        self.assertFalse(event_rule.eval_conditions(data))

netbox/extras/tests/test_event_rules.py

@@ -109,6 +109,47 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
             Tag(name='Baz', slug='baz'),
         ))
 
+    def test_eventrule_snapshot_changed_condition(self):
+        """
+        An event rule using the 'changed' operator fires only when the attribute
+        transitions to the target value, not on subsequent updates that leave it
+        unchanged.  Exercises the full process_event_rules() path.
+        """
+        webhook = Webhook.objects.get(name='Webhook 1')
+        webhook_type = ObjectType.objects.get_for_model(Webhook)
+        site_type = ObjectType.objects.get_for_model(Site)
+        event_rule = EventRule.objects.create(
+            name='Status Change Rule',
+            event_types=[OBJECT_UPDATED],
+            action_type=EventRuleActionChoices.WEBHOOK,
+            action_object_type=webhook_type,
+            action_object_id=webhook.pk,
+            conditions={
+                'and': [
+                    {'attr': 'status.value', 'value': SiteStatusChoices.STATUS_ACTIVE},
+                    {'attr': 'status', 'op': 'changed'},
+                ]
+            }
+        )
+        event_rule.object_types.set([site_type])
+
+        site = Site.objects.create(name='Site Snapshot', slug='site-snapshot', status=SiteStatusChoices.STATUS_PLANNED)
+        url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
+        self.add_permissions('dcim.change_site')
+
+        # planned → active: the 'changed' condition is satisfied; rule must fire
+        response = self.client.patch(url, {'status': SiteStatusChoices.STATUS_ACTIVE}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        rule_jobs = [j for j in self.queue.jobs if j.kwargs['event_rule'] == event_rule]
+        self.assertEqual(len(rule_jobs), 1, 'Expected rule to fire on status transition to active')
+        self.queue.empty()
+
+        # description update while status stays active: 'changed' condition fails; rule must not fire
+        response = self.client.patch(url, {'description': 'Updated'}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        rule_jobs = [j for j in self.queue.jobs if j.kwargs['event_rule'] == event_rule]
+        self.assertEqual(len(rule_jobs), 0, 'Expected rule not to fire when status is unchanged')
+
     def test_eventrule_conditions(self):
         """
         Test evaluation of EventRule conditions.