shopdb-flask/tests/test_plugins/test_printer_assignment.py

"""Machine-level printer assignment and its propagation to the controlling PC.

The whole point of the feature: the assignment is a property of the MACHINE, so
a reimaged bay PC gets its printers back from the asset register with no backup
and no restore step. A PC that carries its own assignment keeps it and shadows
the machine's, because an exception recorded on the PC is a deliberate one.

Two surfaces are exercised:
  GET  /api/printers/for-host/<hostname>            what a host should install
  PUT  /api/printers/assignments/for-asset/<id>     reconcile an asset's set

Relationship types match `flask seed reference-data`: `usesprinter` means "this
printer is installed here" (many), `defaultprinter` means "which of them is the
default" (at most one), both directional source -> printer, both propagating
read-time through `controls`.
"""

import pytest

from shopdb.core.models import (
    Asset,
    AssetRelationship,
    AssetType,
    RelationshipType,
)
from plugins.printers.models import Printer

HOST_URL = '/api/printers/for-host/%s'
ASSIGN_URL = '/api/printers/assignments/for-asset/%d'

SHOPFLOOR_HOST = 'SHOPPC01'
OFFICE_HOST = 'OFFICEPC01'


def _rows(response):
    """The assignment set out of a for-host payload.

    Normalized in one place: the endpoint may return the bare list under `data`
    or wrap it in a `printers` key, and the settled semantics under test are the
    same either way.
    """
    payload = response.get_json()['data']
    if isinstance(payload, dict):
        payload = payload.get('printers') or []
    return payload


def _printerids(response):
    return {row['printerid'] for row in _rows(response)}


def _defaultprinterid(response):
    """The one printer flagged default, or None.

    Asserts the at-most-one rule on the way out: two defaults reaching a client
    means the PC picks whichever it saw last, which is the bug this endpoint
    exists to make impossible.
    """
    flagged = [row['printerid'] for row in _rows(response) if row.get('isdefault')]
    assert len(flagged) <= 1, 'more than one printer came back flagged default'
    return flagged[0] if flagged else None


def _active_defaults(pc):
    """Active defaultprinter rows on an asset, read straight from the table.

    The unique constraint is (source, target, type) and does NOT stop two rows
    with two different targets, so the one-default rule only holds if the write
    path enforces it. Counted here rather than inferred from the read side.
    """
    dp_type = RelationshipType.query.filter_by(relationshiptype='defaultprinter').first()
    return AssetRelationship.query.filter_by(
        sourceassetid=pc.assetid,
        relationshiptypeid=dp_type.relationshiptypeid,
        isactive=True,
    ).all()


@pytest.fixture
def scene(db):
    """A bay PC controlling a machine, an office PC controlling nothing, and
    three printers. No assignments yet: each test builds the ones it needs.
    """
    from plugins.computers.models import Computer

    pc_type = AssetType(assettype='computer', pluginname='computers', tablename='computers')
    machine_type = AssetType(assettype='machine', pluginname='machines', tablename='machines')
    printer_type = AssetType(assettype='printer', pluginname='printers', tablename='printers')
    uses_type = RelationshipType(
        relationshiptype='usesprinter',
        description='Asset to a printer installed on it',
        isdirectional=True,
    )
    default_type = RelationshipType(
        relationshiptype='defaultprinter',
        description='Asset to its default printer',
        isdirectional=True,
    )
    controls_type = RelationshipType(
        relationshiptype='controls',
        description='Operational authority over another asset',
        isdirectional=True,
    )
    db.session.add_all([pc_type, machine_type, printer_type,
                        uses_type, default_type, controls_type])
    db.session.flush()

    shopfloorpc = Asset(assetnumber='1001', name='Bay PC',
                        assettypeid=pc_type.assettypeid, isactive=True)
    officepc = Asset(assetnumber='1002', name='Office PC',
                     assettypeid=pc_type.assettypeid, isactive=True)
    machine = Asset(assetnumber='2001', name='Lathe',
                    assettypeid=machine_type.assettypeid, isactive=True)
    db.session.add_all([shopfloorpc, officepc, machine])
    db.session.flush()

    printers = {}
    for suffix, name in (('A', 'Bay label printer'),
                         ('B', 'Bay laser printer'),
                         ('C', 'Office laser printer')):
        asset = Asset(assetnumber='PRN-%s' % suffix, name=name,
                      assettypeid=printer_type.assettypeid, isactive=True)
        db.session.add(asset)
        db.session.flush()
        printer = Printer(assetid=asset.assetid, windowsname='PRINTER-%s' % suffix,
                          isnetwork=True, hostname='printer-%s' % suffix.lower())
        db.session.add(printer)
        printers[suffix] = {'asset': asset, 'printer': printer}

    db.session.add_all([
        Computer(assetid=shopfloorpc.assetid, hostname=SHOPFLOOR_HOST),
        Computer(assetid=officepc.assetid, hostname=OFFICE_HOST),
    ])
    db.session.commit()

    return {
        'shopfloorpc': shopfloorpc,
        'officepc': officepc,
        'machine': machine,
        'printers': printers,
        'uses_type': uses_type,
        'default_type': default_type,
        'controls_type': controls_type,
    }


