Files
shopdb-flask/tests/test_plugins/test_measuringtools.py
cproudlock afce24379c
All checks were successful
CI / naming (push) Successful in 3s
CI / frontend (push) Successful in 13s
CI / migrations-mysql (push) Successful in 39s
CI / backend (push) Successful in 3m42s
Reading anything needs a permission
A sweep of the URL map found 165 GET routes that answered anybody, with no
token at all: the whole asset inventory, every machine, PC, printer and network
device, the USB checkout history and who is holding which drive, the staff
directory with its photographs, every core report. Guest access had been argued
about one feature-flag at a time while the front door stood open.

Each of those now carries the read permission it belongs to. That also makes
the guest role mean something: a site can open exactly the reads it chooses,
and nothing follows by accident.

Thirty-three endpoints stay public, and each is named with its reason rather
than left to be inferred: the app shell and its bootstrap, the settings
allowlist (the handler already filters an anonymous caller down to branding and
a few keys), images fetched by <img src> which carries no Authorization header,
what a display kiosk renders before anybody signs in, the parts and USB kiosks,
the client scripts that run as SYSTEM before a user session exists, and a
secure-send link whose unguessable token IS the authorisation.

tests/test_core/test_public_read_surface.py is the standing gate, the companion
to the mutating-route sweep that has guarded writes for months. A new public
read fails CI until somebody adds it to that list in front of a reviewer. Four
further reads need a login but no permission, because the handler decides from
who you are - your own user record, your own API tokens, the role catalogue the
grid renders - and the same test proves each still refuses an anonymous caller.
Verified not vacuous by ungating one route and watching the sweep name it.

The employees plugin gained employees.view; the directory is names, SSOs,
photographs and reporting lines, and it had no permission of any kind.

Around ninety tests asserted payload shapes against endpoints that answered
anybody. They are about what the answer says, not about who may ask, so they
are signed in now rather than the gate being loosened. The ones that assert an
anonymous caller is refused were left exactly as they were.
2026-08-27 14:26:06 -04:00

491 lines
22 KiB
Python

