Tech Tools section with a code generator, toner report that names the part, toner forecasting, warranty proof of cover, application uploads
This commit is contained in:
165
tests/test_core/test_application_uploads.py
Normal file
165
tests/test_core/test_application_uploads.py
Normal file
@@ -0,0 +1,165 @@
|
||||
"""Tests for application image and installer uploads.
|
||||
|
||||
The form used to take an image FILENAME and expect someone to drop the file into
|
||||
the frontend's own directory by hand, and an install path typed from memory.
|
||||
Both are uploads now, following the model-image trio.
|
||||
|
||||
The two differ deliberately and that is most of what is pinned here: the image
|
||||
is public because application tiles render before anything is authenticated, the
|
||||
installer is not, because it is licensed vendor software.
|
||||
"""
|
||||
|
||||
import io
|
||||
|
||||
from shopdb.core.models import Application
|
||||
|
||||
|
||||
def _app(db, name='Test App'):
|
||||
app = Application(appname=name)
|
||||
db.session.add(app)
|
||||
db.session.commit()
|
||||
return app
|
||||
|
||||
|
||||
def _png():
|
||||
# 1x1 PNG; enough to exercise the path without a fixture file.
|
||||
return io.BytesIO(
|
||||
b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01'
|
||||
b'\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00'
|
||||
b'\x01\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82')
|
||||
|
||||
|
||||
def test_image_upload_sets_the_served_url(client, db, auth_headers):
|
||||
app = _app(db)
|
||||
|
||||
response = client.post(f'/api/applications/{app.appid}/image',
|
||||
data={'file': (_png(), 'logo.png')},
|
||||
content_type='multipart/form-data',
|
||||
headers=auth_headers)
|
||||
|
||||
assert response.status_code == 200, response.get_json()
|
||||
assert response.get_json()['data']['image'] == \
|
||||
f'/api/applications/image/application-{app.appid}.png'
|
||||
|
||||
|
||||
def test_image_is_readable_without_a_token(client, db, auth_headers):
|
||||
"""Tiles and lists render the image before anything is authenticated."""
|
||||
app = _app(db)
|
||||
client.post(f'/api/applications/{app.appid}/image',
|
||||
data={'file': (_png(), 'logo.png')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
|
||||
anon = client.get(f'/api/applications/image/application-{app.appid}.png')
|
||||
|
||||
assert anon.status_code == 200
|
||||
assert anon.mimetype == 'image/png'
|
||||
|
||||
|
||||
def test_image_rejects_a_non_image(client, db, auth_headers):
|
||||
app = _app(db)
|
||||
|
||||
response = client.post(f'/api/applications/{app.appid}/image',
|
||||
data={'file': (io.BytesIO(b'MZ'), 'payload.exe')},
|
||||
content_type='multipart/form-data',
|
||||
headers=auth_headers)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'Unsupported image type' in response.get_data(as_text=True)
|
||||
|
||||
|
||||
def test_installer_upload_sets_the_install_path(client, db, auth_headers):
|
||||
app = _app(db)
|
||||
|
||||
response = client.post(f'/api/applications/{app.appid}/package',
|
||||
data={'file': (io.BytesIO(b'fake installer'), 'setup.msi')},
|
||||
content_type='multipart/form-data',
|
||||
headers=auth_headers)
|
||||
|
||||
assert response.status_code == 200, response.get_json()
|
||||
body = response.get_json()['data']
|
||||
assert body['installpath'] == \
|
||||
f'/api/applications/package/application-{app.appid}.msi'
|
||||
# The name the vendor shipped is reported back, since the stored copy is
|
||||
# renamed and the uploader needs to recognise what they sent.
|
||||
assert body['uploadedfilename'] == 'setup.msi'
|
||||
|
||||
|
||||
def test_installer_download_needs_a_token(client, db, auth_headers):
|
||||
"""Licensed software: an open URL would publish it to the whole network."""
|
||||
app = _app(db)
|
||||
client.post(f'/api/applications/{app.appid}/package',
|
||||
data={'file': (io.BytesIO(b'fake installer'), 'setup.msi')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
path = f'/api/applications/package/application-{app.appid}.msi'
|
||||
|
||||
assert client.get(path).status_code == 401
|
||||
|
||||
authed = client.get(path, headers=auth_headers)
|
||||
assert authed.status_code == 200
|
||||
assert authed.data == b'fake installer'
|
||||
# Sent as an attachment so a browser saves it rather than rendering it.
|
||||
assert 'attachment' in authed.headers.get('Content-Disposition', '')
|
||||
|
||||
|
||||
def test_installer_rejects_an_unlisted_type(client, db, auth_headers):
|
||||
app = _app(db)
|
||||
|
||||
response = client.post(f'/api/applications/{app.appid}/package',
|
||||
data={'file': (io.BytesIO(b'x'), 'driver.dll')},
|
||||
content_type='multipart/form-data',
|
||||
headers=auth_headers)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'Unsupported installer type' in response.get_data(as_text=True)
|
||||
|
||||
|
||||
def test_installer_rejects_an_oversize_file(client, db, auth_headers, monkeypatch):
|
||||
"""The cap keeps an ISO out of the instance directory."""
|
||||
from shopdb.core.api import applications as applications_api
|
||||
monkeypatch.setattr(applications_api, 'MAX_PACKAGE_BYTES', 1024)
|
||||
app = _app(db)
|
||||
|
||||
response = client.post(f'/api/applications/{app.appid}/package',
|
||||
data={'file': (io.BytesIO(b'x' * 4096), 'big.msi')},
|
||||
content_type='multipart/form-data',
|
||||
headers=auth_headers)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'the limit is' in response.get_data(as_text=True)
|
||||
|
||||
|
||||
def test_removing_an_installer_clears_only_an_uploaded_path(client, db, auth_headers):
|
||||
"""A share path was typed by a person and is not ours to wipe."""
|
||||
app = _app(db)
|
||||
client.post(f'/api/applications/{app.appid}/package',
|
||||
data={'file': (io.BytesIO(b'installer'), 'setup.msi')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
|
||||
cleared = client.delete(f'/api/applications/{app.appid}/package',
|
||||
headers=auth_headers)
|
||||
assert cleared.get_json()['data']['installpath'] is None
|
||||
|
||||
app.installpath = r'\\share\installers\vendor.msi'
|
||||
db.session.commit()
|
||||
again = client.delete(f'/api/applications/{app.appid}/package',
|
||||
headers=auth_headers)
|
||||
assert again.get_json()['data']['installpath'] == r'\\share\installers\vendor.msi'
|
||||
|
||||
|
||||
def test_replacing_an_image_does_not_orphan_the_old_extension(client, db, auth_headers):
|
||||
"""One image per application, whatever the extension arrives as."""
|
||||
import glob
|
||||
import os
|
||||
from flask import current_app
|
||||
app = _app(db)
|
||||
|
||||
client.post(f'/api/applications/{app.appid}/image',
|
||||
data={'file': (_png(), 'logo.png')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
client.post(f'/api/applications/{app.appid}/image',
|
||||
data={'file': (io.BytesIO(b'<svg/>'), 'logo.svg')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
|
||||
imagedir = os.path.join(current_app.instance_path, 'applicationimages')
|
||||
files = glob.glob(os.path.join(imagedir, f'application-{app.appid}.*'))
|
||||
assert len(files) == 1 and files[0].endswith('.svg')
|
||||
@@ -71,6 +71,8 @@ EXPECTED_HEAD_REVISION['printedparts'] = 'printedparts0004txnrev'
|
||||
# notifications indexes businessunitid, then adds the per-type grace window and
|
||||
# the shared board category.
|
||||
EXPECTED_HEAD_REVISION['notifications'] = 'notifications0005boardorder'
|
||||
# warranty adds the proof-of-cover document columns on top of its anchor.
|
||||
EXPECTED_HEAD_REVISION['warranty'] = 'warranty0002proof'
|
||||
|
||||
# Plugins built after the cutover: their 0001 baseline really creates tables the
|
||||
# core chain never owned.
|
||||
@@ -239,9 +241,21 @@ def test_upgrade_all_on_fresh_db_is_clean_and_idempotent(tmp_path, monkeypatch):
|
||||
first = app.extensions['plugin_manager'].upgrade_all_plugins()
|
||||
second = app.extensions['plugin_manager'].upgrade_all_plugins()
|
||||
|
||||
assert set(first) == set(PLUGIN_TABLE_OWNERS)
|
||||
assert all(status == 'ok' for status in first.values()), first
|
||||
assert all(status == 'ok' for status in second.values()), second
|
||||
# A plugin that owns no tables (Tech Tools: every tool is
|
||||
# client-side) carries no chain, so upgrade-all reports it
|
||||
# 'no-migrations'. Assert that explicitly rather than letting it
|
||||
# widen the 'ok' check, which would also swallow a table-owning
|
||||
# plugin whose chain silently went missing.
|
||||
schemaless = set(first) - set(PLUGIN_TABLE_OWNERS)
|
||||
for name in schemaless:
|
||||
assert first[name] == 'no-migrations', (name, first[name])
|
||||
assert second[name] == 'no-migrations', (name, second[name])
|
||||
|
||||
migrated = {name: status for name, status in first.items()
|
||||
if name in PLUGIN_TABLE_OWNERS}
|
||||
assert set(migrated) == set(PLUGIN_TABLE_OWNERS)
|
||||
assert all(status == 'ok' for status in migrated.values()), migrated
|
||||
assert all(second[name] == 'ok' for name in PLUGIN_TABLE_OWNERS), second
|
||||
|
||||
insp = inspect(db.engine)
|
||||
for plugin in PLUGIN_TABLE_OWNERS:
|
||||
|
||||
130
tests/test_plugins/test_supply_history.py
Normal file
130
tests/test_plugins/test_supply_history.py
Normal file
@@ -0,0 +1,130 @@
|
||||
"""Tests for toner burn-rate and replacement counting.
|
||||
|
||||
The analysis is pure arithmetic over a list of readings, so it is tested that
|
||||
way - no Zabbix, no fixtures. What matters is that it refuses to guess: the
|
||||
cases where an estimate would be dishonest are the ones most likely to reach a
|
||||
purchasing decision.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from plugins.printers.services.supply_history import (
|
||||
analyse, burn_rate, current_run, find_replacements, normalise, soonest,
|
||||
)
|
||||
|
||||
START = datetime(2026, 6, 1, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def series(levels, hours=24):
|
||||
"""[(clock, value)] one reading per `hours`, as Zabbix returns them."""
|
||||
return [(str(int((START + timedelta(hours=i * hours)).timestamp())), str(v))
|
||||
for i, v in enumerate(levels)]
|
||||
|
||||
|
||||
def test_counts_one_replacement_per_upward_step():
|
||||
points = normalise(series([80, 60, 40, 20, 100, 80, 60, 95, 70]))
|
||||
|
||||
assert len(find_replacements(points)) == 2
|
||||
|
||||
|
||||
def test_ignores_a_wobble_that_is_not_a_replacement():
|
||||
"""SNMP rounding and a gauge settling both nudge a reading upward."""
|
||||
points = normalise(series([60, 58, 61, 57, 55]))
|
||||
|
||||
assert find_replacements(points) == []
|
||||
|
||||
|
||||
def test_estimate_uses_only_the_current_cartridge():
|
||||
"""Fitting across a swap averages a spent cartridge with a fresh one."""
|
||||
points = normalise(series([90, 60, 30, 100, 90, 80, 70]))
|
||||
|
||||
run = current_run(points)
|
||||
|
||||
assert [level for _, level in run] == [100, 90, 80, 70]
|
||||
|
||||
|
||||
def test_days_left_from_a_steady_drain():
|
||||
# 10 points a day apart, 5% a day, ending at 55%
|
||||
result = analyse(series([100, 95, 90, 85, 80, 75, 70, 65, 60, 55]))
|
||||
|
||||
assert result['burnrateperday'] == 5.0
|
||||
assert result['daysleft'] == 11 # 55 / 5
|
||||
assert result['reason'] is None
|
||||
assert result['basisdays'] == 9.0
|
||||
|
||||
|
||||
def test_no_estimate_from_two_readings():
|
||||
"""Two points through a coarse gauge can prove any rate at all."""
|
||||
result = analyse(series([100, 90]))
|
||||
|
||||
assert result['daysleft'] is None
|
||||
assert result['reason'] == 'not enough history yet'
|
||||
|
||||
|
||||
def test_no_estimate_while_the_gauge_has_not_moved():
|
||||
"""A printer reporting in 10% steps sits on a plateau for a fortnight."""
|
||||
result = analyse(series([70, 70, 70, 70, 70, 70]))
|
||||
|
||||
assert result['daysleft'] is None
|
||||
assert result['reason'] == 'level has not moved enough to estimate'
|
||||
|
||||
|
||||
def test_a_recent_replacement_says_so_rather_than_guessing():
|
||||
result = analyse(series([40, 30, 20, 10, 100, 98]))
|
||||
|
||||
assert result['daysleft'] is None
|
||||
assert result['reason'] == 'replaced recently'
|
||||
assert result['replacements'] == 1
|
||||
assert result['lastreplaced'] is not None
|
||||
|
||||
|
||||
def test_empty_history_is_reported_not_crashed():
|
||||
result = analyse([])
|
||||
|
||||
assert result['reason'] == 'no history'
|
||||
assert result['currentlevel'] is None
|
||||
assert result['replacements'] == 0
|
||||
|
||||
|
||||
def test_junk_rows_are_dropped_not_fatal():
|
||||
points = normalise([('notaclock', '50'), ('1780000000', 'n/a'),
|
||||
('1780000000', '50')])
|
||||
|
||||
assert len(points) == 1
|
||||
|
||||
|
||||
def test_days_left_never_goes_negative():
|
||||
result = analyse(series([20, 15, 10, 5, 0]))
|
||||
|
||||
assert result['daysleft'] == 0
|
||||
|
||||
|
||||
def test_printer_sorts_by_its_soonest_supply():
|
||||
"""A colour MFP is as urgent as its most pressing cartridge."""
|
||||
supplies = [{'daysleft': 40}, {'daysleft': 6}, {'daysleft': None}]
|
||||
|
||||
assert soonest(supplies) == 6
|
||||
|
||||
|
||||
def test_soonest_is_none_when_nothing_can_be_estimated():
|
||||
assert soonest([{'daysleft': None}, {'daysleft': None}]) is None
|
||||
|
||||
|
||||
def test_burn_rate_needs_elapsed_time():
|
||||
"""Several readings in the same second is not a rate."""
|
||||
clock = str(int(START.timestamp()))
|
||||
points = normalise([(clock, '90'), (clock, '80'), (clock, '70'), (clock, '60')])
|
||||
|
||||
assert burn_rate(points) is None
|
||||
|
||||
|
||||
def test_forecast_endpoint_says_so_when_zabbix_is_absent(client, db):
|
||||
"""A dashboard must be told the data is missing, not shown an empty list
|
||||
that reads as 'nothing runs out soon'."""
|
||||
response = client.get('/api/printers/supplies/forecast')
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.get_json()['data']
|
||||
assert body['available'] is False
|
||||
assert 'not configured' in body['reason']
|
||||
assert body['printers'] == []
|
||||
55
tests/test_plugins/test_tools.py
Normal file
55
tests/test_plugins/test_tools.py
Normal file
@@ -0,0 +1,55 @@
|
||||
"""Tests for the Tech Tools plugin.
|
||||
|
||||
The plugin owns no API and no tables, so the whole backend contract is: it
|
||||
loads, it advertises one nav entry, and it claims no database. That last part
|
||||
matters - a plugin that accidentally returns models would pull the tools
|
||||
section into the migration and prune-schema machinery it has no business in.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
|
||||
from plugins.tools.plugin import ToolsPlugin
|
||||
|
||||
|
||||
PLUGIN_DIR = os.path.join(os.path.dirname(os.path.dirname(
|
||||
os.path.dirname(os.path.abspath(__file__)))), 'plugins', 'tools')
|
||||
|
||||
|
||||
def test_meta_matches_manifest():
|
||||
"""The manifest is the single source of truth per ADR-002."""
|
||||
with open(os.path.join(PLUGIN_DIR, 'manifest.json'), encoding='utf-8') as handle:
|
||||
manifest = json.load(handle)
|
||||
|
||||
meta = ToolsPlugin().meta
|
||||
assert meta.name == manifest['name']
|
||||
assert meta.version == manifest['version']
|
||||
assert meta.core_version == manifest['core_version']
|
||||
|
||||
|
||||
def test_no_backend_surface():
|
||||
"""Client-side only: no blueprint to register, no tables to migrate."""
|
||||
plugin = ToolsPlugin()
|
||||
assert plugin.get_blueprint() is None
|
||||
assert plugin.get_models() == []
|
||||
|
||||
|
||||
def test_navigation_item():
|
||||
items = ToolsPlugin().get_navigation_items()
|
||||
assert len(items) == 1
|
||||
item = items[0]
|
||||
assert item['name'] == 'Tech Tools'
|
||||
assert item['route'] == '/tools'
|
||||
# The icon has to exist in AppLayout's iconMap or the nav entry renders
|
||||
# with no glyph.
|
||||
assert item['icon'] == 'wrench'
|
||||
|
||||
|
||||
def test_nav_icon_is_mapped_in_the_frontend():
|
||||
"""Guard the one cross-file coupling: nav icon name -> AppLayout iconMap."""
|
||||
layout = os.path.join(os.path.dirname(PLUGIN_DIR), '..', 'frontend', 'src',
|
||||
'views', 'AppLayout.vue')
|
||||
with open(os.path.normpath(layout), encoding='utf-8') as handle:
|
||||
source = handle.read()
|
||||
for item in ToolsPlugin().get_navigation_items():
|
||||
assert f"'{item['icon']}':" in source
|
||||
110
tests/test_plugins/test_warranty_proof.py
Normal file
110
tests/test_plugins/test_warranty_proof.py
Normal file
@@ -0,0 +1,110 @@
|
||||
"""Tests for the warranty proof-of-cover document.
|
||||
|
||||
A provider lookup answers whether a unit is covered; it does not produce the
|
||||
invoice or the extended-warranty certificate. Those had nowhere to live, so
|
||||
they stayed in somebody's mailbox. These pin the parts that are easy to get
|
||||
wrong: the document is authenticated on the way out, one per warranty, and the
|
||||
vendor's own filename survives so a download is recognisable.
|
||||
"""
|
||||
|
||||
import io
|
||||
|
||||
from plugins.warranty.models import Warranty
|
||||
|
||||
|
||||
def _warranty(db, vendor='Dell'):
|
||||
w = Warranty(vendor=vendor, servicetag='ABC123', provider='manual')
|
||||
db.session.add(w)
|
||||
db.session.commit()
|
||||
return w
|
||||
|
||||
|
||||
def test_upload_sets_url_and_keeps_the_original_filename(client, db, auth_headers):
|
||||
w = _warranty(db)
|
||||
|
||||
response = client.post(f'/api/warranty/{w.warrantyid}/proof',
|
||||
data={'file': (io.BytesIO(b'%PDF-1.4 invoice'),
|
||||
'Dell invoice 4471.pdf')},
|
||||
content_type='multipart/form-data',
|
||||
headers=auth_headers)
|
||||
|
||||
assert response.status_code == 200, response.get_json()
|
||||
body = response.get_json()['data']
|
||||
assert body['proofurl'] == f'/api/warranty/proof/warranty-{w.warrantyid}.pdf'
|
||||
assert body['prooffilename'] == 'Dell invoice 4471.pdf'
|
||||
|
||||
|
||||
def test_download_needs_a_token_and_uses_the_vendor_filename(client, db, auth_headers):
|
||||
"""An invoice carries pricing and a service tag - not open like an asset photo."""
|
||||
w = _warranty(db)
|
||||
client.post(f'/api/warranty/{w.warrantyid}/proof',
|
||||
data={'file': (io.BytesIO(b'%PDF-1.4 invoice'), 'Dell invoice 4471.pdf')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
path = f'/api/warranty/proof/warranty-{w.warrantyid}.pdf'
|
||||
|
||||
assert client.get(path).status_code == 401
|
||||
|
||||
authed = client.get(path, headers=auth_headers)
|
||||
assert authed.status_code == 200
|
||||
assert authed.data == b'%PDF-1.4 invoice'
|
||||
disposition = authed.headers.get('Content-Disposition', '')
|
||||
assert 'attachment' in disposition
|
||||
assert 'Dell invoice 4471.pdf' in disposition
|
||||
|
||||
|
||||
def test_rejects_an_unsupported_type(client, db, auth_headers):
|
||||
w = _warranty(db)
|
||||
|
||||
response = client.post(f'/api/warranty/{w.warrantyid}/proof',
|
||||
data={'file': (io.BytesIO(b'MZ'), 'setup.exe')},
|
||||
content_type='multipart/form-data',
|
||||
headers=auth_headers)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'Unsupported document type' in response.get_data(as_text=True)
|
||||
|
||||
|
||||
def test_rejects_an_oversize_document(client, db, auth_headers, monkeypatch):
|
||||
from plugins.warranty.api import routes
|
||||
monkeypatch.setattr(routes, 'MAX_PROOF_BYTES', 512)
|
||||
w = _warranty(db)
|
||||
|
||||
response = client.post(f'/api/warranty/{w.warrantyid}/proof',
|
||||
data={'file': (io.BytesIO(b'x' * 2048), 'scan.pdf')},
|
||||
content_type='multipart/form-data',
|
||||
headers=auth_headers)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'the limit is' in response.get_data(as_text=True)
|
||||
|
||||
|
||||
def test_replacing_keeps_one_document_per_warranty(client, db, auth_headers):
|
||||
import glob
|
||||
import os
|
||||
from flask import current_app
|
||||
w = _warranty(db)
|
||||
|
||||
client.post(f'/api/warranty/{w.warrantyid}/proof',
|
||||
data={'file': (io.BytesIO(b'first'), 'scan.png')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
client.post(f'/api/warranty/{w.warrantyid}/proof',
|
||||
data={'file': (io.BytesIO(b'%PDF second'), 'invoice.pdf')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
|
||||
proofdir = os.path.join(current_app.instance_path, 'warrantyproofs')
|
||||
files = glob.glob(os.path.join(proofdir, f'warranty-{w.warrantyid}.*'))
|
||||
assert len(files) == 1 and files[0].endswith('.pdf')
|
||||
|
||||
|
||||
def test_removing_clears_both_columns(client, db, auth_headers):
|
||||
w = _warranty(db)
|
||||
client.post(f'/api/warranty/{w.warrantyid}/proof',
|
||||
data={'file': (io.BytesIO(b'%PDF'), 'invoice.pdf')},
|
||||
content_type='multipart/form-data', headers=auth_headers)
|
||||
|
||||
response = client.delete(f'/api/warranty/{w.warrantyid}/proof',
|
||||
headers=auth_headers)
|
||||
|
||||
body = response.get_json()['data']
|
||||
assert body['proofurl'] is None
|
||||
assert body['prooffilename'] is None
|
||||
Reference in New Issue
Block a user