ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Printer assignment end to end, and a container that keeps a site's files

Browse Source 
Assign printers to a MACHINE and let whichever PC controls it install them, with
the observed half kept strictly apart from the assigned half: a collector report
can never write an assignment row. Adoption is explicit. Adds a singular
relationship type so defaultprinter can mean exactly one, client scripts that
converge a bay's queues and correct a drifted one in place, and a wave gate so a
fleet-wide driver rollout cannot stampede its own share.

On the container side: the compose stack never persisted the instance directory,
so a rebuild discarded plugins.json and every upload. It is a volume now, the
image ships instance/ owned by the app user so the mountpoint is writable, and
the database moves to MySQL 8.4 LTS, which the Windows runbook has recommended
since 8.0 went end of life. The image also could not build at all: the frontend
stage never copied the plugin staging script it runs.

Adds a topology migration guide covering IIS to Docker and back.
This commit is contained in:
Cameron Proudlock
2026-08-20 07:46:35 -04:00
parent 09e9d5f047
commit d267813999
45 changed files with 6475 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
tests/test_plugins/test_collector_client_script.py Normal file
View File

@@ -0,0 +1,122 @@
"""The collector reporter, generated with a site's own values.
A site used to receive a script with the reference site's server and VLANs in
it, which it had to find and edit. The server now stamps its own values into the
parameter defaults, so the file downloads ready to deploy.
What must stay true:
- ONLY the defaults are substituted. The copy in the repo stays runnable, so
there is never a second version to drift from the first.
- The collector key is NEVER in the file. It lands on every shop-floor PC, and a
token spread across hundreds of bays cannot be rotated quietly.
- Everything stamped is still overridable, because a bay may need to differ from
its site.
"""
import pytest
from shopdb.core.models import Setting
SCRIPT = '/api/computers/client-script'
def _set(db, key, value):
row = Setting.query.filter_by(key=key).first()
if row is None:
row = Setting(key=key, value=value, valuetype='string', category='computers')
db.session.add(row)
else:
row.value = value
db.session.commit()
def test_an_anonymous_caller_gets_nothing(client):
"""It states a site's URL and internal ranges - configuration, not a
handout."""
assert client.get(SCRIPT).status_code in (401, 422)
def test_it_serves_the_reporter_as_a_download(client, auth_headers):
resp = client.get(SCRIPT, headers=auth_headers)
assert resp.status_code == 200, resp.get_data(as_text=True)[:200]
assert 'attachment' in resp.headers['Content-Disposition']
assert 'Report-AssetToShopDB.ps1' in resp.headers['Content-Disposition']
body = resp.get_data(as_text=True)
assert 'param(' in body
assert 'api/collector/computers' in body
def test_it_publishes_a_hash_of_what_it_served(client, auth_headers):
"""So a deployment can verify what it fetched, like the installer does."""
import hashlib
resp = client.get(SCRIPT, headers=auth_headers)
digest = hashlib.sha256(resp.get_data()).hexdigest()
assert resp.headers['X-Script-Sha256'] == digest
def test_it_stamps_the_sites_own_url(client, db, auth_headers):
_set(db, 'site_base_url', 'https://shopdb.example.net')
body = client.get(SCRIPT, headers=auth_headers).get_data(as_text=True)
assert "[string]$ApiUrl = 'https://shopdb.example.net/api/collector/computers'" in body
assert 'for https://shopdb.example.net' in body
def test_a_site_with_no_url_configured_still_gets_a_usable_script(client, db,
auth_headers):
"""Blank site_base_url falls back to the origin the admin is talking to,
which is by definition a reachable address for this server."""
_set(db, 'site_base_url', '')
body = client.get(SCRIPT, headers=auth_headers).get_data(as_text=True)
assert "[string]$ApiUrl = ''" not in body
assert 'api/collector/computers' in body
def test_it_stamps_the_sites_ranges(client, db, auth_headers):
"""The replacement for the two VLANs that used to be source code."""
_set(db, 'computers_routableranges', '10.20.0.0/23,10.21.4.0/26')
body = client.get(SCRIPT, headers=auth_headers).get_data(as_text=True)
assert "[string]$AllowedRanges = '10.20.0.0/23,10.21.4.0/26'" in body
def test_no_ranges_configured_leaves_the_default_route_fallback(client, db,
auth_headers):
"""An unconfigured site must not be given someone else's addressing: the
script falls back to the NIC carrying the default route."""
_set(db, 'computers_routableranges', '')
body = client.get(SCRIPT, headers=auth_headers).get_data(as_text=True)
assert "[string]$AllowedRanges = ''" in body
assert 'default route' in body
def test_the_collector_key_is_never_in_the_file(client, db, auth_headers):
"""The one thing that must not be stamped in. A token in a file on every bay
is a token nobody can rotate quietly."""
_set(db, 'site_base_url', 'https://shopdb.example.net')
body = client.get(SCRIPT, headers=auth_headers).get_data(as_text=True)
assert 'shopdb_pat_' not in body
assert "$ApiKey = ''" in body or "[string]$ApiKey = ''" in body
# It says where the key comes from instead.
assert 'CollectorKey' in body
def test_the_body_is_not_rewritten_only_the_defaults(client, auth_headers):
"""The repo copy stays runnable. If generation started editing the body,
the file on disk and the file a site runs would diverge."""
from plugins.computers.api.routes import _client_script_path
with open(_client_script_path(), 'r', encoding='utf-8') as handle:
source = handle.read()
body = client.get(SCRIPT, headers=auth_headers).get_data(as_text=True)
# Every line of the source survives except the two parameter defaults.
changed = [line for line in source.splitlines()
if line.strip() and line not in body.splitlines()]
assert all('$ApiUrl' in line or '$AllowedRanges' in line for line in changed), changed
def test_it_names_the_version_that_generated_it(client, auth_headers):
"""So a script found on a bay can be traced to the server that made it."""
from shopdb import __version__
body = client.get(SCRIPT, headers=auth_headers).get_data(as_text=True)
assert __version__ in body

