diff --git a/CHANGELOG.md b/CHANGELOG.md index 887e11a..3d041ad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,23 @@ ADR-007 and ADR-002. ### Added +- Personal API tokens (PATs) so scripts and integrations authenticate without + the hourly-expiring login JWT (immediate consumer: long legacy-import runs + that die when the JWT expires mid-run). New core `apitokens` table + migration + `7d21_apitokens` (stores only the sha256 hash of each secret; the full secret + `shopdb_pat_<40 hex>` is shown ONCE at creation). New core blueprint + `/api/apitokens` (list own / admin `?all=true`; create; rename or deactivate; + revoke). A `Bearer shopdb_pat_...` header is recognized before any JWT decode + by a before_request shim that mints a request-scoped JWT for the token's + owner, so the entire existing auth+authz stack (jwt_required, + require_permission, require_role, import mode, current_user) authenticates the + PAT as its owner with zero decorator changes; an invalid, revoked, or expired + PAT gets a clean 401. `lastusedat` is stamped on use (throttled to at most one + write per 60s). Any authenticated user manages their own tokens; admins may + list or revoke anyone's. New Settings > API Tokens page (`ApiTokensList.vue`) + with a create modal that reveals the secret once (copy button) and an admin + All Tokens section. Docs: `docs/IMPORT-API.md` and `docs/CONFIG.md` updated to + recommend a PAT for imports. Core feature; no plugin-contract change. - Vendor-model photos on asset detail heroes: computers and printers now surface the linked model's `imageurl` in their extension payloads (the field machines already exposed), and the machine, PC, printer, network diff --git a/docs/CONFIG.md b/docs/CONFIG.md index e1eefe7..5e51785 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -221,6 +221,16 @@ them under `instance/branding/`. | `saml_auto_create_users` | `true` | Auto-create users on first SAML login. | | `saml_admin_group` | (empty) | SAML group name that grants the admin role. | +**Personal API tokens.** Besides login JWTs and SAML, a user may create +personal API tokens (PATs) for scripts and integrations, from Settings > API +Tokens (or `POST /api/apitokens`). A PAT is sent like a JWT +(`Authorization: Bearer shopdb_pat_...`), authenticates as its owning user +across the whole API, and does not carry the hourly `JWT_ACCESS_TOKEN_EXPIRES` +limit (it never expires unless an explicit expiry is set). Only the sha256 hash +is stored; the secret is shown once at creation. This is the recommended +credential for long-running imports (see `docs/IMPORT-API.md`). There is no env +var to configure; PATs are managed entirely through the API/UI. + ### identifiers (dynamic) One boolean key per asset identifier per asset type, keyed diff --git a/docs/IMPORT-API.md b/docs/IMPORT-API.md index 33c9261..1a99b6e 100644 --- a/docs/IMPORT-API.md +++ b/docs/IMPORT-API.md @@ -28,15 +28,34 @@ Contents: ### Admin token -Every write needs a JWT, and import mode additionally needs an admin. Get one: +Every write needs authentication, and import mode additionally needs an admin. + +A large import can outlast a login JWT: `access_token` expires after one hour, +so a long run dies mid-import with 401s. Use a **personal API token (PAT)** +instead. A PAT never expires (unless you set an expiry), acts as the user that +created it, and is sent exactly like a JWT. Create one as an admin (via the +Settings > API Tokens page, or the API): ```bash -curl -s http://localhost:5001/api/auth/login \ +# Bootstrap: a short login JWT is fine just to mint the long-lived PAT. +JWT=$(curl -s http://localhost:5001/api/auth/login \ -H 'Content-Type: application/json' \ - -d '{"username":"","password":""}' | jq -r '.data.access_token' + -d '{"username":"","password":""}' | jq -r '.data.access_token') + +# The full secret (shopdb_pat_...) is returned ONCE. Save it now. +curl -s http://localhost:5001/api/apitokens \ + -H "Authorization: Bearer $JWT" \ + -H 'Content-Type: application/json' \ + -d '{"name":"legacy import runner"}' | jq -r '.data.secret' ``` -Send it on every request as `Authorization: Bearer `. +Send the PAT on every request as `Authorization: Bearer shopdb_pat_...`. It +authenticates the whole import surface (every create/update/delete plus import +mode) as its owning admin, exactly as a login JWT would, but without the hourly +expiry. Revoke it from the same Settings page (or `DELETE /api/apitokens/`) +when the import is done. + +A short-lived login JWT still works for quick one-off calls if you prefer. ### Import mode: the `X-Import-Mode` header @@ -378,27 +397,26 @@ Each import-relevant list endpoint has an exact-match filter for its natural key ### Worked example -A small, dependency-free importer (`requests`) that logs in, does the -lookup-then-upsert loop in import mode, supports a `--dry-run` flag, and reports -errors without aborting the whole run: +A small, dependency-free importer (`requests`) that authenticates with a PAT +(so a multi-hour run cannot expire mid-import), does the lookup-then-upsert loop +in import mode, supports a `--dry-run` flag, and reports errors without aborting +the whole run: ```python import argparse +import os import requests BASE = "http://localhost:5001" class ImportClient: - def __init__(self, username, password, dryrun=False): + def __init__(self, token=None, dryrun=False): self.session = requests.Session() self.dryrun = dryrun - resp = self.session.post( - f"{BASE}/api/auth/login", - json={"username": username, "password": password}, - ) - resp.raise_for_status() - token = resp.json()["data"]["access_token"] + # A personal API token (shopdb_pat_...) does not expire like a login + # JWT, so it survives a long import. See section 1 to mint one. + token = token or os.environ["SHOPDB_TOKEN"] # X-Import-Mode makes createddate/modifieddate passthrough take effect. self.session.headers.update({ "Authorization": f"Bearer {token}", @@ -448,12 +466,12 @@ def import_vendors(client, legacyrows): if __name__ == "__main__": parser = argparse.ArgumentParser() - parser.add_argument("--user", required=True) - parser.add_argument("--password", required=True) + # PAT from the SHOPDB_TOKEN env var, or pass --token explicitly. + parser.add_argument("--token", default=None) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() - client = ImportClient(args.user, args.password, dryrun=args.dry_run) + client = ImportClient(args.token, dryrun=args.dry_run) # read legacy rows from prodscratch (read-only) and call the import_* fns # in the order of section 2, keeping a legacy-id -> new-id map as you go. ``` diff --git a/frontend/src/api/index.js b/frontend/src/api/index.js index 3abf387..9d1a289 100644 --- a/frontend/src/api/index.js +++ b/frontend/src/api/index.js @@ -906,6 +906,24 @@ export const usersApi = { } } +// Personal API tokens: authenticate scripts/integrations as a user without +// the hourly-expiring login JWT. The secret is returned ONCE, on create. +export const apitokensApi = { + // all=true (admin) lists everyone's tokens; otherwise just the caller's. + list(params = {}) { + return api.get('/apitokens', { params }) + }, + create(data) { + return api.post('/apitokens', data) + }, + update(id, data) { + return api.put(`/apitokens/${id}`, data) + }, + remove(id) { + return api.delete(`/apitokens/${id}`) + } +} + // Network API (devices, subnets, and VLANs) export const networkApi = { // Network devices diff --git a/frontend/src/components/ShopFloorMap.vue b/frontend/src/components/ShopFloorMap.vue index a54f93c..c6918a1 100644 --- a/frontend/src/components/ShopFloorMap.vue +++ b/frontend/src/components/ShopFloorMap.vue @@ -132,7 +132,9 @@ const assetTypeColorsMap = { 'computer': '#2196F3', // Blue 'printer': '#4CAF50', // Green 'network device': '#FF9800', // Orange - 'network_device': '#FF9800' // Orange (alternate key) + 'network_device': '#FF9800', // Orange (alternate key) + 'measuring_tool': '#9C27B0', // Purple + 'measuring tool': '#9C27B0' // Purple (normalized key) } // Get asset type color with case-insensitive lookup @@ -229,6 +231,7 @@ function getSubtypeId(asset) { if (typeLower === 'computer') return asset.typedata.computertypeid if (typeLower === 'network device') return asset.typedata.networkdevicetypeid if (typeLower === 'printer') return asset.typedata.printertypeid + if (typeLower === 'measuring tool') return asset.typedata.measuringtooltypeid return null } diff --git a/frontend/src/router/routes/core.js b/frontend/src/router/routes/core.js index ea35ef3..34b094d 100644 --- a/frontend/src/router/routes/core.js +++ b/frontend/src/router/routes/core.js @@ -240,6 +240,12 @@ export default [ component: () => import('../../views/settings/AuthenticationSettings.vue'), meta: { requiresAuth: true, requiresAdmin: true } }, + { + path: 'settings/apitokens', + name: 'apitokens-settings', + component: () => import('../../views/settings/ApiTokensList.vue'), + meta: { requiresAuth: true, requiresAdmin: true } + }, { path: 'settings/assetidentifiers', name: 'asset-identifiers-settings', diff --git a/frontend/src/router/routes/measuringtools.js b/frontend/src/router/routes/measuringtools.js index 52a6a78..fb60917 100644 --- a/frontend/src/router/routes/measuringtools.js +++ b/frontend/src/router/routes/measuringtools.js @@ -19,6 +19,14 @@ export default [ component: () => import('../../views/measuringtools/MeasuringToolForm.vue'), meta: { requiresAuth: true, plugin: 'measuringtools' } }, + { + // Resolve a tool from its core asset id (search rows / cross-links carry + // assetid, not the extension id). Shares the detail component. + path: 'measuringtools/by-asset/:assetid', + name: 'measuringtool-by-asset', + component: () => import('../../views/measuringtools/MeasuringToolDetail.vue'), + meta: { plugin: 'measuringtools' } + }, { path: 'measuringtools/:id', name: 'measuringtool-detail', diff --git a/frontend/src/stores/auth.js b/frontend/src/stores/auth.js index 8c91009..a6a0e95 100644 --- a/frontend/src/stores/auth.js +++ b/frontend/src/stores/auth.js @@ -15,9 +15,10 @@ export const useAuthStore = defineStore('auth', { isAdmin: (state) => state.user?.roles?.includes('admin') || false, // Full name from the employee directory (falls back to username/SSO). displayName: (state) => state.user?.directoryname || state.user?.username || '', - // Employee photo URL if the directory has one for this SSO. - avatarUrl: (state) => state.user?.directorypicture - ? `/static/employees/${state.user.directorypicture}` : null + // Employee photo URL if the directory has one for this SSO. The directory + // resolver already returns a usable URL (self-hosted upload or external HR + // path), so it is used as-is. + avatarUrl: (state) => state.user?.directoryphotourl || null }, actions: { @@ -77,7 +78,7 @@ export const useAuthStore = defineStore('auth', { const emp = response.data?.data if (emp) { this.user.directoryname = `${emp.First_Name || ''} ${emp.Last_Name || ''}`.trim() || null - this.user.directorypicture = emp.Picture || null + this.user.directoryphotourl = emp.photourl || null } } catch (err) { /* directory unavailable - fall back to username */ } } diff --git a/frontend/src/utils/mapColors.js b/frontend/src/utils/mapColors.js index 89c4bdf..230aedb 100644 --- a/frontend/src/utils/mapColors.js +++ b/frontend/src/utils/mapColors.js @@ -7,7 +7,9 @@ export const assetTypeColorsMap = { computer: '#2196F3', // Blue printer: '#4CAF50', // Green 'network device': '#FF9800', // Orange - network_device: '#FF9800' // Orange (alternate key) + network_device: '#FF9800', // Orange (alternate key) + measuring_tool: '#9C27B0', // Purple + 'measuring tool': '#9C27B0' // Purple (normalized key) } const DEFAULT_COLOR = '#BDBDBD' @@ -37,6 +39,7 @@ export function getSubtypeId(asset) { if (typeLower === 'computer') return asset.typedata.computertypeid if (typeLower === 'network device') return asset.typedata.networkdevicetypeid if (typeLower === 'printer') return asset.typedata.printertypeid + if (typeLower === 'measuring tool') return asset.typedata.measuringtooltypeid return null } diff --git a/frontend/src/views/MapEditor.vue b/frontend/src/views/MapEditor.vue index fa9b3a6..39330ff 100644 --- a/frontend/src/views/MapEditor.vue +++ b/frontend/src/views/MapEditor.vue @@ -18,6 +18,7 @@ + @@ -105,7 +106,7 @@ + + diff --git a/frontend/src/views/settings/settingsNav.js b/frontend/src/views/settings/settingsNav.js index 7fd4a4c..1a615ae 100644 --- a/frontend/src/views/settings/settingsNav.js +++ b/frontend/src/views/settings/settingsNav.js @@ -4,7 +4,7 @@ // Order: site identity + the reference-data catalogs users touch daily come // first, then the platform/system groups (integrations, communication, search, // plugins, access) cluster together at the end. -import { Factory, MapPin, Tag, Package, Droplets, Monitor, MonitorSmartphone, Laptop, Cog, Building, Globe, Link, History, FileText, Users, Puzzle, Bell, Network, Home, Wrench, Printer, Router, Palette, SlidersHorizontal, Contact, Mail, ShieldCheck, KeyRound, Fingerprint, Search } from 'lucide-vue-next' +import { Factory, MapPin, Tag, Package, Droplets, Monitor, MonitorSmartphone, Laptop, Cog, Building, Globe, Link, History, FileText, Users, Puzzle, Bell, Network, Home, Wrench, Printer, Router, Palette, SlidersHorizontal, Contact, Mail, ShieldCheck, KeyRound, Fingerprint, Search, Key } from 'lucide-vue-next' export const settingsGroups = [ { @@ -110,6 +110,7 @@ export const settingsGroups = [ title: 'Access & Security', cards: [ { to: '/settings/authentication', icon: KeyRound, title: 'Authentication', description: 'SAML single sign-on (SSO) with your IdP, local login, and auto-create users' }, + { to: '/settings/apitokens', icon: Key, title: 'API Tokens', description: 'Personal access tokens for scripts and integrations (e.g. long-running imports) that outlive login sessions' }, { to: '/settings/users', icon: Users, title: 'Users & Roles', description: 'Manage user accounts and permissions' }, { to: '/settings/audit', icon: History, title: 'Audit & Logging', description: 'Audit log retention period and history purge policy' }, { to: '/settings/auditlogs', icon: FileText, title: 'Audit Logs', description: 'View system activity and change history' }, diff --git a/frontend/src/views/warranty/WarrantiesList.vue b/frontend/src/views/warranty/WarrantiesList.vue index 3576338..4e83353 100644 --- a/frontend/src/views/warranty/WarrantiesList.vue +++ b/frontend/src/views/warranty/WarrantiesList.vue @@ -208,7 +208,7 @@ function formatDate(d) { return new Date(d + 'T00:00:00').toLocaleDateString() } // Route to the right detail page by asset type. function assetLink(a) { - const map = { computer: '/pcs/', printer: '/printers/', network_device: '/network/', machine: '/machines/' } + const map = { computer: '/pcs/', printer: '/printers/', network_device: '/network/', machine: '/machines/', measuring_tool: '/measuringtools/by-asset/' } const base = map[a.assettypename] || '/assets/' return base + a.assetid } diff --git a/migrations/versions/7d21_apitokens.py b/migrations/versions/7d21_apitokens.py new file mode 100644 index 0000000..72e373f --- /dev/null +++ b/migrations/versions/7d21_apitokens.py @@ -0,0 +1,61 @@ +"""Personal API tokens (apitokens) + +Adds the apitokens table: personal access tokens that let scripts and +integrations authenticate as a user without the hourly-expiring login JWT. +Only the sha256 hash of each secret is stored. + +Idempotent guard so it is safe on a partially-migrated box; real downgrade. + +Revision ID: 7d21_apitokens +Revises: 7d20_relationshiptypepropagations +Create Date: 2026-07-12 + +""" +from alembic import op +import sqlalchemy as sa + + +revision = '7d21_apitokens' +down_revision = '7d20_relationshiptypepropagations' +branch_labels = None +depends_on = None + + +def upgrade(): + bind = op.get_bind() + insp = sa.inspect(bind) + + if 'apitokens' in insp.get_table_names(): + return + + op.create_table( + 'apitokens', + sa.Column('tokenid', sa.Integer(), primary_key=True), + sa.Column('userid', sa.Integer(), nullable=False), + sa.Column('name', sa.String(length=100), nullable=False), + sa.Column('tokenprefix', sa.String(length=16), nullable=True), + sa.Column('tokenhash', sa.String(length=64), nullable=False), + sa.Column('expiresat', sa.DateTime(), nullable=True), + sa.Column('lastusedat', sa.DateTime(), nullable=True), + sa.Column('createddate', sa.DateTime(), nullable=False), + sa.Column('modifieddate', sa.DateTime(), nullable=False), + sa.Column('isactive', sa.Boolean(), nullable=False, server_default='1'), + sa.ForeignKeyConstraint(['userid'], ['users.userid']), + sa.UniqueConstraint('tokenhash', name='uq_apitoken_tokenhash'), + ) + op.create_index('ix_apitokens_userid', 'apitokens', ['userid']) + op.create_index('ix_apitokens_tokenprefix', 'apitokens', ['tokenprefix']) + op.create_index('ix_apitokens_tokenhash', 'apitokens', ['tokenhash']) + + +def downgrade(): + bind = op.get_bind() + insp = sa.inspect(bind) + + if 'apitokens' not in insp.get_table_names(): + return + + op.drop_index('ix_apitokens_tokenhash', table_name='apitokens') + op.drop_index('ix_apitokens_tokenprefix', table_name='apitokens') + op.drop_index('ix_apitokens_userid', table_name='apitokens') + op.drop_table('apitokens') diff --git a/plugins/measuringtools/api/routes.py b/plugins/measuringtools/api/routes.py index 2d3e9a2..36ad821 100644 --- a/plugins/measuringtools/api/routes.py +++ b/plugins/measuringtools/api/routes.py @@ -348,6 +348,32 @@ def delete_tool(tool_id: int): return success_response(message='Measuring tool deleted') +# ============================================================================= +# Map overlay (ADR-010 calibration-due badge) +# ============================================================================= + +@measuringtools_bp.route('/map-overlay', methods=['GET']) +@jwt_required(optional=True) +def map_overlay(): + """Calibration-status overlay for active measuring tools. + + The map places markers itself from the assets feed; this overlay only + supplies the per-asset calibration decoration. Consumers join by assetid, + so no map coordinates are returned. Status is derived at read time. + """ + today = date.today() + query = db.session.query(MeasuringTool).join(Asset).filter(Asset.isactive == True) + data = [] + for tool in query.all(): + status = derive_status(tool.nextcalibrationdate, today) + data.append({ + 'assetid': tool.assetid, + 'calibrationstatus': status, + 'statuscolor': STATUS_COLORS.get(status, STATUS_COLORS['unknown']), + }) + return success_response(data) + + # ============================================================================= # Calibration report (for the Reports hub) # ============================================================================= diff --git a/plugins/measuringtools/plugin.py b/plugins/measuringtools/plugin.py index 0c06ea3..88f0b70 100644 --- a/plugins/measuringtools/plugin.py +++ b/plugins/measuringtools/plugin.py @@ -112,12 +112,15 @@ class MeasuringToolsPlugin(BasePlugin): def get_asset_presentation(self) -> List[Dict]: # ADR-010 pilot. Tells core how to render + link the measuring_tool # asset type in global-search rows and cross-links. + # The consumer only substitutes {assetid} (search/cross-link rows carry + # the core asset id, not the extension id), so link through the by-asset + # resolver route rather than the id-keyed detail route. return [ { 'assettype': 'measuring_tool', 'icon': 'ruler', 'label': 'Measuring Tool', - 'route': '/measuringtools/{assetid}', + 'route': '/measuringtools/by-asset/{assetid}', }, ] diff --git a/shopdb/__init__.py b/shopdb/__init__.py index d0684e4..25e7def 100644 --- a/shopdb/__init__.py +++ b/shopdb/__init__.py @@ -86,6 +86,11 @@ def create_app(config_name: str = None) -> Flask: # Register core blueprints register_blueprints(app) + # Personal API token auth shim: recognize `Bearer shopdb_pat_...` before + # any JWT decode and mint a request-scoped JWT for the token's owner. + from .utils.apitoken_auth import install_apitoken_auth + install_apitoken_auth(app) + # Register CLI commands register_cli_commands(app) @@ -128,6 +133,7 @@ CORE_BLUEPRINT_NAMES = ( 'customfields', 'setup', 'pluginui', + 'apitokens', ) diff --git a/shopdb/core/api/__init__.py b/shopdb/core/api/__init__.py index 80bc8bb..2a699d3 100644 --- a/shopdb/core/api/__init__.py +++ b/shopdb/core/api/__init__.py @@ -22,6 +22,7 @@ from .users import users_bp from .customfields import customfields_bp from .setup import setup_bp from .pluginui import pluginui_bp +from .apitokens import apitokens_bp __all__ = [ 'auth_bp', @@ -46,4 +47,5 @@ __all__ = [ 'customfields_bp', 'setup_bp', 'pluginui_bp', + 'apitokens_bp', ] diff --git a/shopdb/core/api/apitokens.py b/shopdb/core/api/apitokens.py new file mode 100644 index 0000000..ba8f1f0 --- /dev/null +++ b/shopdb/core/api/apitokens.py @@ -0,0 +1,120 @@ +"""Personal API token management endpoints. + +Any authenticated user may manage their OWN tokens; an admin may list or revoke +anyone's. Endpoints are jwt_required (a token must be bootstrapped from a real +login or an existing token). The full secret is returned ONCE, on create. +""" + +from flask import Blueprint, request +from flask_jwt_extended import jwt_required, current_user + +from shopdb.extensions import db +from shopdb.core.models import ApiToken, AuditLog +from shopdb.utils.responses import success_response, error_response, ErrorCodes +from shopdb.utils.import_mode import parse_import_datetime + +apitokens_bp = Blueprint('apitokens', __name__) + + +@apitokens_bp.route('', methods=['GET']) +@jwt_required() +def list_apitokens(): + """List the caller's own tokens. Admins may pass ?all=true for everyone's. + + Never returns hashes or secrets. + """ + wants_all = request.args.get('all', 'false').lower() == 'true' + is_admin = current_user.hasrole('admin') + + query = ApiToken.query + if wants_all and is_admin: + include_owner = True + else: + query = query.filter(ApiToken.userid == current_user.userid) + include_owner = False + + query = query.order_by(ApiToken.createddate.desc()) + tokens = [t.to_dict(include_owner=include_owner) for t in query.all()] + return success_response(tokens) + + +@apitokens_bp.route('', methods=['POST']) +@jwt_required() +def create_apitoken(): + """Create a token for the caller. Returns the full secret ONCE.""" + data = request.get_json() or {} + + name = (data.get('name') or '').strip() + if not name: + return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required') + + expiresat = None + if data.get('expiresat'): + expiresat = parse_import_datetime(data.get('expiresat')) + if expiresat is None: + return error_response(ErrorCodes.VALIDATION_ERROR, + 'expiresat is not a valid date/datetime') + + secret = ApiToken.generate_secret() + token = ApiToken( + userid=current_user.userid, + name=name, + tokenprefix=ApiToken.prefix_of(secret), + tokenhash=ApiToken.hash_secret(secret), + expiresat=expiresat, + ) + db.session.add(token) + db.session.flush() + + AuditLog.log('created', 'ApiToken', entityid=token.tokenid, entityname=name) + db.session.commit() + + result = token.to_dict() + # The secret appears here and NOWHERE else, ever. Not stored, not logged. + result['secret'] = secret + result['warning'] = ('Save this token now. It will not be shown again. ' + 'Store it somewhere safe.') + return success_response(result, message='Token created', http_code=201) + + +@apitokens_bp.route('/', methods=['PUT']) +@jwt_required() +def update_apitoken(tokenid: int): + """Rename or deactivate a token. Own token, or any if admin.""" + token = db.session.get(ApiToken, tokenid) + if token is None: + return error_response(ErrorCodes.NOT_FOUND, 'Token not found', http_code=404) + + if token.userid != current_user.userid and not current_user.hasrole('admin'): + return error_response(ErrorCodes.FORBIDDEN, + 'You may only manage your own tokens', http_code=403) + + data = request.get_json() or {} + if 'name' in data: + newname = (data.get('name') or '').strip() + if not newname: + return error_response(ErrorCodes.VALIDATION_ERROR, 'name cannot be empty') + token.name = newname + if 'isactive' in data: + token.isactive = bool(data['isactive']) + + db.session.commit() + return success_response(token.to_dict(), message='Token updated') + + +@apitokens_bp.route('/', methods=['DELETE']) +@jwt_required() +def revoke_apitoken(tokenid: int): + """Revoke (deactivate) a token. Own token, or any if admin.""" + token = db.session.get(ApiToken, tokenid) + if token is None: + return error_response(ErrorCodes.NOT_FOUND, 'Token not found', http_code=404) + + if token.userid != current_user.userid and not current_user.hasrole('admin'): + return error_response(ErrorCodes.FORBIDDEN, + 'You may only manage your own tokens', http_code=403) + + token.isactive = False + AuditLog.log('deleted', 'ApiToken', entityid=token.tokenid, entityname=token.name) + db.session.commit() + return success_response(message='Token revoked') diff --git a/shopdb/core/api/assets.py b/shopdb/core/api/assets.py index 340a835..1a02965 100644 --- a/shopdb/core/api/assets.py +++ b/shopdb/core/api/assets.py @@ -913,6 +913,14 @@ def get_assets_map(): ) except (ImportError, AttributeError): pass + try: + from plugins.measuringtools.models import MeasuringTool + eager_options.append( + subqueryload(Asset.measuringtool) + .joinedload(MeasuringTool.measuringtooltype) + ) + except (ImportError, AttributeError): + pass query = Asset.query.options(*eager_options).filter( Asset.isactive == True, @@ -941,7 +949,10 @@ def get_assets_map(): # Filter by subtype (depends on asset type) - case-insensitive matching if subtype_id := request.args.get('subtype'): subtype_id = int(subtype_id) - asset_type_lower = selected_assettype.lower() if selected_assettype else '' + # Normalize the underscore DB form (measuring_tool, network_device) to + # the space form the branches below compare against. + asset_type_lower = ( + selected_assettype.lower().replace('_', ' ') if selected_assettype else '') if asset_type_lower == 'machine': try: from plugins.machines.models import Machine @@ -974,6 +985,15 @@ def get_assets_map(): ) except ImportError: pass + elif asset_type_lower == 'measuring tool': + try: + from plugins.measuringtools.models import MeasuringTool + query = query.join( + MeasuringTool, MeasuringTool.assetid == Asset.assetid).filter( + MeasuringTool.measuringtooltypeid == subtype_id + ) + except ImportError: + pass # Filter by business unit if bu_id := request.args.get('businessunitid'): @@ -1100,6 +1120,14 @@ def get_assets_map(): except ImportError: subtypes['Printer'] = [] + try: + from plugins.measuringtools.models import MeasuringToolType + measuringtool_types = MeasuringToolType.query.filter( + MeasuringToolType.isactive == True).order_by(MeasuringToolType.name).all() + subtypes['Measuring Tool'] = [{'id': mt.measuringtooltypeid, 'name': mt.name, 'color': mt.color} for mt in measuringtool_types] + except ImportError: + subtypes['Measuring Tool'] = [] + return success_response({ 'assets': data, 'total': len(data), diff --git a/shopdb/core/api/dashboard.py b/shopdb/core/api/dashboard.py index 79cac0c..44bc2dd 100644 --- a/shopdb/core/api/dashboard.py +++ b/shopdb/core/api/dashboard.py @@ -15,6 +15,7 @@ _TYPE_CATEGORY = { 'computer': 'PC', 'printer': 'Printer', 'network_device': 'Network', + 'measuring_tool': 'Measuring Tool', } @@ -45,7 +46,8 @@ def get_dashboard(): pc_count = _count_by_type('computer') network_count = _count_by_type('network_device') printer_count = _count_by_type('printer') - total = machine_count + pc_count + network_count + printer_count + measuringtool_count = _count_by_type('measuring_tool') + total = machine_count + pc_count + network_count + printer_count + measuringtool_count # Count by status status_counts = db.session.query( @@ -70,6 +72,7 @@ def get_dashboard(): 'totalpc': pc_count, 'totalnetwork': network_count, 'totalprinter': printer_count, + 'totalmeasuringtool': measuringtool_count, 'activeassets': status_dict.get('In Use', 0), 'inrepair': status_dict.get('In Repair', 0), # Structured data @@ -78,6 +81,7 @@ def get_dashboard(): 'pcs': pc_count, 'networkdevices': network_count, 'printers': printer_count, + 'measuringtools': measuringtool_count, 'total': total }, 'bystatus': status_dict, diff --git a/shopdb/core/models/__init__.py b/shopdb/core/models/__init__.py index 60b5b74..90e4fa3 100644 --- a/shopdb/core/models/__init__.py +++ b/shopdb/core/models/__init__.py @@ -17,6 +17,7 @@ from .supportteam import SupportTeam, SupportTeamContact from .setting import Setting from .auditlog import AuditLog from .customfield import CustomField, CustomFieldValue +from .apitoken import ApiToken __all__ = [ # Base @@ -62,4 +63,6 @@ __all__ = [ # Custom fields 'CustomField', 'CustomFieldValue', + # Personal API tokens + 'ApiToken', ] diff --git a/shopdb/core/models/apitoken.py b/shopdb/core/models/apitoken.py new file mode 100644 index 0000000..b0245dc --- /dev/null +++ b/shopdb/core/models/apitoken.py @@ -0,0 +1,89 @@ +"""Personal API token model. + +A personal API token (PAT) lets a script or integration authenticate as a +user without the hourly-expiring login JWT. The secret is shown ONCE at +creation; only its sha256 hash is stored. The token acts as its owning user, +so the existing role/permission decorators authorize it unchanged. +""" + +import hashlib +import secrets +from datetime import datetime, timezone + +from shopdb.extensions import db +from .base import BaseModel + + +def _utcnow(): + # naive UTC to match the other DB DateTime columns (stored without tzinfo) + return datetime.now(timezone.utc).replace(tzinfo=None) + + +# Wire label on the full secret. Scripts send "Authorization: Bearer ". +TOKEN_SECRET_PREFIX = 'shopdb_pat_' +# Hex chars of randomness after the label (secrets.token_hex(20) => 40 hex). +_TOKEN_RANDOM_BYTES = 20 +# How many leading random-hex chars we keep in the clear for display/lookup. +_TOKEN_PREFIX_LEN = 8 + + +class ApiToken(BaseModel): + """Personal API token. Stores only the hash of the secret.""" + __tablename__ = 'apitokens' + + tokenid = db.Column(db.Integer, primary_key=True) + # The token acts as this user; NOT NULL so authz always has a principal. + userid = db.Column(db.Integer, db.ForeignKey('users.userid'), + nullable=False, index=True) + # What the token is for (e.g. "legacy import runner"). + name = db.Column(db.String(100), nullable=False) + # First few random-hex chars, kept clear so a user can tell tokens apart. + tokenprefix = db.Column(db.String(16), nullable=True, index=True) + # sha256 hex of the full secret. Unique so a hash lookup finds one row. + tokenhash = db.Column(db.String(64), unique=True, nullable=False, index=True) + # Null expiresat means the token never expires. + expiresat = db.Column(db.DateTime, nullable=True) + # Last time the token authenticated a request (throttled write). + lastusedat = db.Column(db.DateTime, nullable=True) + + user = db.relationship('User', backref=db.backref('apitokens', lazy='dynamic')) + + @staticmethod + def generate_secret() -> str: + """Return a fresh full secret: shopdb_pat_<40 hex>. Never stored.""" + return TOKEN_SECRET_PREFIX + secrets.token_hex(_TOKEN_RANDOM_BYTES) + + @staticmethod + def hash_secret(secret: str) -> str: + """sha256 hex of the full secret. The token has 160 bits of entropy, + so a plain hash lookup (not a slow password hash) is appropriate.""" + return hashlib.sha256(secret.encode('utf-8')).hexdigest() + + @staticmethod + def prefix_of(secret: str) -> str: + """The clear display prefix (leading random-hex chars) of a secret.""" + randompart = secret[len(TOKEN_SECRET_PREFIX):] + return randompart[:_TOKEN_PREFIX_LEN] + + @property + def is_expired(self) -> bool: + """True when expiresat is set and in the past.""" + return self.expiresat is not None and self.expiresat < _utcnow() + + def to_dict(self, include_owner: bool = False) -> dict: + """Serialize for the API. NEVER includes the hash or the secret.""" + result = { + 'tokenid': self.tokenid, + 'userid': self.userid, + 'name': self.name, + 'tokenprefix': self.tokenprefix, + 'displayprefix': f'{TOKEN_SECRET_PREFIX}{self.tokenprefix or ""}', + 'expiresat': self.expiresat.isoformat() + 'Z' if self.expiresat else None, + 'lastusedat': self.lastusedat.isoformat() + 'Z' if self.lastusedat else None, + 'isactive': self.isactive, + 'isexpired': self.is_expired, + 'createddate': self.createddate.isoformat() + 'Z' if self.createddate else None, + } + if include_owner: + result['username'] = self.user.username if self.user else None + return result diff --git a/shopdb/core/models/asset.py b/shopdb/core/models/asset.py index fe413d1..a836b7b 100644 --- a/shopdb/core/models/asset.py +++ b/shopdb/core/models/asset.py @@ -234,6 +234,8 @@ class Asset(BaseModel, SoftDeleteMixin, AuditMixin): result['pluginid'] = self.network_device.networkdeviceid elif hasattr(self, 'printer') and self.printer: result['pluginid'] = self.printer.printerid + elif hasattr(self, 'measuringtool') and self.measuringtool: + result['pluginid'] = self.measuringtool.measuringtoolid # Include inherited location if this asset has no location data if include_inherited_location: @@ -271,4 +273,7 @@ class Asset(BaseModel, SoftDeleteMixin, AuditMixin): # Check for printer extension if hasattr(self, 'printer') and self.printer: return self.printer.to_dict() + # Check for measuring-tool extension + if hasattr(self, 'measuringtool') and self.measuringtool: + return self.measuringtool.to_dict() return None diff --git a/shopdb/utils/apitoken_auth.py b/shopdb/utils/apitoken_auth.py new file mode 100644 index 0000000..8de7e25 --- /dev/null +++ b/shopdb/utils/apitoken_auth.py @@ -0,0 +1,108 @@ +"""Personal API token (PAT) authentication shim. + +A request may send `Authorization: Bearer shopdb_pat_...`. This is recognized +BEFORE any JWT decode: a before_request hook validates the PAT (hash lookup, +active, not expired, active owner) and, on success, mints a short internal +request-scoped JWT for the token's user and swaps it into the request's +Authorization header. + +Why mint a JWT instead of only stashing the user on g: every write route in +this app stacks a mandatory @jwt_required() ABOVE @require_permission / +@require_role. That mandatory decorator decodes the Authorization header +itself, so the ONLY way a PAT reaches the whole existing auth+authz stack +(jwt_required, require_permission, require_role, import_mode, current_user, +get_jwt_identity) unchanged is to present a genuine JWT downstream. The minted +token lives only in this request's environ and is never returned to the client. + +Result: a PAT authenticates any route a login JWT would, acting as its owner, +with zero changes to the authz decorators or import-mode helpers. +""" + +from datetime import datetime, timezone + +from flask import g, request +from flask_jwt_extended import create_access_token + +from shopdb.extensions import db +from shopdb.core.models.apitoken import ApiToken, TOKEN_SECRET_PREFIX +from shopdb.utils.responses import error_response, ErrorCodes + + +# Only rewrite lastusedat when it is older than this, to avoid a DB write on +# every single request a busy integration makes. +_LASTUSED_THROTTLE_SECONDS = 60 + + +def _utcnow(): + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def _extract_pat_secret(): + """Return the PAT secret from the Authorization header, or None.""" + header = request.headers.get('Authorization', '') + parts = header.split() + if len(parts) == 2 and parts[0] == 'Bearer' \ + and parts[1].startswith(TOKEN_SECRET_PREFIX): + return parts[1] + return None + + +def _resolve_pat(secret): + """Validate a PAT secret. Return (token, user) or None.""" + from shopdb.core.models import User + + token = ApiToken.query.filter_by( + tokenhash=ApiToken.hash_secret(secret), isactive=True).first() + if token is None or token.is_expired: + return None + user = db.session.get(User, token.userid) + if user is None or not user.isactive: + return None + return token, user + + +def _touch_lastused(token): + """Throttled lastusedat write. Independent commit; nothing else is pending + this early in the request, so it cannot clobber route work.""" + now = _utcnow() + if token.lastusedat is None \ + or (now - token.lastusedat).total_seconds() > _LASTUSED_THROTTLE_SECONDS: + token.lastusedat = now + db.session.commit() + + +def install_apitoken_auth(app): + """Register the before_request PAT shim on the app.""" + + @app.before_request + def _apitoken_before_request(): + secret = _extract_pat_secret() + if secret is None: + return + resolved = _resolve_pat(secret) + if resolved is None: + # The caller clearly meant to use a PAT (shopdb_pat_ prefix) but it + # is unknown, revoked, or expired. Reject with a clear 401 instead + # of letting the JWT decoder emit a confusing 422 on the non-JWT. + return error_response( + ErrorCodes.UNAUTHORIZED, + 'Invalid, revoked, or expired API token', + http_code=401) + token, user = resolved + + # Read claim inputs before the (possible) commit expires the instance. + claims = { + 'username': user.username, + 'roles': [role.rolename for role in user.roles], + } + # Expose the token/user for audit and introspection if a handler wants it. + g.apitokenid = token.tokenid + g.apitokenuser = user + + _touch_lastused(token) + + # Mint a request-scoped JWT for the owner and swap it into the header + # so the whole downstream auth stack authenticates as that user. + access_token = create_access_token( + identity=str(user.userid), additional_claims=claims) + request.environ['HTTP_AUTHORIZATION'] = f'Bearer {access_token}' diff --git a/tests/test_core/test_apitokens.py b/tests/test_core/test_apitokens.py new file mode 100644 index 0000000..58ac265 --- /dev/null +++ b/tests/test_core/test_apitokens.py @@ -0,0 +1,178 @@ +"""Personal API token tests. + +Covers: create returns the secret once and stores only a hash; a PAT +authenticates a permission-gated write as its owner; a PAT is rejected when its +owner lacks the permission; expired and revoked tokens are rejected; lastusedat +updates on use; a non-owner member cannot revoke someone else's token; an admin +lists everyone's tokens with ?all=true; and import mode works over a PAT for an +admin. +""" + +from datetime import datetime, timedelta, timezone + +from shopdb.core.models import ApiToken, Vendor +from shopdb.extensions import db as _db + + +def _naive_utcnow(): + return datetime.now(timezone.utc).replace(tzinfo=None) + + +def _create_token(client, headers, name='test token', expiresat=None): + body = {'name': name} + if expiresat is not None: + body['expiresat'] = expiresat + response = client.post('/api/apitokens', json=body, headers=headers) + return response + + +def _pat_headers(secret): + return {'Authorization': f'Bearer {secret}'} + + +def test_create_returns_secret_once_and_stores_hash(client, db, auth_headers): + response = _create_token(client, auth_headers, name='import runner') + assert response.status_code == 201 + data = response.get_json()['data'] + + secret = data['secret'] + assert secret.startswith('shopdb_pat_') + assert 'warning' in data + # The stored row must not carry the raw secret; only its hash. + token = ApiToken.query.filter_by(tokenid=data['tokenid']).first() + assert token is not None + assert token.tokenhash == ApiToken.hash_secret(secret) + assert secret not in (token.tokenhash, token.tokenprefix or '') + assert token.tokenprefix and token.tokenprefix in secret + + +def test_pat_authenticates_permission_write_as_owner(client, db, admin_user, + auth_headers): + """A PAT owned by an admin can create a vendor (admin-gated write).""" + secret = _create_token(client, auth_headers).get_json()['data']['secret'] + + response = client.post('/api/vendors', json={'vendor': 'PAT Vendor'}, + headers=_pat_headers(secret)) + assert response.status_code == 201 + assert Vendor.query.filter_by(vendor='PAT Vendor').first() is not None + + +def test_pat_403_when_owner_lacks_permission(client, db, member_user, + member_headers): + """A PAT owned by a role-less member is forbidden from an admin write.""" + secret = _create_token(client, member_headers).get_json()['data']['secret'] + + response = client.post('/api/vendors', json={'vendor': 'Nope'}, + headers=_pat_headers(secret)) + assert response.status_code == 403 + assert Vendor.query.filter_by(vendor='Nope').first() is None + + +def test_expired_token_rejected(client, db, admin_user, auth_headers): + secret = _create_token(client, auth_headers).get_json()['data']['secret'] + token = ApiToken.query.filter_by( + tokenhash=ApiToken.hash_secret(secret)).first() + token.expiresat = _naive_utcnow() - timedelta(days=1) + _db.session.commit() + + response = client.post('/api/vendors', json={'vendor': 'Expired'}, + headers=_pat_headers(secret)) + assert response.status_code == 401 + assert Vendor.query.filter_by(vendor='Expired').first() is None + + +def test_revoked_token_rejected(client, db, admin_user, auth_headers): + secret = _create_token(client, auth_headers).get_json()['data']['secret'] + token = ApiToken.query.filter_by( + tokenhash=ApiToken.hash_secret(secret)).first() + tokenid = token.tokenid + + revoke = client.delete(f'/api/apitokens/{tokenid}', headers=auth_headers) + assert revoke.status_code == 200 + + response = client.post('/api/vendors', json={'vendor': 'Revoked'}, + headers=_pat_headers(secret)) + assert response.status_code == 401 + assert Vendor.query.filter_by(vendor='Revoked').first() is None + + +def test_lastusedat_updates_on_use(client, db, admin_user, auth_headers): + secret = _create_token(client, auth_headers).get_json()['data']['secret'] + token = ApiToken.query.filter_by( + tokenhash=ApiToken.hash_secret(secret)).first() + assert token.lastusedat is None + + client.get('/api/apitokens', headers=_pat_headers(secret)) + _db.session.expire_all() + token = ApiToken.query.filter_by( + tokenhash=ApiToken.hash_secret(secret)).first() + assert token.lastusedat is not None + + +def test_member_cannot_revoke_other_users_token(client, db, admin_user, + auth_headers, member_headers): + """A role-less member cannot revoke a token owned by a different user.""" + secret = _create_token(client, auth_headers).get_json()['data']['secret'] + tokenid = ApiToken.query.filter_by( + tokenhash=ApiToken.hash_secret(secret)).first().tokenid + + response = client.delete(f'/api/apitokens/{tokenid}', headers=member_headers) + assert response.status_code == 403 + # Still active. + assert _db.session.get(ApiToken, tokenid).isactive is True + + +def test_member_can_manage_own_token(client, db, member_user, member_headers): + """By design any authed user manages their OWN tokens.""" + create = _create_token(client, member_headers, name='mine') + assert create.status_code == 201 + tokenid = create.get_json()['data']['tokenid'] + + revoke = client.delete(f'/api/apitokens/{tokenid}', headers=member_headers) + assert revoke.status_code == 200 + assert _db.session.get(ApiToken, tokenid).isactive is False + + +def test_admin_all_true_lists_everyone(client, db, admin_user, auth_headers, + member_user, member_headers): + _create_token(client, auth_headers, name='admin token') + _create_token(client, member_headers, name='member token') + + # Own-only (default) for admin: just the admin's token. + own = client.get('/api/apitokens', headers=auth_headers).get_json()['data'] + assert all(t['userid'] == admin_user.userid for t in own) + + # all=true: both users' tokens, with owner usernames. + everyone = client.get('/api/apitokens?all=true', + headers=auth_headers).get_json()['data'] + userids = {t['userid'] for t in everyone} + assert admin_user.userid in userids and member_user.userid in userids + assert any(t.get('username') for t in everyone) + + +def test_member_all_true_ignored(client, db, member_user, member_headers, + admin_user, auth_headers): + """A non-admin passing ?all=true still only sees their own tokens.""" + _create_token(client, auth_headers, name='admin token') + _create_token(client, member_headers, name='member token') + + result = client.get('/api/apitokens?all=true', + headers=member_headers).get_json()['data'] + assert all(t['userid'] == member_user.userid for t in result) + + +def test_import_mode_works_over_pat(client, db, admin_user, auth_headers): + """An admin PAT plus X-Import-Mode backdates createddate on a write.""" + secret = _create_token(client, auth_headers).get_json()['data']['secret'] + + headers = _pat_headers(secret) + headers['X-Import-Mode'] = 'true' + response = client.post( + '/api/vendors', + json={'vendor': 'Legacy Vendor', 'createddate': '2019-01-02 03:04:05'}, + headers=headers) + assert response.status_code == 201 + + vendor = Vendor.query.filter_by(vendor='Legacy Vendor').first() + assert vendor is not None + assert vendor.createddate == datetime(2019, 1, 2, 3, 4, 5) diff --git a/tests/test_core/test_authz.py b/tests/test_core/test_authz.py index f53500d..6d7788f 100644 --- a/tests/test_core/test_authz.py +++ b/tests/test_core/test_authz.py @@ -37,7 +37,14 @@ EXEMPT_BLUEPRINTS = {'auth', 'collector', 'setup'} # role-less member MAY edit their own record, so it does not fit the # 403-for-every-member contract this sweep asserts. The other-user 403 is # covered by test_member_cannot_update_other_user below. -EXEMPT_ENDPOINTS = {'knowledgebase.track_click', 'users.update_user'} +# apitokens.create_apitoken / update_apitoken / revoke_apitoken - personal +# API tokens. By design ANY authenticated user may create and manage their +# OWN tokens (own-resource logic, not a flat deny), so a role-less member +# gets 201/200 here, not the 403 this sweep asserts. The non-owner 403 is +# covered by test_apitokens.py (member cannot revoke another user's token). +EXEMPT_ENDPOINTS = {'knowledgebase.track_click', 'users.update_user', + 'apitokens.create_apitoken', 'apitokens.update_apitoken', + 'apitokens.revoke_apitoken'} @pytest.fixture(autouse=True) diff --git a/tests/test_core/test_pluginui.py b/tests/test_core/test_pluginui.py index a727cbe..b218601 100644 --- a/tests/test_core/test_pluginui.py +++ b/tests/test_core/test_pluginui.py @@ -133,7 +133,7 @@ def test_asset_presentation_aggregate_enabled_plugins(app, client, auth_headers, entry = next((e for e in entries if e.get('assettype') == 'measuring_tool'), None) assert entry is not None assert entry['plugin'] == 'measuringtools' - assert entry['route'] == '/measuringtools/{assetid}' + assert entry['route'] == '/measuringtools/by-asset/{assetid}' def test_asset_presentation_skip_disabled_plugin(app, client, auth_headers, monkeypatch): diff --git a/tests/test_plugins/test_measuringtools.py b/tests/test_plugins/test_measuringtools.py index 6367a3f..acc25b4 100644 --- a/tests/test_plugins/test_measuringtools.py +++ b/tests/test_plugins/test_measuringtools.py @@ -19,7 +19,8 @@ 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 +from plugins.measuringtools.models import ( + derive_status, DUESOON_WINDOW_DAYS, STATUS_COLORS) # ============================================================================= @@ -320,3 +321,126 @@ def test_calibration_report_shape(client, auth_headers): 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') + 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= 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}') + 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') + 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') + assert response.status_code == 200, response.get_json() + rows = response.get_json()['data'] + row = next(r for r in rows if r['assetid'] == assetid) + assert set(row) == {'assetid', 'calibrationstatus', 'statuscolor'} + assert row['calibrationstatus'] == 'overdue' + assert row['statuscolor'] == 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}'