def _relate(db, source, target, reltype):
    db.session.add(AssetRelationship(
        sourceassetid=source.assetid,
        targetassetid=target.assetid,
        relationshiptypeid=reltype.relationshiptypeid,
    ))
    db.session.commit()


def _assign(db, scene, owner, suffixes, default=None):
    """Write usesprinter rows (and one defaultprinter row) straight to the table."""
    for suffix in suffixes:
        _relate(db, owner, scene['printers'][suffix]['asset'], scene['uses_type'])
    if default:
        _relate(db, owner, scene['printers'][default]['asset'], scene['default_type'])


def _controls(db, scene):
    _relate(db, scene['shopfloorpc'], scene['machine'], scene['controls_type'])


def _printerid(scene, suffix):
    return scene['printers'][suffix]['printer'].printerid


def _assetid(scene, suffix):
    return scene['printers'][suffix]['asset'].assetid


def test_pc_with_own_assignment_gets_exactly_that(client, db, scene):
    """A PC's own rows resolve as-is.

    If this breaks, an assignment recorded against the PC itself either does not
    reach the host or arrives padded with printers nobody assigned - and the
    client installs queues on a bay that never asked for them.
    """
    _assign(db, scene, scene['shopfloorpc'], ['A', 'B'], default='A')

    response = client.get(HOST_URL % SHOPFLOOR_HOST)

    assert response.status_code == 200
    assert _printerids(response) == {_printerid(scene, 'A'), _printerid(scene, 'B')}
    assert _defaultprinterid(response) == _printerid(scene, 'A')


def test_pc_without_assignment_inherits_from_the_machine_it_controls(client, db, scene):
    """The reimage case, and the reason the feature exists.

    A rebuilt bay PC has no rows of its own. It must still come back with the
    machine's printers through its `controls` edge. If inheritance is lost, every
    reimage costs a technician visit again and the asset register stops being the
    source of truth for what a bay prints on.
    """
    _assign(db, scene, scene['machine'], ['A', 'B'], default='B')
    _controls(db, scene)

    response = client.get(HOST_URL % SHOPFLOOR_HOST)

    assert response.status_code == 200
    assert _printerids(response) == {_printerid(scene, 'A'), _printerid(scene, 'B')}
    assert _defaultprinterid(response) == _printerid(scene, 'B')


def test_own_assignment_overrides_the_machine_rather_than_merging(client, db, scene):
    """Own rows shadow inherited ones. They do not add to them.

    A PC row is how a site records a deliberate exception ("this bay's PC prints
    to the office laser instead"). Merging would silently reinstate exactly the
    printers the exception was written to remove, and no amount of editing the PC
    would ever get rid of them.
    """
    _assign(db, scene, scene['machine'], ['A', 'B'], default='A')
    _controls(db, scene)
    _assign(db, scene, scene['shopfloorpc'], ['C'], default='C')

    response = client.get(HOST_URL % SHOPFLOOR_HOST)

    assert response.status_code == 200
    assert _printerids(response) == {_printerid(scene, 'C')}
    assert _defaultprinterid(response) == _printerid(scene, 'C')


def test_office_pc_controlling_no_machine_resolves_empty(client, db, scene):
    """A PC with no machine and no assignment is a valid answer, not an error.

    Most office PCs control nothing. The walk must end quietly and return an
    empty set: a 404 or a 500 here would make the client script log a failure on
    every cycle on every office PC, and real failures would drown in it.
    """
    response = client.get(HOST_URL % OFFICE_HOST)

    assert response.status_code == 200
    assert _rows(response) == []
    assert _defaultprinterid(response) is None


