add operator issue reports for endpoint agent
Operators had no way to flag a problem PC from the tray. This adds a report path: the tray captures a reason, free-text note, and optional screenshot and the agent uploads it on the collector token to a new operatorreports table linked to the asset. Ships the endpoints 0009 migration so the table exists before the code queries it; the running DB had drifted behind the code and returned HTTP 500 on upload.
This commit is contained in:
benjamin.baker
237
tests/test_plugins/test_endpoints_operator_reports.py
Normal file
237
tests/test_plugins/test_endpoints_operator_reports.py
Normal file
@@ -0,0 +1,237 @@
|
||||
"""Operator issue report upload, retention purge, and hardened serve.
|
||||
|
||||
The tray captures a screenshot and a free-text note; the agent posts them to
|
||||
POST /api/endpoints/operatorreports on the collector key. The bytes land on
|
||||
disk under the operator-reports uploads dir and the row holds metadata. The
|
||||
human-facing panel and image-serve routes are JWT + endpoints.view gated, and
|
||||
the serve route carries the inert-upload headers. A retention sweep purges rows
|
||||
and files older than the configured window.
|
||||
"""
|
||||
|
||||
import io
|
||||
import os
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
from shopdb.core.models import Asset, AssetType, Setting
|
||||
|
||||
|
||||
UPLOAD_URL = '/api/endpoints/operatorreports'
|
||||
PANEL_URL = '/api/endpoints/panel/{}/operatorreports'
|
||||
IMAGE_URL = '/api/endpoints/operatorreports/{}/image'
|
||||
COLLECTOR_KEY = 'testcollectorkey'
|
||||
|
||||
|
||||
def _collector_headers(app):
|
||||
app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY
|
||||
return {'X-API-Key': COLLECTOR_KEY}
|
||||
|
||||
|
||||
def _computer(db, hostname):
|
||||
"""Create a computer Asset + Computer; return the asset."""
|
||||
from plugins.computers.models import Computer, ComputerType
|
||||
|
||||
atype = AssetType.query.filter_by(assettype='computer').first()
|
||||
if not atype:
|
||||
atype = AssetType(assettype='computer')
|
||||
db.session.add(atype)
|
||||
db.session.flush()
|
||||
|
||||
ctype = ComputerType.query.filter_by(computertype='CMM PC').first()
|
||||
if not ctype:
|
||||
ctype = ComputerType(computertype='CMM PC')
|
||||
db.session.add(ctype)
|
||||
db.session.flush()
|
||||
|
||||
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
|
||||
isactive=True)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
comp = Computer(assetid=asset.assetid, hostname=hostname,
|
||||
computertypeid=ctype.computertypeid)
|
||||
db.session.add(comp)
|
||||
db.session.commit()
|
||||
return asset
|
||||
|
||||
|
||||
def _png_bytes():
|
||||
"""A tiny valid 1x1 PNG."""
|
||||
import base64
|
||||
return base64.b64decode(
|
||||
'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk'
|
||||
'YPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==')
|
||||
|
||||
|
||||
def test_upload_creates_report_and_stores_screenshot(client, db, app):
|
||||
"""A collector-key upload with a screenshot creates a row and a file."""
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
|
||||
asset = _computer(db, 'OPREPORT1')
|
||||
data = {
|
||||
'hostname': 'OPREPORT1',
|
||||
'reason': 'Other (free text)',
|
||||
'details': 'the CMM app will not start',
|
||||
'screenshot': (io.BytesIO(_png_bytes()), 'shot.png'),
|
||||
}
|
||||
response = client.post(UPLOAD_URL, data=data,
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 201
|
||||
body = response.get_json()['data']
|
||||
assert body['assetid'] == asset.assetid
|
||||
assert body['details'] == 'the CMM app will not start'
|
||||
assert body['hasimage'] is True
|
||||
|
||||
report = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
assert report is not None
|
||||
assert report.imagefilename
|
||||
stored = os.path.join(app.instance_path, 'operatorreports',
|
||||
report.imagefilename)
|
||||
assert os.path.isfile(stored)
|
||||
|
||||
|
||||
def test_upload_without_screenshot_is_allowed(client, db, app):
|
||||
"""A note-only report (no screenshot) is valid; hasimage is false."""
|
||||
_computer(db, 'OPREPORT2')
|
||||
response = client.post(
|
||||
UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT2', 'details': 'noisy fan'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 201
|
||||
assert response.get_json()['data']['hasimage'] is False
|
||||
|
||||
|
||||
def test_upload_unknown_hostname_is_validation_error(client, db, app):
|
||||
"""A hostname with no computer row is rejected, not silently stored."""
|
||||
response = client.post(
|
||||
UPLOAD_URL,
|
||||
data={'hostname': 'NOSUCHBOX', 'details': 'x'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_upload_requires_collector_auth(client, db, app):
|
||||
"""Without the collector key the upload refuses (no JWT fallback)."""
|
||||
_computer(db, 'OPREPORT3')
|
||||
app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY
|
||||
response = client.post(
|
||||
UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT3', 'details': 'x'},
|
||||
content_type='multipart/form-data')
|
||||
assert response.status_code in (401, 403)
|
||||
|
||||
|
||||
def test_panel_lists_reports_newest_first(client, db, app, auth_headers):
|
||||
"""The panel route returns this asset's reports, newest first."""
|
||||
asset = _computer(db, 'OPREPORT4')
|
||||
for note in ('first', 'second'):
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT4', 'details': note},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
|
||||
response = client.get(PANEL_URL.format(asset.assetid), headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
reports = response.get_json()['data']['reports']
|
||||
assert [r['details'] for r in reports] == ['second', 'first']
|
||||
|
||||
|
||||
def test_served_image_is_hardened(client, db, app, auth_headers):
|
||||
"""The screenshot serve route carries the inert-upload headers."""
|
||||
from shopdb.utils.uploads import UPLOAD_CSP
|
||||
|
||||
asset = _computer(db, 'OPREPORT5')
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT5', 'details': 'x',
|
||||
'screenshot': (io.BytesIO(_png_bytes()), 'shot.png')},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
report = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
|
||||
response = client.get(IMAGE_URL.format(report.operatorreportid),
|
||||
headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
assert response.headers.get('Content-Security-Policy') == UPLOAD_CSP
|
||||
assert response.headers.get('X-Content-Type-Options') == 'nosniff'
|
||||
|
||||
|
||||
def test_image_serve_requires_jwt(client, db, app):
|
||||
"""The screenshot serve route is not reachable without a user JWT."""
|
||||
asset = _computer(db, 'OPREPORT6')
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT6', 'details': 'x',
|
||||
'screenshot': (io.BytesIO(_png_bytes()), 'shot.png')},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
report = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
|
||||
response = client.get(IMAGE_URL.format(report.operatorreportid))
|
||||
assert response.status_code in (401, 403)
|
||||
|
||||
|
||||
def test_retention_purge_removes_expired_reports_and_files(client, db, app):
|
||||
"""A report past the retention window is purged, file and all, on upload."""
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
from plugins.endpoints.api.routes import (
|
||||
OPERATOR_REPORT_RETENTION_SETTING,
|
||||
)
|
||||
|
||||
db.session.add(Setting(key=OPERATOR_REPORT_RETENTION_SETTING, value='7'))
|
||||
db.session.commit()
|
||||
|
||||
asset = _computer(db, 'OPREPORT7')
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT7', 'details': 'old',
|
||||
'screenshot': (io.BytesIO(_png_bytes()), 'shot.png')},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
old = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
old_file = os.path.join(app.instance_path, 'operatorreports',
|
||||
old.imagefilename)
|
||||
assert os.path.isfile(old_file)
|
||||
|
||||
# Backdate it past the window, then upload a fresh report to trigger the
|
||||
# opportunistic sweep.
|
||||
old.createdat = old.createdat - timedelta(days=8)
|
||||
db.session.commit()
|
||||
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT7', 'details': 'new'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
|
||||
assert db.session.get(OperatorReport, old.operatorreportid) is None
|
||||
assert not os.path.isfile(old_file)
|
||||
remaining = OperatorReport.query.filter_by(assetid=asset.assetid).all()
|
||||
assert [r.details for r in remaining] == ['new']
|
||||
|
||||
|
||||
def test_retention_disabled_keeps_old_reports(client, db, app):
|
||||
"""A retention of 0 disables the sweep; old rows survive."""
|
||||
from plugins.endpoints.models import OperatorReport
|
||||
from plugins.endpoints.api.routes import (
|
||||
OPERATOR_REPORT_RETENTION_SETTING,
|
||||
)
|
||||
|
||||
db.session.add(Setting(key=OPERATOR_REPORT_RETENTION_SETTING, value='0'))
|
||||
db.session.commit()
|
||||
|
||||
asset = _computer(db, 'OPREPORT8')
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT8', 'details': 'old'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
old = OperatorReport.query.filter_by(assetid=asset.assetid).first()
|
||||
old.createdat = old.createdat - timedelta(days=365)
|
||||
db.session.commit()
|
||||
|
||||
client.post(UPLOAD_URL,
|
||||
data={'hostname': 'OPREPORT8', 'details': 'new'},
|
||||
content_type='multipart/form-data',
|
||||
headers=_collector_headers(app))
|
||||
|
||||
assert db.session.get(OperatorReport, old.operatorreportid) is not None
|
||||
@@ -69,11 +69,12 @@ def test_endpoints_asset_panels_are_computer_scoped():
|
||||
"""Every asset panel targets computer assets only, with an endpoint."""
|
||||
plugin = EndpointsPlugin()
|
||||
panels = plugin.get_asset_panels()
|
||||
assert len(panels) == 6
|
||||
assert len(panels) == 7
|
||||
for panel in panels:
|
||||
assert panel['assettypes'] == ['computer']
|
||||
assert panel['endpoint'].startswith('/api/endpoints/asset/{assetid}')
|
||||
assert panel['render'] in ('keyvalue', 'list', 'table')
|
||||
assert panel['endpoint'].startswith('/api/endpoints/')
|
||||
assert '{assetid}' in panel['endpoint']
|
||||
assert panel['render'] in ('keyvalue', 'list', 'table', 'operatorreports')
|
||||
|
||||
|
||||
def test_services_and_processes_are_split_into_two_filtered_cards():
|
||||
|
||||
Reference in New Issue
Block a user