"""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 # ============================================================================= # 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(): """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): response = client.get('/api/measuringtools/types') 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): 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}') assert fetched.status_code == 200 assert fetched.get_json()['data']['measuringtool']['measuringtoolid'] == tool_id 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') 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) # -- Report shape ------------------------------------------------------------- def test_calibration_report_shape(client, auth_headers): response = client.get('/api/measuringtools/report/calibration') 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