Add ADR-006 generic plugin collector contract

Implements the plugin collector contract from ADR-006 so any plugin can
accept idempotent inventory ingest, not just PCs.

- base.py: add apply_collector_payload hook (companion to get_collector_schema),
  raises NotImplementedError by default for plugins that declare a schema but
  do not implement the upsert.
- collector.py: generic POST /api/collector/<plugin> dispatch with per-plugin
  API key (COLLECTOR_API_KEY_<PLUGINNAME> with COLLECTOR_API_KEY fallback),
  schema-driven identity validation, idempotent upsert, ADR-006 response
  contract (status, action, assetid, identityvalue, warnings), audit log.
  JWT-protected GET /api/collector/_schemas lists registered schemas. Legacy
  /pc, /apps, /heartbeat, /bulk kept for back-compat.
- computers plugin: implements get_collector_schema (identityfield hostname)
  and apply_collector_payload (create-or-update Asset+Computer, serialnumber,
  loggedinuser, lastboottime, primary IP communication, installed apps).
- tests: 7 collector-contract tests (auth, 404, validation, create/idempotent
  update, per-plugin key precedence, JWT schema listing).

A single dynamic dispatch route is used instead of per-plugin blueprint
registration, avoiding Flask's register-blueprint-after-first-request error.

144 tests pass, naming/style check green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

plugins/computers/plugin.py

@@ -65,6 +65,104 @@ class ComputersPlugin(BasePlugin):
         """Initialize plugin with Flask app."""
         logger.info(f"Computers plugin initialized (v{self.meta.version})")
 
+    # -- ADR-006 collector contract -----------------------------------------
+
+    def get_collector_schema(self) -> Optional[Dict]:
+        """Schema for the PC collector payload (matched by hostname)."""
+        return {
+            'identityfield': 'hostname',
+            'fields': {
+                'hostname': {'type': 'string', 'required': True},
+                'serialnumber': {'type': 'string'},
+                'currentuser': {'type': 'string'},
+                'lastboottime': {'type': 'string', 'format': 'date-time'},
+                'ipaddress': {'type': 'string'},
+                'installedsoftware': {
+                    'type': 'array',
+                    'items': {'name': 'string', 'version': 'string'},
+                },
+            },
+        }
+
+    def apply_collector_payload(self, payload: Dict) -> Dict:
+        """Idempotent upsert of a PC from a collector payload (by hostname)."""
+        from datetime import datetime
+        from shopdb.core.models import (
+            Asset, AssetType, Application, Communication, CommunicationType,
+        )
+
+        warnings = []
+        hostname = (payload.get('hostname') or '').strip()
+        if not hostname:
+            raise ValueError('hostname is required')
+
+        comp = Computer.query.filter(Computer.hostname.ilike(hostname)).first()
+        if not comp:
+            comp = (Computer.query.join(Asset, Asset.assetid == Computer.assetid)
+                    .filter(Asset.assetnumber.ilike(hostname)).first())
+
+        action = 'updated'
+        if not comp:
+            atype = AssetType.query.filter_by(assettype='computer').first()
+            asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
+                          statusid=1)
+            db.session.add(asset)
+            db.session.flush()
+            comp = Computer(assetid=asset.assetid, hostname=hostname)
+            db.session.add(comp)
+            db.session.flush()
+            action = 'created'
+
+        comp.lastreporteddate = datetime.utcnow()
+        if payload.get('lastboottime'):
+            try:
+                comp.lastboottime = datetime.fromisoformat(
+                    payload['lastboottime'].replace('Z', '+00:00'))
+            except (ValueError, AttributeError):
+                warnings.append('lastboottime not parseable')
+        if payload.get('currentuser'):
+            comp.loggedinuser = payload['currentuser']
+        if payload.get('serialnumber') and comp.asset:
+            comp.asset.serialnumber = payload['serialnumber']
+
+        if payload.get('ipaddress'):
+            ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
+            primary = Communication.query.filter_by(
+                assetid=comp.assetid, isprimary=True).first()
+            if primary:
+                primary.ipaddress = payload['ipaddress']
+            elif ip_comtype:
+                db.session.add(Communication(
+                    assetid=comp.assetid, comtypeid=ip_comtype.comtypeid,
+                    ipaddress=payload['ipaddress'], isprimary=True))
+
+        for app_data in payload.get('installedsoftware', []) or []:
+            name = app_data.get('name')
+            if not name:
+                continue
+            app = Application.query.filter(Application.appname.ilike(name)).first()
+            if not app:
+                warnings.append(f'unknown application: {name}')
+                continue
+            installed = ComputerInstalledApp.query.filter_by(
+                computerid=comp.computerid, appid=app.appid).first()
+            version = app_data.get('version')
+            if installed:
+                installed.installedversion = version
+                installed.isactive = True
+            else:
+                db.session.add(ComputerInstalledApp(
+                    computerid=comp.computerid, appid=app.appid,
+                    installedversion=version))
+
+        db.session.commit()
+        return {
+            'action': action,
+            'assetid': comp.assetid,
+            'identityvalue': hostname,
+            'warnings': warnings,
+        }
+
     def on_install(self, app: Flask) -> None:
         """Called when plugin is installed."""
         with app.app_context():

