Make alertscope actually target fleet alerts

alertscope was stored, serialised and editable in the authoring form but
never read when alerts were handed to an endpoint, so every fleet alert
went to every machine regardless of what was entered. The control implied
a capability nothing enforced.

get_fleet_alerts now resolves the polling endpoint's facts and filters
each alert through its scope. Supported keys: hostname, assettype,
computertype, location, businessunit. A key may hold one value or a list
(any match), matching is case-insensitive since these are operator-typed
names, and multiple keys narrow together.

Unrecognised keys are ignored rather than treated as match-nothing. A
typo like {"hostnam": "BAY-01"} would otherwise mute the alert fleet-wide
and look identical to a delivery failure; sending wider than intended is
the safe direction for something whose job is to warn people. A garbled
non-dict scope falls back to the whole fleet for the same reason. The
real fix for typos belongs at authoring time, not delivery.

BusinessUnit is not exported from shopdb.api, so its name is resolved by
table rather than importing the model and breaking the plugin import
contract.

Ten tests cover the narrowing cases and the fail-open behaviour.
This commit is contained in:
benjamin.baker
2026-09-09 15:40:40 -04:00
parent e0a7a51d46
commit 494b837bae
2 changed files with 236 additions and 0 deletions

View File

@@ -309,3 +309,144 @@ def test_non_fleet_notification_is_never_delivered(client, db, app):
assert resp.status_code == 200
assert resp.get_json()['data'] == []
# --- targeting -------------------------------------------------------------
#
# alertscope narrows a fleet alert to part of the floor. It was previously
# stored, serialised and editable in the authoring form but never read here, so
# every alert went to every endpoint no matter what was entered - a control that
# looked like it worked and did not. These pin the behaviour it should have had.
def _ids(client, app, hostname):
resp = client.get(f'{ALERTS_URL}?hostname={hostname}',
headers=_collector_headers(app))
assert resp.status_code == 200, resp.get_data(as_text=True)
return [a['notificationId'] for a in resp.get_json()['data']]
def test_no_scope_still_means_the_whole_fleet(client, db, app):
"""The default is unchanged: an unscoped alert reaches everyone."""
_computer(db, 'BAY-01')
_computer(db, 'BAY-02')
alert = _fleet_alert(db)
assert _ids(client, app, 'BAY-01') == [alert.notificationid]
assert _ids(client, app, 'BAY-02') == [alert.notificationid]
def test_an_empty_scope_means_the_whole_fleet(client, db, app):
"""An empty dict is not a match-nothing."""
_computer(db, 'BAY-01')
alert = _fleet_alert(db)
alert.alertscope = {}
db.session.commit()
assert _ids(client, app, 'BAY-01') == [alert.notificationid]
def test_hostname_scope_targets_one_endpoint(client, db, app):
"""The narrowest case, and the one the test card drives."""
_computer(db, 'BAY-01')
_computer(db, 'BAY-02')
alert = _fleet_alert(db)
alert.alertscope = {'hostname': 'BAY-01'}
db.session.commit()
assert _ids(client, app, 'BAY-01') == [alert.notificationid]
assert _ids(client, app, 'BAY-02') == []
def test_hostname_scope_is_case_insensitive(client, db, app):
"""Operators type these by hand; case must not decide delivery."""
_computer(db, 'BAY-01')
alert = _fleet_alert(db)
alert.alertscope = {'hostname': 'bay-01'}
db.session.commit()
assert _ids(client, app, 'BAY-01') == [alert.notificationid]
def test_a_scope_may_list_several_values(client, db, app):
"""Any value in a list matches, so a scope can name a handful of bays."""
_computer(db, 'BAY-01')
_computer(db, 'BAY-02')
_computer(db, 'BAY-03')
alert = _fleet_alert(db)
alert.alertscope = {'hostname': ['BAY-01', 'BAY-03']}
db.session.commit()
assert _ids(client, app, 'BAY-01') == [alert.notificationid]
assert _ids(client, app, 'BAY-02') == []
assert _ids(client, app, 'BAY-03') == [alert.notificationid]
def test_computertype_scope_targets_a_class_of_machine(client, db, app):
"""Targeting by what the machine is, not which one it is."""
_computer(db, 'BAY-01')
alert = _fleet_alert(db)
alert.alertscope = {'computertype': 'CMM PC'}
db.session.commit()
assert _ids(client, app, 'BAY-01') == [alert.notificationid]
alert.alertscope = {'computertype': 'Kiosk'}
db.session.commit()
assert _ids(client, app, 'BAY-01') == []
def test_several_scope_keys_must_all_match(client, db, app):
"""Keys narrow together - it is an AND, not an OR."""
_computer(db, 'BAY-01')
alert = _fleet_alert(db)
alert.alertscope = {'hostname': 'BAY-01', 'computertype': 'CMM PC'}
db.session.commit()
assert _ids(client, app, 'BAY-01') == [alert.notificationid]
# Right host, wrong type: the alert must not land.
alert.alertscope = {'hostname': 'BAY-01', 'computertype': 'Kiosk'}
db.session.commit()
assert _ids(client, app, 'BAY-01') == []
def test_an_unknown_scope_key_is_ignored_not_fatal(client, db, app):
"""A typo must not silently mute an alert everywhere.
Scopes are authored by hand. If {"hostnam": "BAY-01"} matched nothing, the
alert would vanish fleet-wide and look like a delivery failure. Sending it
wider than intended is the safe direction for something whose job is to warn
people, so unrecognised keys are ignored.
"""
_computer(db, 'BAY-01')
alert = _fleet_alert(db)
alert.alertscope = {'hostnam': 'BAY-01'}
db.session.commit()
assert _ids(client, app, 'BAY-01') == [alert.notificationid]
def test_a_non_dict_scope_falls_back_to_the_whole_fleet(client, db, app):
"""Garbled scope JSON must not stop an alert reaching the floor."""
_computer(db, 'BAY-01')
alert = _fleet_alert(db)
alert.alertscope = 'BAY-01'
db.session.commit()
assert _ids(client, app, 'BAY-01') == [alert.notificationid]
def test_scoping_does_not_leak_into_acknowledgement(client, db, app):
"""A targeted alert is acknowledged the same way any other one is."""
asset = _computer(db, 'BAY-01')
alert = _fleet_alert(db)
alert.alertscope = {'hostname': 'BAY-01'}
db.session.commit()
ack = client.post(ACK_URL.format(alert.notificationid),
json={'hostname': 'BAY-01', 'assetid': asset.assetid},
headers=_collector_headers(app))
assert ack.status_code == 201
assert _ids(client, app, 'BAY-01') == []