Relationship inverse guard, audit command, widened direction fixer

tests/test_core/test_relationship_inverse_guard.py

@@ -0,0 +1,100 @@
+"""A directional relationship points ONE way.
+
+Prod grew rows saying both `PC controls 2005` and `2005 controls PC`, and a CMM
+PC showing `<- controls from CMM4` beside its own outgoing link. Only one
+direction can be true - a PC drives a machine, never the reverse - but the
+duplicate check was keyed on (source, target, type), so the inverse inserted
+cleanly and both rows then rendered on both asset pages.
+
+Symmetric types are exempt on purpose: they store both directions and the card
+collapses them to one entry per peer.
+"""
+
+from shopdb.extensions import db
+from shopdb.core.models import Asset, AssetType
+from shopdb.core.models.relationship import RelationshipType
+
+
+def _asset(assetnumber):
+    assettype = AssetType.query.filter_by(assettype='machine').first()
+    if not assettype:
+        assettype = AssetType(assettype='machine')
+        db.session.add(assettype)
+        db.session.flush()
+    asset = Asset(assetnumber=assetnumber, assettypeid=assettype.assettypeid)
+    db.session.add(asset)
+    db.session.flush()
+    return asset
+
+
+def _reltype(name, directional=True):
+    existing = RelationshipType.query.filter_by(relationshiptype=name).first()
+    if existing:
+        existing.isdirectional = directional
+        db.session.flush()
+        return existing
+    reltype = RelationshipType(relationshiptype=name, isdirectional=directional)
+    db.session.add(reltype)
+    db.session.flush()
+    return reltype
+
+
+def _create(client, headers, source, target, typeid):
+    return client.post('/api/assets/relationships',
+                       json={'sourceassetid': source, 'targetassetid': target,
+                             'relationshiptypeid': typeid},
+                       headers=headers)
+
+
+def test_inverse_of_a_directional_link_is_refused(client, db, auth_headers):
+    pc, machine = _asset('PC-INV-1'), _asset('MACH-INV-1')
+    controls = _reltype('controls')
+    db.session.commit()
+
+    first = _create(client, auth_headers, pc.assetid, machine.assetid,
+                    controls.relationshiptypeid)
+    assert first.status_code == 201, first.get_json()
+
+    inverse = _create(client, auth_headers, machine.assetid, pc.assetid,
+                      controls.relationshiptypeid)
+    assert inverse.status_code == 409, inverse.get_json()
+    message = inverse.get_json()['data']['error']['message'].lower()
+    assert 'reverse' in message
+    assert 'one way' in message
+
+
+def test_the_same_direction_is_still_refused_as_a_duplicate(client, db, auth_headers):
+    pc, machine = _asset('PC-INV-2'), _asset('MACH-INV-2')
+    controls = _reltype('controls')
+    db.session.commit()
+
+    assert _create(client, auth_headers, pc.assetid, machine.assetid,
+                   controls.relationshiptypeid).status_code == 201
+    again = _create(client, auth_headers, pc.assetid, machine.assetid,
+                    controls.relationshiptypeid)
+    assert again.status_code == 409
+    assert 'already exists' in again.get_json()['data']['error']['message'].lower()
+
+
+def test_symmetric_types_may_store_both_directions(client, db, auth_headers):
+    """Dualpath bays link both ways by design - the guard must not touch them."""
+    baya, bayb = _asset('BAY-INV-A'), _asset('BAY-INV-B')
+    dualpath = _reltype('Dualpath', directional=False)
+    db.session.commit()
+
+    assert _create(client, auth_headers, baya.assetid, bayb.assetid,
+                   dualpath.relationshiptypeid).status_code == 201
+    both = _create(client, auth_headers, bayb.assetid, baya.assetid,
+                   dualpath.relationshiptypeid)
+    assert both.status_code == 201, both.get_json()
+
+
+def test_an_asset_cannot_relate_to_itself(client, db, auth_headers):
+    lonely = _asset('SELF-INV-1')
+    controls = _reltype('controls')
+    db.session.commit()
+
+    resp = _create(client, auth_headers, lonely.assetid, lonely.assetid,
+                   controls.relationshiptypeid)
+    assert resp.status_code == 400, resp.get_json()
+    assert 'itself' in resp.get_json()['data']['error']['message'].lower()