Wire relationship directionality and dualpath controls propagation
Symmetric relationship types (isdirectional flag, migration 7d19) show
one entry per peer on the relationships card - a Dualpath pair no
longer lists its partner twice - and directional types read naturally
instead of Outgoing/Incoming. Deleting a collapsed entry removes every
underlying direction row.
Propagation is now real (migration 7d20): relationship types declare
propagation-through pairs in relationshiptypepropagations (M:N,
replacing the never-consumed single column); creating a controls link
on either bay of a Dualpath pair auto-creates it on the partner,
mirrored across both endpoints because live data stores controls as
bay -> PC. flask relationships propagate backfills existing data (29
rows fanned out on the WJ dataset, idempotent).
This also completes the tree that commit 1d21bf0 accidentally split
(core/models/__init__ imported RelationshipTypePropagation ahead of the
file that defines it), returning CI to green.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
cproudlock
106
tests/test_core/test_relationship_directional.py
Normal file
106
tests/test_core/test_relationship_directional.py
Normal file
@@ -0,0 +1,106 @@
|
||||
"""Relationship-type directionality: CRUD carries isdirectional and the
|
||||
per-asset relationships endpoint tags each row with its type directionality.
|
||||
|
||||
The frontend uses isdirectional to collapse symmetric-type rows (Dualpath and
|
||||
other physical links) into one direction-blind card entry, and to render
|
||||
directional types (controls, partof...) with source->target phrasing. That
|
||||
dedupe is view logic verified live; here we lock the backend surface it needs.
|
||||
"""
|
||||
|
||||
from shopdb.extensions import db
|
||||
from shopdb.core.models import Asset, AssetType
|
||||
from shopdb.core.models.relationship import RelationshipType, AssetRelationship
|
||||
|
||||
|
||||
def _make_asset(assetnumber, typeid):
|
||||
asset = Asset(assetnumber=assetnumber, assettypeid=typeid)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
return asset
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Relationship type CRUD carries isdirectional
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_create_type_defaults_directional(client, db, auth_headers):
|
||||
resp = client.post('/api/assets/relationshiptypes',
|
||||
json={'relationshiptype': 'Backup For'},
|
||||
headers=auth_headers)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
assert resp.get_json()['data']['isdirectional'] is True
|
||||
|
||||
|
||||
def test_create_symmetric_type(client, db, auth_headers):
|
||||
resp = client.post('/api/assets/relationshiptypes',
|
||||
json={'relationshiptype': 'Dualpath',
|
||||
'isdirectional': False},
|
||||
headers=auth_headers)
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
assert resp.get_json()['data']['isdirectional'] is False
|
||||
|
||||
|
||||
def test_update_toggles_directional(client, db, auth_headers):
|
||||
typeid = client.post('/api/assets/relationshiptypes',
|
||||
json={'relationshiptype': 'USB'},
|
||||
headers=auth_headers).get_json()['data']['relationshiptypeid']
|
||||
resp = client.put(f'/api/assets/relationshiptypes/{typeid}',
|
||||
json={'isdirectional': False}, headers=auth_headers)
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json()['data']['isdirectional'] is False
|
||||
|
||||
|
||||
def test_list_includes_directional(client, db, auth_headers):
|
||||
client.post('/api/assets/relationshiptypes',
|
||||
json={'relationshiptype': 'WiFi', 'isdirectional': False},
|
||||
headers=auth_headers)
|
||||
resp = client.get('/api/assets/relationshiptypes')
|
||||
assert resp.status_code == 200
|
||||
row = next(t for t in resp.get_json()['data']
|
||||
if t['relationshiptype'] == 'WiFi')
|
||||
assert row['isdirectional'] is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Per-asset relationships endpoint tags each row with isdirectional
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_relationships_row_carries_isdirectional(client, db, auth_headers):
|
||||
atype = AssetType(assettype='machine')
|
||||
db.session.add(atype)
|
||||
db.session.flush()
|
||||
|
||||
symmetric = RelationshipType(relationshiptype='Dualpath', isdirectional=False)
|
||||
directional = RelationshipType(relationshiptype='controls', isdirectional=True)
|
||||
db.session.add_all([symmetric, directional])
|
||||
db.session.flush()
|
||||
|
||||
a = _make_asset('M-A', atype.assettypeid)
|
||||
b = _make_asset('M-B', atype.assettypeid)
|
||||
|
||||
# symmetric link stored in both directions (legacy shape)
|
||||
db.session.add(AssetRelationship(sourceassetid=a.assetid, targetassetid=b.assetid,
|
||||
relationshiptypeid=symmetric.relationshiptypeid))
|
||||
db.session.add(AssetRelationship(sourceassetid=b.assetid, targetassetid=a.assetid,
|
||||
relationshiptypeid=symmetric.relationshiptypeid))
|
||||
# directional link a controls b
|
||||
db.session.add(AssetRelationship(sourceassetid=a.assetid, targetassetid=b.assetid,
|
||||
relationshiptypeid=directional.relationshiptypeid))
|
||||
db.session.commit()
|
||||
|
||||
resp = client.get(f'/api/assets/{a.assetid}/relationships')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()['data']
|
||||
|
||||
for row in data['outgoing']:
|
||||
assert 'isdirectional' in row
|
||||
for row in data['incoming']:
|
||||
assert 'isdirectional' in row
|
||||
|
||||
# symmetric type reports False both directions; directional reports True
|
||||
sym_rows = [r for r in data['outgoing'] + data['incoming']
|
||||
if r['relationshiptypename'] == 'Dualpath']
|
||||
assert sym_rows and all(r['isdirectional'] is False for r in sym_rows)
|
||||
ctrl_rows = [r for r in data['outgoing']
|
||||
if r['relationshiptypename'] == 'controls']
|
||||
assert ctrl_rows and all(r['isdirectional'] is True for r in ctrl_rows)
|
||||
249
tests/test_core/test_relationship_propagation.py
Normal file
249
tests/test_core/test_relationship_propagation.py
Normal file
@@ -0,0 +1,249 @@
|
||||
"""Relationship propagation across symmetric rails (ADR-001, wired).
|
||||
|
||||
A Dualpath pair is one physical dual-bay machine with a single controller PC
|
||||
and a bay-selector switch, so BOTH bay records must carry the controls link to
|
||||
that PC. Creating PC controls bayA where bayA has a Dualpath partner bayB
|
||||
auto-creates PC controls bayB. The propagation set is M:N
|
||||
(relationshiptypepropagations): controls propagates through Dualpath
|
||||
(symmetric, consumed) and through partof (directional, declared but inert).
|
||||
"""
|
||||
|
||||
from shopdb.extensions import db
|
||||
from shopdb.core.models import Asset, AssetType
|
||||
from shopdb.core.models.relationship import (
|
||||
RelationshipType,
|
||||
RelationshipTypePropagation,
|
||||
AssetRelationship,
|
||||
)
|
||||
|
||||
|
||||
def _make_asset(assetnumber, typeid):
|
||||
asset = Asset(assetnumber=assetnumber, assettypeid=typeid)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
return asset
|
||||
|
||||
|
||||
def _setup_types():
|
||||
atype = AssetType(assettype='machine')
|
||||
db.session.add(atype)
|
||||
db.session.flush()
|
||||
|
||||
controls = RelationshipType(relationshiptype='controls', isdirectional=True)
|
||||
dualpath = RelationshipType(relationshiptype='Dualpath', isdirectional=False)
|
||||
partof = RelationshipType(relationshiptype='partof', isdirectional=True)
|
||||
db.session.add_all([controls, dualpath, partof])
|
||||
db.session.flush()
|
||||
return atype, controls, dualpath, partof
|
||||
|
||||
|
||||
def _rel_exists(sourceid, targetid, typeid):
|
||||
return AssetRelationship.query.filter_by(
|
||||
sourceassetid=sourceid, targetassetid=targetid, relationshiptypeid=typeid
|
||||
).first() is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Symmetric through-type propagates (controls fans across a Dualpath pair)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_controls_propagates_across_dualpath_pair(client, db, auth_headers):
|
||||
atype, controls, dualpath, partof = _setup_types()
|
||||
db.session.add(RelationshipTypePropagation(
|
||||
relationshiptypeid=controls.relationshiptypeid,
|
||||
throughtypeid=dualpath.relationshiptypeid))
|
||||
|
||||
pc = _make_asset('PC-1', atype.assettypeid)
|
||||
baya = _make_asset('BAY-A', atype.assettypeid)
|
||||
bayb = _make_asset('BAY-B', atype.assettypeid)
|
||||
|
||||
# the two bays are Dualpath partners (one physical machine)
|
||||
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
|
||||
relationshiptypeid=dualpath.relationshiptypeid))
|
||||
db.session.commit()
|
||||
|
||||
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
|
||||
'sourceassetid': pc.assetid,
|
||||
'targetassetid': baya.assetid,
|
||||
'relationshiptypeid': controls.relationshiptypeid,
|
||||
})
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
assert resp.get_json()['data']['propagatedcount'] == 1
|
||||
|
||||
# PC now controls BOTH bays
|
||||
assert _rel_exists(pc.assetid, baya.assetid, controls.relationshiptypeid)
|
||||
assert _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
|
||||
|
||||
|
||||
def test_propagation_finds_partner_stored_either_direction(client, db, auth_headers):
|
||||
# Dualpath row stored partner->target; propagation must still catch it.
|
||||
atype, controls, dualpath, partof = _setup_types()
|
||||
db.session.add(RelationshipTypePropagation(
|
||||
relationshiptypeid=controls.relationshiptypeid,
|
||||
throughtypeid=dualpath.relationshiptypeid))
|
||||
|
||||
pc = _make_asset('PC-2', atype.assettypeid)
|
||||
baya = _make_asset('BAY-A2', atype.assettypeid)
|
||||
bayb = _make_asset('BAY-B2', atype.assettypeid)
|
||||
# note: bayb is the SOURCE of the Dualpath row, target is baya
|
||||
db.session.add(AssetRelationship(sourceassetid=bayb.assetid, targetassetid=baya.assetid,
|
||||
relationshiptypeid=dualpath.relationshiptypeid))
|
||||
db.session.commit()
|
||||
|
||||
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
|
||||
'sourceassetid': pc.assetid,
|
||||
'targetassetid': baya.assetid,
|
||||
'relationshiptypeid': controls.relationshiptypeid,
|
||||
})
|
||||
assert resp.status_code == 201
|
||||
assert _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
|
||||
|
||||
|
||||
def test_controls_propagates_when_bay_is_source(client, db, auth_headers):
|
||||
# Matches real dev data: controls is stored bay -> PC (source is the bay).
|
||||
# The bay's Dualpath partner must also get a controls -> PC row.
|
||||
atype, controls, dualpath, partof = _setup_types()
|
||||
db.session.add(RelationshipTypePropagation(
|
||||
relationshiptypeid=controls.relationshiptypeid,
|
||||
throughtypeid=dualpath.relationshiptypeid))
|
||||
|
||||
pc = _make_asset('PC-S', atype.assettypeid)
|
||||
baya = _make_asset('BAY-AS', atype.assettypeid)
|
||||
bayb = _make_asset('BAY-BS', atype.assettypeid)
|
||||
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
|
||||
relationshiptypeid=dualpath.relationshiptypeid))
|
||||
db.session.commit()
|
||||
|
||||
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
|
||||
'sourceassetid': baya.assetid,
|
||||
'targetassetid': pc.assetid,
|
||||
'relationshiptypeid': controls.relationshiptypeid,
|
||||
})
|
||||
assert resp.status_code == 201, resp.get_json()
|
||||
assert resp.get_json()['data']['propagatedcount'] == 1
|
||||
# partner bay now also controls the PC, same direction (bay -> PC)
|
||||
assert _rel_exists(bayb.assetid, pc.assetid, controls.relationshiptypeid)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Idempotent: pre-existing propagated row is not duplicated
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_propagation_idempotent(client, db, auth_headers):
|
||||
atype, controls, dualpath, partof = _setup_types()
|
||||
db.session.add(RelationshipTypePropagation(
|
||||
relationshiptypeid=controls.relationshiptypeid,
|
||||
throughtypeid=dualpath.relationshiptypeid))
|
||||
|
||||
pc = _make_asset('PC-3', atype.assettypeid)
|
||||
baya = _make_asset('BAY-A3', atype.assettypeid)
|
||||
bayb = _make_asset('BAY-B3', atype.assettypeid)
|
||||
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
|
||||
relationshiptypeid=dualpath.relationshiptypeid))
|
||||
# the mirrored row already exists
|
||||
db.session.add(AssetRelationship(sourceassetid=pc.assetid, targetassetid=bayb.assetid,
|
||||
relationshiptypeid=controls.relationshiptypeid))
|
||||
db.session.commit()
|
||||
|
||||
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
|
||||
'sourceassetid': pc.assetid,
|
||||
'targetassetid': baya.assetid,
|
||||
'relationshiptypeid': controls.relationshiptypeid,
|
||||
})
|
||||
assert resp.status_code == 201
|
||||
assert resp.get_json()['data']['propagatedcount'] == 0
|
||||
|
||||
count = AssetRelationship.query.filter_by(
|
||||
sourceassetid=pc.assetid, targetassetid=bayb.assetid,
|
||||
relationshiptypeid=controls.relationshiptypeid).count()
|
||||
assert count == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Non-propagating type creates no extras
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_non_propagating_type_no_extras(client, db, auth_headers):
|
||||
atype, controls, dualpath, partof = _setup_types()
|
||||
# controls has NO propagation rows here
|
||||
|
||||
pc = _make_asset('PC-4', atype.assettypeid)
|
||||
baya = _make_asset('BAY-A4', atype.assettypeid)
|
||||
bayb = _make_asset('BAY-B4', atype.assettypeid)
|
||||
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
|
||||
relationshiptypeid=dualpath.relationshiptypeid))
|
||||
db.session.commit()
|
||||
|
||||
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
|
||||
'sourceassetid': pc.assetid,
|
||||
'targetassetid': baya.assetid,
|
||||
'relationshiptypeid': controls.relationshiptypeid,
|
||||
})
|
||||
assert resp.status_code == 201
|
||||
assert resp.get_json()['data']['propagatedcount'] == 0
|
||||
assert not _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Directional through-type (partof) does NOT propagate (deferred)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_directional_through_type_does_not_propagate(client, db, auth_headers):
|
||||
atype, controls, dualpath, partof = _setup_types()
|
||||
# controls propagates only through partof, which is directional -> inert
|
||||
db.session.add(RelationshipTypePropagation(
|
||||
relationshiptypeid=controls.relationshiptypeid,
|
||||
throughtypeid=partof.relationshiptypeid))
|
||||
|
||||
pc = _make_asset('PC-5', atype.assettypeid)
|
||||
parent = _make_asset('PARENT-5', atype.assettypeid)
|
||||
child = _make_asset('CHILD-5', atype.assettypeid)
|
||||
db.session.add(AssetRelationship(sourceassetid=child.assetid, targetassetid=parent.assetid,
|
||||
relationshiptypeid=partof.relationshiptypeid))
|
||||
db.session.commit()
|
||||
|
||||
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
|
||||
'sourceassetid': pc.assetid,
|
||||
'targetassetid': parent.assetid,
|
||||
'relationshiptypeid': controls.relationshiptypeid,
|
||||
})
|
||||
assert resp.status_code == 201
|
||||
assert resp.get_json()['data']['propagatedcount'] == 0
|
||||
assert not _rel_exists(pc.assetid, child.assetid, controls.relationshiptypeid)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI backfill creates missing rows and is idempotent run twice
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_cli_backfill_propagates_and_is_idempotent(client, db, auth_headers, runner):
|
||||
atype, controls, dualpath, partof = _setup_types()
|
||||
db.session.add(RelationshipTypePropagation(
|
||||
relationshiptypeid=controls.relationshiptypeid,
|
||||
throughtypeid=dualpath.relationshiptypeid))
|
||||
|
||||
pc = _make_asset('PC-6', atype.assettypeid)
|
||||
baya = _make_asset('BAY-A6', atype.assettypeid)
|
||||
bayb = _make_asset('BAY-B6', atype.assettypeid)
|
||||
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
|
||||
relationshiptypeid=dualpath.relationshiptypeid))
|
||||
# legacy-import shape: controls exists only on the primary bay
|
||||
db.session.add(AssetRelationship(sourceassetid=pc.assetid, targetassetid=baya.assetid,
|
||||
relationshiptypeid=controls.relationshiptypeid))
|
||||
db.session.commit()
|
||||
|
||||
assert not _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
|
||||
|
||||
result = runner.invoke(args=['relationships', 'propagate'])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert 'Propagated 1' in result.output
|
||||
assert _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
|
||||
|
||||
# second run creates nothing more
|
||||
result2 = runner.invoke(args=['relationships', 'propagate'])
|
||||
assert result2.exit_code == 0, result2.output
|
||||
assert 'Propagated 0' in result2.output
|
||||
count = AssetRelationship.query.filter_by(
|
||||
sourceassetid=pc.assetid, targetassetid=bayb.assetid,
|
||||
relationshiptypeid=controls.relationshiptypeid).count()
|
||||
assert count == 1
|
||||
Reference in New Issue
Block a user