"""Tests for the measuringtools plugin.
The plugin ships default-disabled, so the shared session app does not register
its blueprint. This module builds its own app, registers the blueprint, runs the
plugin's on_install seeding, and restores the process-wide plugin_manager
singleton afterwards (create_app repoints it) so other test modules are
unaffected - the same snapshot/restore dance used by test_plugin_migrations.
Coverage: derived calibration status (all four buckets + the 30-day boundary),
type CRUD with the in-use delete guard, tool create/update in one merged
payload, and the calibration report shape.
"""
from datetime import date, timedelta
import pytest
from werkzeug.security import generate_password_hash
from shopdb import create_app
from shopdb.extensions import db as _db
from shopdb.plugins import plugin_manager
from plugins.measuringtools.models import (
derive_status, DUESOON_WINDOW_DAYS, STATUS_COLORS)
@pytest.fixture(autouse=True)
def _signed_in(client, auth_headers):
"""Reading needs a permission now, and these tests are about what the
payload says rather than about who may see it. Signing the client in once
beats threading headers through every helper in the module."""
client.environ_base['HTTP_AUTHORIZATION'] = auth_headers['Authorization']
yield
client.environ_base.pop('HTTP_AUTHORIZATION', None)
# =============================================================================
# Pure unit tests: derived status (no app needed)
# =============================================================================
def test_derive_status_unknown_when_no_date():
assert derive_status(None) == 'unknown'
def test_derive_status_overdue_in_past():
assert derive_status(date.today() - timedelta(days=1)) == 'overdue'
def test_derive_status_current_far_future():
assert derive_status(date.today() + timedelta(days=DUESOON_WINDOW_DAYS + 1)) == 'current'
def test_derive_status_duesoon_within_window():
assert derive_status(date.today() + timedelta(days=DUESOON_WINDOW_DAYS - 1)) == 'duesoon'
def test_derive_status_duesoon_on_boundary():
"""Exactly DUESOON_WINDOW_DAYS out is still 'due soon' (inclusive)."""
assert derive_status(date.today() + timedelta(days=DUESOON_WINDOW_DAYS)) == 'duesoon'
def test_derive_status_duesoon_today(auth_headers):
"""Due today is not yet overdue - it falls in the due-soon window."""
assert derive_status(date.today()) == 'duesoon'
# =============================================================================
# API tests (self-contained app with the blueprint registered)
# =============================================================================
@pytest.fixture(scope='module')
def mt_app():
"""A testing app with the measuringtools blueprint registered + seeded."""
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.measuringtools.plugin import MeasuringToolsPlugin
plugin = MeasuringToolsPlugin()
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()
_seed(application, plugin)
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
def _seed(application, plugin):
"""Seed permissions, an admin, a status, and the plugin's own reference data."""
from shopdb.core.models import Permission, User, Role, AssetStatus
Permission.seed()
role = Role(rolename='admin', description='Administrator')
_db.session.add(role)
_db.session.add(AssetStatus(status='In Use', description='In use'))
_db.session.flush()
user = User(username='testadmin', email='admin@test.local',
passwordhash=generate_password_hash('testpass'))
user.roles.append(role)
_db.session.add(user)
_db.session.commit()
# Seeds AssetType 'measuring_tool' + the starter tool types.
plugin.on_install(application)
@pytest.fixture
def client(mt_app):
return mt_app.test_client()
@pytest.fixture
def auth_headers(client):
response = client.post('/api/auth/login',
json={'username': 'testadmin', 'password': 'testpass'})
assert response.status_code == 200, response.get_json()
token = response.get_json()['data']['access_token']
return {'Authorization': f'Bearer {token}'}
def _status_id(client):
return 1 # only status seeded
# -- Type CRUD + in-use guard -------------------------------------------------
def test_starter_types_seeded(client, auth_headers):
response = client.get('/api/measuringtools/types', headers=auth_headers)
assert response.status_code == 200
names = {t['name'] for t in response.get_json()['data']}
assert {'Caliper', 'Micrometer', 'Thread Gage', 'Other'} <= names
def test_type_create_requires_auth(client):
# This module signs the client in for every test; this one is about what
# happens when nobody is signed in, so it takes that back.
client.environ_base.pop('HTTP_AUTHORIZATION', None)
response = client.post('/api/measuringtools/types', json={'name': 'Nope'})
assert response.status_code == 401
def test_type_crud_lifecycle(client, auth_headers):
created = client.post('/api/measuringtools/types', headers=auth_headers,
json={'name': 'Pin Gage', 'description': 'Pin gage set',
'color': '#123456'})
assert created.status_code == 201, created.get_json()
type_id = created.get_json()['data']['measuringtooltypeid']
updated = client.put(f'/api/measuringtools/types/{type_id}', headers=auth_headers,
json={'description': 'Updated'})
assert updated.status_code == 200
assert updated.get_json()['data']['description'] == 'Updated'
dup = client.post('/api/measuringtools/types', headers=auth_headers,
json={'name': 'Pin Gage'})
assert dup.status_code == 409
deleted = client.delete(f'/api/measuringtools/types/{type_id}', headers=auth_headers)
assert deleted.status_code == 200
def test_type_delete_blocked_when_in_use(client, auth_headers):
type_created = client.post('/api/measuringtools/types', headers=auth_headers,
json={'name': 'Depth Gage'})
type_id = type_created.get_json()['data']['measuringtooltypeid']
tool_created = client.post('/api/measuringtools', headers=auth_headers,
json={'assetnumber': 'MT-INUSE-1',
'measuringtooltypeid': type_id,
'statusid': _status_id(client)})
assert tool_created.status_code == 201, tool_created.get_json()
blocked = client.delete(f'/api/measuringtools/types/{type_id}', headers=auth_headers)
assert blocked.status_code == 409
assert 'still use this type' in blocked.get_json()['data']['error']['message']
# -- Tool create/update merged payload ----------------------------------------
def test_tool_create_and_get_merged(client, auth_headers):
caliper = client.get('/api/measuringtools/types').get_json()['data']
caliper_id = next(t['measuringtooltypeid'] for t in caliper if t['name'] == 'Caliper')
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-001',
'name': 'Bench caliper',
'serialnumber': 'SN-CAL-9',
'gaugelabreference': 'GL-42',
'measuringtooltypeid': caliper_id,
'statusid': _status_id(client),
'calibrationintervaldays': 365,
'lastcalibrationdate': '2026-01-01',
'nextcalibrationdate': '2027-01-01',
'calibrationprovider': 'Metro Cal Lab',
})
assert created.status_code == 201, created.get_json()
payload = created.get_json()['data']
# Asset core fields at the top level, extension nested.
assert payload['assetnumber'] == 'MT-001'
assert payload['serialnumber'] == 'SN-CAL-9'
assert payload['gaugelabreference'] == 'GL-42'
ext = payload['measuringtool']
assert ext['measuringtooltypename'] == 'Caliper'
assert ext['calibrationprovider'] == 'Metro Cal Lab'
assert ext['calibrationstatus'] == 'current'
tool_id = ext['measuringtoolid']
fetched = client.get(f'/api/measuringtools/{tool_id}', headers=auth_headers)
assert fetched.status_code == 200
assert fetched.get_json()['data']['measuringtool']['measuringtoolid'] == tool_id
def test_tool_payload_carries_location_code_and_name(mt_app, client, auth_headers):
"""A tool at an operation location exposes locationcode (leading token)
and locationname so a printed label can encode the inspection operation."""
with mt_app.app_context():
from shopdb.core.models import Location
location = Location(locationname='0615 Blisk Inspection')
_db.session.add(location)
_db.session.commit()
location_id = location.locationid
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-LOC-1', 'statusid': _status_id(client),
'locationid': location_id,
})
assert created.status_code == 201, created.get_json()
payload = created.get_json()['data']
assert payload['locationname'] == '0615 Blisk Inspection'
assert payload['locationcode'] == '0615'
def test_tool_payload_location_code_none_when_unplaced(client, auth_headers):
"""A tool with no location degrades gracefully: locationcode is None."""
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-NOLOC-1', 'statusid': _status_id(client),
})
assert created.status_code == 201, created.get_json()
assert created.get_json()['data']['locationcode'] is None
def test_tool_update_merged_payload(client, auth_headers):
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-UPD-1', 'statusid': _status_id(client),
'nextcalibrationdate': '2030-01-01',
})
tool_id = created.get_json()['data']['measuringtool']['measuringtoolid']
updated = client.put(f'/api/measuringtools/{tool_id}', headers=auth_headers, json={
'name': 'Renamed tool', # asset core field
'calibrationprovider': 'In-house', # extension field
'nextcalibrationdate': str(date.today() - timedelta(days=5)), # -> overdue
})
assert updated.status_code == 200, updated.get_json()
data = updated.get_json()['data']
assert data['name'] == 'Renamed tool'
assert data['measuringtool']['calibrationprovider'] == 'In-house'
assert data['measuringtool']['calibrationstatus'] == 'overdue'
def test_tool_create_duplicate_assetnumber_conflicts(client, auth_headers):
body = {'assetnumber': 'MT-DUP', 'statusid': _status_id(client)}
first = client.post('/api/measuringtools', headers=auth_headers, json=body)
assert first.status_code == 201
second = client.post('/api/measuringtools', headers=auth_headers, json=body)
assert second.status_code == 409
def test_list_filter_by_calibrationstatus(client, auth_headers):
client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-OVERDUE-1', 'statusid': _status_id(client),
'nextcalibrationdate': str(date.today() - timedelta(days=10)),
})
response = client.get('/api/measuringtools?calibrationstatus=overdue', headers=auth_headers)
assert response.status_code == 200
rows = response.get_json()['data']
assert rows, 'expected at least one overdue tool'
assert all(r['measuringtool']['calibrationstatus'] == 'overdue' for r in rows)
# -- Maintenance reference identifier -----------------------------------------
def test_maintenancereference_roundtrip(client, auth_headers):
"""Maintenance reference persists on create and update."""
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-MNT-1', 'statusid': _status_id(client),
'maintenancereference': 'MNT-9',
})
assert created.status_code == 201, created.get_json()
assert created.get_json()['data']['maintenancereference'] == 'MNT-9'
tool_id = created.get_json()['data']['measuringtool']['measuringtoolid']
updated = client.put(f'/api/measuringtools/{tool_id}', headers=auth_headers,
json={'maintenancereference': 'MNT-10'})
assert updated.status_code == 200
assert updated.get_json()['data']['maintenancereference'] == 'MNT-10'
# -- Global search ------------------------------------------------------------
def _mt_search_hits(client, auth_headers, term):
resp = client.get(f'/api/search?q={term}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
return [r for r in resp.get_json()['data']['results']
if r.get('type') == 'measuring_tool']
def test_search_finds_tool_by_assetnumber(mt_app, client, auth_headers, monkeypatch):
"""A measuring tool appears in global search by asset number, routed right."""
client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-SEARCH-AN', 'statusid': _status_id(client)})
pm = mt_app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
hits = _mt_search_hits(client, auth_headers, 'MT-SEARCH-AN')
assert hits
assert hits[0]['url'].startswith('/measuringtools/')
def test_search_finds_tool_by_gaugelabreference(mt_app, client, auth_headers, monkeypatch):
"""A gage-tag lookup by gaugelabreference finds the tool."""
client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-SEARCH-GL', 'statusid': _status_id(client),
'gaugelabreference': 'GLREF-778'})
pm = mt_app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
hits = _mt_search_hits(client, auth_headers, 'GLREF-778')
assert hits
assert hits[0]['url'].startswith('/measuringtools/')
def test_search_excludes_tools_when_domain_disabled(mt_app, client, auth_headers, monkeypatch):
"""search_measuring_tool_enabled=false hides measuring tools from search."""
from shopdb.core.models import Setting
client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-SEARCH-OFF', 'statusid': _status_id(client),
'gaugelabreference': 'GLREF-OFF-1'})
pm = mt_app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
Setting.set('search_measuring_tool_enabled', False, valuetype='boolean',
category='search')
try:
assert _mt_search_hits(client, auth_headers, 'MT-SEARCH-OFF') == []
assert _mt_search_hits(client, auth_headers, 'GLREF-OFF-1') == []
finally:
Setting.set('search_measuring_tool_enabled', True, valuetype='boolean',
category='search')
# -- Report shape -------------------------------------------------------------
def test_calibration_report_shape(client, auth_headers):
response = client.get('/api/measuringtools/report/calibration', headers=auth_headers)
assert response.status_code == 200
data = response.get_json()['data']
assert set(data['counts']) == {'overdue', 'duesoon', 'current', 'unknown'}
assert set(data['buckets']) == {'overdue', 'duesoon', 'current', 'unknown'}
# Buckets and counts agree.
for key, rows in data['buckets'].items():
assert data['counts'][key] == len(rows)
assert 'statuscolors' in data
# -- Core integration: asset serialization ------------------------------------
def _caliper_id(client):
types = client.get('/api/measuringtools/types').get_json()['data']
return next(t['measuringtooltypeid'] for t in types if t['name'] == 'Caliper')
def test_asset_todict_carries_measuringtool_typedata_and_pluginid(mt_app, client, auth_headers):
"""Asset.to_dict resolves the measuringtool extension (typedata + pluginid)."""
from shopdb.core.models import Asset
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-TD-1', 'statusid': _status_id(client),
'measuringtooltypeid': _caliper_id(client),
})
assert created.status_code == 201, created.get_json()
tool_id = created.get_json()['data']['measuringtool']['measuringtoolid']
assetid = created.get_json()['data']['assetid']
with mt_app.app_context():
asset = _db.session.get(Asset, assetid)
result = asset.to_dict(include_type_data=True)
assert result['pluginid'] == tool_id
assert result['typedata']['measuringtooltypename'] == 'Caliper'
assert result['typedata']['measuringtoolid'] == tool_id
# -- Core integration: shop-floor map -----------------------------------------
def test_map_lists_measuringtool_subtypes(client, auth_headers):
"""The map filter dropdown carries a Measuring Tool subtype list with color."""
response = client.get('/api/assets/map', headers=auth_headers)
assert response.status_code == 200, response.get_json()
subtypes = response.get_json()['data']['filters']['subtypes']
assert 'Measuring Tool' in subtypes
names = {s['name'] for s in subtypes['Measuring Tool']}
assert 'Caliper' in names
assert all('color' in s for s in subtypes['Measuring Tool'])
def test_map_honors_measuringtool_subtype_filter(client, auth_headers):
"""?assettype=measuring_tool&subtype=<id> returns only that subtype's tools."""
types = client.get('/api/measuringtools/types').get_json()['data']
caliper_id = next(t['measuringtooltypeid'] for t in types if t['name'] == 'Caliper')
micrometer_id = next(t['measuringtooltypeid'] for t in types if t['name'] == 'Micrometer')
client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-MAP-CAL', 'statusid': _status_id(client),
'measuringtooltypeid': caliper_id, 'mapx': 10, 'mapy': 20})
client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-MAP-MIC', 'statusid': _status_id(client),
'measuringtooltypeid': micrometer_id, 'mapx': 30, 'mapy': 40})
response = client.get(
f'/api/assets/map?assettype=measuring_tool&subtype={caliper_id}', headers=auth_headers)
assert response.status_code == 200, response.get_json()
numbers = {a['assetnumber'] for a in response.get_json()['data']['assets']}
assert 'MT-MAP-CAL' in numbers
assert 'MT-MAP-MIC' not in numbers
def test_map_item_carries_measuringtool_typedata(client, auth_headers):
"""A mapped tool's item carries the extension typedata for marker coloring."""
client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-MAP-TD', 'statusid': _status_id(client),
'measuringtooltypeid': _caliper_id(client), 'mapx': 55, 'mapy': 66})
response = client.get('/api/assets/map?assettype=measuring_tool', headers=auth_headers)
item = next(a for a in response.get_json()['data']['assets']
if a['assetnumber'] == 'MT-MAP-TD')
assert item['typedata']['measuringtooltypename'] == 'Caliper'
# -- Core integration: dashboard ----------------------------------------------
def test_dashboard_counts_include_measuringtools(client, auth_headers):
"""Dashboard total and counts include active measuring tools."""
before = client.get('/api/dashboard').get_json()['data']
client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-DASH-1', 'statusid': _status_id(client)})
after = client.get('/api/dashboard').get_json()['data']
assert after['counts']['measuringtools'] == before['counts']['measuringtools'] + 1
assert after['totalmeasuringtool'] == before['totalmeasuringtool'] + 1
assert after['counts']['total'] == before['counts']['total'] + 1
# -- Map overlay endpoint (ADR-010) -------------------------------------------
def test_map_overlay_shape_and_derivation(client, auth_headers):
"""map-overlay returns per-asset derived calibration status + color."""
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-OVL-1', 'statusid': _status_id(client),
'nextcalibrationdate': str(date.today() - timedelta(days=3))}) # overdue
assetid = created.get_json()['data']['assetid']
response = client.get('/api/measuringtools/map-overlay', headers=auth_headers)
assert response.status_code == 200, response.get_json()
rows = response.get_json()['data']
# ADR-010 overlay contract: [{assetid, color, label}], only due/overdue tools.
row = next(r for r in rows if r['assetid'] == assetid)
assert set(row) == {'assetid', 'color', 'label'}
assert row['label'] == 'Overdue'
assert row['color'] == STATUS_COLORS['overdue']
def test_map_overlay_excludes_inactive(client, auth_headers):
"""A soft-deleted tool drops out of the overlay."""
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-OVL-DEL', 'statusid': _status_id(client)})
tool_id = created.get_json()['data']['measuringtool']['measuringtoolid']
assetid = created.get_json()['data']['assetid']
client.delete(f'/api/measuringtools/{tool_id}', headers=auth_headers)
rows = client.get('/api/measuringtools/map-overlay').get_json()['data']
assert all(r['assetid'] != assetid for r in rows)
# -- Presentation route token (ADR-010) ---------------------------------------
def test_asset_presentation_route_token():
"""Presentation route links through the by-asset resolver (only {assetid})."""
from plugins.measuringtools.plugin import MeasuringToolsPlugin
entries = MeasuringToolsPlugin().get_asset_presentation()
entry = next(e for e in entries if e['assettype'] == 'measuring_tool')
assert entry['route'] == '/measuringtools/by-asset/{assetid}'