Security hardening, multi-site portability, and a documentation and API-surface sweep
This commit is contained in:
@@ -206,12 +206,19 @@ def test_partmarker_kind_is_share_backed_and_not_renderable():
|
||||
assert kind.parse(b'anything') is None
|
||||
|
||||
|
||||
def test_partmarker_sharedir_is_under_the_sfld_backups_root():
|
||||
path = registry.getkind('partmarker').sharedir('lathe', '3204')
|
||||
assert path.startswith(registry.DEFAULTSHAREROOT)
|
||||
def test_partmarker_sharedir_is_under_the_configured_backups_root():
|
||||
root = r'\\fileserver\shopfloor\backups'
|
||||
path = registry.getkind('partmarker').sharedir('lathe', '3204', shareroot=root)
|
||||
assert path.startswith(root)
|
||||
assert path.endswith(r'lathe\3204\partmarker')
|
||||
|
||||
|
||||
def test_sharedir_is_empty_when_no_root_is_configured():
|
||||
"""No default root ships (ADR-015), so an unconfigured site gets nothing
|
||||
rather than a path composed from another site's file server."""
|
||||
assert registry.getkind('partmarker').sharedir('lathe', '3204') == ''
|
||||
|
||||
|
||||
def test_getkind_is_case_insensitive_and_returns_none_when_unknown():
|
||||
assert registry.getkind('NTLARS') is not None
|
||||
assert registry.getkind('nosuchkind') is None
|
||||
@@ -981,3 +988,21 @@ def test_the_panel_shows_verified_beside_captured(bk_app, bk_plugin):
|
||||
listpanel = next(p for p in panels if p.get('render') == 'list')
|
||||
labels = [m['label'] for m in listpanel['map']['meta']]
|
||||
assert 'Captured' in labels and 'Verified' in labels
|
||||
|
||||
|
||||
def test_retention_reads_the_key_the_settings_page_writes(app, db):
|
||||
"""The plugin helper namespaces to `plugin.backups.<key>`; the declared key
|
||||
is bare. Reading through the helper meant retention was configurable and
|
||||
inert - it always fell back to 0, which means keep everything."""
|
||||
from shopdb.core.models import Setting
|
||||
from plugins.backups.plugin import BackupsPlugin
|
||||
|
||||
declared = {entry['key'] for entry in BackupsPlugin().get_settings_defaults()}
|
||||
assert 'backups_retentioncount' in declared
|
||||
|
||||
Setting.set('backups_retentioncount', 5)
|
||||
db.session.commit()
|
||||
assert int(Setting.get('backups_retentioncount', 0)) == 5
|
||||
# The helper cannot see it, which is the whole defect: it looks under
|
||||
# plugin.backups.backups_retentioncount, which nothing writes.
|
||||
assert int(BackupsPlugin().get_setting('backups_retentioncount', 0) or 0) == 0
|
||||
|
||||
232
tests/test_plugins/test_collector_measuringtool_adoption.py
Normal file
232
tests/test_plugins/test_collector_measuringtool_adoption.py
Normal file
@@ -0,0 +1,232 @@
|
||||
"""The collector ADOPTS an existing instrument instead of minting a twin.
|
||||
|
||||
Minting derived the asset number from the HOSTNAME, so a permanent instrument
|
||||
inherited the identity of whichever PC drove it that week. And because the
|
||||
collector keyed idempotency on its OWN label, it could not see a tool it had not
|
||||
created - on prod that left 43 legacy MT-#### tools shadowed by minted
|
||||
<HOST>-CMM twins, three records deep for some CMMs.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from shopdb.core.models import Asset, AssetType
|
||||
from shopdb.core.models.relationship import RelationshipType, AssetRelationship
|
||||
from plugins.measuringtools.models import MeasuringTool
|
||||
|
||||
KEY = 'collector-adoption-key'
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def collector_key(app):
|
||||
old = app.config.get('COLLECTOR_API_KEY')
|
||||
app.config['COLLECTOR_API_KEY'] = KEY
|
||||
yield KEY
|
||||
app.config['COLLECTOR_API_KEY'] = old
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rig(db):
|
||||
"""Asset types, a controls relationship type, and nothing else."""
|
||||
for name in ('computer', 'measuring_tool'):
|
||||
if not AssetType.query.filter_by(assettype=name).first():
|
||||
db.session.add(AssetType(assettype=name))
|
||||
if not RelationshipType.query.filter_by(relationshiptype='controls').first():
|
||||
db.session.add(RelationshipType(relationshiptype='controls'))
|
||||
db.session.commit()
|
||||
|
||||
|
||||
def _tool(db, assetnumber):
|
||||
"""A measuring tool asset the collector did NOT create."""
|
||||
assettype = AssetType.query.filter_by(assettype='measuring_tool').first()
|
||||
asset = Asset(assetnumber=assetnumber, assettypeid=assettype.assettypeid)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
db.session.add(MeasuringTool(assetid=asset.assetid))
|
||||
db.session.commit()
|
||||
return asset
|
||||
|
||||
|
||||
def _report(client, key, hostname, **extra):
|
||||
payload = {'hostname': hostname, 'pctype': 'gea-shopfloor-cmm'}
|
||||
payload.update(extra)
|
||||
return client.post('/api/collector/computers', json=payload,
|
||||
headers={'X-API-Key': key})
|
||||
|
||||
|
||||
def _tools_controlled(hostname):
|
||||
pc = Asset.query.filter_by(assetnumber=hostname).first()
|
||||
controls = RelationshipType.query.filter_by(relationshiptype='controls').first()
|
||||
toolids = {r.assetid for r in MeasuringTool.query.with_entities(MeasuringTool.assetid)}
|
||||
rows = AssetRelationship.query.filter(
|
||||
AssetRelationship.sourceassetid == pc.assetid,
|
||||
AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
|
||||
AssetRelationship.isactive.is_(True)).all()
|
||||
return [Asset.query.get(r.targetassetid).assetnumber
|
||||
for r in rows if r.targetassetid in toolids]
|
||||
|
||||
|
||||
def test_a_named_instrument_is_adopted_not_minted(client, db, rig, collector_key):
|
||||
"""measuringtool-id.txt names the instrument; nothing new is created."""
|
||||
_tool(db, 'MT-6227')
|
||||
assert _report(client, collector_key, 'FB7FB1V3',
|
||||
measuringtoolid='MT-6227').status_code in (200, 201)
|
||||
|
||||
assert _tools_controlled('FB7FB1V3') == ['MT-6227']
|
||||
assert Asset.query.filter_by(assetnumber='FB7FB1V3-CMM').first() is None
|
||||
|
||||
|
||||
def test_a_tool_already_controlled_is_adopted(client, db, rig, collector_key):
|
||||
"""The legacy MT-#### case: linked by the import, invisible to the collector.
|
||||
|
||||
The link must pre-date the first report - that is the real sequence, and it
|
||||
is the whole point. Reporting first would mint the twin before adoption ever
|
||||
had a chance, which is exactly the bug.
|
||||
"""
|
||||
from plugins.computers.models import Computer
|
||||
tool = _tool(db, 'MT-6224')
|
||||
computertype = AssetType.query.filter_by(assettype='computer').first()
|
||||
pc = Asset(assetnumber='FJPX1GT3', assettypeid=computertype.assettypeid)
|
||||
db.session.add(pc)
|
||||
db.session.flush()
|
||||
db.session.add(Computer(assetid=pc.assetid, hostname='FJPX1GT3'))
|
||||
controls = RelationshipType.query.filter_by(relationshiptype='controls').first()
|
||||
db.session.add(AssetRelationship(
|
||||
sourceassetid=pc.assetid, targetassetid=tool.assetid,
|
||||
relationshiptypeid=controls.relationshiptypeid, label=None))
|
||||
db.session.commit()
|
||||
|
||||
assert _report(client, collector_key, 'FJPX1GT3').status_code in (200, 201)
|
||||
assert _tools_controlled('FJPX1GT3') == ['MT-6224']
|
||||
assert Asset.query.filter_by(assetnumber='FJPX1GT3-CMM').first() is None
|
||||
|
||||
|
||||
def test_the_reported_bay_id_is_adopted(client, db, rig, collector_key):
|
||||
"""The CMM case: cmmid.txt already reports CMM4, which is a real asset."""
|
||||
_tool(db, 'CMM4')
|
||||
assert _report(client, collector_key, 'FB7FB1V3',
|
||||
machinenumber='CMM4').status_code in (200, 201)
|
||||
|
||||
assert 'CMM4' in _tools_controlled('FB7FB1V3')
|
||||
assert Asset.query.filter_by(assetnumber='FB7FB1V3-CMM').first() is None
|
||||
|
||||
|
||||
def test_minting_still_happens_when_nothing_exists(client, db, rig, collector_key):
|
||||
"""A genuinely new instrument has no identity to adopt - last resort."""
|
||||
assert _report(client, collector_key, 'FNEWPC01').status_code in (200, 201)
|
||||
assert _tools_controlled('FNEWPC01') == ['FNEWPC01-CMM']
|
||||
|
||||
|
||||
def test_a_named_instrument_that_does_not_exist_warns_and_mints_nothing_wrong(
|
||||
client, db, rig, collector_key):
|
||||
"""A typo must not mint a phantom instrument nobody can account for."""
|
||||
resp = _report(client, collector_key, 'FTYPO001', measuringtoolid='MT-NOPE')
|
||||
assert resp.status_code in (200, 201)
|
||||
body = resp.get_json()['data']
|
||||
assert any('MT-NOPE' in w for w in body.get('warnings', []))
|
||||
|
||||
|
||||
def test_adoption_is_idempotent(client, db, rig, collector_key):
|
||||
_tool(db, 'MT-6331')
|
||||
for _ in range(3):
|
||||
_report(client, collector_key, 'F9LQSDB4', measuringtoolid='MT-6331')
|
||||
assert _tools_controlled('F9LQSDB4') == ['MT-6331']
|
||||
|
||||
|
||||
def test_a_named_instrument_supersedes_a_minted_twin(
|
||||
client, db, rig, collector_key):
|
||||
"""The migration case: the twin already exists when the file names the real
|
||||
one.
|
||||
|
||||
This is the state prod is actually in - 43 minted twins already linked - so
|
||||
adoption that only works on a clean PC fixes nothing. The twin's link has to
|
||||
be archived, or the PC reads as controlling two instruments.
|
||||
"""
|
||||
_tool(db, 'MT-6227')
|
||||
_report(client, collector_key, 'FB7FB1V3')
|
||||
assert _tools_controlled('FB7FB1V3') == ['FB7FB1V3-CMM']
|
||||
|
||||
_report(client, collector_key, 'FB7FB1V3', measuringtoolid='MT-6227')
|
||||
assert _tools_controlled('FB7FB1V3') == ['MT-6227']
|
||||
|
||||
|
||||
def test_a_named_id_that_is_not_a_measuring_tool_is_refused(
|
||||
client, db, rig, collector_key):
|
||||
"""A machine number in measuringtool-id.txt must not link the machine.
|
||||
|
||||
The name resolving is not enough: linked under the measuring-tool label it
|
||||
would read as an instrument everywhere downstream, on a row the machine sync
|
||||
also owns.
|
||||
"""
|
||||
machinetype = AssetType.query.filter_by(assettype='machine').first()
|
||||
if machinetype is None:
|
||||
machinetype = AssetType(assettype='machine')
|
||||
db.session.add(machinetype)
|
||||
db.session.flush()
|
||||
db.session.add(Asset(assetnumber='2335', assettypeid=machinetype.assettypeid))
|
||||
db.session.commit()
|
||||
|
||||
resp = _report(client, collector_key, 'FTYPO777', measuringtoolid='2335')
|
||||
assert resp.status_code in (200, 201)
|
||||
body = resp.get_json()['data']
|
||||
assert any('2335' in w for w in body.get('warnings', []))
|
||||
assert '2335' not in _all_controlled('FTYPO777')
|
||||
|
||||
|
||||
def test_a_cmm_reporting_its_bay_survives_a_second_cycle(
|
||||
client, db, rig, collector_key):
|
||||
"""The instrument IS the reported bay, so both syncs want the same triple.
|
||||
|
||||
Adoption used to relabel the machine sync's row. The machine sync finds its
|
||||
rows BY that label, so cycle 2 built a second row for the same
|
||||
(pc, tool, controls) triple and MySQL rejected it on the unique constraint:
|
||||
200, then 500 on every report after. The machine link must survive as a
|
||||
machine link.
|
||||
"""
|
||||
_tool(db, 'CMM4')
|
||||
for _ in range(3):
|
||||
resp = _report(client, collector_key, 'FB7FB1V3', machinenumber='CMM4')
|
||||
assert resp.status_code in (200, 201)
|
||||
|
||||
assert _all_controlled('FB7FB1V3') == ['CMM4']
|
||||
assert Asset.query.filter_by(assetnumber='FB7FB1V3-CMM').first() is None
|
||||
|
||||
pc = Asset.query.filter_by(assetnumber='FB7FB1V3').first()
|
||||
tool = Asset.query.filter_by(assetnumber='CMM4').first()
|
||||
rows = AssetRelationship.query.filter_by(
|
||||
sourceassetid=pc.assetid, targetassetid=tool.assetid).all()
|
||||
assert len(rows) == 1
|
||||
assert rows[0].label == 'collector:machine'
|
||||
|
||||
|
||||
def test_a_reported_identifier_is_not_a_wildcard(client, db, rig, collector_key):
|
||||
"""`_` and `%` in a reported identifier must match themselves, nothing else.
|
||||
|
||||
These values come from a text file on a shopfloor PC. Resolved with ilike,
|
||||
'MT-600_' matched MT-6001 and a bare '%' matched whatever active asset came
|
||||
first - an identifier resolving to an asset it does not name, then linked to
|
||||
as though it did.
|
||||
"""
|
||||
_tool(db, 'MT-6001')
|
||||
resp = _report(client, collector_key, 'FWILD001', measuringtoolid='MT-600_')
|
||||
assert resp.status_code in (200, 201)
|
||||
assert 'MT-6001' not in _tools_controlled('FWILD001')
|
||||
|
||||
resp = _report(client, collector_key, 'FWILD002', measuringtoolid='%')
|
||||
assert resp.status_code in (200, 201)
|
||||
assert 'MT-6001' not in _tools_controlled('FWILD002')
|
||||
|
||||
|
||||
def _all_controlled(hostname):
|
||||
"""Every active controls target, tool or not - the refusal cases need to see
|
||||
a link the tool-only helper would filter out."""
|
||||
pc = Asset.query.filter_by(assetnumber=hostname).first()
|
||||
controls = RelationshipType.query.filter_by(relationshiptype='controls').first()
|
||||
rows = AssetRelationship.query.filter(
|
||||
AssetRelationship.sourceassetid == pc.assetid,
|
||||
AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
|
||||
AssetRelationship.isactive.is_(True)).all()
|
||||
targets = [r.targetassetid for r in rows]
|
||||
if not targets:
|
||||
return []
|
||||
return sorted(a.assetnumber for a in
|
||||
Asset.query.filter(Asset.assetid.in_(targets)).all())
|
||||
97
tests/test_plugins/test_collector_partmarker_links.py
Normal file
97
tests/test_plugins/test_collector_partmarker_links.py
Normal file
@@ -0,0 +1,97 @@
|
||||
"""A part-marker PC keeps reporting after the first cycle.
|
||||
|
||||
Both links a part-marker PC produces are rows in assetrelationships, which is
|
||||
unique on (source, target, type). Both were looked up by LABEL, so a row made by
|
||||
hand or by the legacy import was invisible and the insert violated the
|
||||
constraint; and the minted marker's asset number is derived from the PC, so it
|
||||
could already be taken. Either way the PC reported 200 on the cycle that minted
|
||||
the marker and 500 on every cycle after - the same shape as the CMM adoption bug
|
||||
in test_collector_measuringtool_adoption.py.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from shopdb.core.models import Asset, AssetType
|
||||
from shopdb.core.models.relationship import RelationshipType, AssetRelationship
|
||||
|
||||
KEY = 'partmarker-links-key'
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def collector_key(app):
|
||||
old = app.config.get('COLLECTOR_API_KEY')
|
||||
app.config['COLLECTOR_API_KEY'] = KEY
|
||||
yield KEY
|
||||
app.config['COLLECTOR_API_KEY'] = old
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def rig(db):
|
||||
for name in ('computer', 'machine'):
|
||||
if not AssetType.query.filter_by(assettype=name).first():
|
||||
db.session.add(AssetType(assettype=name))
|
||||
for name in ('controls', 'partof'):
|
||||
if not RelationshipType.query.filter_by(relationshiptype=name).first():
|
||||
db.session.add(RelationshipType(relationshiptype=name))
|
||||
db.session.commit()
|
||||
|
||||
|
||||
def _report(client, key, hostname, **extra):
|
||||
payload = {'hostname': hostname, 'pctype': 'gea-shopfloor-partmarker'}
|
||||
payload.update(extra)
|
||||
return client.post('/api/collector/computers', json=payload,
|
||||
headers={'X-API-Key': key})
|
||||
|
||||
|
||||
def _operation(db, assetnumber):
|
||||
machinetype = AssetType.query.filter_by(assettype='machine').first()
|
||||
asset = Asset(assetnumber=assetnumber, assettypeid=machinetype.assettypeid)
|
||||
db.session.add(asset)
|
||||
db.session.commit()
|
||||
return asset
|
||||
|
||||
|
||||
def test_repeat_cycles_do_not_500(client, db, rig, collector_key):
|
||||
_operation(db, '0613')
|
||||
codes = [_report(client, collector_key, 'FMARK001', machinenumber='0613').status_code
|
||||
for _ in range(3)]
|
||||
assert codes == [codes[0]] * 3, codes
|
||||
assert codes[0] in (200, 201), codes
|
||||
|
||||
|
||||
def test_a_partof_row_under_another_label_is_not_duplicated(
|
||||
client, db, rig, collector_key):
|
||||
"""A partof row made by hand or by the legacy import carries no label."""
|
||||
operation = _operation(db, '0615')
|
||||
# First cycle mints the marker and links it.
|
||||
assert _report(client, collector_key, 'FMARK002',
|
||||
machinenumber='0615').status_code in (200, 201)
|
||||
marker = Asset.query.filter_by(assetnumber='FMARK002-PARTMARKER').first()
|
||||
assert marker is not None
|
||||
partof = RelationshipType.query.filter_by(relationshiptype='partof').first()
|
||||
row = AssetRelationship.query.filter_by(
|
||||
sourceassetid=marker.assetid, targetassetid=operation.assetid,
|
||||
relationshiptypeid=partof.relationshiptypeid).first()
|
||||
assert row is not None
|
||||
# Somebody relabels it (or it came from the import unlabelled).
|
||||
row.label = None
|
||||
db.session.commit()
|
||||
|
||||
resp = _report(client, collector_key, 'FMARK002', machinenumber='0615')
|
||||
assert resp.status_code in (200, 201), resp.status_code
|
||||
rows = AssetRelationship.query.filter_by(
|
||||
sourceassetid=marker.assetid, targetassetid=operation.assetid,
|
||||
relationshiptypeid=partof.relationshiptypeid).all()
|
||||
assert len(rows) == 1
|
||||
|
||||
|
||||
def test_a_taken_marker_asset_number_does_not_500(client, db, rig, collector_key):
|
||||
"""The minted number is derived from the PC, so it can already exist."""
|
||||
machinetype = AssetType.query.filter_by(assettype='machine').first()
|
||||
db.session.add(Asset(assetnumber='FMARK003-PARTMARKER',
|
||||
assettypeid=machinetype.assettypeid))
|
||||
db.session.commit()
|
||||
_operation(db, '0617')
|
||||
|
||||
resp = _report(client, collector_key, 'FMARK003', machinenumber='0617')
|
||||
assert resp.status_code in (200, 201), resp.status_code
|
||||
@@ -125,7 +125,7 @@ def _set_allowlist(app, value):
|
||||
def test_ip_allowlist_allows_without_token(client, db, app):
|
||||
# An allowlisted caller reaches the manifest with NO token (vault trust).
|
||||
_seed_and_publish(app)
|
||||
_set_allowlist(app, '127.0.0.0/8, 10.134.48.0/23') # test client is 127.0.0.1
|
||||
_set_allowlist(app, '127.0.0.0/8, 192.0.2.0/24') # test client is 127.0.0.1
|
||||
resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm')
|
||||
assert resp.status_code == 200, resp.data
|
||||
assert b'Alpha' in resp.data
|
||||
@@ -152,9 +152,9 @@ def test_spoofed_forwarded_for_does_not_bypass_allowlist(client, db, app):
|
||||
# whose real IP (127.0.0.1) is NOT allowlisted must NOT gain token-less access
|
||||
# by forging X-Forwarded-For to an allowlisted address.
|
||||
_seed_and_publish(app)
|
||||
_set_allowlist(app, '10.134.48.0/23') # test client 127.0.0.1 is NOT in it
|
||||
_set_allowlist(app, '192.0.2.0/24') # test client 127.0.0.1 is NOT in it
|
||||
resp = client.get('/api/geenforce/manifest?pctype=gea-shopfloor-cmm',
|
||||
headers={'X-Forwarded-For': '10.134.48.10'})
|
||||
headers={'X-Forwarded-For': '192.0.2.10'})
|
||||
assert resp.status_code == 401, 'spoofed X-Forwarded-For bypassed the allowlist'
|
||||
|
||||
|
||||
|
||||
@@ -263,3 +263,42 @@ def test_reported_subtype_keeps_an_unknown_value_verbatim(db):
|
||||
stored = ManifestEnforcementReport.query.filter_by(
|
||||
hostname='WJKIOSKODD', iscurrent=True).one()
|
||||
assert stored.subtype == 'Lobbby'
|
||||
|
||||
|
||||
# -- what the gate compares against ------------------------------------------
|
||||
#
|
||||
# The gate used manifestversion as the library floor. That is the manifest's own
|
||||
# 'Version' field: for a share-imported manifest it happens to share the
|
||||
# library's numbering, but a code-authored scope sets it freely -
|
||||
# seed_display_scope writes '2.0' - so the fleet was compared against a number
|
||||
# that says nothing about library features, and the gate passed on the very
|
||||
# scope that most needed it.
|
||||
|
||||
def test_a_declared_minlibversion_is_what_the_fleet_is_judged_against(db):
|
||||
scope = _scope(manifestversion='2.0')
|
||||
scope.minlibversion = '2.6'
|
||||
db.session.flush()
|
||||
assert service.required_libversion(scope) == '2.6'
|
||||
|
||||
_report('KIOSK01', '2.4')
|
||||
with pytest.raises(service.LibVersionTooOldError):
|
||||
service.publish_scope('gea-shopfloor-cmm', 'runtime')
|
||||
|
||||
|
||||
def test_the_floor_is_derived_from_the_features_the_manifest_uses(db):
|
||||
"""A scope written before minlibversion existed is still gated on what it
|
||||
contains: a version-gated entry needs lib 2.6, whatever its own Version
|
||||
field claims."""
|
||||
from plugins.geenforce.models import ManifestEntry
|
||||
|
||||
scope = _scope(manifestversion='2.0')
|
||||
scope.entries.append(ManifestEntry(
|
||||
name='PC-DMIS 2023.2', entrytype='MSI', sortorder=0,
|
||||
cmmversion='2023'))
|
||||
db.session.flush()
|
||||
assert service.required_libversion(scope) == '2.6'
|
||||
|
||||
|
||||
def test_a_scope_using_nothing_special_falls_back_to_its_own_version(db):
|
||||
scope = _scope(manifestversion='2.4')
|
||||
assert service.required_libversion(scope) == '2.4'
|
||||
|
||||
108
tests/test_plugins/test_network_assetnumber_generation.py
Normal file
108
tests/test_plugins/test_network_assetnumber_generation.py
Normal file
@@ -0,0 +1,108 @@
|
||||
"""Asset numbers generated from the device type's prefix.
|
||||
|
||||
Every network device on this fleet already followed one convention applied by
|
||||
hand - AP-<name>, SW-<name>, SVR-<name>, IDF-<name>, 45 records, no exceptions.
|
||||
Generation codifies it so the same value is not typed twice, and must reproduce
|
||||
those records exactly rather than inventing a new scheme.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from shopdb.extensions import db
|
||||
from shopdb.core.models import AssetType
|
||||
from plugins.network.models import NetworkDeviceType
|
||||
|
||||
|
||||
def _devicetype(db, name, prefix):
|
||||
"""A device type, with the core asset type the create path needs present."""
|
||||
if not AssetType.query.filter_by(assettype='network_device').first():
|
||||
db.session.add(AssetType(assettype='network_device'))
|
||||
devicetype = NetworkDeviceType(networkdevicetype=name, prefix=prefix)
|
||||
db.session.add(devicetype)
|
||||
db.session.commit()
|
||||
return devicetype
|
||||
from plugins.network.services.assetnumbers import generate, sanitize
|
||||
|
||||
|
||||
@pytest.mark.parametrize('prefix,name,expected', [
|
||||
('AP', 'AIRwdMUSwestj02', 'AP-AIRwdMUSwestj02'),
|
||||
('SW', 'AIRsdMUSwestj01', 'SW-AIRsdMUSwestj01'),
|
||||
('SVR', 'AVEWP4546V01', 'SVR-AVEWP4546V01'),
|
||||
])
|
||||
def test_reproduces_the_existing_convention(prefix, name, expected):
|
||||
assert generate(prefix, name) == expected
|
||||
|
||||
|
||||
def test_a_type_with_no_prefix_gives_the_bare_name():
|
||||
"""Nullable on purpose - nobody invents a prefix for a type that wants none."""
|
||||
assert generate(None, 'Bare Name') == 'Bare-Name'
|
||||
assert generate('', 'Bare Name') == 'Bare-Name'
|
||||
|
||||
|
||||
def test_an_already_prefixed_name_is_not_stacked():
|
||||
"""IDF-03 under type IDF is IDF-03, never IDF-IDF-03. Real row on this data."""
|
||||
assert generate('IDF', 'IDF-03') == 'IDF-03'
|
||||
assert generate('idf', 'idf-03') == 'idf-03'
|
||||
|
||||
|
||||
def test_characters_that_break_a_business_key_are_stripped():
|
||||
"""assetnumber lands in URLs and in joins. The existing IDF-Telco-Demarc-#1
|
||||
row shows a '#' reaching a unique key, which is what this prevents."""
|
||||
assert generate('IDF', 'Telco Demarc #1') == 'IDF-Telco-Demarc-1'
|
||||
assert generate('APNL', 'Door 4 / West') == 'APNL-Door-4-West'
|
||||
assert sanitize(' spaced out ') == 'spaced-out'
|
||||
|
||||
|
||||
def test_nothing_to_build_from_returns_empty():
|
||||
"""So the caller demands an explicit value instead of inventing one."""
|
||||
assert generate('AP', '') == ''
|
||||
assert generate('AP', None) == ''
|
||||
|
||||
|
||||
def test_create_generates_when_assetnumber_is_blank(client, db, auth_headers):
|
||||
devicetype = _devicetype(db, 'Access Panel', 'APNL')
|
||||
|
||||
resp = client.post('/api/network', json={
|
||||
'name': 'Dock Door 4',
|
||||
'networkdevicetypeid': devicetype.networkdevicetypeid,
|
||||
}, headers=auth_headers)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
assert resp.get_json()['data']['assetnumber'] == 'APNL-Dock-Door-4'
|
||||
|
||||
|
||||
def test_an_explicit_assetnumber_always_wins(client, db, auth_headers):
|
||||
"""Adopt an external identifier where one exists - a vendor tag beats a
|
||||
generated value, which is why generation only fills a BLANK field."""
|
||||
devicetype = _devicetype(db, 'Access Panel 2', 'APNL')
|
||||
|
||||
resp = client.post('/api/network', json={
|
||||
'assetnumber': 'VENDORTAG-99',
|
||||
'name': 'Dock Door 5',
|
||||
'networkdevicetypeid': devicetype.networkdevicetypeid,
|
||||
}, headers=auth_headers)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
assert resp.get_json()['data']['assetnumber'] == 'VENDORTAG-99'
|
||||
|
||||
|
||||
def test_no_name_and_no_assetnumber_is_refused(client, db, auth_headers):
|
||||
devicetype = _devicetype(db, 'Access Panel 3', 'APNL')
|
||||
|
||||
resp = client.post('/api/network', json={
|
||||
'networkdevicetypeid': devicetype.networkdevicetypeid,
|
||||
}, headers=auth_headers)
|
||||
assert resp.status_code == 400
|
||||
assert 'assetnumber is required' in resp.get_json()['data']['error']['message']
|
||||
|
||||
|
||||
def test_prefix_round_trips_through_the_type_api(client, db, auth_headers):
|
||||
created = client.post('/api/network/types', json={
|
||||
'networkdevicetype': 'Door Controller', 'prefix': 'DOOR',
|
||||
}, headers=auth_headers)
|
||||
assert created.status_code == 201, created.get_json()
|
||||
assert created.get_json()['data']['prefix'] == 'DOOR'
|
||||
|
||||
typeid = created.get_json()['data']['networkdevicetypeid']
|
||||
updated = client.put(f'/api/network/types/{typeid}', json={'prefix': 'DR'},
|
||||
headers=auth_headers)
|
||||
assert updated.status_code == 200, updated.get_json()
|
||||
assert updated.get_json()['data']['prefix'] == 'DR'
|
||||
Reference in New Issue
Block a user