b1e5dd734d
Three map and aggregation defects. The first is a behaviour change you asked for; the other two are the 2026-09-03 run's findings. A dualpath secondary bay was excluded from /api/assets/map unconditionally while the collapse setting is on, so positioning it was a silent no-op: the form took the pin, stored mapx/mapy, and the map never drew it. One marker per pair is right when the pair sits in one place, and that stays the default - but a dual-bay machine whose second bay is physically elsewhere is exactly what positioning it separately means. Now: a secondary with its own position draws its own marker; one without is represented by the primary's, which carries the combined '2007 / 2008' label. Nothing extra is needed for the fallback, because the feed already requires mapx/mapy - an unpositioned bay never reaches the loop. The combined label is now suppressed when the partner draws its own marker. Labelling two separate pins '2007 / 2008' would say one machine is in two places, which is worse than either bay being unlabelled. Verified against a live instance: unpositioned secondary gives one marker labelled with both bays; positioned gives two markers each with its own name; clearing the position puts it back to one. That third case matters - a one-way door would be its own bug. machines/dashboard/summary applied the dualpath exclusion to total and bytype but not to bystatus, so the status bars summed to more than the headline above them, by exactly the number of secondary bays. A breakdown that does not add up to its own total is an afternoon of reconciling. The measuring-tool calibrationstatus filter ran AFTER pagination. It is a derived value so it cannot be a SQL filter, but applying it to the current page meant asking for 'overdue' with one overdue tool in sixteen returned an empty page 1, a total of 16, eight pages, and the row itself on page 3. When that filter is in play the matching set is now built and filtered before paging - the same tradeoff the old comment named, spent on correct answers instead of a cheaper query. Seven tests. The pagination one verified against the committed version, which fails it. Scope: product
647 lines
29 KiB
Python
647 lines
29 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}'
|
|
|
|
|
|
# --- a date that is not a date -----------------------------------------------
|
|
#
|
|
# _parse_date used to swallow the ValueError and return None, so a typo in the
|
|
# calibration date answered 200 "Measuring tool updated", stored NULL, flipped
|
|
# the tool to 'unknown', and dropped it out of the calibration report and the
|
|
# map overlay. Clearing a date and mistyping one are different intents and now
|
|
# get different answers.
|
|
|
|
def test_a_mistyped_calibration_date_is_refused_not_stored_as_null(
|
|
client, auth_headers):
|
|
created = client.post('/api/measuringtools', headers=auth_headers, json={
|
|
'assetnumber': 'MT-DATE-1',
|
|
'name': 'Date guard caliper',
|
|
'statusid': _status_id(client),
|
|
'nextcalibrationdate': '2027-01-01',
|
|
})
|
|
assert created.status_code == 201, created.get_json()
|
|
toolid = created.get_json()['data']['measuringtool']['measuringtoolid']
|
|
|
|
refused = client.put(f'/api/measuringtools/{toolid}', headers=auth_headers,
|
|
json={'nextcalibrationdate': 'not-a-date'})
|
|
assert refused.status_code == 400, refused.get_json()
|
|
assert 'nextcalibrationdate' in \
|
|
refused.get_json()['data']['error']['message']
|
|
|
|
# The stored date is untouched - the whole point of refusing.
|
|
after = client.get(f'/api/measuringtools/{toolid}',
|
|
headers=auth_headers).get_json()['data']
|
|
assert after['measuringtool']['nextcalibrationdate'] == '2027-01-01'
|
|
assert after['measuringtool']['calibrationstatus'] != 'unknown'
|
|
|
|
|
|
def test_an_explicitly_empty_calibration_date_still_clears_it(client,
|
|
auth_headers):
|
|
"""Clearing is a real intent and must keep working, or the fix above just
|
|
trades one broken behaviour for another."""
|
|
created = client.post('/api/measuringtools', headers=auth_headers, json={
|
|
'assetnumber': 'MT-DATE-2',
|
|
'name': 'Clearable caliper',
|
|
'statusid': _status_id(client),
|
|
'nextcalibrationdate': '2027-01-01',
|
|
})
|
|
toolid = created.get_json()['data']['measuringtool']['measuringtoolid']
|
|
|
|
cleared = client.put(f'/api/measuringtools/{toolid}', headers=auth_headers,
|
|
json={'nextcalibrationdate': ''})
|
|
assert cleared.status_code == 200, cleared.get_json()
|
|
after = client.get(f'/api/measuringtools/{toolid}',
|
|
headers=auth_headers).get_json()['data']
|
|
assert after['measuringtool']['nextcalibrationdate'] is None
|
|
|
|
|
|
def test_a_mistyped_date_is_refused_at_create_too(client, auth_headers):
|
|
refused = client.post('/api/measuringtools', headers=auth_headers, json={
|
|
'assetnumber': 'MT-DATE-3',
|
|
'name': 'Never created',
|
|
'statusid': _status_id(client),
|
|
'lastcalibrationdate': '01/02/2026',
|
|
})
|
|
assert refused.status_code == 400, refused.get_json()
|
|
listed = client.get('/api/measuringtools?search=MT-DATE-3',
|
|
headers=auth_headers).get_json()['data']
|
|
assert listed == [], 'nothing may be created by a refused request'
|
|
|
|
|
|
# --- an audited field holding a date must not kill the request ---------------
|
|
#
|
|
# PUT /api/machines/<id> returned 500 and rolled the whole edit back for any
|
|
# machine with a maintenance date: the route builds
|
|
# {'old': getattr(row, field)} straight off the model, and json.dumps raises on
|
|
# a date. A rename that never touched the date failed too. The audit log
|
|
# refusing to record a change must not be what stops the change.
|
|
|
|
def test_an_edit_is_saved_even_when_an_audited_field_holds_a_date(
|
|
client, auth_headers):
|
|
created = client.post('/api/measuringtools', headers=auth_headers, json={
|
|
'assetnumber': 'MT-AUDIT-1',
|
|
'name': 'Before rename',
|
|
'statusid': _status_id(client),
|
|
'lastcalibrationdate': '2026-01-15',
|
|
'nextcalibrationdate': '2027-01-15',
|
|
})
|
|
assert created.status_code == 201, created.get_json()
|
|
toolid = created.get_json()['data']['measuringtool']['measuringtoolid']
|
|
|
|
# A rename that does not touch the dates. The date is only in the CHANGES
|
|
# dict because the interval moves with it.
|
|
renamed = client.put(f'/api/measuringtools/{toolid}', headers=auth_headers,
|
|
json={'name': 'After rename',
|
|
'nextcalibrationdate': '2028-01-15'})
|
|
assert renamed.status_code == 200, renamed.get_json()
|
|
|
|
after = client.get(f'/api/measuringtools/{toolid}',
|
|
headers=auth_headers).get_json()['data']
|
|
assert after['name'] == 'After rename'
|
|
assert after['measuringtool']['nextcalibrationdate'] == '2028-01-15'
|
|
|
|
|
|
# --- a derived filter must page over the MATCHES ----------------------------
|
|
#
|
|
# calibrationstatus was applied to the rows of the current page, so a request
|
|
# for 'overdue' returned whatever matched on that page while the total counted
|
|
# every tool. With one overdue tool in sixteen: page 1 empty, total 16, eight
|
|
# pages, and the row itself on page 3. The filter hid matching rows and
|
|
# miscounted the rest.
|
|
|
|
def test_a_calibration_filter_pages_over_the_matches_not_the_page(
|
|
client, auth_headers):
|
|
from datetime import date, timedelta
|
|
|
|
overdue = (date.today() - timedelta(days=5)).isoformat()
|
|
current = (date.today() + timedelta(days=400)).isoformat()
|
|
|
|
# One overdue tool, well past the first page of a small page size.
|
|
for index in range(6):
|
|
created = client.post('/api/measuringtools', headers=auth_headers, json={
|
|
'assetnumber': 'MT-PAGE-%02d' % index,
|
|
'name': 'Paged tool %d' % index,
|
|
'statusid': _status_id(client),
|
|
'nextcalibrationdate': overdue if index == 5 else current,
|
|
})
|
|
assert created.status_code == 201, created.get_json()
|
|
|
|
# Scoped with search as well, so earlier tests in this module cannot
|
|
# change the expected set - the point is paging over matches, not how many
|
|
# overdue tools the whole fixture happens to hold.
|
|
response = client.get(
|
|
'/api/measuringtools?calibrationstatus=overdue&search=MT-PAGE&perpage=2',
|
|
headers=auth_headers)
|
|
assert response.status_code == 200, response.get_json()
|
|
body = response.get_json()
|
|
rows = body['data']
|
|
pagination = body['meta']['pagination']
|
|
|
|
assert [row['assetnumber'] for row in rows] == ['MT-PAGE-05'], \
|
|
'the overdue tool must be on page 1 of a filtered list'
|
|
assert pagination['total'] == 1, \
|
|
'the total counts the MATCHES, not every tool'
|
|
assert pagination['totalpages'] == 1
|
|
|
|
|
|
def test_an_unfiltered_list_still_pages_normally(client, auth_headers):
|
|
for index in range(3):
|
|
client.post('/api/measuringtools', headers=auth_headers, json={
|
|
'assetnumber': 'MT-PLAIN-%02d' % index,
|
|
'name': 'Plain tool %d' % index,
|
|
'statusid': _status_id(client),
|
|
})
|
|
|
|
response = client.get('/api/measuringtools?perpage=2', headers=auth_headers)
|
|
assert response.status_code == 200
|
|
pagination = response.get_json()['meta']['pagination']
|
|
assert pagination['perpage'] == 2
|
|
assert pagination['total'] >= 3
|