shopdb-flask/tests/test_plugins/test_guides.py

"""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'

def test_a_section_counts_as_something_to_tick():
    """A reader crosses off a step or a heading to keep their place, so those
    count towards progress too. A guide converted from an article often has
    fifty headings and no steps at all, and "0 steps to tick" on a card that is
    entirely tickable reads as a broken feature."""
    from plugins.guides.services.blocks import checklistitemcount
    blocks = [
        _block('b1', 'heading', text='Before you start'),
        _block('b2', 'paragraph', text='Not tickable.'),
        _block('b3', 'step', title='Do the thing'),
        _block('b4', 'checklist', items=[{'id': 'i1', 'text': 'One'},
                                         {'id': 'i2', 'text': 'Two'}]),
        _block('b5', 'code', text='echo hi'),
    ]
    # heading + step + two checklist items
    assert checklistitemcount(blocks) == 4


def test_restoring_the_current_version_says_nothing_changed(gd_app, gd_clean):
    """save() dedups on the content hash, so restoring the version that is
    already current writes nothing. Reporting "Restored version N as a new
    draft version" described a revision that did not exist, and the unchanged
    revision list quietly disagreed with the message."""
    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='only')])
        _db.session.commit()
        before = guide.revision

        authoring.restore(guide, first)
        _db.session.commit()

        assert guide.revision == before, 'nothing new was written'
        assert GuideRevision.query.filter_by(guideid=guide.guideid).count() == 1