Merge remote-tracking branch 'origin/main' into wilm
# Conflicts: # frontend/src/views/settings/SiteSettings.vue
This commit is contained in:
benjamin.baker
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()
|
||||
@@ -266,3 +266,75 @@ def test_an_ip_search_links_to_the_matched_subnet(client, subnetrow):
|
||||
assert len(rows) == 1
|
||||
assert rows[0]['id'] == subnetrow
|
||||
assert rows[0]['url'] == f'/networks/{subnetrow}'
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Guides. A guide is not an asset either, so the generic asset search cannot see
|
||||
# one - and the draft rule makes the omission worse than invisibility: a
|
||||
# half-written procedure reachable through the search box is one somebody
|
||||
# follows.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture
|
||||
def guides(db):
|
||||
"""One published guide and one draft, with distinctive text.
|
||||
|
||||
The published one carries its distinguishing words in the BODY, not the
|
||||
title, because that is how people look for a procedure - by something
|
||||
written halfway down it.
|
||||
"""
|
||||
from plugins.guides.models import Guide
|
||||
from plugins.guides.services import authoring
|
||||
|
||||
published = Guide(title='Zzmarker controller swap', slug='zzmarker-swap',
|
||||
status='draft')
|
||||
draft = Guide(title='Zzdraft unfinished procedure', slug='zzdraft-unfinished',
|
||||
status='draft')
|
||||
db.session.add_all([published, draft])
|
||||
db.session.flush()
|
||||
|
||||
authoring.save(published, title=published.title,
|
||||
summary='Restoring after a swap', tags='zztelesis, zzbackups',
|
||||
blocks=[{'id': 'g1', 'type': 'paragraph',
|
||||
'text': 'Check the zzmappeddrive before starting.'}])
|
||||
authoring.publish(published)
|
||||
authoring.save(draft, title=draft.title, summary='', tags='zztelesis',
|
||||
blocks=[{'id': 'g2', 'type': 'paragraph',
|
||||
'text': 'Half written zzmappeddrive notes.'}])
|
||||
db.session.commit()
|
||||
return {'published': published.guideid, 'draft': draft.guideid}
|
||||
|
||||
|
||||
def test_a_published_guide_is_found_by_title(client, guides):
|
||||
rows = [r for r in search(client, 'Zzmarker') if r['type'] == 'guide']
|
||||
assert len(rows) == 1
|
||||
assert rows[0]['id'] == guides['published']
|
||||
assert rows[0]['url'] == '/guides/zzmarker-swap'
|
||||
|
||||
|
||||
def test_a_guide_is_found_by_tag(client, guides):
|
||||
"""Tags are how a technician narrows to a family of procedures."""
|
||||
assert 'Zzmarker controller swap' in titles(client, 'zzbackups', 'guide')
|
||||
|
||||
|
||||
def test_a_guide_is_found_by_words_in_its_body(client, guides):
|
||||
"""People search for what they remember reading, not for the title.
|
||||
|
||||
The body is stored as a JSON blob of blocks, so this is the check that the
|
||||
blob is actually reachable rather than merely present.
|
||||
"""
|
||||
assert 'Zzmarker controller swap' in titles(client, 'zzmappeddrive', 'guide')
|
||||
|
||||
|
||||
def test_a_draft_guide_is_never_returned(client, guides):
|
||||
"""The draft rule has to hold on EVERY path that reaches the row.
|
||||
|
||||
A draft is deliberately invisible to readers, and search is where that rule
|
||||
would otherwise leak: somebody finds a half-written procedure through the
|
||||
search box and follows it. The draft here shares a tag and a body word with
|
||||
the published guide, so it would ride along on both queries if the filter
|
||||
were missing.
|
||||
"""
|
||||
assert 'Zzdraft unfinished procedure' not in titles(client, 'zzmappeddrive', 'guide')
|
||||
assert 'Zzdraft unfinished procedure' not in titles(client, 'zztelesis', 'guide')
|
||||
assert titles(client, 'Zzdraft', 'guide') == []
|
||||
|
||||
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():
|
||||
|
||||
@@ -54,6 +54,9 @@ EXPECTED_HEAD_REVISION['computers'] = 'computers0003azurefields'
|
||||
# 0002 adds lastseenat - when a config was last CONFIRMED unchanged, which dedup
|
||||
# otherwise throws away.
|
||||
EXPECTED_HEAD_REVISION['backups'] = 'backups0003clearlastseen'
|
||||
# guides is a NEW plugin, so its 0001 builds tables rather than stamping an
|
||||
# anchor over core-owned ones.
|
||||
EXPECTED_HEAD_REVISION['guides'] = 'guides0001baseline'
|
||||
# geenforce adds the content-addressed blob store (manifestblobs) on top of its
|
||||
# baseline.
|
||||
EXPECTED_HEAD_REVISION['geenforce'] = 'geenforce0004minlib'
|
||||
@@ -283,18 +286,21 @@ def test_upgrade_all_on_fresh_db_is_clean_and_idempotent(tmp_path, monkeypatch):
|
||||
|
||||
|
||||
|
||||
# Plugins whose whole chain is a single baseline. Their tables existing already
|
||||
# is the ONLY divergence possible between schema and stamp, so upgrade-all must
|
||||
# handle it end to end. A longer chain can also meet an ALTER whose column is
|
||||
# already present, which no amount of guarding a create_table fixes - see the
|
||||
# test below for why that is a different problem.
|
||||
SINGLE_BASELINE_PLUGINS = tuple(
|
||||
plugin for plugin in POST_CUTOVER_PLUGINS
|
||||
if len(list((PLUGINS_DIR / plugin / 'migrations' / 'versions').glob('0*.py'))) == 1
|
||||
)
|
||||
# EVERY post-cutover chain, not just the single-baseline ones. This was once
|
||||
# restricted to plugins whose whole chain is one baseline, on the reasoning that
|
||||
# a longer chain meets a second problem - an ALTER whose column is already
|
||||
# present - that guarding a create_table does not fix. True, but the conclusion
|
||||
# was wrong: the answer is to guard the ALTERs too, not to stop testing the
|
||||
# chains that have them. While this was narrowed, printedparts 0003 sat
|
||||
# unguarded for a month and would fail upgrade-all on any create-all-built
|
||||
# database, taking 0004 down with it, with no test able to see it.
|
||||
#
|
||||
# The invariant is now the whole one: a post-cutover chain must run to its head
|
||||
# against a database whose tables were built from the models and stamped
|
||||
# nowhere, and must do it twice.
|
||||
|
||||
|
||||
@pytest.mark.parametrize('plugin', SINGLE_BASELINE_PLUGINS)
|
||||
@pytest.mark.parametrize('plugin', POST_CUTOVER_PLUGINS)
|
||||
def test_baseline_skips_tables_that_already_exist(plugin, tmp_path, monkeypatch):
|
||||
"""A baseline must stamp, not fail, when its table is already there.
|
||||
|
||||
|
||||
@@ -1416,3 +1416,112 @@ def test_the_card_omits_the_snapshot_row_when_no_path_is_passed(bk_app, bk_plugi
|
||||
from plugins.backups.services import partmarkerinfo
|
||||
card = partmarkerinfo.build(_markermanifest(['A.PAT']), 1)
|
||||
assert 'Snapshot' not in [f['label'] for f in card['fields']]
|
||||
|
||||
|
||||
def test_an_empty_source_hostname_still_dedups(bk_app, bk_plugin):
|
||||
"""A client posting sourcehostname='' must not defeat dedup forever.
|
||||
|
||||
samesource() maps '' to IS NULL for the QUERY, but the insert stored the raw
|
||||
'' - so the rows a client wrote could never match its own next post. Every
|
||||
post inserted, the table grew without bound, and the revision history filled
|
||||
with identical entries nobody could read.
|
||||
"""
|
||||
from plugins.backups.models import BackupRevision
|
||||
with bk_app.app_context():
|
||||
first = bk_plugin.apply_collector_payload(_payload(sourcehostname=''))
|
||||
assert first['action'] == 'created'
|
||||
|
||||
second = bk_plugin.apply_collector_payload(_payload(sourcehostname=''))
|
||||
assert second['action'] == 'noop', 'an identical re-post must dedup'
|
||||
|
||||
rows = BackupRevision.query.all()
|
||||
assert len(rows) == 1
|
||||
assert rows[0].sourcehostname is None, \
|
||||
"'' is stored as NULL, matching how samesource() queries it"
|
||||
|
||||
|
||||
def test_a_padded_source_hostname_is_the_same_source(bk_app, bk_plugin):
|
||||
"""Whitespace must not fork a machine's history into two chains."""
|
||||
with bk_app.app_context():
|
||||
bk_plugin.apply_collector_payload(_payload(sourcehostname='GGBX0NH3ESF'))
|
||||
second = bk_plugin.apply_collector_payload(
|
||||
_payload(sourcehostname=' GGBX0NH3ESF '))
|
||||
assert second['action'] == 'noop'
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Credential redaction. The DNC registry carries an FTP password in cleartext;
|
||||
# stored verbatim it reached the database, the revision panel and the download
|
||||
# route, so anyone with backups.view could read the fleet's password off a page.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# A configured install that also carries the FTP credential, which is what a
|
||||
# real marker/DNC bay actually has in its registry.
|
||||
SECRETREG = CONFIGUREDREG.replace(
|
||||
'"MachineNo"="3204"',
|
||||
'"MachineNo"="3204"\r\n"FtpPasswd"="hunter2"\r\n"FtpUser"="dnc"')
|
||||
|
||||
|
||||
def test_a_credential_never_enters_the_projection():
|
||||
"""Redaction happens at parse, so the secret is never persisted.
|
||||
|
||||
Redacting on display instead would leave it at rest and one query away.
|
||||
"""
|
||||
import json
|
||||
from plugins.backups.services import ntlars
|
||||
|
||||
projection = ntlars.parse(_asbytes(SECRETREG))
|
||||
|
||||
assert 'hunter2' not in json.dumps(projection)
|
||||
assert projection['redacted'] == ['General\\FtpPasswd']
|
||||
values = dict(projection['keys'][0]['values'])
|
||||
for key in projection['keys']:
|
||||
values.update(key['values'])
|
||||
assert values['FtpPasswd']['data'] == ntlars.REDACTEDMARKER
|
||||
assert values['FtpUser']['data'] == 'dnc', 'only credentials are withheld'
|
||||
|
||||
|
||||
def test_a_redacted_value_is_commented_out_not_imported():
|
||||
"""Importing the marker as a password yields a machine that cannot log in.
|
||||
|
||||
It would LOOK restored, and the technician would go debugging the network.
|
||||
A commented-out value fails at the one moment somebody is reading the file.
|
||||
"""
|
||||
from plugins.backups.services import ntlars
|
||||
|
||||
text = ntlars.render(ntlars.parse(_asbytes(SECRETREG))).decode('utf-16-le')
|
||||
|
||||
assert 'hunter2' not in text
|
||||
assert ntlars.REDACTEDMARKER not in text
|
||||
assert 'FtpPasswd' in text, 'the omission must be named, not silent'
|
||||
assert '"FtpPasswd"="' not in text, 'must not be an importable value line'
|
||||
assert '"FtpUser"="dnc"' in text, 'everything else still restores'
|
||||
|
||||
|
||||
def test_the_download_route_cannot_hand_back_a_password(bk_app, bk_plugin):
|
||||
"""End to end: post a config with a password, download it back."""
|
||||
with bk_app.app_context():
|
||||
result = bk_plugin.apply_collector_payload(_payload(reg=SECRETREG))
|
||||
assert result['action'] == 'created'
|
||||
|
||||
from plugins.backups.models import BackupRevision
|
||||
revision = _db.session.get(BackupRevision, result['backuprevisionid'])
|
||||
stored = json.dumps(revision.payload)
|
||||
assert 'hunter2' not in stored
|
||||
|
||||
rendered = ntlars_render(revision.payload)
|
||||
assert 'hunter2' not in rendered
|
||||
|
||||
|
||||
def ntlars_render(projection):
|
||||
from plugins.backups.services import ntlars
|
||||
return ntlars.render(projection).decode('utf-16-le')
|
||||
|
||||
|
||||
def test_secret_name_matching_covers_the_usual_spellings():
|
||||
from plugins.backups.services.ntlars import issecretname
|
||||
for name in ('FtpPasswd', 'Password', 'ftp_pwd', 'ApiKey', 'AuthToken',
|
||||
'ClientSecret'):
|
||||
assert issecretname(name), name
|
||||
for name in ('FtpUser', 'FMSHost', 'Retries', 'Passive'):
|
||||
assert not issecretname(name), name
|
||||
|
||||
561
tests/test_plugins/test_guides.py
Normal file
561
tests/test_plugins/test_guides.py
Normal file
@@ -0,0 +1,561 @@
|
||||
"""Guides: block validation, revisioning, and the draft/publish split.
|
||||
|
||||
The behaviour worth protecting hardest is the last one. A technician following a
|
||||
procedure on a shop floor must not have it change under them because somebody is
|
||||
editing it upstairs, and the only thing standing between them and that is the
|
||||
published pointer.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from shopdb import create_app
|
||||
from shopdb.extensions import db as _db
|
||||
from shopdb.plugins import plugin_manager
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Block validation - pure, no app context needed. These rules exist to stop a
|
||||
# BAD SAVE becoming a guide that will not render weeks later, when the author
|
||||
# has moved on and the only evidence is a blank page.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _block(blockid='b1', blocktype='paragraph', **extra):
|
||||
return dict({'id': blockid, 'type': blocktype}, **extra)
|
||||
|
||||
|
||||
def test_a_well_formed_block_list_passes():
|
||||
from plugins.guides.services.blocks import validate
|
||||
blocks = [
|
||||
_block('b1', 'heading', text='Before you start'),
|
||||
_block('b2', 'paragraph', text='Check the **cable** first.'),
|
||||
_block('b3', 'checklist', items=[{'id': 'i1', 'text': 'Power on'}]),
|
||||
]
|
||||
assert validate(blocks) == blocks
|
||||
|
||||
|
||||
def test_a_block_without_an_id_is_refused():
|
||||
"""Ids are what make a tick rememberable.
|
||||
|
||||
Without one the checkbox silently loses its state on the next visit, which
|
||||
reads to a user as "the checkboxes do not work" - a bug report with no
|
||||
stack trace and no obvious cause.
|
||||
"""
|
||||
from plugins.guides.services.blocks import validate, BlockError
|
||||
with pytest.raises(BlockError) as exc:
|
||||
validate([{'type': 'paragraph', 'text': 'no id'}])
|
||||
assert 'stable id' in str(exc.value)
|
||||
|
||||
|
||||
def test_duplicate_ids_are_refused():
|
||||
from plugins.guides.services.blocks import validate, BlockError
|
||||
with pytest.raises(BlockError) as exc:
|
||||
validate([_block('same'), _block('same')])
|
||||
assert 'duplicate' in str(exc.value)
|
||||
|
||||
|
||||
def test_an_unknown_block_type_is_refused_and_lists_the_known_ones():
|
||||
from plugins.guides.services.blocks import validate, BlockError
|
||||
with pytest.raises(BlockError) as exc:
|
||||
validate([_block('b1', 'carousel')])
|
||||
message = str(exc.value)
|
||||
assert 'carousel' in message and 'checklist' in message
|
||||
|
||||
|
||||
def test_checklist_items_need_their_own_ids():
|
||||
"""Same reason as blocks: the item id is what a tick is stored against."""
|
||||
from plugins.guides.services.blocks import validate, BlockError
|
||||
with pytest.raises(BlockError) as exc:
|
||||
validate([_block('b1', 'checklist', items=[{'text': 'no id'}])])
|
||||
assert 'item' in str(exc.value)
|
||||
|
||||
|
||||
def test_the_error_names_which_block_is_wrong():
|
||||
"""An author fixing a rejected save should not have to count blocks."""
|
||||
from plugins.guides.services.blocks import validate, BlockError
|
||||
with pytest.raises(BlockError) as exc:
|
||||
validate([_block('b1'), _block('b2'), {'type': 'paragraph'}])
|
||||
assert str(exc.value).startswith('block 3:')
|
||||
|
||||
|
||||
def test_content_hash_covers_title_and_tags_not_just_blocks():
|
||||
"""Renaming or retagging a guide is a change a reader notices.
|
||||
|
||||
Hashing only the body would drop those edits out of the history entirely.
|
||||
"""
|
||||
from plugins.guides.services.blocks import contenthash
|
||||
blocks = [_block('b1', 'paragraph', text='same body')]
|
||||
base = contenthash('Title', 'Summary', 'pxe, share', blocks)
|
||||
assert contenthash('Different', 'Summary', 'pxe, share', blocks) != base
|
||||
assert contenthash('Title', 'Summary', 'pxe', blocks) != base
|
||||
assert contenthash('Title', 'Different', 'pxe, share', blocks) != base
|
||||
|
||||
|
||||
def test_content_hash_ignores_tag_order_case_and_spacing():
|
||||
"""Otherwise reordering tags in the editor would manufacture a version."""
|
||||
from plugins.guides.services.blocks import contenthash
|
||||
blocks = [_block()]
|
||||
assert (contenthash('T', '', 'PXE, Share', blocks)
|
||||
== contenthash('T', '', 'share ,pxe', blocks))
|
||||
|
||||
|
||||
def test_checklist_item_count_counts_only_checklists():
|
||||
from plugins.guides.services.blocks import checklistitemcount
|
||||
blocks = [
|
||||
_block('b1', 'checklist', items=[{'id': 'i1'}, {'id': 'i2'}]),
|
||||
_block('b2', 'bullets', items=['not tickable']),
|
||||
_block('b3', 'checklist', items=[{'id': 'i3'}]),
|
||||
]
|
||||
assert checklistitemcount(blocks) == 3
|
||||
|
||||
|
||||
def test_plaintext_flattens_a_guide_for_search():
|
||||
"""People search for something written halfway down, not for the title."""
|
||||
from plugins.guides.services.blocks import plaintext
|
||||
blocks = [
|
||||
_block('b1', 'heading', text='Mapped drive'),
|
||||
_block('b2', 'checklist', items=[{'id': 'i1', 'text': 'Run the sync'}]),
|
||||
_block('b3', 'table', rows=[['left', 'right']]),
|
||||
]
|
||||
flat = plaintext(blocks)
|
||||
assert 'Mapped drive' in flat and 'Run the sync' in flat and 'right' in flat
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Authoring: revisions, dedup and publishing. Needs an app context.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture(scope='module')
|
||||
def gd_app():
|
||||
saved = (plugin_manager._app, plugin_manager._db, plugin_manager.registry,
|
||||
plugin_manager.loader, plugin_manager.migration_manager,
|
||||
plugin_manager._registered_prefixes)
|
||||
application = create_app('testing')
|
||||
|
||||
from plugins.guides.plugin import GuidesPlugin
|
||||
plugin = GuidesPlugin()
|
||||
pm = application.extensions['plugin_manager']
|
||||
if plugin.meta.api_prefix not in pm._registered_prefixes:
|
||||
pm._register_plugin_components(plugin)
|
||||
|
||||
with application.app_context():
|
||||
_db.create_all()
|
||||
yield application
|
||||
_db.session.remove()
|
||||
_db.drop_all()
|
||||
|
||||
(plugin_manager._app, plugin_manager._db, plugin_manager.registry,
|
||||
plugin_manager.loader, plugin_manager.migration_manager,
|
||||
plugin_manager._registered_prefixes) = saved
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def gd_clean(gd_app):
|
||||
from plugins.guides.models import Guide, GuideRevision
|
||||
with gd_app.app_context():
|
||||
_db.session.query(GuideRevision).delete()
|
||||
_db.session.query(Guide).delete()
|
||||
_db.session.commit()
|
||||
yield
|
||||
|
||||
|
||||
def _newguide(title='Replacing a marker controller'):
|
||||
from plugins.guides.models import Guide
|
||||
guide = Guide(title=title, slug=title.lower().replace(' ', '-'),
|
||||
status='draft')
|
||||
_db.session.add(guide)
|
||||
_db.session.flush()
|
||||
return guide
|
||||
|
||||
|
||||
def test_the_first_save_creates_version_one(gd_app, gd_clean):
|
||||
from plugins.guides.services import authoring
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
guide, revision = authoring.save(
|
||||
guide, title='Replacing a marker controller',
|
||||
blocks=[_block('b1', 'paragraph', text='Power it off first.')])
|
||||
_db.session.commit()
|
||||
assert revision is not None
|
||||
assert revision.revision == 1
|
||||
assert guide.revision == 1
|
||||
|
||||
|
||||
def test_saving_identical_content_writes_no_revision(gd_app, gd_clean):
|
||||
"""Autosave would otherwise produce a hundred versions a day.
|
||||
|
||||
A history full of identical entries hides the three edits that mattered,
|
||||
which is the same failure the backups plugin dedups away.
|
||||
"""
|
||||
from plugins.guides.models import GuideRevision
|
||||
from plugins.guides.services import authoring
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
blocks = [_block('b1', 'paragraph', text='Unchanged.')]
|
||||
authoring.save(guide, title='T', blocks=blocks)
|
||||
_db.session.commit()
|
||||
|
||||
guide, second = authoring.save(guide, title='T', blocks=blocks)
|
||||
_db.session.commit()
|
||||
|
||||
assert second is None, 'an unchanged save must not manufacture a version'
|
||||
assert GuideRevision.query.filter_by(guideid=guide.guideid).count() == 1
|
||||
assert guide.revision == 1
|
||||
|
||||
|
||||
def test_a_real_edit_creates_the_next_version(gd_app, gd_clean):
|
||||
from plugins.guides.models import GuideRevision
|
||||
from plugins.guides.services import authoring
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
authoring.save(guide, title='T',
|
||||
blocks=[_block('b1', 'paragraph', text='One.')])
|
||||
_db.session.commit()
|
||||
guide, second = authoring.save(
|
||||
guide, blocks=[_block('b1', 'paragraph', text='Two.')])
|
||||
_db.session.commit()
|
||||
|
||||
assert second.revision == 2
|
||||
assert GuideRevision.query.filter_by(guideid=guide.guideid).count() == 2
|
||||
|
||||
|
||||
def test_readers_see_nothing_until_it_is_published(gd_app, gd_clean):
|
||||
"""A half-written procedure on a shop floor is worse than no procedure."""
|
||||
from plugins.guides.services import authoring
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
authoring.save(guide, title='T', blocks=[_block()])
|
||||
_db.session.commit()
|
||||
assert authoring.readable(guide) is None
|
||||
|
||||
|
||||
def test_editing_a_published_guide_does_not_change_what_readers_see(gd_app, gd_clean):
|
||||
"""THE point of the draft/publish split.
|
||||
|
||||
A technician halfway through a procedure keeps reading the version that was
|
||||
published, even while the next one is being written.
|
||||
"""
|
||||
from plugins.guides.services import authoring
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
authoring.save(guide, title='T',
|
||||
blocks=[_block('b1', 'paragraph', text='Published text.')])
|
||||
authoring.publish(guide)
|
||||
_db.session.commit()
|
||||
|
||||
authoring.save(guide,
|
||||
blocks=[_block('b1', 'paragraph', text='Work in progress.')])
|
||||
_db.session.commit()
|
||||
|
||||
served = authoring.readable(guide)
|
||||
assert served.blocks[0]['text'] == 'Published text.'
|
||||
assert guide.revision == 2, 'the working copy moved on'
|
||||
assert served.revision == 1, 'what readers get did not'
|
||||
|
||||
|
||||
def test_publishing_promotes_the_latest_saved_revision(gd_app, gd_clean):
|
||||
from plugins.guides.services import authoring
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
authoring.save(guide, title='T', blocks=[_block('b1', 'paragraph', text='v1')])
|
||||
authoring.save(guide, blocks=[_block('b1', 'paragraph', text='v2')])
|
||||
authoring.publish(guide)
|
||||
_db.session.commit()
|
||||
assert authoring.readable(guide).blocks[0]['text'] == 'v2'
|
||||
assert guide.status == 'published'
|
||||
|
||||
|
||||
def test_publishing_an_empty_guide_is_refused(gd_app, gd_clean):
|
||||
from plugins.guides.services import authoring
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
_db.session.commit()
|
||||
with pytest.raises(ValueError):
|
||||
authoring.publish(guide)
|
||||
|
||||
|
||||
def test_unpublishing_withdraws_without_destroying_the_history(gd_app, gd_clean):
|
||||
"""For a procedure that turns out to be wrong.
|
||||
|
||||
The versions stay, so it can be corrected and published again, and so there
|
||||
is still a record of what people were being told in the meantime.
|
||||
"""
|
||||
from plugins.guides.models import GuideRevision
|
||||
from plugins.guides.services import authoring
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
authoring.save(guide, title='T', blocks=[_block()])
|
||||
authoring.publish(guide)
|
||||
_db.session.commit()
|
||||
|
||||
authoring.unpublish(guide)
|
||||
_db.session.commit()
|
||||
|
||||
assert authoring.readable(guide) is None
|
||||
assert guide.status == 'draft'
|
||||
assert GuideRevision.query.filter_by(guideid=guide.guideid).count() == 1
|
||||
|
||||
|
||||
def test_restoring_writes_a_new_version_rather_than_rewriting_history(gd_app, gd_clean):
|
||||
"""A history you can edit is not a record of anything.
|
||||
|
||||
Restoring brings old content forward as a NEW version, so what was
|
||||
published in between stays visible.
|
||||
"""
|
||||
from plugins.guides.models import GuideRevision
|
||||
from plugins.guides.services import authoring
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
_, first = authoring.save(guide, title='T',
|
||||
blocks=[_block('b1', 'paragraph', text='original')])
|
||||
authoring.save(guide, blocks=[_block('b1', 'paragraph', text='mistake')])
|
||||
_db.session.commit()
|
||||
|
||||
authoring.restore(guide, first)
|
||||
_db.session.commit()
|
||||
|
||||
rows = (GuideRevision.query.filter_by(guideid=guide.guideid)
|
||||
.order_by(GuideRevision.revision).all())
|
||||
assert [r.revision for r in rows] == [1, 2, 3]
|
||||
assert rows[2].blocks[0]['text'] == 'original'
|
||||
assert 'Restored from version 1' in (rows[2].note or '')
|
||||
assert rows[1].blocks[0]['text'] == 'mistake', 'the mistake stays on record'
|
||||
|
||||
|
||||
def test_a_corrupt_blocks_column_renders_as_empty_rather_than_erroring(gd_app, gd_clean):
|
||||
"""A guide that will not render is a page a person cannot read.
|
||||
|
||||
An empty guide with its title and tags intact is a better failure than an
|
||||
error page, and the corruption is still visible - the editor shows nothing.
|
||||
"""
|
||||
with gd_app.app_context():
|
||||
guide = _newguide()
|
||||
guide.blocksjson = 'not json at all'
|
||||
_db.session.commit()
|
||||
assert guide.blocks == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The API surface. Everything above this line tests services directly; the
|
||||
# draft-invisibility rule lives in the ROUTES, and until these existed no test
|
||||
# executed a single one of them. That is how three routes shipped serving
|
||||
# unpublished drafts to any reader who could count upwards through ids.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _user_with(db, username, permissions, rolename=None):
|
||||
"""A user holding exactly these guides permissions and nothing else."""
|
||||
from werkzeug.security import generate_password_hash
|
||||
from shopdb.core.models import User, Role, Permission
|
||||
|
||||
role = Role(rolename=rolename or (username + 'role'), description=username)
|
||||
for name in permissions:
|
||||
# Shared across fixtures: two users in one test both want guides.view,
|
||||
# and permissions.name is unique.
|
||||
permission = Permission.query.filter_by(name=name).first()
|
||||
if permission is None:
|
||||
permission = Permission(name=name, description=name, category='guides')
|
||||
db.session.add(permission)
|
||||
db.session.flush()
|
||||
role.permissions.append(permission)
|
||||
db.session.add(role)
|
||||
db.session.flush()
|
||||
|
||||
user = User(username=username, email=username + '@test.local',
|
||||
passwordhash=generate_password_hash('testpass'))
|
||||
user.roles.append(role)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
return user
|
||||
|
||||
|
||||
def _login(client, username):
|
||||
response = client.post('/api/auth/login',
|
||||
json={'username': username, 'password': 'testpass'})
|
||||
assert response.status_code == 200, response.get_json()
|
||||
return {'Authorization':
|
||||
'Bearer ' + response.get_json()['data']['access_token']}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def reader(client, db):
|
||||
"""Can read guides. Cannot author them."""
|
||||
_user_with(db, 'guidereader', ['guides.view'])
|
||||
return _login(client, 'guidereader')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def author(client, db):
|
||||
"""Can write their own guides, but holds no guides.edit over others'."""
|
||||
_user_with(db, 'guideauthor', ['guides.view', 'guides.create'])
|
||||
return _login(client, 'guideauthor')
|
||||
|
||||
|
||||
def _draft(db, title='Unreleased procedure', published=False):
|
||||
"""A guide with one saved revision, published or not."""
|
||||
from plugins.guides.models import Guide
|
||||
from plugins.guides.services import authoring
|
||||
guide = Guide(title=title, slug=title.lower().replace(' ', '-'),
|
||||
status='draft')
|
||||
db.session.add(guide)
|
||||
db.session.flush()
|
||||
guide, revision = authoring.save(
|
||||
guide, title=title,
|
||||
blocks=[_block('b1', 'paragraph', text='secret draft wording')])
|
||||
if published:
|
||||
authoring.publish(guide)
|
||||
db.session.commit()
|
||||
return guide, revision
|
||||
|
||||
|
||||
def test_a_reader_cannot_open_an_unpublished_guide_by_slug(client, db, reader):
|
||||
guide, _ = _draft(db)
|
||||
response = client.get('/api/guides/' + guide.slug, headers=reader)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_a_reader_cannot_read_an_unpublished_revision_by_id(client, db, reader):
|
||||
"""Revision ids are sequential, so this route is walkable by hand.
|
||||
|
||||
Gating it on guides.view alone meant anyone who could read one guide could
|
||||
read every draft on the site by counting upwards.
|
||||
"""
|
||||
guide, revision = _draft(db)
|
||||
response = client.get('/api/guides/revisions/{}'.format(revision.guiderevisionid),
|
||||
headers=reader)
|
||||
assert response.status_code == 404
|
||||
assert 'secret draft wording' not in response.get_data(as_text=True)
|
||||
|
||||
|
||||
def test_a_reader_may_read_the_published_revision_by_id(client, db, reader):
|
||||
guide, revision = _draft(db, title='Published procedure', published=True)
|
||||
response = client.get('/api/guides/revisions/{}'.format(revision.guiderevisionid),
|
||||
headers=reader)
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_a_reader_cannot_list_a_guides_revision_history(client, db, reader):
|
||||
guide, _ = _draft(db, title='Published procedure', published=True)
|
||||
response = client.get('/api/guides/{}/revisions'.format(guide.guideid),
|
||||
headers=reader)
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_search_does_not_match_a_reader_against_draft_wording(client, db, reader):
|
||||
"""A hit a reader cannot open is worse than no hit.
|
||||
|
||||
The guide row holds the WORKING copy, so matching it would surface a
|
||||
published guide on a word that exists only in somebody's unpublished edit -
|
||||
and the word is not on the page when the reader opens it.
|
||||
"""
|
||||
from plugins.guides.services import authoring
|
||||
guide, revision = _draft(db, title='Published procedure', published=True)
|
||||
authoring.save(guide, blocks=[_block('b1', 'paragraph',
|
||||
text='unpublished sentence about widgets')])
|
||||
db.session.commit()
|
||||
|
||||
response = client.get('/api/guides?q=widgets', headers=reader)
|
||||
assert response.status_code == 200
|
||||
assert response.get_json()['data'] == []
|
||||
|
||||
# The author of the draft still finds their own work.
|
||||
response = client.get('/api/guides?q=secret', headers=reader)
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_an_author_may_edit_their_own_draft_but_not_someone_elses(client, db, author):
|
||||
guide, _ = _draft(db)
|
||||
response = client.get('/api/guides/{}/working'.format(guide.guideid),
|
||||
headers=author)
|
||||
assert response.status_code == 403, 'not their guide'
|
||||
|
||||
|
||||
def test_an_image_upload_refuses_a_type_that_is_not_an_image(client, db, author):
|
||||
import io
|
||||
data = {'file': (io.BytesIO(b'MZ not a picture'), 'payload.exe')}
|
||||
response = client.post('/api/guides/images', data=data, headers=author,
|
||||
content_type='multipart/form-data')
|
||||
assert response.status_code == 400
|
||||
assert 'Unsupported image type' in response.get_json()['data']['error']['message']
|
||||
|
||||
|
||||
def test_an_image_in_a_draft_is_not_served_to_a_reader(client, db, reader, author):
|
||||
"""An image is content, and a draft's images are draft content."""
|
||||
import io
|
||||
from plugins.guides.models import GuideImage
|
||||
|
||||
png = (b'\x89PNG\r\n\x1a\n' + b'0' * 32)
|
||||
upload = client.post(
|
||||
'/api/guides/images',
|
||||
data={'file': (io.BytesIO(png), 'screenshot.png')},
|
||||
headers=author, content_type='multipart/form-data')
|
||||
assert upload.status_code == 200, upload.get_json()
|
||||
imageid = upload.get_json()['data']['guideimageid']
|
||||
|
||||
# Unattached: it belongs to whoever uploaded it, not to every reader.
|
||||
assert client.get('/api/guides/images/{}'.format(imageid),
|
||||
headers=reader).status_code == 404
|
||||
assert client.get('/api/guides/images/{}'.format(imageid),
|
||||
headers=author).status_code == 200
|
||||
|
||||
# Attached to an unpublished guide: still not the reader's business.
|
||||
guide, _ = _draft(db)
|
||||
db.session.get(GuideImage, imageid).guideid = guide.guideid
|
||||
db.session.commit()
|
||||
assert client.get('/api/guides/images/{}'.format(imageid),
|
||||
headers=reader).status_code == 404
|
||||
|
||||
|
||||
def test_an_image_in_a_published_guide_is_served_and_hardened(client, db, reader, author):
|
||||
"""The other half: a published screenshot must actually reach the reader.
|
||||
|
||||
Guarding this route without checking the published case is how you trade a
|
||||
leak for every image on the site rendering as a broken icon.
|
||||
"""
|
||||
import io
|
||||
from plugins.guides.models import Guide, GuideImage
|
||||
from plugins.guides.services import authoring
|
||||
|
||||
png = (b'\x89PNG\r\n\x1a\n' + b'0' * 32)
|
||||
upload = client.post(
|
||||
'/api/guides/images',
|
||||
data={'file': (io.BytesIO(png), 'screenshot.png')},
|
||||
headers=author, content_type='multipart/form-data')
|
||||
imageid = upload.get_json()['data']['guideimageid']
|
||||
|
||||
guide = Guide(title='With a picture', slug='with-a-picture', status='draft')
|
||||
db.session.add(guide)
|
||||
db.session.flush()
|
||||
guide, revision = authoring.save(
|
||||
guide, title='With a picture',
|
||||
blocks=[{'id': 'b1', 'type': 'image', 'imageid': imageid,
|
||||
'url': '/api/guides/images/{}'.format(imageid)}])
|
||||
authoring.publish(guide)
|
||||
db.session.get(GuideImage, imageid).guideid = guide.guideid
|
||||
db.session.commit()
|
||||
|
||||
response = client.get('/api/guides/images/{}'.format(imageid), headers=reader)
|
||||
assert response.status_code == 200
|
||||
assert response.data == png
|
||||
# Uploaded bytes are attacker-influenced; they must be inert in this origin.
|
||||
assert response.headers.get('X-Content-Type-Options') == 'nosniff'
|
||||
assert 'Content-Security-Policy' in response.headers
|
||||
|
||||
|
||||
def test_renaming_a_guide_does_not_move_its_url(client, db, author):
|
||||
"""The stored slug exists so a pasted link keeps working.
|
||||
|
||||
The editor autosaves on a timer, so re-slugging on title change moved the
|
||||
URL while somebody was still typing the title.
|
||||
"""
|
||||
created = client.post('/api/guides',
|
||||
json={'title': 'Replacing a marker controler',
|
||||
'blocks': []}, headers=author)
|
||||
assert created.status_code == 200, created.get_json()
|
||||
guideid = created.get_json()['data']['guideid']
|
||||
slug = created.get_json()['data']['slug']
|
||||
|
||||
fixed = client.put('/api/guides/{}'.format(guideid),
|
||||
json={'title': 'Replacing a marker controller',
|
||||
'blocks': []}, headers=author)
|
||||
assert fixed.status_code == 200
|
||||
assert fixed.get_json()['data']['slug'] == slug, 'the URL must not move'
|
||||
assert fixed.get_json()['data']['title'] == 'Replacing a marker controller'
|
||||
Reference in New Issue
Block a user