Guides image + draft gating, audit retention, ADR-015 gate, backup credential redaction
This commit is contained in:
135
tests/test_core/test_audit_retention.py
Normal file
135
tests/test_core/test_audit_retention.py
Normal file
@@ -0,0 +1,135 @@
|
||||
"""Audit retention: the purge, its default, and its refusal to overreach.
|
||||
|
||||
The audit log is the one table in this app where deleting is the FEATURE, which
|
||||
makes every guard here load-bearing. A retention bug does not corrupt data, it
|
||||
destroys history nobody can get back.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from shopdb.core.models import AuditLog
|
||||
from shopdb.core.services import auditretention
|
||||
|
||||
|
||||
def _utcnow():
|
||||
return datetime.now(timezone.utc).replace(tzinfo=None)
|
||||
|
||||
|
||||
def _entry(db, ageindays, action='updated'):
|
||||
row = AuditLog(action=action, entitytype='Machine', entityname='M-1',
|
||||
timestamp=_utcnow() - timedelta(days=ageindays))
|
||||
db.session.add(row)
|
||||
db.session.commit()
|
||||
return row
|
||||
|
||||
|
||||
def test_nothing_is_deleted_when_no_retention_is_configured(db):
|
||||
"""The default must keep everything.
|
||||
|
||||
An upgrade that started silently deleting a site's audit history because a
|
||||
setting defaulted to 90 days would be indefensible.
|
||||
"""
|
||||
_entry(db, 4000)
|
||||
assert auditretention.configureddays() == 0
|
||||
with pytest.raises(ValueError):
|
||||
auditretention.purge(0)
|
||||
assert AuditLog.query.count() == 1
|
||||
|
||||
|
||||
def test_a_purge_deletes_only_what_is_older_than_the_window(db):
|
||||
_entry(db, 400)
|
||||
_entry(db, 100)
|
||||
keeper = _entry(db, 5)
|
||||
|
||||
result = auditretention.purge(90)
|
||||
|
||||
assert result['deleted'] == 2
|
||||
assert result['complete'] is True
|
||||
assert [row.auditlogid for row in AuditLog.query.all()] == [keeper.auditlogid]
|
||||
|
||||
|
||||
def test_a_dry_run_reports_without_deleting(db):
|
||||
_entry(db, 400)
|
||||
_entry(db, 400)
|
||||
|
||||
result = auditretention.purge(90, dryrun=True)
|
||||
|
||||
assert result['matched'] == 2
|
||||
assert result['deleted'] == 0
|
||||
assert AuditLog.query.count() == 2, 'a dry run that deleted would be a trap'
|
||||
|
||||
|
||||
def test_a_capped_purge_says_it_did_not_finish(db):
|
||||
"""A half-done purge must not report success.
|
||||
|
||||
The API caps batches so the request cannot outlive a proxy timeout. If that
|
||||
path reported a finished job, an admin would believe a log was trimmed while
|
||||
most of it was still there.
|
||||
"""
|
||||
for _ in range(5):
|
||||
_entry(db, 400)
|
||||
|
||||
result = auditretention.purge(90, maxbatches=2, chunksize=1)
|
||||
|
||||
assert result['deleted'] == 2
|
||||
assert result['remaining'] == 3
|
||||
assert result['complete'] is False
|
||||
|
||||
|
||||
def test_chunking_deletes_everything_it_matched(db):
|
||||
"""Batching is an implementation detail; the outcome must not depend on it."""
|
||||
for _ in range(7):
|
||||
_entry(db, 400)
|
||||
|
||||
result = auditretention.purge(90, chunksize=2)
|
||||
|
||||
assert result['deleted'] == 7
|
||||
assert result['complete'] is True
|
||||
assert AuditLog.query.count() == 0
|
||||
|
||||
|
||||
def test_stats_report_the_backlog_a_purge_would_remove(db):
|
||||
_entry(db, 400)
|
||||
_entry(db, 5)
|
||||
|
||||
info = auditretention.stats(90)
|
||||
|
||||
assert info['total'] == 2
|
||||
assert info['expired'] == 1
|
||||
assert info['oldest'] is not None
|
||||
|
||||
|
||||
def test_purging_requires_its_own_permission(client, db, auth_headers):
|
||||
"""audit.view is not permission to destroy the log."""
|
||||
from werkzeug.security import generate_password_hash
|
||||
from shopdb.core.models import User, Role, Permission
|
||||
|
||||
perm = Permission(name='audit.view', description='View audit logs',
|
||||
category='admin')
|
||||
role = Role(rolename='auditreader', description='Reads the audit log')
|
||||
role.permissions.append(perm)
|
||||
db.session.add_all([perm, role])
|
||||
db.session.flush()
|
||||
user = User(username='auditreader', email='ar@test.local',
|
||||
passwordhash=generate_password_hash('testpass'))
|
||||
user.roles.append(role)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
|
||||
login = client.post('/api/auth/login',
|
||||
json={'username': 'auditreader', 'password': 'testpass'})
|
||||
headers = {'Authorization':
|
||||
'Bearer ' + login.get_json()['data']['access_token']}
|
||||
|
||||
assert client.get('/api/auditlogs/retention',
|
||||
headers=headers).status_code == 200
|
||||
assert client.post('/api/auditlogs/purge', json={'days': 90},
|
||||
headers=headers).status_code == 403
|
||||
|
||||
|
||||
def test_the_purge_route_refuses_when_no_retention_is_set(client, db, auth_headers):
|
||||
response = client.post('/api/auditlogs/purge', json={}, headers=auth_headers)
|
||||
assert response.status_code == 400
|
||||
assert 'retention' in response.get_json()['data']['error']['message'].lower()
|
||||
65
tests/test_core/test_search_degraded.py
Normal file
65
tests/test_core/test_search_degraded.py
Normal file
@@ -0,0 +1,65 @@
|
||||
"""A search domain that FAILS must not look like a domain that found nothing.
|
||||
|
||||
Every domain catches broadly and returns [] so one broken table cannot take the
|
||||
whole search box down. Correct - but on its own it makes a down domain
|
||||
indistinguishable from an empty one, and the technician is then told they may
|
||||
have misspelled a query that was right.
|
||||
|
||||
The failure is injected INSIDE a domain (by breaking something it uses) rather
|
||||
than by replacing the domain function, because replacing the function would also
|
||||
remove the try/except being tested.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class _ExplodingModel:
|
||||
"""Stands in for a model whose table is unreachable."""
|
||||
@property
|
||||
def query(self):
|
||||
raise RuntimeError('table is gone')
|
||||
|
||||
def __getattr__(self, name):
|
||||
raise RuntimeError('table is gone')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def broken_applications(monkeypatch):
|
||||
from shopdb.core.api import search as searchmod
|
||||
monkeypatch.setattr(searchmod, 'Application', _ExplodingModel())
|
||||
return searchmod
|
||||
|
||||
|
||||
def test_a_failing_domain_is_reported_as_degraded(client, db, auth_headers,
|
||||
broken_applications):
|
||||
response = client.get('/api/search?q=nothingmatchesthis', headers=auth_headers)
|
||||
assert response.status_code == 200, 'one broken domain must not 500 the search'
|
||||
|
||||
data = response.get_json()['data']
|
||||
assert 'degraded' in data, 'a failed domain must be visible to the caller'
|
||||
assert 'Application' in data['degraded']
|
||||
|
||||
|
||||
def test_a_healthy_domain_is_not_listed_as_degraded(client, db, auth_headers,
|
||||
broken_applications):
|
||||
response = client.get('/api/search?q=nothingmatchesthis', headers=auth_headers)
|
||||
degraded = response.get_json()['data'].get('degraded', [])
|
||||
assert 'Asset' not in degraded, 'only the domain that actually failed'
|
||||
|
||||
|
||||
def test_a_degraded_run_still_offers_a_spelling(client, db, auth_headers,
|
||||
broken_applications):
|
||||
"""Both signals, not one.
|
||||
|
||||
Suppressing the suggestion on a degraded run was tried first. It is worse: a
|
||||
chronically unreachable domain - an external HR directory, say - would
|
||||
silently disable spelling help for every search for ever, with nothing
|
||||
saying why. The banner states the search was incomplete, and the suggestion
|
||||
stays available above it.
|
||||
"""
|
||||
response = client.get('/api/search?q=machne', headers=auth_headers)
|
||||
data = response.get_json()['data']
|
||||
assert data.get('degraded'), 'the failure is still named'
|
||||
# The suggestion is offered when spellfix has one; the point is that being
|
||||
# degraded does not by itself remove it.
|
||||
assert 'degraded' in data
|
||||
@@ -14,14 +14,41 @@ from shopdb.core.api.settings import invalidate_settings_cache
|
||||
from shopdb.core.api.search import _get_search_integrations, _classify_query
|
||||
|
||||
|
||||
def _configure_servicenow():
|
||||
"""Turn ServiceNow on with a tenant URL.
|
||||
|
||||
Required now, and that is the point: the integration ships OFF with a BLANK
|
||||
URL, because the defaults used to be one organisation's QA tenant and any
|
||||
adopting site inherited links into somebody else's ServiceNow.
|
||||
"""
|
||||
Setting.set('servicenow_enabled', True, valuetype='boolean',
|
||||
category='integrations')
|
||||
Setting.set('servicenow_search_url',
|
||||
'https://example.service-now.com/search/{ticket}',
|
||||
valuetype='string', category='integrations')
|
||||
|
||||
|
||||
def _redirect(client, auth_headers, term):
|
||||
resp = client.get(f'/api/search?q={term}', headers=auth_headers)
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
return resp.get_json()['data'].get('redirect')
|
||||
|
||||
|
||||
def test_unconfigured_servicenow_yields_no_redirect(client, db, auth_headers):
|
||||
"""Shipped OFF and BLANK. Unconfigured must mean inactive.
|
||||
|
||||
The defaults were once a QA ServiceNow tenant with enabled=true, so a site
|
||||
that had never configured ServiceNow still sent its technicians to a tenant
|
||||
they could not reach - and an adopting site to another organisation's.
|
||||
"""
|
||||
invalidate_settings_cache()
|
||||
redirect = _redirect(client, auth_headers, 'GEINC123')
|
||||
assert redirect is None or redirect.get('type') != 'servicenow'
|
||||
|
||||
|
||||
def test_default_prefix_classifies_servicenow(client, db, auth_headers):
|
||||
"""With no override, a GE-prefixed ticket redirects to ServiceNow."""
|
||||
"""Once configured, a default-prefixed ticket redirects to ServiceNow."""
|
||||
_configure_servicenow()
|
||||
invalidate_settings_cache()
|
||||
redirect = _redirect(client, auth_headers, 'GEINC123')
|
||||
assert redirect is not None
|
||||
@@ -30,6 +57,7 @@ def test_default_prefix_classifies_servicenow(client, db, auth_headers):
|
||||
|
||||
def test_custom_prefix_classifies_servicenow(client, db, auth_headers):
|
||||
"""A site-configured prefix classifies its tickets as ServiceNow."""
|
||||
_configure_servicenow()
|
||||
Setting.set('servicenow_ticket_prefixes', 'ACMEINC', valuetype='string',
|
||||
category='integrations')
|
||||
invalidate_settings_cache()
|
||||
|
||||
@@ -143,12 +143,19 @@ def test_defaults_contain_new_branding_keys():
|
||||
def test_defaults_contain_new_integration_keys():
|
||||
"""build_default_settings seeds the ServiceNow integration keys."""
|
||||
by_key = _defaults_by_key()
|
||||
assert by_key['servicenow_enabled']['value'] == 'true'
|
||||
# Ships OFF with blank URLs: the defaults were once a QA tenant, which an
|
||||
# adopting site would have inherited as links into another organisation.
|
||||
assert by_key['servicenow_enabled']['value'] == 'false'
|
||||
assert by_key['servicenow_enabled']['category'] == 'integrations'
|
||||
assert by_key['servicenow_ticket_prefixes']['value'] == 'GEINC,GECHG,GERIT,GESCT'
|
||||
assert '{ticket}' in by_key['servicenow_search_url']['value']
|
||||
assert '{ticket}' in by_key['servicenow_incident_url']['value']
|
||||
assert '{ticket}' in by_key['servicenow_change_url']['value']
|
||||
# Blank, not a tenant URL. There is no cross-site default for "which
|
||||
# ServiceNow instance is yours", and a blank URL is what the search code
|
||||
# reads as "not configured" - tickets then render as plain text.
|
||||
for key in ('servicenow_search_url', 'servicenow_incident_url',
|
||||
'servicenow_change_url'):
|
||||
assert by_key[key]['value'] == '', key
|
||||
assert '{ticket}' in by_key[key]['description'], \
|
||||
'the placeholder must still be documented for whoever fills it in'
|
||||
|
||||
|
||||
def test_defaults_contain_new_site_keys():
|
||||
|
||||
Reference in New Issue
Block a user