Cam Proudlock
a9d0dfff4a
Flask + Vue rewrite of the classic-ASP shopdb, built as a plugin framework so sister GE Aerospace sites can adopt it. Product version 0.7.0-pre-0.8.0, plugin contract 0.11.0, 966 backend tests. Key entry points: - docs/INSTALL-WINDOWS-IIS.md - tested Windows + IIS install runbook (own-site and subpath deployment methods) - docs/PILOT-DEPLOY.md - production pilot: stand up + import + verify - docs/IMPORT-API.md - schema-agnostic legacy data import contract - scripts/site_imports/wjf/ - West Jefferson reference import loader - docs/adr/ - twelve accepted architecture decision records
271 lines
9.4 KiB
Python
271 lines
9.4 KiB
Python
"""Tests for the public shopdb.api namespace exposed to plugins.
|
|
|
|
Pins audit_log, resolve_asset_position, and the BasePlugin
|
|
get_setting/set_setting helpers as part of the contract surface.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from shopdb.api import audit_log, resolve_asset_position
|
|
from shopdb.core.models import AuditLog, Setting
|
|
|
|
|
|
def test_audit_log_creates_row(db, client):
|
|
"""audit_log writes an AuditLog row with the standard fields."""
|
|
with client.application.test_request_context('/'):
|
|
entry = audit_log(
|
|
action='created',
|
|
entitytype='Computer',
|
|
entityid=42,
|
|
entityname='PC-1234',
|
|
changes={'before': {}, 'after': {'hostname': 'PC-1234'}},
|
|
)
|
|
|
|
assert entry is not None
|
|
assert entry.action == 'created'
|
|
assert entry.entitytype == 'Computer'
|
|
assert entry.entityid == 42
|
|
assert entry.entityname == 'PC-1234'
|
|
|
|
saved = AuditLog.query.filter_by(entityid=42, entitytype='Computer').first()
|
|
assert saved is not None
|
|
|
|
|
|
def test_resolve_asset_position_returns_none_when_no_data():
|
|
"""An asset with no coords and no location returns None."""
|
|
class FakeAsset:
|
|
mapx = None
|
|
mapy = None
|
|
location = None
|
|
|
|
assert resolve_asset_position(FakeAsset()) is None
|
|
|
|
|
|
def test_resolve_asset_position_uses_self_when_set():
|
|
"""Asset-specific coords win over everything else."""
|
|
class FakeLocation:
|
|
mapx = 10
|
|
mapy = 20
|
|
|
|
class FakeAsset:
|
|
mapx = 100
|
|
mapy = 200
|
|
location = FakeLocation()
|
|
|
|
result = resolve_asset_position(FakeAsset())
|
|
assert result == {'mapx': 100, 'mapy': 200, 'positionsource': 'self'}
|
|
|
|
|
|
def test_resolve_asset_position_falls_back_to_location():
|
|
"""When asset has no coords, falls back to location coords."""
|
|
class FakeLocation:
|
|
mapx = 50
|
|
mapy = 75
|
|
|
|
class FakeAsset:
|
|
mapx = None
|
|
mapy = None
|
|
location = FakeLocation()
|
|
|
|
result = resolve_asset_position(FakeAsset())
|
|
assert result == {'mapx': 50, 'mapy': 75, 'positionsource': 'location'}
|
|
|
|
|
|
def test_resolve_asset_position_handles_asset_without_mapx_attr():
|
|
"""Assets that don't yet have mapx/mapy columns degrade gracefully."""
|
|
class FakeLocation:
|
|
mapx = 1
|
|
mapy = 2
|
|
|
|
class FakeAsset:
|
|
location = FakeLocation()
|
|
|
|
result = resolve_asset_position(FakeAsset())
|
|
assert result == {'mapx': 1, 'mapy': 2, 'positionsource': 'location'}
|
|
|
|
|
|
# --- Relationship-walk path (priority 2 in the chain) ----------------------
|
|
|
|
def _make_rel(rtype_name, neighbor, inheritsposition=True, isactive=True):
|
|
"""Build a fake AssetRelationship-shaped object pointing at neighbor.
|
|
`neighbor` is wired as both source and target so the walk helper finds
|
|
it regardless of direction; tests pick which list to attach it to."""
|
|
class FakeRelType:
|
|
relationshiptype = rtype_name
|
|
|
|
class FakeRel:
|
|
pass
|
|
|
|
r = FakeRel()
|
|
r.relationshiptype = FakeRelType()
|
|
r.inheritsposition = inheritsposition
|
|
r.isactive = isactive
|
|
r.targetasset = neighbor
|
|
r.sourceasset = neighbor
|
|
return r
|
|
|
|
|
|
def _make_asset(assetid, mapx=None, mapy=None, outgoing=None, incoming=None, location=None):
|
|
class FakeAsset:
|
|
pass
|
|
|
|
a = FakeAsset()
|
|
a.assetid = assetid
|
|
a.mapx = mapx
|
|
a.mapy = mapy
|
|
a.outgoing_relationships = outgoing or []
|
|
a.incoming_relationships = incoming or []
|
|
a.location = location
|
|
return a
|
|
|
|
|
|
def test_resolve_asset_position_walks_partof_edge():
|
|
"""Priority 2: inheritsposition=true partof edge resolves from neighbor."""
|
|
parent = _make_asset(assetid=2, mapx=300, mapy=400)
|
|
rel = _make_rel('partof', parent)
|
|
child = _make_asset(assetid=1, outgoing=[rel])
|
|
|
|
result = resolve_asset_position(child)
|
|
assert result == {'mapx': 300, 'mapy': 400, 'positionsource': 'related'}
|
|
|
|
|
|
def test_resolve_asset_position_walks_controls_after_partof():
|
|
"""Priority 2 ordering: partof beats controls when both have coords."""
|
|
partof_neighbor = _make_asset(assetid=10, mapx=11, mapy=12)
|
|
controls_neighbor = _make_asset(assetid=20, mapx=99, mapy=99)
|
|
rel_partof = _make_rel('partof', partof_neighbor)
|
|
rel_controls = _make_rel('controls', controls_neighbor)
|
|
asset = _make_asset(assetid=1, outgoing=[rel_controls, rel_partof])
|
|
|
|
result = resolve_asset_position(asset)
|
|
assert result == {'mapx': 11, 'mapy': 12, 'positionsource': 'related'}
|
|
|
|
|
|
def test_resolve_asset_position_skips_non_inheritable_type():
|
|
"""connectedto edges are never walked even if inheritsposition is true."""
|
|
neighbor = _make_asset(assetid=2, mapx=5, mapy=6)
|
|
rel = _make_rel('connectedto', neighbor, inheritsposition=True)
|
|
asset = _make_asset(assetid=1, outgoing=[rel])
|
|
|
|
assert resolve_asset_position(asset) is None
|
|
|
|
|
|
def test_resolve_asset_position_skips_when_inheritsposition_false():
|
|
"""An edge with inheritsposition=false is not walked."""
|
|
neighbor = _make_asset(assetid=2, mapx=5, mapy=6)
|
|
rel = _make_rel('partof', neighbor, inheritsposition=False)
|
|
asset = _make_asset(assetid=1, outgoing=[rel])
|
|
|
|
assert resolve_asset_position(asset) is None
|
|
|
|
|
|
def test_resolve_asset_position_walks_recursively():
|
|
"""The walk recurses: child -> middle -> root, where only root has coords."""
|
|
root = _make_asset(assetid=3, mapx=1, mapy=2)
|
|
middle = _make_asset(assetid=2, outgoing=[_make_rel('partof', root)])
|
|
child = _make_asset(assetid=1, outgoing=[_make_rel('partof', middle)])
|
|
|
|
result = resolve_asset_position(child)
|
|
assert result == {'mapx': 1, 'mapy': 2, 'positionsource': 'related'}
|
|
|
|
|
|
def test_resolve_asset_position_breaks_cycles():
|
|
"""A cycle A<->B with no coords anywhere returns None without recursing
|
|
forever."""
|
|
a = _make_asset(assetid=1)
|
|
b = _make_asset(assetid=2)
|
|
a.outgoing_relationships = [_make_rel('partof', b)]
|
|
b.outgoing_relationships = [_make_rel('partof', a)]
|
|
|
|
assert resolve_asset_position(a) is None
|
|
|
|
|
|
def test_resolve_asset_position_depth_cap_is_three():
|
|
"""Past depth 3 the walk gives up. Build a chain of 5 nodes where only
|
|
the last has coords; expect None."""
|
|
coords_node = _make_asset(assetid=5, mapx=99, mapy=99)
|
|
n4 = _make_asset(assetid=4, outgoing=[_make_rel('partof', coords_node)])
|
|
n3 = _make_asset(assetid=3, outgoing=[_make_rel('partof', n4)])
|
|
n2 = _make_asset(assetid=2, outgoing=[_make_rel('partof', n3)])
|
|
root = _make_asset(assetid=1, outgoing=[_make_rel('partof', n2)])
|
|
|
|
assert resolve_asset_position(root) is None
|
|
|
|
|
|
def test_resolve_asset_position_self_beats_related():
|
|
"""Priority 1 beats priority 2: asset's own coords win even when a
|
|
related neighbor would also resolve."""
|
|
neighbor = _make_asset(assetid=2, mapx=99, mapy=99)
|
|
rel = _make_rel('partof', neighbor)
|
|
asset = _make_asset(assetid=1, mapx=1, mapy=2, outgoing=[rel])
|
|
|
|
result = resolve_asset_position(asset)
|
|
assert result == {'mapx': 1, 'mapy': 2, 'positionsource': 'self'}
|
|
|
|
|
|
def test_resolve_asset_position_related_beats_location():
|
|
"""Priority 2 beats priority 3: a related neighbor's coords win over
|
|
the asset's location coords."""
|
|
class FakeLocation:
|
|
mapx = 500
|
|
mapy = 600
|
|
|
|
neighbor = _make_asset(assetid=2, mapx=10, mapy=20)
|
|
rel = _make_rel('partof', neighbor)
|
|
asset = _make_asset(assetid=1, outgoing=[rel], location=FakeLocation())
|
|
|
|
result = resolve_asset_position(asset)
|
|
assert result == {'mapx': 10, 'mapy': 20, 'positionsource': 'related'}
|
|
|
|
|
|
def test_resolve_asset_position_inactive_edge_skipped():
|
|
"""Soft-deleted (isactive=false) relationships are not walked."""
|
|
neighbor = _make_asset(assetid=2, mapx=5, mapy=6)
|
|
rel = _make_rel('partof', neighbor, isactive=False)
|
|
asset = _make_asset(assetid=1, outgoing=[rel])
|
|
|
|
assert resolve_asset_position(asset) is None
|
|
|
|
|
|
def test_plugin_get_setting_returns_default_when_unset(db, app):
|
|
"""A plugin reading an unset setting gets the default."""
|
|
from shopdb.plugins import plugin_manager
|
|
with app.app_context():
|
|
printers = plugin_manager.get_plugin('printers')
|
|
if printers is None:
|
|
pytest.skip('printers plugin not loaded')
|
|
assert printers.get_setting('nonexistentkey', default='fallback') == 'fallback'
|
|
|
|
|
|
def test_plugin_set_and_get_setting_roundtrip(db, app):
|
|
"""A plugin can set and read its own setting; key is namespaced."""
|
|
from shopdb.plugins import plugin_manager
|
|
with app.app_context():
|
|
printers = plugin_manager.get_plugin('printers')
|
|
if printers is None:
|
|
pytest.skip('printers plugin not loaded')
|
|
|
|
printers.set_setting('zabbix_url', 'http://zabbix.example.com')
|
|
|
|
assert printers.get_setting('zabbix_url') == 'http://zabbix.example.com'
|
|
|
|
raw = Setting.query.filter_by(key='plugin.printers.zabbix_url').first()
|
|
assert raw is not None
|
|
assert raw.value == 'http://zabbix.example.com'
|
|
|
|
|
|
def test_plugin_setting_is_namespaced_per_plugin(db, app):
|
|
"""Two plugins using the same key do not collide."""
|
|
from shopdb.plugins import plugin_manager
|
|
with app.app_context():
|
|
printers = plugin_manager.get_plugin('printers')
|
|
computers = plugin_manager.get_plugin('computers')
|
|
if printers is None or computers is None:
|
|
pytest.skip('required plugins not loaded')
|
|
|
|
printers.set_setting('shared_key', 'printers_value')
|
|
computers.set_setting('shared_key', 'computers_value')
|
|
|
|
assert printers.get_setting('shared_key') == 'printers_value'
|
|
assert computers.get_setting('shared_key') == 'computers_value'
|