shopdb/core/api/collector.py

@@ -58,6 +58,119 @@ def _parse_boot(value):
         return None
 
 
+def _plugin_api_key(pluginname):
+    """Per-plugin collector key, falling back to the shared key (ADR-006).
+
+    Looks up COLLECTOR_API_KEY_<PLUGINNAME> (uppercased) first so each
+    collector can carry its own credential, then COLLECTOR_API_KEY.
+    """
+    per_plugin = current_app.config.get(
+        f'COLLECTOR_API_KEY_{pluginname.upper()}')
+    return per_plugin or current_app.config.get('COLLECTOR_API_KEY')
+
+
+def _collector_plugins():
+    """Map of pluginname -> (plugin, schema) for plugins accepting collector input."""
+    pm = current_app.extensions.get('plugin_manager')
+    result = {}
+    if not pm:
+        return result
+    for name, plugin in pm.get_all_plugins().items():
+        if not pm.registry.is_enabled(name):
+            continue
+        try:
+            schema = plugin.get_collector_schema()
+        except Exception:
+            schema = None
+        if schema:
+            result[name] = (plugin, schema)
+    return result
+
+
+@collector_bp.route('/<pluginname>', methods=['POST'])
+def generic_collect(pluginname):
+    """Generic collector ingest for any plugin (ADR-006).
+
+    Per-plugin API key auth, schema-driven validation of the identity field,
+    idempotent upsert via the plugin's apply_collector_payload. Returns the
+    ADR-006 response contract: status, action, assetid, identityvalue, warnings.
+    """
+    from shopdb.core.models import AuditLog
+
+    plugins = _collector_plugins()
+    if pluginname not in plugins:
+        return error_response(
+            ErrorCodes.NOT_FOUND,
+            f'No collector registered for plugin {pluginname}',
+            http_code=404)
+
+    plugin, schema = plugins[pluginname]
+
+    expected_key = _plugin_api_key(pluginname)
+    if not expected_key:
+        return error_response(ErrorCodes.INTERNAL_ERROR,
+                              'Collector API key not configured', http_code=500)
+    api_key = request.headers.get('X-API-Key') or request.args.get('api_key')
+    if api_key != expected_key:
+        return error_response(ErrorCodes.UNAUTHORIZED, 'Invalid API key',
+                              http_code=401)
+
+    payload = request.get_json(silent=True)
+    if not payload or not isinstance(payload, dict):
+        return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
+
+    identityfield = schema.get('identityfield')
+    identityvalue = (payload.get(identityfield) or '').strip() if identityfield else ''
+    if identityfield and not identityvalue:
+        return error_response(ErrorCodes.VALIDATION_ERROR,
+                              f'{identityfield} is required')
+
+    try:
+        outcome = plugin.apply_collector_payload(payload)
+    except NotImplementedError:
+        return error_response(
+            ErrorCodes.INTERNAL_ERROR,
+            f'Plugin {pluginname} does not implement apply_collector_payload',
+            http_code=500)
+    except ValueError as exc:
+        db.session.rollback()
+        return error_response(ErrorCodes.VALIDATION_ERROR, str(exc))
+    except Exception as exc:
+        db.session.rollback()
+        current_app.logger.exception('Collector upsert failed for %s', pluginname)
+        return error_response(ErrorCodes.INTERNAL_ERROR, str(exc), http_code=500)
+
+    action = outcome.get('action', 'noop')
+    AuditLog.log(
+        action if action in ('created', 'updated') else 'updated',
+        'Collector',
+        entityid=outcome.get('assetid'),
+        entityname=outcome.get('identityvalue', identityvalue),
+        details=f'collector:{pluginname} action={action}',
+    )
+    db.session.commit()
+
+    return success_response({
+        'status': 'ok',
+        'action': action,
+        'assetid': outcome.get('assetid'),
+        'identityvalue': outcome.get('identityvalue', identityvalue),
+        'warnings': outcome.get('warnings', []),
+    }, message=f'{pluginname} collector {action}')
+
+
+@collector_bp.route('/_schemas', methods=['GET'])
+def collector_schemas():
+    """List collector schemas for all enabled plugins (JWT-protected)."""
+    from flask_jwt_extended import verify_jwt_in_request
+    verify_jwt_in_request()
+
+    schemas = {
+        name: schema for name, (plugin, schema) in _collector_plugins().items()
+    }
+    return success_response({'schemas': schemas})
+
+
 @collector_bp.route('/pc', methods=['POST'])
 @require_api_key
 def update_pc_info():
