shopdb-flask/tests/test_plugins/test_endpoints_script_runs.py

"""The Run Scripts tab's history feed.

A successful script produces a green badge and a one-line message; nothing in
that tells an operator whether the script did any work. These tests pin the
contract the tab relies on: only script runs, newest first, capped, and
carrying the agent's structured detail so the exit code and run context are
visible without opening the database.
"""

import json

from werkzeug.security import generate_password_hash


def _endpoint(db, hostname='scriptbox'):
    """An Asset + Computer pair with a reported state, as the routes expect."""
    from shopdb.api import Asset, AssetType
    from plugins.computers.models import Computer
    from plugins.endpoints.models import EndpointState

    assettype = AssetType.query.filter_by(assettype='computer').first()
    if assettype is None:
        assettype = AssetType(assettype='computer', pluginname='computers',
                              tablename='computers')
        db.session.add(assettype)
        db.session.flush()

    asset = Asset(assetnumber=hostname, name=hostname,
                  assettypeid=assettype.assettypeid, isactive=True)
    db.session.add(asset)
    db.session.flush()
    db.session.add(Computer(assetid=asset.assetid, hostname=hostname))
    db.session.add(EndpointState(assetid=asset.assetid, hostname=hostname))
    db.session.commit()
    return asset.assetid


def _viewer(db, client, username='scriptviewer'):
    from shopdb.core.models import User, Role, Permission

    Permission.seed()
    db.session.flush()
    role = Role.query.filter_by(rolename='scriptviewerrole').first()
    if role is None:
        role = Role(rolename='scriptviewerrole')
        db.session.add(role)
        db.session.flush()
    role.permissions = Permission.query.filter(
        Permission.name.in_(['endpoints.view'])).all()
    user = User(username=username, email=username + '@test.local',
                passwordhash=generate_password_hash('testpass'))
    user.roles.append(role)
    db.session.add(user)
    db.session.commit()

    response = client.post('/api/auth/login',
                           json={'username': username, 'password': 'testpass'})
    token = response.get_json()['data']['access_token']
    return {'Authorization': f'Bearer {token}'}


def _script(db, name='test-open-notepad'):
    from shopdb.core.models import Script

    script = Script(scriptname=name, runcontext='system')
    db.session.add(script)
    db.session.commit()
    return script.scriptid


def _command(db, assetid, target, createdat, action='run_script',
             status='succeeded', resultjson=None, resultmessage=None):
    from plugins.endpoints.models import EndpointCommand

    command = EndpointCommand(assetid=assetid, action=action, target=target,
                              status=status, resultmessage=resultmessage,
                              resultjson=resultjson, createdat=createdat)
    db.session.add(command)
    db.session.commit()
    return command.endpointcommandid


def test_only_script_runs_are_returned(app, client, db):
    """A restart sitting in the same feed is not a script run.

    The tab asks a narrower question than the activity panel, so filtering has
    to happen server-side; a mixed feed sliced client-side would quietly show
    fewer than five script runs whenever other commands interleave.
    """
    from datetime import datetime

    with app.app_context():
        assetid = _endpoint(db)
        scriptid = _script(db)
        headers = _viewer(db, client)
        _command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0))
        _command(db, assetid, 'restart', datetime(2026, 1, 1, 11, 0),
                 action='restart')

        response = client.get(f'/api/endpoints/{assetid}/scriptruns',
                              headers=headers)
        rows = response.get_json()['data']

        assert response.status_code == 200
        assert [row['action'] for row in rows] == ['run_script']


def test_newest_first_and_capped_at_five(app, client, db):
    """Six runs, five rows, most recent at the top."""
    from datetime import datetime

    with app.app_context():
        assetid = _endpoint(db)
        scriptid = _script(db)
        headers = _viewer(db, client)
        for hour in range(6):
            _command(db, assetid, f'script:{scriptid}',
                     datetime(2026, 1, 1, 10 + hour, 0))

        response = client.get(f'/api/endpoints/{assetid}/scriptruns',
                              headers=headers)
        rows = response.get_json()['data']

        assert len(rows) == 5
        stamps = [row['createdat'] for row in rows]
        assert stamps == sorted(stamps, reverse=True)


def test_agent_detail_is_exposed(app, client, db):
    """The exit code and run context are the evidence behind a green badge."""
    from datetime import datetime

    with app.app_context():
        assetid = _endpoint(db)
        scriptid = _script(db)
        headers = _viewer(db, client)
        _command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0),
                 resultmessage='Script completed (system context)',
                 resultjson=json.dumps({'kind': 'script', 'exitcode': 0,
                                        'runcontext': 'system'}))

        response = client.get(f'/api/endpoints/{assetid}/scriptruns',
                              headers=headers)
        row = response.get_json()['data'][0]

        assert row['resultdetail']['exitcode'] == 0
        assert row['resultdetail']['runcontext'] == 'system'


def test_the_script_name_is_resolved(app, client, db):
    """'script:5' is the wire target, not what the operator picked."""
    from datetime import datetime

    with app.app_context():
        assetid = _endpoint(db)
        scriptid = _script(db, name='rebuild-shopfloor-user')
        headers = _viewer(db, client)
        _command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0))

        response = client.get(f'/api/endpoints/{assetid}/scriptruns',
                              headers=headers)
        row = response.get_json()['data'][0]

        assert row['scriptname'] == 'rebuild-shopfloor-user'


def test_malformed_detail_does_not_break_the_list(app, client, db):
    """resultjson is agent-supplied text and can arrive truncated.

    A diagnostics panel failing to parse must degrade to "no detail", not take
    down the history that surrounds it.
    """
    from datetime import datetime

    with app.app_context():
        assetid = _endpoint(db)
        scriptid = _script(db)
        headers = _viewer(db, client)
        _command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0),
                 resultjson='{"kind": "script", "exitcode":')

        response = client.get(f'/api/endpoints/{assetid}/scriptruns',
                              headers=headers)

        assert response.status_code == 200
        assert response.get_json()['data'][0]['resultdetail'] is None


def test_view_permission_is_required(app, client, db):
    """History reveals what has been run on a machine, so it is gated."""
    from datetime import datetime

    with app.app_context():
        assetid = _endpoint(db)
        scriptid = _script(db)
        _command(db, assetid, f'script:{scriptid}', datetime(2026, 1, 1, 10, 0))

        response = client.get(f'/api/endpoints/{assetid}/scriptruns')

        assert response.status_code in (401, 403)