5fa5160420
Addresses findings from a 6-lens review against the project skills (defining-asset-contract, enforcing-plugin-contract, hardening-flask-config, integrating-plugin-hooks, pinning-flask-behavior, simplifying-python). Security (hardening-flask-config): - Load per-plugin COLLECTOR_API_KEY_<PLUGIN> from env in create_app. from_object only copies class attributes, so per-plugin keys (ADR-006) were dead in real deploys and silently fell back to the shared key. - EMPLOYEE_DB_USER/PASSWORD no longer default to root/rootpassword (no safe default for a secret; unset fails loud). Documented in .env.example + DEPLOY.md. - COLLECTOR_API_KEY + per-plugin + EMPLOYEE_DB_* added to .env.example/DEPLOY.md. Hook isolation (integrating-plugin-hooks): - collector _collector_plugins and dashboard get_navigation now re-raise in dev/test and log+isolate in prod, instead of silently swallowing a broken plugin hook. Plugin loader (enforcing-plugin-contract): - enable_plugin/install_plugin read dependencies+version from the manifest instead of instantiating the plugin class. - _register_plugin_components rejects a second plugin claiming an already-used api_prefix (reset per app in init_app). Tests (pinning-flask-behavior): - test_identifiers.py: gauge/maintenance round-trip on computer/printer/network create+update; per-type seed yields the 12 identifier keys. - contract tests for apply_collector_payload presence + schema-declarers-implement. - security tests for per-plugin key env loading + no employee-db password default. Docs/contract sync (defining-asset-contract): - PLUGIN-HOOKS.md documents apply_collector_payload; stale 0.2.0 -> 0.3.0. - ADR-006 documents apply_collector_payload + single-dispatch rationale. - ADR-001 enumerates the expanded shopdb.api import surface. Simplify (simplifying-python): - De-duplicate the 21-entry settings defaults: shared build_default_settings() used by both the /settings/seed route and the CLI (were drifting copies). - Remove dead AssetStatus import + redundant AssetType local import in computers plugin; comment the statusid=1 collector default. 153 tests pass (was 145), naming/style green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
307 lines
11 KiB
Python
307 lines
11 KiB
Python
"""Computers plugin main class."""
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import List, Dict, Optional, Type
|
|
|
|
from flask import Flask, Blueprint
|
|
import click
|
|
|
|
from shopdb.plugins.base import BasePlugin, PluginMeta
|
|
from shopdb.api import db, AssetType
|
|
|
|
from .models import Computer, ComputerType, ComputerInstalledApp
|
|
from .api import computers_bp
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ComputersPlugin(BasePlugin):
|
|
"""
|
|
Computers plugin - manages PC, server, and workstation assets.
|
|
|
|
Computers include shopfloor PCs, engineer workstations, servers, etc.
|
|
Uses the new Asset architecture with Computer extension table.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._manifest = self._load_manifest()
|
|
|
|
def _load_manifest(self) -> Dict:
|
|
"""Load plugin manifest from JSON file."""
|
|
manifestpath = Path(__file__).parent / 'manifest.json'
|
|
if manifestpath.exists():
|
|
with open(manifestpath, 'r') as f:
|
|
return json.load(f)
|
|
return {}
|
|
|
|
@property
|
|
def meta(self) -> PluginMeta:
|
|
"""Return plugin metadata."""
|
|
return PluginMeta(
|
|
name=self._manifest.get('name', 'computers'),
|
|
version=self._manifest.get('version', '1.0.0'),
|
|
description=self._manifest.get(
|
|
'description',
|
|
'Computer management for PCs, servers, and workstations'
|
|
),
|
|
author=self._manifest.get('author', 'ShopDB Team'),
|
|
dependencies=self._manifest.get('dependencies', []),
|
|
core_version=self._manifest.get('core_version', '>=1.0.0'),
|
|
api_prefix=self._manifest.get('api_prefix', '/api/computers'),
|
|
)
|
|
|
|
def get_blueprint(self) -> Optional[Blueprint]:
|
|
"""Return Flask Blueprint with API routes."""
|
|
return computers_bp
|
|
|
|
def get_models(self) -> List[Type]:
|
|
"""Return list of SQLAlchemy model classes."""
|
|
return [Computer, ComputerType, ComputerInstalledApp]
|
|
|
|
def init_app(self, app: Flask, db_instance) -> None:
|
|
"""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.api import Asset, 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()
|
|
# statusid=1 is the first seeded asset status ("In Use"); a
|
|
# collector-discovered PC is by definition in use.
|
|
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():
|
|
self._ensure_asset_type()
|
|
self._ensure_computer_types()
|
|
logger.info("Computers plugin installed")
|
|
|
|
def _ensure_asset_type(self) -> None:
|
|
"""Ensure computer asset type exists."""
|
|
existing = AssetType.query.filter_by(assettype='computer').first()
|
|
if not existing:
|
|
at = AssetType(
|
|
assettype='computer',
|
|
pluginname='computers',
|
|
tablename='computers',
|
|
description='PCs, servers, and workstations',
|
|
icon='desktop'
|
|
)
|
|
db.session.add(at)
|
|
logger.debug("Created asset type: computer")
|
|
db.session.commit()
|
|
|
|
def _ensure_computer_types(self) -> None:
|
|
"""Ensure basic computer types exist."""
|
|
computer_types = [
|
|
('Shopfloor PC', 'PC located on the shop floor for machine operation', 'desktop'),
|
|
('Engineer Workstation', 'Engineering workstation for CAD/CAM work', 'laptop'),
|
|
('CMM PC', 'PC dedicated to CMM operation', 'desktop'),
|
|
('Server', 'Server system', 'server'),
|
|
('Kiosk', 'Kiosk or info display PC', 'tv'),
|
|
('Laptop', 'Laptop computer', 'laptop'),
|
|
('Virtual Machine', 'Virtual machine', 'cloud'),
|
|
('Other', 'Other computer type', 'desktop'),
|
|
]
|
|
|
|
for name, description, icon in computer_types:
|
|
existing = ComputerType.query.filter_by(computertype=name).first()
|
|
if not existing:
|
|
ct = ComputerType(
|
|
computertype=name,
|
|
description=description,
|
|
icon=icon
|
|
)
|
|
db.session.add(ct)
|
|
logger.debug(f"Created computer type: {name}")
|
|
|
|
db.session.commit()
|
|
|
|
def on_uninstall(self, app: Flask) -> None:
|
|
"""Called when plugin is uninstalled."""
|
|
logger.info("Computers plugin uninstalled")
|
|
|
|
def get_cli_commands(self) -> List:
|
|
"""Return CLI commands for this plugin."""
|
|
|
|
@click.group('computers')
|
|
def computerscli():
|
|
"""Computers plugin commands."""
|
|
pass
|
|
|
|
@computerscli.command('list-types')
|
|
def list_types():
|
|
"""List all computer types."""
|
|
from flask import current_app
|
|
|
|
with current_app.app_context():
|
|
types = ComputerType.query.filter_by(isactive=True).all()
|
|
if not types:
|
|
click.echo('No computer types found.')
|
|
return
|
|
|
|
click.echo('Computer Types:')
|
|
for t in types:
|
|
click.echo(f" [{t.computertypeid}] {t.computertype}")
|
|
|
|
@computerscli.command('stats')
|
|
def stats():
|
|
"""Show computer statistics."""
|
|
from flask import current_app
|
|
from shopdb.api import Asset
|
|
|
|
with current_app.app_context():
|
|
total = db.session.query(Computer).join(Asset).filter(
|
|
Asset.isactive == True
|
|
).count()
|
|
|
|
click.echo(f"Total active computers: {total}")
|
|
|
|
# Shopfloor count
|
|
shopfloor = db.session.query(Computer).join(Asset).filter(
|
|
Asset.isactive == True,
|
|
Computer.isshopfloor == True
|
|
).count()
|
|
|
|
click.echo(f" Shopfloor PCs: {shopfloor}")
|
|
click.echo(f" Other: {total - shopfloor}")
|
|
|
|
@computerscli.command('find')
|
|
@click.argument('hostname')
|
|
def find_by_hostname(hostname):
|
|
"""Find a computer by hostname."""
|
|
from flask import current_app
|
|
|
|
with current_app.app_context():
|
|
comp = Computer.query.filter(
|
|
Computer.hostname.ilike(f'%{hostname}%')
|
|
).first()
|
|
|
|
if not comp:
|
|
click.echo(f'No computer found matching hostname: {hostname}')
|
|
return
|
|
|
|
click.echo(f'Found: {comp.hostname}')
|
|
click.echo(f' Asset: {comp.asset.assetnumber}')
|
|
click.echo(f' Type: {comp.computertype.computertype if comp.computertype else "N/A"}')
|
|
click.echo(f' OS: {comp.operatingsystem.osname if comp.operatingsystem else "N/A"}')
|
|
click.echo(f' Logged in: {comp.loggedinuser or "N/A"}')
|
|
|
|
return [computerscli]
|
|
|
|
def get_dashboard_widgets(self) -> List[Dict]:
|
|
"""Return dashboard widget definitions."""
|
|
return [
|
|
{
|
|
'name': 'Computer Status',
|
|
'component': 'ComputerStatusWidget',
|
|
'endpoint': '/api/computers/dashboard/summary',
|
|
'size': 'medium',
|
|
'position': 6,
|
|
},
|
|
]
|
|
|
|
def get_navigation_items(self) -> List[Dict]:
|
|
"""Return navigation menu items."""
|
|
return [
|
|
{
|
|
'name': 'PCs',
|
|
'icon': 'desktop',
|
|
'route': '/pcs',
|
|
'position': 15,
|
|
},
|
|
]
|