@@ -257,6 +370,8 @@ def collector_status():
         'status': 'ok',
         'timestamp': datetime.utcnow().isoformat(),
         'endpoints': [
+            'POST /api/collector/<plugin>',
+            'GET /api/collector/_schemas',
             'POST /api/collector/pc',
             'POST /api/collector/apps',
             'POST /api/collector/heartbeat',

shopdb/plugins/base.py

@@ -120,6 +120,21 @@ class BasePlugin(ABC):
         """
         return None
 
+    def apply_collector_payload(self, payload: Dict) -> Dict:
+        """Idempotently upsert an asset from a validated collector payload.
+
+        Called by the generic /api/collector/<pluginname> endpoint after the
+        payload passed schema validation. Plugins that return a schema from
+        get_collector_schema must implement this. Return a dict with at least:
+        - 'action': 'created' | 'updated' | 'noop'
+        - 'assetid': the affected asset id (or None)
+        - 'warnings': list[str]
+        """
+        raise NotImplementedError(
+            f"{self.meta.name} declares a collector schema but does not "
+            f"implement apply_collector_payload"
+        )
+
     def on_install(self, app: Flask) -> None:
         """Called when plugin is installed via CLI."""
         pass

tests/test_core/test_collector_contract.py

@@ -0,0 +1,117 @@
+"""Tests for the generic plugin collector contract (ADR-006).
+
+Covers the auto-dispatched /api/collector/<plugin> endpoint, per-plugin API
+key auth, idempotent upsert by identity field, and the JWT-protected
+/api/collector/_schemas listing.
+"""
+
+import pytest
+
+KEY = 'testcollectorkey'
+
+
+@pytest.fixture
+def collector_key(app):
+    """Set the shared collector API key for the test."""
+    old = app.config.get('COLLECTOR_API_KEY')
+    app.config['COLLECTOR_API_KEY'] = KEY
+    yield KEY
+    app.config['COLLECTOR_API_KEY'] = old
+
+
+@pytest.fixture
+def computer_assettype(db):
+    """Seed the computer asset type needed for create-on-ingest."""
+    from shopdb.core.models import AssetType
+
+    at = AssetType(assettype='computer', pluginname='computers',
+                   tablename='computers', description='PCs')
+    db.session.add(at)
+    db.session.commit()
+    return at
+
+
+def test_schemas_requires_jwt(client, db):
+    """The schema listing rejects unauthenticated callers."""
+    response = client.get('/api/collector/_schemas')
+    assert response.status_code == 401
+
+
+def test_schemas_lists_computers(client, db, auth_headers):
+    """Computers plugin exposes a collector schema keyed by hostname."""
+    response = client.get('/api/collector/_schemas', headers=auth_headers)
+    assert response.status_code == 200
+    schemas = response.get_json()['data']['schemas']
+    assert 'computers' in schemas
+    assert schemas['computers']['identityfield'] == 'hostname'
+
+
+def test_missing_key_rejected(client, db, collector_key, computer_assettype):
+    """No API key -> 401."""
+    response = client.post('/api/collector/computers',
+                           json={'hostname': 'WJPC001'})
+    assert response.status_code == 401
+
+
+def test_unknown_plugin_404(client, db, collector_key):
+    """A plugin with no collector schema returns 404."""
+    response = client.post('/api/collector/nosuchplugin',
+                           json={'hostname': 'x'},
+                           headers={'X-API-Key': KEY})
+    assert response.status_code == 404
+
+
+def test_missing_identity_rejected(client, db, collector_key, computer_assettype):
+    """Missing the identity field is a validation error."""
+    response = client.post('/api/collector/computers',
+                           json={'currentuser': 'someone'},
+                           headers={'X-API-Key': KEY})
+    assert response.status_code == 400
+
+
+def test_create_then_idempotent_update(client, db, collector_key,
+                                       computer_assettype):
+    """First post creates; second post with same hostname updates (no dup)."""
+    from plugins.computers.models import Computer
+
+    payload = {'hostname': 'WJPC100', 'currentuser': 'alice',
+               'serialnumber': 'SN-100'}
+
+    first = client.post('/api/collector/computers', json=payload,
+                        headers={'X-API-Key': KEY})
+    assert first.status_code == 200, first.get_json()
+    data = first.get_json()['data']
+    assert data['action'] == 'created'
+    assert data['assetid'] is not None
+    assert data['identityvalue'] == 'WJPC100'
+
+    payload['currentuser'] = 'bob'
+    second = client.post('/api/collector/computers', json=payload,
+                         headers={'X-API-Key': KEY})
+    assert second.status_code == 200
+    assert second.get_json()['data']['action'] == 'updated'
+
+    with client.application.app_context():
+        comps = Computer.query.filter(Computer.hostname.ilike('WJPC100')).all()
+        assert len(comps) == 1
+        assert comps[0].loggedinuser == 'bob'
+
+
+def test_per_plugin_key_overrides_shared(client, db, app, computer_assettype):
+    """COLLECTOR_API_KEY_<PLUGIN> takes precedence over the shared key."""
+    app.config['COLLECTOR_API_KEY'] = 'sharedkey'
+    app.config['COLLECTOR_API_KEY_COMPUTERS'] = 'computerskey'
+    try:
+        # Shared key now rejected for this plugin.
+        rejected = client.post('/api/collector/computers',
+                               json={'hostname': 'WJPC200'},
+                               headers={'X-API-Key': 'sharedkey'})
+        assert rejected.status_code == 401
+
+        accepted = client.post('/api/collector/computers',
+                               json={'hostname': 'WJPC200'},
+                               headers={'X-API-Key': 'computerskey'})
+        assert accepted.status_code == 200
+    finally:
+        app.config.pop('COLLECTOR_API_KEY_COMPUTERS', None)
+        app.config['COLLECTOR_API_KEY'] = None