342
tests/test_plugins/test_printer_assignment.py Normal file
View File

@@ -0,0 +1,342 @@
"""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

120
tests/test_plugins/test_printer_driver_resolution.py Normal file
View File

@@ -0,0 +1,120 @@
"""Which driver a printer installs with.
One row per model was unworkable: HP and Xerox universal drivers cover 41 of the
reference site's 44 printers, so binding a driver to a single model meant 21
near-duplicate rows pointing at one package - a table nobody keeps true, and the
reason 42 of 44 printers could not resolve a driver at all.
The order is most-specific-first, and each step exists for a printer that really
is on this floor: a plotter and a card printer need their own driver, everything
else takes its make's universal one.
"""
import pytest
from shopdb.core.models import Vendor, Model
from plugins.printers.models import Printer, PrinterDriver
from plugins.printers.api.asset_routes import _printer_driver
def _universal():
return (PrinterDriver.query
.filter(PrinterDriver.modelnumberid.is_(None),
PrinterDriver.isactive == True)
.order_by(PrinterDriver.name).all())
@pytest.fixture
def fleet(db):
hp = Vendor(vendor='HP')
xerox = Vendor(vendor='Xerox')
db.session.add_all([hp, xerox])
db.session.flush()
laserjet = Model(modelnumber='LaserJet M602', vendorid=hp.vendorid)
designjet = Model(modelnumber='DesignJet T1700', vendorid=hp.vendorid)
db.session.add_all([laserjet, designjet])
db.session.flush()
upd = PrinterDriver(name='HP Universal Print Driver', drivername='HP Universal Printing PS',
location=r'\\server\share\hp_upd', vendorid=hp.vendorid, isactive=True)
plotter = PrinterDriver(name='HP DesignJet T1700', drivername='HP DesignJet T1700dr V4',
location=r'\\server\share\designjet', vendorid=hp.vendorid,
modelnumberid=designjet.modelnumberid, isactive=True)
gpd = PrinterDriver(name='Xerox Global Print Driver PCL6',
drivername='Xerox Global Print Driver PCL6',
location=r'\\server\share\xerox', vendorid=xerox.vendorid, isactive=True)
db.session.add_all([upd, plotter, gpd])
db.session.commit()
return {'hp': hp, 'xerox': xerox, 'laserjet': laserjet, 'designjet': designjet,
'upd': upd, 'plotter': plotter, 'gpd': gpd}
def test_a_printer_takes_its_makes_universal_driver(fleet):
"""The case that covers most of a floor: no per-model row exists, and none
should have to."""
printer = Printer(vendorid=fleet['hp'].vendorid, modelnumberid=fleet['laserjet'].modelnumberid)
assert _printer_driver(printer, _universal()).name == 'HP Universal Print Driver'
def test_a_model_specific_driver_beats_the_universal_one(fleet):
"""A plotter is not a LaserJet. If the universal driver won here, the
DesignJet would be installed with a driver that cannot drive it."""
printer = Printer(vendorid=fleet['hp'].vendorid, modelnumberid=fleet['designjet'].modelnumberid)
assert _printer_driver(printer, _universal()).name == 'HP DesignJet T1700'
def test_vendors_do_not_bleed_into_each_other(fleet):
"""A Xerox must never resolve to the HP driver, whatever the ordering."""
printer = Printer(vendorid=fleet['xerox'].vendorid, modelnumberid=None)
assert _printer_driver(printer, _universal()).name == 'Xerox Global Print Driver PCL6'
def test_a_printer_with_no_vendor_resolves_to_nothing(fleet):
"""Better nothing than a guess: installing the wrong driver is worse than
reporting that a printer has none."""
printer = Printer(vendorid=None, modelnumberid=None)
assert _printer_driver(printer, _universal()) is None
def test_a_make_with_no_driver_row_resolves_to_nothing(db, fleet):
"""A vendor nobody has added a driver for is unresolved, not misresolved."""
zebra = Vendor(vendor='Zebra')
db.session.add(zebra)
db.session.commit()
printer = Printer(vendorid=zebra.vendorid, modelnumberid=None)
assert _printer_driver(printer, _universal()) is None
def test_the_pre_vendorid_naming_convention_still_resolves(db, fleet):
"""A site that populated printerdrivers before the column existed matched on
the vendor word in the driver's NAME. That must keep working, or an upgrade
silently takes every driver away."""
epson = Vendor(vendor='Epson')
db.session.add(epson)
db.session.flush()
db.session.add(PrinterDriver(name='Epson ColorWorks universal', drivername='EPSON TM-C3500',
location=r'\\server\share\epson', isactive=True))
db.session.commit()
# vendor attached, not just vendorid: the legacy path reads the vendor's
# NAME, which only exists through the relationship.
printer = Printer(vendorid=epson.vendorid, modelnumberid=None)
printer.vendor = epson
assert _printer_driver(printer, _universal()).name == 'Epson ColorWorks universal'
def test_a_row_that_names_a_vendor_is_not_matched_by_its_text(db, fleet):
"""A driver whose vendorid is set and does NOT match must not then be picked
up by the name convention: a mis-set vendor would resolve to the wrong
package, which is worse than resolving to none."""
brother = Vendor(vendor='Brother')
db.session.add(brother)
db.session.flush()
# Named for Brother, but bound to HP by id - the id is the truth.
db.session.add(PrinterDriver(name='Brother universal', drivername='Brother Universal',
location=r'\\server\share\brother',
vendorid=fleet['hp'].vendorid, isactive=True))
db.session.commit()
printer = Printer(vendorid=brother.vendorid, modelnumberid=None)
printer.vendor = brother
assert _printer_driver(printer, _universal()) is None

836
tests/test_plugins/test_printer_observations.py Normal file
View File

@@ -0,0 +1,836 @@
"""What a bay ACTUALLY has, and how it compares to what it was assigned.
ShopDB has always known what a host SHOULD have (usesprinter/defaultprinter rows
on the machine, read through /api/printers/for-host). This is the other half: a
PC reports the queues it really carries to POST /api/collector/printers, and
/api/printers/observed/for-asset/<id> puts the two sides next to each other.
Three settled rules are what these tests exist to defend:
Observed and assigned stay apart. A collector report never writes an
assignment row. The moment a drifted bay's own state is allowed to become
what that bay is told to install, enforcement means nothing and every
configuration error becomes permanent the next time the PC checks in.
A report REPLACES that host's rows. This is current state, not history: the
latest report is the whole truth for the host, so "what does this bay have"
stays a filter and never becomes a question about time.
An unmatched queue is UNKNOWN, never a guess. A wrong match seeds a wrong
assignment, and a wrong assignment is worse than no assignment because the
client then installs it on every cycle.
Two surfaces are exercised:
POST /api/collector/printers what the host reports it has
GET /api/printers/observed/for-asset/<assetid> observed against assigned
Seeding has its own route, POST /api/printers/assignments/seed-from-observed/<id>,
because a rollout adopts many machines at once and doing that through the editor
would be one round trip per bay. It is still not a second WRITE path: it calls the
same _reconcile_edges the editor's PUT does, so both are validated identically and
an assignment can only be written one way.
What makes it safe is that it is explicit. Nothing calls it on a schedule, and a
queue that resolves to no known printer is refused rather than guessed into an
assignment.
"""
import json
import pytest
from shopdb.core.models import (
Asset,
AssetRelationship,
AssetType,
Communication,
CommunicationType,
Model,
RelationshipType,
Vendor,
)
from plugins.printers.models import Printer, PrinterDriver, PrinterObservedQueue
COLLECT_URL = '/api/collector/printers'
OBSERVED_URL = '/api/printers/observed/for-asset/%d'
ASSIGN_URL = '/api/printers/assignments/for-asset/%d'
HOST_URL = '/api/printers/for-host/%s'
KEY = 'testcollectorkey'
BAY_HOST = 'BAYPC01'
SECOND_BAY_HOST = 'BAYPC02'
OFFICE_HOST = 'OFFICEPC01'
# Addresses only, no site meaning: the port address is the match key under test.
ADDRESS_A = '10.20.0.11'
ADDRESS_B = '10.20.0.12'
ADDRESS_C = '10.20.0.13'
ADDRESS_NOBODY = '10.20.0.99'
DRIVER_NAME = 'HP Universal Printing PS'
@pytest.fixture
def collector_key(app):
"""Set the shared collector key. A site may scope a printers-only key
instead (COLLECTOR_API_KEY_PRINTERS); the shared key is the documented
fallback and is what the reporter script falls back to as well."""
old = app.config.get('COLLECTOR_API_KEY')
app.config['COLLECTOR_API_KEY'] = KEY
yield KEY
app.config['COLLECTOR_API_KEY'] = old
@pytest.fixture
def scene(db):
"""Two bay PCs controlling one machine, an office PC controlling nothing,
and three printers each reachable at its own address.
No assignments and no observations: every test builds the pair it needs, so
a classification can never be an accident of the fixture.
"""
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',
isdirectional=True)
ip_comtype = CommunicationType(comtype='IP')
db.session.add_all([pc_type, machine_type, printer_type, uses_type,
default_type, controls_type, ip_comtype])
db.session.flush()
vendor = Vendor(vendor='HP')
db.session.add(vendor)
db.session.flush()
model = Model(modelnumber='LaserJet M602', vendorid=vendor.vendorid)
db.session.add(model)
db.session.flush()
# Model-bound so the driver ShopDB would install is unambiguous: driver
# drift is only meaningful against a driver the assigned side actually names.
db.session.add(PrinterDriver(name='HP Universal Print Driver',
drivername=DRIVER_NAME,
location=r'\\server\share\hp',
vendorid=vendor.vendorid,
modelnumberid=model.modelnumberid,
isactive=True))
baypc = Asset(assetnumber='1001', name='Bay PC',
assettypeid=pc_type.assettypeid, isactive=True)
secondbaypc = Asset(assetnumber='1002', name='Second Bay PC',
assettypeid=pc_type.assettypeid, isactive=True)
officepc = Asset(assetnumber='1003', 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([baypc, secondbaypc, officepc, machine])
db.session.flush()
printers = {}
for suffix, name, address in (('A', 'Bay label printer', ADDRESS_A),
('B', 'Bay laser printer', ADDRESS_B),
('C', 'Office laser printer', ADDRESS_C)):
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,
hostname='printer-%s' % suffix.lower(),
vendorid=vendor.vendorid,
modelnumberid=model.modelnumberid,
isnetwork=True)
db.session.add(printer)
db.session.add(Communication(assetid=asset.assetid,
comtypeid=ip_comtype.comtypeid,
ipaddress=address, isprimary=True))
printers[suffix] = {'asset': asset, 'printer': printer,
'address': address}
db.session.add_all([
Computer(assetid=baypc.assetid, hostname=BAY_HOST),
Computer(assetid=secondbaypc.assetid, hostname=SECOND_BAY_HOST),
Computer(assetid=officepc.assetid, hostname=OFFICE_HOST),
])
db.session.commit()
return {
'baypc': baypc,
'secondbaypc': secondbaypc,
'officepc': officepc,
'machine': machine,
'printers': printers,
'uses_type': uses_type,
'default_type': default_type,
'controls_type': controls_type,
}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _queue(name, address=None, drivername=DRIVER_NAME, isdefault=False,
portname=None):
"""One entry of the reported queues array, spelled as the client sends it."""
return {
'queuename': name,
'drivername': drivername,
'portname': portname or ('IP_%s' % address if address else 'LPT1'),
'portaddress': address,
'isdefault': isdefault,
}
def _report(client, hostname, queues):
return client.post(COLLECT_URL, json={'hostname': hostname,
'queues': queues},
headers={'X-API-Key': KEY})
def _stored(hostname):
"""Observed rows held for a host, read straight from the table.
Read here rather than through the comparison endpoint because replacement is
a property of the STORE: a read that filtered by newest timestamp would hide
an append-only table growing behind it.
"""
return PrinterObservedQueue.query.filter(
PrinterObservedQueue.hostname.ilike(hostname)).all()
def _queuenames(hostname):
return {row.queuename for row in _stored(hostname)}
def _assignment_rows(scene):
"""Every active usesprinter/defaultprinter row in the database.
Not scoped to one asset on purpose: a collector report must not create an
assignment ANYWHERE, including on an asset the test never named.
"""
typeids = [scene['uses_type'].relationshiptypeid,
scene['default_type'].relationshiptypeid]
rows = AssetRelationship.query.filter(
AssetRelationship.relationshiptypeid.in_(typeids),
AssetRelationship.isactive == True).all() # noqa: E712
return {(row.sourceassetid, row.targetassetid, row.relationshiptypeid)
for row in rows}
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):
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, pc):
_relate(db, pc, scene['machine'], scene['controls_type'])
def _assetid(scene, suffix):
return scene['printers'][suffix]['asset'].assetid
def _hostblocks(response):
"""{hostname (lowercased): block} out of a comparison payload.
Normalized in one place because a machine answers with one block per
controlling PC while a PC has only itself, and the endpoint may reasonably
return the single case unwrapped. The semantics under test are the same
either way; which of the two shapes it is, is not.
"""
payload = response.get_json()['data']
blocks = payload.get('hosts')
if blocks is None:
blocks = [payload]
return {(block.get('hostname') or '').lower(): block for block in blocks}
def _oneblock(response, hostname):
blocks = _hostblocks(response)
assert hostname.lower() in blocks, \
'no block for %s in %s' % (hostname, sorted(blocks))
return blocks[hostname.lower()]
def _classified(block):
"""{classification: {identity}} for one host block.
Identity is the printer assetid when the row resolved to a printer, and the
queue name when it did not - which is exactly the distinction the UNKNOWN
rule is about. `missing` rows describe an assigned printer that was never
observed, so they may arrive in the queue list or in a list of their own.
"""
rows = list(block.get('queues') or [])
rows.extend(block.get('missing') or [])
result = {}
for row in rows:
assetid = row.get('printerassetid')
identity = assetid if assetid is not None else row.get('queuename')
result.setdefault(row.get('classification', 'missing'), set()).add(identity)
return result
def _seedcandidate(block):
seed = block.get('seedcandidate')
assert seed is not None, 'block carries no seedcandidate: %s' % sorted(block)
return seed
def _skippedtext(seed):
"""Everything the seed candidate says it left out, as one lowercase blob.
The shape of the skip report is not what matters; that an operator can see
WHICH queues were not seeded is. A seed that silently drops the queues it
could not match looks identical to a bay that has nothing else installed.
"""
skipped = (seed.get('skippedqueuenames') if 'skippedqueuenames' in seed
else seed.get('skipped'))
assert skipped is not None, \
'seedcandidate reports nothing about what it skipped: %s' % sorted(seed)
return json.dumps(skipped).lower()
# ---------------------------------------------------------------------------
# Collection
# ---------------------------------------------------------------------------
def test_a_report_stores_the_hosts_queues(client, db, collector_key, scene):
"""The queues a bay reports land verbatim, port address included.
Port address is the primary match key and the only unambiguous one. If it is
dropped or rewritten on the way in, every later comparison falls back to
matching on a queue name - a naming convention - and a renamed queue starts
reading as a different printer.
"""
response = _report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_A, isdefault=True),
_queue('PRINTER-B', ADDRESS_B),
])
assert response.status_code == 200, response.get_json()
rows = {row.queuename: row for row in _stored(BAY_HOST)}
assert set(rows) == {'PRINTER-A', 'PRINTER-B'}
assert rows['PRINTER-A'].portaddress == ADDRESS_A
assert rows['PRINTER-A'].drivername == DRIVER_NAME
assert rows['PRINTER-A'].isdefault is True
assert rows['PRINTER-B'].isdefault is False
def test_a_second_report_replaces_the_first(client, db, collector_key, scene):
"""The latest report is the whole truth for that host.
Accumulating instead would grow a row per queue per GE-Enforce cycle forever
and, worse, answer "what does this bay have" with every queue it has ever
had - so a printer removed from a bay would look installed for the rest of
the site's life.
"""
first = _report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_A),
_queue('PRINTER-B', ADDRESS_B),
])
assert first.status_code == 200, first.get_json()
second = _report(client, BAY_HOST, [_queue('PRINTER-A', ADDRESS_A)])
assert second.status_code == 200, second.get_json()
assert _queuenames(BAY_HOST) == {'PRINTER-A'}
def test_a_report_replaces_only_the_reporting_host(client, db, collector_key, scene):
"""One bay's report must not touch another bay's rows.
Replacement keyed on anything wider than the hostname turns every cycle into
a race: whichever PC reported last would be the only one ShopDB believes has
any printers at all.
"""
_report(client, BAY_HOST, [_queue('PRINTER-A', ADDRESS_A)])
_report(client, SECOND_BAY_HOST, [_queue('PRINTER-B', ADDRESS_B)])
assert _queuenames(BAY_HOST) == {'PRINTER-A'}
assert _queuenames(SECOND_BAY_HOST) == {'PRINTER-B'}
def test_an_empty_queue_list_clears_the_host(client, db, collector_key, scene):
"""A host that genuinely has no printers reports that, and it takes effect.
This is the counterpart of the rule below: [] is a real observation and must
wipe the previous set, or a printer removed from a bay stays visible in
ShopDB forever.
"""
_report(client, BAY_HOST, [_queue('PRINTER-A', ADDRESS_A)])
cleared = _report(client, BAY_HOST, [])
assert cleared.status_code == 200, cleared.get_json()
assert _stored(BAY_HOST) == []
def test_a_report_with_no_queues_key_is_rejected_and_changes_nothing(
client, db, collector_key, scene):
"""Absent is not empty, and the difference is the whole safety margin.
A client whose enumeration failed must send nothing. If a malformed report
with no queues key were treated as "this host has none", one client bug
would erase the observed state of the fleet host by host, quietly, at
collector cadence.
"""
_report(client, BAY_HOST, [_queue('PRINTER-A', ADDRESS_A)])
response = client.post(COLLECT_URL, json={'hostname': BAY_HOST},
headers={'X-API-Key': KEY})
assert response.status_code == 400
assert _queuenames(BAY_HOST) == {'PRINTER-A'}
def test_an_unknown_hostname_warns_instead_of_failing(client, db, collector_key,
scene):
"""A bay ShopDB has no PC record for still gets to report.
Reporting before enrollment is normal on a fresh build, and the rows are
keyed by hostname so they resolve the moment the record appears. Failing the
call instead would make the client log an error on every cycle on every
unenrolled bay, and real failures would drown in it.
"""
response = _report(client, 'NOSUCHHOST', [_queue('PRINTER-A', ADDRESS_A)])
assert response.status_code == 200, response.get_json()
data = response.get_json()['data']
assert data['warnings'], 'an unresolvable hostname reported no warning'
assert any('nosuchhost' in warning.lower() for warning in data['warnings'])
assert _queuenames('NOSUCHHOST') == {'PRINTER-A'}
def test_a_report_never_changes_an_assignment(client, db, collector_key, scene):
"""The separation this whole design rests on.
The bay is assigned printer A and reports B and C instead - the exact drift
the feature exists to show. Not one assignment row may move. If observed
state could write the assigned side, a misconfigured bay would rewrite its
own orders on its next check-in, drift would self-heal into permanence, and
/api/printers/for-host would stop meaning "what this host should have".
"""
_assign(db, scene, scene['machine'], ['A'], default='A')
_controls(db, scene, scene['baypc'])
before = _assignment_rows(scene)
response = _report(client, BAY_HOST, [
_queue('PRINTER-B', ADDRESS_B, isdefault=True),
_queue('PRINTER-C', ADDRESS_C),
])
assert response.status_code == 200, response.get_json()
assert _assignment_rows(scene) == before
# And the host is still told to install exactly what it was told before.
resolved = client.get(HOST_URL % BAY_HOST)
assert resolved.status_code == 200
assigned = resolved.get_json()['data']['printers']
assert [row['assetid'] for row in assigned] == [_assetid(scene, 'A')]
# ---------------------------------------------------------------------------
# Comparison
# ---------------------------------------------------------------------------
def test_comparison_requires_authentication(client, db, scene):
"""Observed state is internal detail, not a machine-readable public feed.
The collector endpoint has its own key auth for unattended clients; this
read is for people, so it goes through the normal login. Left open, a bay's
installed-software-adjacent inventory would be readable by anyone who can
reach the API.
"""
response = client.get(OBSERVED_URL % scene['baypc'].assetid)
assert response.status_code == 401
def test_comparison_classifies_matching_missing_and_extra(
client, db, collector_key, scene, auth_headers):
"""The three plain answers, in one bay.
A is assigned and observed (matching), B is assigned and absent (missing),
C is observed and never assigned (extra). Collapsing any of these into the
others is what makes a comparison view worthless: missing is a bay that
never converged, extra is a printer somebody added by hand, and reading one
as the other sends a technician to the wrong problem.
"""
_assign(db, scene, scene['machine'], ['A', 'B'], default='A')
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_A, isdefault=True),
_queue('PRINTER-C', ADDRESS_C),
])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
assert response.status_code == 200, response.get_json()
classified = _classified(_oneblock(response, BAY_HOST))
assert classified.get('matching') == {_assetid(scene, 'A')}
assert classified.get('missing') == {_assetid(scene, 'B')}
assert classified.get('extra') == {_assetid(scene, 'C')}
def test_a_queue_pointing_at_the_wrong_address_is_drifted(
client, db, collector_key, scene, auth_headers):
"""Right printer, wrong port: drifted, not matching.
The queue carries the assigned printer's name but prints to an address that
is not that printer's. Called matching, the bay reads as converged while its
jobs come out somewhere else - the failure that is invisible from the server
and obvious to whoever is standing at the machine.
"""
_assign(db, scene, scene['machine'], ['A'], default='A')
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_NOBODY, isdefault=True),
])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
assert response.status_code == 200, response.get_json()
classified = _classified(_oneblock(response, BAY_HOST))
assert classified.get('drifted') == {_assetid(scene, 'A')}
assert not classified.get('matching')
def test_a_queue_on_the_wrong_driver_is_drifted(
client, db, collector_key, scene, auth_headers):
"""Right printer, right port, driver nobody assigned: still drifted.
Drift is the whole reason Set-ShopdbPrinters repairs queues instead of only
creating them. A queue left on a driver the register does not name is the
case that prints, badly - wrong tray, wrong duplex, wrong paper - so it must
not read as converged.
"""
_assign(db, scene, scene['machine'], ['A'], default='A')
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_A, drivername='Some Other Driver',
isdefault=True),
])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
assert response.status_code == 200, response.get_json()
classified = _classified(_oneblock(response, BAY_HOST))
assert classified.get('drifted') == {_assetid(scene, 'A')}
assert not classified.get('matching')
def test_port_address_beats_a_colliding_queue_name(
client, db, collector_key, scene, auth_headers):
"""When the two match keys disagree, the address wins.
A queue name is a convention a technician typed; an address identifies a
device. A bay that named its queue after one printer while pointing it at
another is precisely the mistake this view exists to surface, and matching
on the name would report the mistake as agreement.
"""
_assign(db, scene, scene['machine'], ['A'], default='A')
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [_queue('PRINTER-B', ADDRESS_A, isdefault=True)])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
assert response.status_code == 200, response.get_json()
block = _oneblock(response, BAY_HOST)
resolved = {row.get('queuename'): row.get('printerassetid')
for row in (block.get('queues') or [])}
assert resolved.get('PRINTER-B') == _assetid(scene, 'A')
def test_a_queue_matching_no_printer_is_unknown_not_guessed(
client, db, collector_key, scene, auth_headers):
"""No match is reported as no match.
Nothing in ShopDB carries this name or this address. A fuzzy fallback that
reached for the nearest printer would put a wrong assetid in front of a
reviewer, and that reviewer's next click writes it into an assignment the
client then installs on every cycle. Unknown costs one conversation; a wrong
match costs a bay.
"""
_assign(db, scene, scene['machine'], ['A'], default='A')
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_A, isdefault=True),
_queue('Reception Copier', ADDRESS_NOBODY),
])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
assert response.status_code == 200, response.get_json()
block = _oneblock(response, BAY_HOST)
classified = _classified(block)
assert classified.get('unknown') == {'Reception Copier'}
# Unknown is not a quiet flavour of extra: extra means "resolved to a
# printer nobody assigned", which is a different conversation.
assert 'Reception Copier' not in classified.get('extra', set())
unmatched = next(row for row in block['queues']
if row.get('queuename') == 'Reception Copier')
assert unmatched.get('printerassetid') is None
def test_a_machine_answers_per_controlling_host(
client, db, collector_key, scene, auth_headers):
"""The assignment lives on the machine; the observations live on the PCs.
A dualpath pair or a part marker legitimately puts two PCs on one machine.
Merging their queues into one list would hide WHICH bay drifted, and the
only actionable thing about drift is which box to walk to.
"""
_assign(db, scene, scene['machine'], ['A'], default='A')
_controls(db, scene, scene['baypc'])
_controls(db, scene, scene['secondbaypc'])
_report(client, BAY_HOST, [_queue('PRINTER-A', ADDRESS_A, isdefault=True)])
_report(client, SECOND_BAY_HOST, [_queue('PRINTER-C', ADDRESS_C)])
response = client.get(OBSERVED_URL % scene['machine'].assetid,
headers=auth_headers)
assert response.status_code == 200, response.get_json()
blocks = _hostblocks(response)
assert {BAY_HOST.lower(), SECOND_BAY_HOST.lower()} <= set(blocks)
converged = _classified(blocks[BAY_HOST.lower()])
drifted = _classified(blocks[SECOND_BAY_HOST.lower()])
assert converged.get('matching') == {_assetid(scene, 'A')}
assert drifted.get('missing') == {_assetid(scene, 'A')}
assert drifted.get('extra') == {_assetid(scene, 'C')}
def test_comparison_reports_when_the_host_last_reported(
client, db, collector_key, scene, auth_headers):
"""A block with no timestamp cannot be trusted.
Observed state is only ever as good as its age: a bay that stopped reporting
six months ago and a bay that reported this morning produce identical
comparisons, and only the timestamp tells a reviewer which one is worth
acting on.
"""
_assign(db, scene, scene['machine'], ['A'], default='A')
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [_queue('PRINTER-A', ADDRESS_A, isdefault=True)])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
block = _oneblock(response, BAY_HOST)
# Server-stamped at ingest, so a bay with a wrong clock cannot report itself
# fresh. Either spelling of the key is the ingest stamp.
stamp = block.get('reportedat') or block.get('observedat')
assert stamp, 'no report timestamp on the host block: %s' % sorted(block)
def test_a_host_that_has_never_reported_is_empty_not_an_error(
client, db, scene, auth_headers):
"""Silence is a legitimate answer.
Most PCs will not have reported yet the day this ships. A 404 or a 500 here
would break the asset page for every one of them, and the page is where the
assignment is edited.
"""
_assign(db, scene, scene['machine'], ['A'], default='A')
_controls(db, scene, scene['baypc'])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
assert response.status_code == 200, response.get_json()
classified = _classified(_oneblock(response, BAY_HOST))
# Everything assigned is missing, and nothing was observed.
assert classified.get('missing') == {_assetid(scene, 'A')}
assert not classified.get('matching')
assert not classified.get('extra')
assert not classified.get('unknown')
# ---------------------------------------------------------------------------
# Seeding an assignment from what was observed
# ---------------------------------------------------------------------------
def test_seedcandidate_offers_matched_queues_and_names_what_it_skipped(
client, db, collector_key, scene, auth_headers):
"""The seed is a proposal made of matches only, and it says what it left out.
Unknown queues are never seeded - that is the never-guess rule reaching the
write path. But dropping them silently is its own failure: the reviewer sees
a short list, assumes the bay only has those, and the real queue goes
unrecorded with nothing anywhere saying it was skipped.
"""
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_A, isdefault=True),
_queue('Reception Copier', ADDRESS_NOBODY),
])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
assert response.status_code == 200, response.get_json()
seed = _seedcandidate(_oneblock(response, BAY_HOST))
assert list(seed['printerassetids']) == [_assetid(scene, 'A')]
assert 'reception copier' in _skippedtext(seed)
def test_seedcandidate_leaves_the_default_unset_when_it_cannot_be_matched(
client, db, collector_key, scene, auth_headers):
"""An unmatched default seeds no default at all.
The alternative is picking one of the matched queues so the field is not
blank, which would change a user's default printer on the strength of a
guess. No default is a state the client already handles quietly; a wrong one
is a support call.
"""
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_A),
_queue('Reception Copier', ADDRESS_NOBODY, isdefault=True),
])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
seed = _seedcandidate(_oneblock(response, BAY_HOST))
assert list(seed['printerassetids']) == [_assetid(scene, 'A')]
assert seed['defaultprinterassetid'] is None
def test_seedcandidate_carries_the_default_when_it_matched(
client, db, collector_key, scene, auth_headers):
"""A matched default is offered, so the common case is one click.
If the default were never proposed, every seeded bay would come back later
for a second edit, and the half-seeded assignments in between are exactly
the state that makes the register untrustworthy.
"""
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_A, isdefault=True),
_queue('PRINTER-B', ADDRESS_B),
])
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
seed = _seedcandidate(_oneblock(response, BAY_HOST))
assert set(seed['printerassetids']) == {_assetid(scene, 'A'),
_assetid(scene, 'B')}
assert seed['defaultprinterassetid'] == _assetid(scene, 'A')
def test_reading_the_seedcandidate_writes_no_assignment(
client, db, collector_key, scene, auth_headers):
"""Offering is not applying.
A candidate that wrote itself on read would make every visit to an asset
page adopt whatever that bay happened to have - the observed side quietly
becoming the assigned side, which is the one thing this design forbids.
"""
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [_queue('PRINTER-A', ADDRESS_A, isdefault=True)])
before = _assignment_rows(scene)
response = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
assert response.status_code == 200, response.get_json()
assert _assignment_rows(scene) == before
stored = client.get(ASSIGN_URL % scene['machine'].assetid,
headers=auth_headers)
assert stored.get_json()['data']['printerassetids'] == []
def test_seeding_writes_the_assignment_only_when_a_person_saves_it(
client, db, collector_key, scene, auth_headers):
"""The seed is saved through the one existing write path, by hand.
Routing it through PUT /api/printers/assignments/for-asset keeps a single
place where an assignment is written, so the reconcile rules - the
default-must-be-in-the-set check, the soft delete, the printer-type
validation - cannot be bypassed by a seed that grew its own endpoint.
"""
_controls(db, scene, scene['baypc'])
_report(client, BAY_HOST, [
_queue('PRINTER-A', ADDRESS_A, isdefault=True),
_queue('PRINTER-B', ADDRESS_B),
_queue('Reception Copier', ADDRESS_NOBODY),
])
observed = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
assert observed.status_code == 200, observed.get_json()
seed = _seedcandidate(_oneblock(observed, BAY_HOST))
# Seeded onto the MACHINE, which is where an assignment belongs: seeding the
# PC would create own rows that permanently shadow the bay's and quietly
# defeat reimage inheritance.
saved = client.put(ASSIGN_URL % scene['machine'].assetid,
headers=auth_headers,
json={'printerassetids': list(seed['printerassetids']),
'defaultprinterassetid': seed['defaultprinterassetid']})
assert saved.status_code == 200, saved.get_json()
resolved = client.get(HOST_URL % BAY_HOST)
installed = resolved.get_json()['data']['printers']
assert {row['assetid'] for row in installed} == {_assetid(scene, 'A'),
_assetid(scene, 'B')}
default = [row['assetid'] for row in installed if row['isdefault']]
assert default == [_assetid(scene, 'A')]
# The queue that matched nothing is still not an assignment, and the bay
# still reports it - drift stays visible instead of being adopted.
after = client.get(OBSERVED_URL % scene['baypc'].assetid,
headers=auth_headers)
classified = _classified(_oneblock(after, BAY_HOST))
assert classified.get('unknown') == {'Reception Copier'}
def test_a_host_that_changes_spelling_does_not_double_its_queues(client, db,
scene,
collector_key):
"""The replace must cover every spelling of one host.
A PC enrolled short can later report its FQDN, or the other way round. The
READ path already treats those as the same machine, so a delete matching
only the exact string left the other spelling's rows behind, and the bay
appeared to have every queue twice - which reads as drift that is not there,
and would be adopted as a duplicate assignment.
"""
_report(client, 'OBSPC01', [_queue('CSF01-HP', address='10.0.0.5')])
_report(client, 'obspc01.example.net', [_queue('CSF01-HP', address='10.0.0.5')])
# Counted across BOTH spellings, because the second report is stored under
# the name it sent. What must be true is that one physical host holds one
# row set, whichever spelling it last used.
held = PrinterObservedQueue.query.filter(
PrinterObservedQueue.hostname.ilike('obspc01%')).all()
assert [row.queuename for row in held] == ['CSF01-HP'], (
'the same queue was stored twice under two spellings of one host')