def test_default_is_optional(client, db, scene):
    """Printers with no default is a legitimate state.

    A bay with three printers and no default exists on the floor. If the API
    forces a default, a write either fails or invents one, and the client then
    changes a user's default printer because ShopDB picked arbitrarily.
    """
    _assign(db, scene, scene['machine'], ['A', 'B', 'C'])
    _controls(db, scene)

    response = client.get(HOST_URL % SHOPFLOOR_HOST)

    assert response.status_code == 200
    assert len(_rows(response)) == 3
    assert _defaultprinterid(response) is None
    assert _active_defaults(scene['machine']) == []


def test_setting_a_default_replaces_the_existing_one(client, db, scene, auth_headers):
    """Two active defaults must be impossible.

    The unique constraint is (source, target, type), so a second default INSERTs
    cleanly and nothing complains. Then the read side returns two, the client
    picks whichever it iterated last, and the bay's default printer flips at
    random between cycles.
    """
    first = client.put(ASSIGN_URL % scene['machine'].assetid, headers=auth_headers, json={
        'printerassetids': [_assetid(scene, 'A'), _assetid(scene, 'B')],
        'defaultprinterassetid': _assetid(scene, 'A'),
    })
    assert first.status_code == 200

    second = client.put(ASSIGN_URL % scene['machine'].assetid, headers=auth_headers, json={
        'printerassetids': [_assetid(scene, 'A'), _assetid(scene, 'B')],
        'defaultprinterassetid': _assetid(scene, 'B'),
    })
    assert second.status_code == 200

    defaults = _active_defaults(scene['machine'])
    assert len(defaults) == 1
    assert defaults[0].targetassetid == _assetid(scene, 'B')

    _controls(db, scene)
    response = client.get(HOST_URL % SHOPFLOOR_HOST)
    assert _defaultprinterid(response) == _printerid(scene, 'B')


def test_default_must_be_one_of_the_assigned_printers(client, db, scene, auth_headers):
    """A default outside the assignment set is rejected.

    Otherwise the client is told to default to a queue it was never told to
    install, fails to set it, and the bay looks broken with nothing in ShopDB
    showing why.
    """
    response = client.put(ASSIGN_URL % scene['machine'].assetid, headers=auth_headers, json={
        'printerassetids': [_assetid(scene, 'A')],
        'defaultprinterassetid': _assetid(scene, 'C'),
    })

    assert response.status_code == 400
    assert _active_defaults(scene['machine']) == []


def test_unassigning_the_default_printer_clears_the_default(client, db, scene, auth_headers):
    """Removing a printer takes its default with it.

    A default row left pointing at an unassigned printer is the dangling case:
    the printer disappears from the install set while the client is still told to
    make it the default. Removal is server-side only - it uninstalls nothing.
    """
    assigned = client.put(ASSIGN_URL % scene['machine'].assetid, headers=auth_headers, json={
        'printerassetids': [_assetid(scene, 'A'), _assetid(scene, 'B')],
        'defaultprinterassetid': _assetid(scene, 'B'),
    })
    assert assigned.status_code == 200

    unassigned = client.put(ASSIGN_URL % scene['machine'].assetid, headers=auth_headers, json={
        'printerassetids': [_assetid(scene, 'A')],
        'defaultprinterassetid': None,
    })
    assert unassigned.status_code == 200
    assert _active_defaults(scene['machine']) == []

    _controls(db, scene)
    response = client.get(HOST_URL % SHOPFLOOR_HOST)
    assert _printerids(response) == {_printerid(scene, 'A')}
    assert _defaultprinterid(response) is None


def test_unknown_hostname_is_a_404(client, db, scene):
    """A host ShopDB has never heard of is an error, not an empty set.

    Empty means "this PC is assigned nothing", which the client treats as a safe
    no-op. A mistyped or unenrolled hostname returning empty looks exactly the
    same, so a bay would sit unconfigured with nothing anywhere saying its
    record is missing.
    """
    response = client.get(HOST_URL % 'NOSUCHHOST')

    assert response.status_code == 404