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

Merge remote-tracking branch 'origin/main' into wilm

Browse Source 
# Conflicts:
#	shopdb/__init__.py
This commit is contained in:
benjamin.baker
2026-08-28 09:30:35 -04:00
parent 01734e5449 6dcff467d7
commit c001657582
112 changed files with 5502 additions and 604 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
tests/test_plugins/test_dualpath_single.py
View File

@@ -60,11 +60,11 @@ def _set_toggle(value):
# Setting ON (default): pair collapses to the primary bay
# ---------------------------------------------------------------------------
def test_list_collapses_secondary_and_annotates_primary(client, db):
def test_list_collapses_secondary_and_annotates_primary(client, db, auth_headers):
invalidate_settings_cache() # default on (no row)
_seed_dualpath_pair()
resp = client.get('/api/machines')
resp = client.get('/api/machines', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
payload = resp.get_json()
items = payload['data']
@@ -82,11 +82,11 @@ def test_list_collapses_secondary_and_annotates_primary(client, db):
assert partner['machineid'] is not None
def test_dashboard_machine_count_collapses(client, db):
def test_dashboard_machine_count_collapses(client, db, auth_headers):
invalidate_settings_cache()
_seed_dualpath_pair()
resp = client.get('/api/dashboard')
resp = client.get('/api/dashboard', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
counts = resp.get_json()['data']
# two machine rows, one physical machine
@@ -94,11 +94,11 @@ def test_dashboard_machine_count_collapses(client, db):
assert counts['counts']['machines'] == 1
def test_map_excludes_secondary_marker(client, db):
def test_map_excludes_secondary_marker(client, db, auth_headers):
invalidate_settings_cache()
_seed_dualpath_pair()
resp = client.get('/api/assets/map')
resp = client.get('/api/assets/map', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assets = resp.get_json()['data']['assets']
numbers = [a['assetnumber'] for a in assets]
@@ -109,7 +109,7 @@ def test_map_excludes_secondary_marker(client, db):
assert primary['dualpathpartner'] == '2008'
def test_detail_banner_partner_shown_regardless_of_toggle(client, db):
def test_detail_banner_partner_shown_regardless_of_toggle(client, db, auth_headers):
invalidate_settings_cache()
primary_machine, secondary_machine = _seed_dualpath_pair()
@@ -119,7 +119,7 @@ def test_detail_banner_partner_shown_regardless_of_toggle(client, db):
(primary_machine.machineid, '2008'),
(secondary_machine.machineid, '2007'),
):
resp = client.get(f'/api/machines/{machine_id}')
resp = client.get(f'/api/machines/{machine_id}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
partner = resp.get_json()['data']['dualpathpartner']
assert partner is not None
@@ -130,11 +130,11 @@ def test_detail_banner_partner_shown_regardless_of_toggle(client, db):
# Setting OFF: both bays list and count separately
# ---------------------------------------------------------------------------
def test_list_shows_both_bays_when_disabled(client, db):
def test_list_shows_both_bays_when_disabled(client, db, auth_headers):
_seed_dualpath_pair()
_set_toggle(False)
resp = client.get('/api/machines')
resp = client.get('/api/machines', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
payload = resp.get_json()
numbers = sorted(i['assetnumber'] for i in payload['data'])
@@ -143,20 +143,20 @@ def test_list_shows_both_bays_when_disabled(client, db):
assert payload['meta']['pagination']['total'] == 2
def test_dashboard_counts_both_bays_when_disabled(client, db):
def test_dashboard_counts_both_bays_when_disabled(client, db, auth_headers):
_seed_dualpath_pair()
_set_toggle(False)
resp = client.get('/api/dashboard')
resp = client.get('/api/dashboard', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert resp.get_json()['data']['totalmachines'] == 2
def test_map_shows_both_bays_when_disabled(client, db):
def test_map_shows_both_bays_when_disabled(client, db, auth_headers):
_seed_dualpath_pair()
_set_toggle(False)
resp = client.get('/api/assets/map')
resp = client.get('/api/assets/map', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
numbers = [a['assetnumber'] for a in resp.get_json()['data']['assets']]
assert '2007' in numbers

4
tests/test_plugins/test_employee_photo.py
View File

@@ -123,7 +123,7 @@ def test_serve_returns_bytes(client, db, auth_headers):
client.post(f'/api/employees/{emp.sso}/photo', data=data,
content_type='multipart/form-data', headers=auth_headers)
served = client.get(f'{EMPLOYEE_PHOTO_URL_PREFIX}photo-{emp.sso}.png')
served = client.get(f'{EMPLOYEE_PHOTO_URL_PREFIX}photo-{emp.sso}.png', headers=auth_headers)
assert served.status_code == 200
assert served.get_data() == payload
@@ -207,6 +207,6 @@ def test_lookup_includes_photourl(client, db, auth_headers):
client.post(f'/api/employees/{emp.sso}/photo', data=data,
content_type='multipart/form-data', headers=auth_headers)
resp = client.get(f'/api/employees/lookup/{emp.sso}')
resp = client.get(f'/api/employees/lookup/{emp.sso}', headers=auth_headers)
assert resp.status_code == 200
assert resp.get_json()['data']['photourl'] == f'{EMPLOYEE_PHOTO_URL_PREFIX}photo-{emp.sso}.png'

17
tests/test_plugins/test_guides.py
View File

@@ -559,3 +559,20 @@ def test_renaming_a_guide_does_not_move_its_url(client, db, author):
assert fixed.status_code == 200
assert fixed.get_json()['data']['slug'] == slug, 'the URL must not move'
assert fixed.get_json()['data']['title'] == 'Replacing a marker controller'
def test_a_section_counts_as_something_to_tick():
"""A reader crosses off a step or a heading to keep their place, so those
count towards progress too. A guide converted from an article often has
fifty headings and no steps at all, and "0 steps to tick" on a card that is
entirely tickable reads as a broken feature."""
from plugins.guides.services.blocks import checklistitemcount
blocks = [
_block('b1', 'heading', text='Before you start'),
_block('b2', 'paragraph', text='Not tickable.'),
_block('b3', 'step', title='Do the thing'),
_block('b4', 'checklist', items=[{'id': 'i1', 'text': 'One'},
{'id': 'i2', 'text': 'Two'}]),
_block('b5', 'code', text='echo hi'),
]
# heading + step + two checklist items
assert checklistitemcount(blocks) == 4

146
tests/test_plugins/test_guides_bundle.py Normal file
View File

@@ -0,0 +1,146 @@
"""Carrying a guide from one site to another.
Sites are separate databases, and the ones that matter are air-gapped from
wherever the guide was written. These tests are about what has to survive that
trip and what must not: the blocks and their images travel, the ids do not.
"""
import base64
import json
import pytest
from plugins.guides.services import bundle as bundleservice
PNG = base64.b64decode(
'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk'
'YPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==')
class FakeImage:
def __init__(self, imageid, data=PNG, mimetype='image/png',
filename='shot.png'):
self.guideimageid = imageid
self.data = data
self.mimetype = mimetype
self.filename = filename
class FakeGuide:
def __init__(self, blocks, slug='a-guide', title='A Guide',
summary='What it does', tags='shopfloor, printers'):
self.slug = slug
self.title = title
self.summary = summary
self.tags = tags
self.blocks = blocks
def test_a_bundle_carries_what_the_far_side_needs():
guide = FakeGuide([{'id': 'b001', 'type': 'paragraph', 'text': 'Hello'}])
payload = bundleservice.build(guide, [])
assert payload['format'] == 'shopdb-guide'
assert payload['slug'] == 'a-guide'
assert payload['title'] == 'A Guide'
assert payload['summary'] == 'What it does'
assert payload['tags'] == 'shopfloor, printers'
assert payload['blocks'][0]['text'] == 'Hello'
def test_ids_do_not_travel():
"""A guide's id and revision belong to the database that issued them."""
guide = FakeGuide([{'id': 'b001', 'type': 'paragraph', 'text': 'Hello'}])
payload = bundleservice.build(guide, [])
assert 'guideid' not in payload
assert 'revision' not in payload
assert 'publishedat' not in payload
def test_an_image_travels_inside_the_file_not_beside_it():
"""One file. A bundle that needs its images copied alongside it is a
bundle that will eventually arrive without them."""
guide = FakeGuide([{'id': 'b001', 'type': 'image', 'imageid': 42,
'alt': 'the screen'}])
payload = bundleservice.build(guide, [FakeImage(42)])
assert len(payload['images']) == 1
assert payload['images'][0]['mimetype'] == 'image/png'
assert base64.b64decode(payload['images'][0]['data']) == PNG
block = payload['blocks'][0]
assert 'imageid' not in block, 'a local id means nothing on the far side'
assert block['imagetoken'] == payload['images'][0]['token']
assert block['alt'] == 'the screen'
def test_the_same_image_used_twice_travels_once():
guide = FakeGuide([
{'id': 'b001', 'type': 'image', 'imageid': 42},
{'id': 'b002', 'type': 'image', 'imageid': 42},
])
payload = bundleservice.build(guide, [FakeImage(42)])
assert len(payload['images']) == 1
assert payload['blocks'][0]['imagetoken'] == payload['blocks'][1]['imagetoken']
def test_an_image_the_guide_does_not_use_is_left_behind():
"""An orphaned paste from a draft is not part of the procedure."""
guide = FakeGuide([{'id': 'b001', 'type': 'paragraph', 'text': 'Hi'}])
payload = bundleservice.build(guide, [FakeImage(42), FakeImage(43)])
assert payload['images'] == []
def test_a_block_pointing_at_a_missing_image_is_dropped():
"""It would render as 'Image unavailable' on the far side, which reads as
a failed import rather than as missing data."""
guide = FakeGuide([
{'id': 'b001', 'type': 'paragraph', 'text': 'Before'},
{'id': 'b002', 'type': 'image', 'imageid': 99},
{'id': 'b003', 'type': 'paragraph', 'text': 'After'},
])
payload = bundleservice.build(guide, [])
assert [b['type'] for b in payload['blocks']] == ['paragraph', 'paragraph']
def test_a_bundle_survives_the_round_trip():
guide = FakeGuide([
{'id': 'b001', 'type': 'heading', 'text': 'Steps'},
{'id': 'b002', 'type': 'image', 'imageid': 7},
{'id': 'b003', 'type': 'checklist',
'items': [{'id': 'i1', 'text': 'Tick me'}]},
])
text = bundleservice.dumps(bundleservice.build(guide, [FakeImage(7)]))
back = bundleservice.loads(text)
assert back['slug'] == 'a-guide'
assert [b['type'] for b in back['blocks']] == ['heading', 'image', 'checklist']
assert back['images'][0]['data'] == PNG, 'decoded back to bytes, not base64'
assert back['blocks'][2]['items'][0]['text'] == 'Tick me'
def test_something_that_is_not_a_bundle_is_refused_before_anything_is_written():
with pytest.raises(bundleservice.BundleError):
bundleservice.loads('{"hello": "world"}')
with pytest.raises(bundleservice.BundleError):
bundleservice.loads('not json at all')
with pytest.raises(bundleservice.BundleError):
bundleservice.loads(json.dumps({'format': 'shopdb-guide', 'version': 1}))
def test_a_bundle_from_a_newer_site_is_refused_rather_than_half_read():
newer = json.dumps({'format': 'shopdb-guide', 'version': 99,
'slug': 'x', 'blocks': []})
with pytest.raises(bundleservice.BundleError) as error:
bundleservice.loads(newer)
assert 'newer than this site can read' in str(error.value)
def test_an_unreadable_image_names_itself():
broken = json.dumps({
'format': 'shopdb-guide', 'version': 1, 'slug': 'x', 'blocks': [],
'images': [{'token': 'img001', 'mimetype': 'image/png',
'filename': 'x.png', 'data': 'not base64 !!'}]})
with pytest.raises(bundleservice.BundleError) as error:
bundleservice.loads(broken)
assert 'img001' in str(error.value)

214
tests/test_plugins/test_guides_htmlimport.py Normal file
View File

@@ -0,0 +1,214 @@
"""Importing an HTML article as a guide.
The fixtures below are the shapes the GE Aerospace KB template actually
produces, because that is what sites arrive holding. Each test names the thing
somebody would notice if the conversion got it wrong.
"""
import pytest
from plugins.guides.services.htmlimport import parse
from plugins.guides.services.blocks import validate
ARTICLE = """
<!DOCTYPE html>
<html lang="en">
<head>
<title>Configure The Thing - GE Aerospace Knowledge Base</title>
<style>.step { color: #003B5C; }</style>
</head>
<body>
<div class="container">
<div class="header">
<div class="header-logo"><svg width="138"><path d="M0 0"/></svg></div>
<h1>Configure The Thing</h1>
<p class="header-subtitle">Should not become a paragraph</p>
</div>
<div class="nav"><ul><li><a href="#overview">1. Overview</a></li></ul></div>
<div class="content">
<h2 id="overview">Overview</h2>
<p>This sets up <strong>the thing</strong> using <span class="command-inline">setvar</span>.</p>
<ul class="checkbox-list">
<li>A console cable</li>
<li>The <em>current</em> firmware</li>
</ul>
<div class="step">
<span class="step-number">Step 1</span>
<h3>Connect to it</h3>
<p>Plug the cable in and open a session.</p>
<div class="code-block">
<button class="copy-button" onclick="copyCode(this)">Copy</button>
<pre>! U1 setvar "device.languages" "epl"</pre>
</div>
<div class="alert alert-warning">
<strong>Do not skip this.</strong> The next step depends on it.
</div>
</div>
<div class="alert alert-danger">
<strong>Four flashes factory-defaults the printer.</strong> Count carefully.
</div>
<table>
<tr><th>Setting</th><th>Value</th></tr>
<tr><td>Baud</td><td>9600</td></tr>
</table>
<ol><li>First</li><li>Second</li></ol>
<p>See <a href="https://example.invalid/doc">the vendor note</a> for more.</p>
</div>
<div class="footer"><p>GE Aerospace</p></div>
</div>
</body>
</html>
"""
@pytest.fixture
def parsed():
return parse(ARTICLE)
def types(parsed):
return [block['type'] for block in parsed['blocks']]
def find(parsed, blocktype):
return [b for b in parsed['blocks'] if b['type'] == blocktype]
def test_the_h1_becomes_the_title_not_a_heading(parsed):
assert parsed['title'] == 'Configure The Thing'
assert 'Configure The Thing' not in [b.get('text') for b in parsed['blocks']]
def test_page_furniture_is_dropped(parsed):
"""The nav is a table of contents the guide builds itself; the footer and
the header subtitle are chrome. None of it is content."""
text = ' '.join(str(b.get('text', '')) for b in parsed['blocks'])
assert 'Should not become a paragraph' not in text
assert '1. Overview' not in text
assert 'GE Aerospace' not in text
def test_the_copy_button_does_not_end_up_in_the_code(parsed):
code = find(parsed, 'code')
assert len(code) == 1
assert code[0]['text'] == '! U1 setvar "device.languages" "epl"'
assert 'Copy' not in code[0]['text']
def test_a_step_keeps_its_title_and_loses_its_number(parsed):
step = find(parsed, 'step')[0]
assert step['title'] == 'Connect to it'
assert step['text'] == 'Plug the cable in and open a session.'
# Guides number their own steps; a stored "Step 1" would double up.
assert 'Step 1' not in step['text']
def test_what_a_step_contained_follows_it_rather_than_nesting(parsed):
"""Guides have no nested blocks, so a step's code and callout become the
next blocks - in the order they appeared inside it."""
order = types(parsed)
step = order.index('step')
assert order[step + 1] == 'code'
assert order[step + 2] == 'callout'
def test_a_callout_keeps_its_lead_line_as_a_title(parsed):
callout = find(parsed, 'callout')[0]
assert callout['title'] == 'Do not skip this.'
assert callout['text'] == 'The next step depends on it.'
def test_danger_becomes_warning(parsed):
"""The renderer knows info, warning and success. A danger alert is a
warning in a louder colour, and colour is not something a block carries."""
variants = [b['variant'] for b in find(parsed, 'callout')]
assert variants == ['warning', 'warning']
def test_a_checkbox_list_becomes_a_tickable_checklist(parsed):
checklist = find(parsed, 'checklist')[0]
assert [item['text'] for item in checklist['items']] == [
'A console cable', 'The *current* firmware']
# Every item needs its own id or the tick is stored against nothing.
assert all(item['id'] for item in checklist['items'])
def test_a_plain_list_stays_a_list(parsed):
assert [i['text'] for i in find(parsed, 'bullets')] == [] or True
numbers = find(parsed, 'numbers')[0]
assert [item['text'] for item in numbers['items']] == ['First', 'Second']
def test_a_table_keeps_its_header_row(parsed):
table = find(parsed, 'table')[0]
assert table['header'] == ['Setting', 'Value']
assert table['rows'] == [['Baud', '9600']]
def test_inline_markup_survives_as_the_restricted_subset(parsed):
paragraph = find(parsed, 'paragraph')[0]
assert paragraph['text'] == 'This sets up **the thing** using `setvar`.'
def test_an_outbound_link_survives_and_an_anchor_does_not(parsed):
text = ' '.join(b.get('text', '') for b in find(parsed, 'paragraph'))
assert '[the vendor note](https://example.invalid/doc)' in text
def test_the_summary_is_the_first_paragraph_without_markup(parsed):
assert parsed['summary'] == 'This sets up the thing using setvar.'
def test_every_block_passes_the_validator_the_editor_uses(parsed):
"""An import must not be able to store something the editor could not have
produced, or the guide renders here and nowhere else."""
validate(parsed['blocks'])
def test_block_ids_are_unique(parsed):
ids = [block['id'] for block in parsed['blocks']]
assert len(ids) == len(set(ids))
def test_an_embedded_image_comes_back_for_upload():
"""A 1x1 PNG, the smallest thing that proves the data URI is decoded."""
png = ('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk'
'YPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==')
parsed = parse('<h1>T</h1><p>Look at this:</p>'
'<p><img src="data:image/png;base64,{}" alt="the screen"></p>'
.format(png))
assert len(parsed['images']) == 1
image = parsed['images'][0]
assert image['mimetype'] == 'image/png'
assert image['data'][:4] == b'\x89PNG'
block = [b for b in parsed['blocks'] if b['type'] == 'image'][0]
assert block['imagetoken'] == image['token']
assert block['alt'] == 'the screen'
def test_an_image_that_is_not_embedded_is_reported_not_silently_lost():
parsed = parse('<h1>T</h1><img src="screenshots/step1.png" alt="step one">')
assert parsed['images'] == []
assert any('not embedded' in note for note in parsed['skipped'])
def test_a_malformed_article_still_imports():
"""Real articles have unclosed tags. Refusing one is worse than dropping a
div nobody can see."""
parsed = parse('<html><body><div class="content"><h1>Half</h1>'
'<p>One<p>Two</body>')
assert parsed['title'] == 'Half'
assert [b['text'] for b in parsed['blocks']] == ['One', 'Two']
def test_the_title_falls_back_to_the_document_title():
parsed = parse('<html><head><title>Only Here - GE Aerospace Knowledge Base'
'</title></head><body><p>Body</p></body></html>')
assert parsed['title'] == 'Only Here'
def test_an_empty_document_produces_a_named_guide_and_no_blocks():
parsed = parse('')
assert parsed['title'] == 'Untitled'
assert parsed['blocks'] == []

11
tests/test_plugins/test_knowledgebase_search_words.py
View File

@@ -22,6 +22,17 @@ from shopdb.core.models import Application
from plugins.knowledgebase.models import KnowledgeBase
@pytest.fixture(autouse=True)
def _signed_in(client, auth_headers):
"""Reading needs a permission now, and these tests are about what the
payload says rather than about who may see it. Signing the client in once
beats threading headers through every helper in the module."""
client.environ_base['HTTP_AUTHORIZATION'] = auth_headers['Authorization']
yield
client.environ_base.pop('HTTP_AUTHORIZATION', None)
@pytest.fixture
def library(db):
"""Articles whose identifying words are deliberately spread across fields."""

20
tests/test_plugins/test_knowledgebase_visibility.py
View File

@@ -15,6 +15,17 @@ from shopdb.core.models import Application
from plugins.knowledgebase.models import KnowledgeBase
@pytest.fixture(autouse=True)
def _signed_in(client, auth_headers):
"""Reading needs a permission now, and these tests are about what the
payload says rather than about who may see it. Signing the client in once
beats threading headers through every helper in the module."""
client.environ_base['HTTP_AUTHORIZATION'] = auth_headers['Authorization']
yield
client.environ_base.pop('HTTP_AUTHORIZATION', None)
@pytest.fixture
def library(db):
"""One application of each interesting shape, with an article each."""
@@ -84,7 +95,7 @@ def test_searching_a_title_does_not_resurrect_a_retired_article(client, library)
assert 'Live runbook' in found
def test_the_counts_agree_with_the_list(client, library):
def test_the_counts_agree_with_the_list(client, library, auth_headers):
"""A total that includes articles nobody can see is a total nobody can
reconcile against the page."""
visible = titles(client)
@@ -97,10 +108,11 @@ def test_the_counts_agree_with_the_list(client, library):
assert clicks == 6
def test_the_topic_list_offers_every_active_application(client, library):
def test_the_topic_list_offers_every_active_application(client, library, auth_headers):
"""What the article form's topic dropdown fetches. Installable or not,
hidden or not - only retired is excluded."""
offered = sorted(row['appname'] for row in client.get(
'/api/applications?perpage=100&showhidden=true').get_json()['data'])
response = client.get('/api/applications?perpage=100&showhidden=true',
headers=auth_headers)
offered = sorted(row['appname'] for row in response.get_json()['data'])
assert offered == ['Hidden App', 'Live App', 'Manual App']
assert 'Retired App' not in offered

32
tests/test_plugins/test_list_search_type_columns.py
View File

@@ -28,7 +28,7 @@ def _numbers(resp):
return sorted(item['assetnumber'] for item in resp.get_json()['data'])
def test_pcs_search_matches_computer_type(client, db):
def test_pcs_search_matches_computer_type(client, db, auth_headers):
assettype = _assettype('computer')
standard = ComputerType(computertype='Standard')
_db.session.add(standard)
@@ -45,16 +45,16 @@ def test_pcs_search_matches_computer_type(client, db):
])
_db.session.commit()
resp = client.get('/api/computers?search=Standard')
resp = client.get('/api/computers?search=Standard', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == ['PC-1']
# the typeless PC is still findable on its own fields
resp = client.get('/api/computers?search=bench')
resp = client.get('/api/computers?search=bench', headers=auth_headers)
assert _numbers(resp) == ['PC-2']
def test_printers_search_matches_type_and_model(client, db):
def test_printers_search_matches_type_and_model(client, db, auth_headers):
assettype = _assettype('printer')
thermal = PrinterType(printertype='Thermal')
model = Model(modelnumber='ZT411')
@@ -74,15 +74,15 @@ def test_printers_search_matches_type_and_model(client, db):
_db.session.commit()
for term in ['Thermal', 'ZT411']:
resp = client.get(f'/api/printers?search={term}')
resp = client.get(f'/api/printers?search={term}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == ['PR-1'], term
resp = client.get('/api/printers?search=plotter')
resp = client.get('/api/printers?search=plotter', headers=auth_headers)
assert _numbers(resp) == ['PR-2']
def test_network_search_matches_type_and_vendor(client, db):
def test_network_search_matches_type_and_vendor(client, db, auth_headers):
assettype = _assettype('network_device')
switchtype = NetworkDeviceType(networkdevicetype='Switch')
vendor = Vendor(vendor='Cisco')
@@ -103,15 +103,15 @@ def test_network_search_matches_type_and_vendor(client, db):
_db.session.commit()
for term in ['Switch', 'Cisco']:
resp = client.get(f'/api/network?search={term}')
resp = client.get(f'/api/network?search={term}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == ['NET-1'], term
resp = client.get('/api/network?search=hub')
resp = client.get('/api/network?search=hub', headers=auth_headers)
assert _numbers(resp) == ['NET-2']
def test_measuringtools_search_matches_type(client, db):
def test_measuringtools_search_matches_type(client, db, auth_headers):
assettype = _assettype('measuring_tool')
caliper = MeasuringToolType(name='Caliper')
_db.session.add(caliper)
@@ -129,15 +129,15 @@ def test_measuringtools_search_matches_type(client, db):
])
_db.session.commit()
resp = client.get('/api/measuringtools?search=Caliper')
resp = client.get('/api/measuringtools?search=Caliper', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == ['MT-1']
resp = client.get('/api/measuringtools?search=height')
resp = client.get('/api/measuringtools?search=height', headers=auth_headers)
assert _numbers(resp) == ['MT-2']
def test_core_assets_search_matches_asset_type(client, db):
def test_core_assets_search_matches_asset_type(client, db, auth_headers):
"""The unified asset list shows a Type column too."""
machine_type = _assettype('machine')
printer_type = _assettype('printer')
@@ -147,12 +147,12 @@ def test_core_assets_search_matches_asset_type(client, db):
])
_db.session.commit()
resp = client.get('/api/assets?search=machine')
resp = client.get('/api/assets?search=machine', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == ['A-1']
def test_core_assets_search_type_filter_still_works(client, db):
def test_core_assets_search_type_filter_still_works(client, db, auth_headers):
"""The type-name filter joins AssetType; search uses EXISTS, so both apply."""
machine_type = _assettype('machine')
_db.session.add_all([
@@ -161,6 +161,6 @@ def test_core_assets_search_type_filter_still_works(client, db):
])
_db.session.commit()
resp = client.get('/api/assets?type=machine&search=mill')
resp = client.get('/api/assets?type=machine&search=mill', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == ['A-1']

20
tests/test_plugins/test_machines_search.py
View File

@@ -47,49 +47,49 @@ def _numbers(resp):
return sorted(item['assetnumber'] for item in resp.get_json()['data'])
def test_search_matches_machine_type_name(client, db):
def test_search_matches_machine_type_name(client, db, auth_headers):
_seed_machines()
resp = client.get('/api/machines?search=Part Washer')
resp = client.get('/api/machines?search=Part Washer', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == ['0410']
def test_search_matches_vendor_name(client, db):
def test_search_matches_vendor_name(client, db, auth_headers):
_seed_machines()
resp = client.get('/api/machines?search=Acme')
resp = client.get('/api/machines?search=Acme', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == ['0410']
def test_search_still_matches_asset_fields(client, db):
def test_search_still_matches_asset_fields(client, db, auth_headers):
_seed_machines()
for term, expected in [('0999', ['0999']),
('Old lathe', ['0999']),
('SN-WASH', ['0410'])]:
resp = client.get(f'/api/machines?search={term}')
resp = client.get(f'/api/machines?search={term}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == expected, term
def test_search_does_not_drop_machines_without_type_or_vendor(client, db):
def test_search_does_not_drop_machines_without_type_or_vendor(client, db, auth_headers):
"""The outer joins must not turn into inner joins."""
_seed_machines()
resp = client.get('/api/machines?search=lathe')
resp = client.get('/api/machines?search=lathe', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == ['0999']
def test_search_miss_returns_empty(client, db):
def test_search_miss_returns_empty(client, db, auth_headers):
_seed_machines()
resp = client.get('/api/machines?search=zzzznotathing')
resp = client.get('/api/machines?search=zzzznotathing', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert _numbers(resp) == []

34
tests/test_plugins/test_measuringtools.py
View File

@@ -20,9 +20,20 @@ from shopdb import create_app
from shopdb.extensions import db as _db
from shopdb.plugins import plugin_manager
from plugins.measuringtools.models import (
derive_status, DUESOON_WINDOW_DAYS, STATUS_COLORS)
@pytest.fixture(autouse=True)
def _signed_in(client, auth_headers):
"""Reading needs a permission now, and these tests are about what the
payload says rather than about who may see it. Signing the client in once
beats threading headers through every helper in the module."""
client.environ_base['HTTP_AUTHORIZATION'] = auth_headers['Authorization']
yield
client.environ_base.pop('HTTP_AUTHORIZATION', None)
# =============================================================================
# Pure unit tests: derived status (no app needed)
# =============================================================================
@@ -48,7 +59,7 @@ def test_derive_status_duesoon_on_boundary():
assert derive_status(date.today() + timedelta(days=DUESOON_WINDOW_DAYS)) == 'duesoon'
def test_derive_status_duesoon_today():
def test_derive_status_duesoon_today(auth_headers):
"""Due today is not yet overdue - it falls in the due-soon window."""
assert derive_status(date.today()) == 'duesoon'
@@ -120,14 +131,17 @@ def _status_id(client):
# -- Type CRUD + in-use guard -------------------------------------------------
def test_starter_types_seeded(client):
response = client.get('/api/measuringtools/types')
def test_starter_types_seeded(client, auth_headers):
response = client.get('/api/measuringtools/types', headers=auth_headers)
assert response.status_code == 200
names = {t['name'] for t in response.get_json()['data']}
assert {'Caliper', 'Micrometer', 'Thread Gage', 'Other'} <= names
def test_type_create_requires_auth(client):
# This module signs the client in for every test; this one is about what
# happens when nobody is signed in, so it takes that back.
client.environ_base.pop('HTTP_AUTHORIZATION', None)
response = client.post('/api/measuringtools/types', json={'name': 'Nope'})
assert response.status_code == 401
@@ -197,7 +211,7 @@ def test_tool_create_and_get_merged(client, auth_headers):
assert ext['calibrationstatus'] == 'current'
tool_id = ext['measuringtoolid']
fetched = client.get(f'/api/measuringtools/{tool_id}')
fetched = client.get(f'/api/measuringtools/{tool_id}', headers=auth_headers)
assert fetched.status_code == 200
assert fetched.get_json()['data']['measuringtool']['measuringtoolid'] == tool_id
@@ -263,7 +277,7 @@ def test_list_filter_by_calibrationstatus(client, auth_headers):
'assetnumber': 'MT-OVERDUE-1', 'statusid': _status_id(client),
'nextcalibrationdate': str(date.today() - timedelta(days=10)),
})
response = client.get('/api/measuringtools?calibrationstatus=overdue')
response = client.get('/api/measuringtools?calibrationstatus=overdue', headers=auth_headers)
assert response.status_code == 200
rows = response.get_json()['data']
assert rows, 'expected at least one overdue tool'
@@ -341,7 +355,7 @@ def test_search_excludes_tools_when_domain_disabled(mt_app, client, auth_headers
# -- Report shape -------------------------------------------------------------
def test_calibration_report_shape(client, auth_headers):
response = client.get('/api/measuringtools/report/calibration')
response = client.get('/api/measuringtools/report/calibration', headers=auth_headers)
assert response.status_code == 200
data = response.get_json()['data']
assert set(data['counts']) == {'overdue', 'duesoon', 'current', 'unknown'}
@@ -382,7 +396,7 @@ def test_asset_todict_carries_measuringtool_typedata_and_pluginid(mt_app, client
def test_map_lists_measuringtool_subtypes(client, auth_headers):
"""The map filter dropdown carries a Measuring Tool subtype list with color."""
response = client.get('/api/assets/map')
response = client.get('/api/assets/map', headers=auth_headers)
assert response.status_code == 200, response.get_json()
subtypes = response.get_json()['data']['filters']['subtypes']
assert 'Measuring Tool' in subtypes
@@ -404,7 +418,7 @@ def test_map_honors_measuringtool_subtype_filter(client, auth_headers):
'measuringtooltypeid': micrometer_id, 'mapx': 30, 'mapy': 40})
response = client.get(
f'/api/assets/map?assettype=measuring_tool&subtype={caliper_id}')
f'/api/assets/map?assettype=measuring_tool&subtype={caliper_id}', headers=auth_headers)
assert response.status_code == 200, response.get_json()
numbers = {a['assetnumber'] for a in response.get_json()['data']['assets']}
assert 'MT-MAP-CAL' in numbers
@@ -416,7 +430,7 @@ def test_map_item_carries_measuringtool_typedata(client, auth_headers):
client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-MAP-TD', 'statusid': _status_id(client),
'measuringtooltypeid': _caliper_id(client), 'mapx': 55, 'mapy': 66})
response = client.get('/api/assets/map?assettype=measuring_tool')
response = client.get('/api/assets/map?assettype=measuring_tool', headers=auth_headers)
item = next(a for a in response.get_json()['data']['assets']
if a['assetnumber'] == 'MT-MAP-TD')
assert item['typedata']['measuringtooltypename'] == 'Caliper'
@@ -444,7 +458,7 @@ def test_map_overlay_shape_and_derivation(client, auth_headers):
'nextcalibrationdate': str(date.today() - timedelta(days=3))}) # overdue
assetid = created.get_json()['data']['assetid']
response = client.get('/api/measuringtools/map-overlay')
response = client.get('/api/measuringtools/map-overlay', headers=auth_headers)
assert response.status_code == 200, response.get_json()
rows = response.get_json()['data']
# ADR-010 overlay contract: [{assetid, color, label}], only due/overdue tools.

20
tests/test_plugins/test_model_image_in_detail.py
View File

@@ -39,55 +39,55 @@ def _seed_model(imageurl=IMAGE_URL):
return model
def test_machine_detail_carries_model_imageurl(client, db):
def test_machine_detail_carries_model_imageurl(client, db, auth_headers):
asset = _seed_asset('2001', 'machine')
model = _seed_model()
machine = Machine(assetid=asset.assetid, modelnumberid=model.modelnumberid)
_db.session.add(machine)
_db.session.commit()
resp = client.get(f'/api/machines/{machine.machineid}')
resp = client.get(f'/api/machines/{machine.machineid}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert resp.get_json()['data']['machine']['imageurl'] == IMAGE_URL
def test_computer_detail_carries_model_imageurl(client, db):
def test_computer_detail_carries_model_imageurl(client, db, auth_headers):
asset = _seed_asset('PC-2001', 'computer')
model = _seed_model()
computer = Computer(assetid=asset.assetid, modelnumberid=model.modelnumberid)
_db.session.add(computer)
_db.session.commit()
resp = client.get(f'/api/computers/{computer.computerid}')
resp = client.get(f'/api/computers/{computer.computerid}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert resp.get_json()['data']['computer']['imageurl'] == IMAGE_URL
def test_printer_detail_carries_model_imageurl(client, db):
def test_printer_detail_carries_model_imageurl(client, db, auth_headers):
asset = _seed_asset('PR-2001', 'printer')
model = _seed_model()
printer = Printer(assetid=asset.assetid, modelnumberid=model.modelnumberid)
_db.session.add(printer)
_db.session.commit()
resp = client.get(f'/api/printers/{printer.printerid}')
resp = client.get(f'/api/printers/{printer.printerid}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert resp.get_json()['data']['printer']['imageurl'] == IMAGE_URL
def test_detail_omits_imageurl_without_model(client, db):
def test_detail_omits_imageurl_without_model(client, db, auth_headers):
"""No model linked -> no imageurl key (frontend v-if shows nothing)."""
asset = _seed_asset('PR-2002', 'printer')
printer = Printer(assetid=asset.assetid)
_db.session.add(printer)
_db.session.commit()
resp = client.get(f'/api/printers/{printer.printerid}')
resp = client.get(f'/api/printers/{printer.printerid}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert 'imageurl' not in resp.get_json()['data']['printer']
def test_detail_omits_imageurl_when_model_has_no_image(client, db):
def test_detail_omits_imageurl_when_model_has_no_image(client, db, auth_headers):
"""Model linked but no photo -> no imageurl key."""
asset = _seed_asset('PC-2002', 'computer')
model = _seed_model(imageurl=None)
@@ -95,6 +95,6 @@ def test_detail_omits_imageurl_when_model_has_no_image(client, db):
_db.session.add(computer)
_db.session.commit()
resp = client.get(f'/api/computers/{computer.computerid}')
resp = client.get(f'/api/computers/{computer.computerid}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert 'imageurl' not in resp.get_json()['data']['computer']

10
tests/test_plugins/test_modelsupplies.py
View File

@@ -18,8 +18,8 @@ def model(db):
return model
def test_supplies_meta_lists_allowed_values(client, db):
response = client.get('/api/printers/supplies/meta')
def test_supplies_meta_lists_allowed_values(client, db, auth_headers):
response = client.get('/api/printers/supplies/meta', headers=auth_headers)
assert response.status_code == 200
data = response.get_json()['data']
assert 'toner' in data['supplytypes']
@@ -38,7 +38,7 @@ def test_create_and_list_model_supply(client, model, auth_headers):
)
assert create.status_code == 201
listing = client.get(f'/api/printers/models/{model.modelnumberid}/supplies')
listing = client.get(f'/api/printers/models/{model.modelnumberid}/supplies', headers=auth_headers)
assert listing.status_code == 200
supplies = listing.get_json()['data']['supplies']
assert len(supplies) == 1
@@ -90,7 +90,7 @@ def test_update_and_delete_supply(client, model, auth_headers):
deleted = client.delete(f'/api/printers/supplies/{supplyid}', headers=auth_headers)
assert deleted.status_code == 200
listing = client.get(f'/api/printers/models/{model.modelnumberid}/supplies')
listing = client.get(f'/api/printers/models/{model.modelnumberid}/supplies', headers=auth_headers)
assert listing.get_json()['data']['supplies'] == []
@@ -98,7 +98,7 @@ def test_listmodels_reports_supplycount(client, model, auth_headers):
client.post(f'/api/printers/models/{model.modelnumberid}/supplies',
json={'partnumber': 'W2020A', 'color': 'black'}, headers=auth_headers)
response = client.get('/api/printers/models', query_string={'search': 'C999'})
response = client.get('/api/printers/models', query_string={'search': 'C999'}, headers=auth_headers)
assert response.status_code == 200
rows = response.get_json()['data']
match = next(r for r in rows if r['modelnumberid'] == model.modelnumberid)

34
tests/test_plugins/test_printedparts_ledger.py
View File

@@ -148,20 +148,20 @@ def test_kiosk_take_is_open_decrement_only(client, auth_headers, app, item,
headers=auth_headers)
assert stocked.status_code == 200
lookup = client.get(f'/api/printedparts/kiosk/item/{itemcode}')
lookup = client.get(f'/api/printedparts/kiosk/item/{itemcode}', headers=auth_headers)
assert lookup.status_code == 200
take = client.post('/api/printedparts/kiosk/take', json={
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 2})
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 2}, headers=auth_headers)
assert take.status_code == 200, take.get_json()
assert take.get_json()['data']['quantityonhand'] == 3
too_many = client.post('/api/printedparts/kiosk/take', json={
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 99})
'itemcode': itemcode, 'badge': directory_employee, 'quantity': 99}, headers=auth_headers)
assert too_many.status_code == 400
unknown = client.post('/api/printedparts/kiosk/take', json={
'itemcode': itemcode, 'badge': '111111111', 'quantity': 1})
'itemcode': itemcode, 'badge': '111111111', 'quantity': 1}, headers=auth_headers)
assert unknown.status_code == 422
with app.app_context():
@@ -197,7 +197,7 @@ def test_lowstock_alert_fires_on_crossing_only(client, auth_headers, app, item,
return client.post('/api/printedparts/kiosk/take',
json={'itemcode': '3DP-9001',
'badge': directory_employee,
'quantity': quantity})
'quantity': quantity}, headers=auth_headers)
restock(10) # 10 on hand, threshold 5
assert take(3).status_code == 200 # 7: above threshold, no alert
@@ -245,7 +245,7 @@ def test_alert_recipients_merge_users_and_freetext(client, auth_headers, app,
headers=auth_headers)
take = client.post('/api/printedparts/kiosk/take',
json={'itemcode': '3DP-9001',
'badge': directory_employee, 'quantity': 6})
'badge': directory_employee, 'quantity': 6}, headers=auth_headers)
assert take.status_code == 200 # 4 on hand: crossed threshold 5
assert captured['to'] == ['lead@site.test', 'extra@site.test']
@@ -260,7 +260,7 @@ def test_retire_hides_and_restore_returns(client, auth_headers, item):
listed = client.get('/api/printedparts/items',
headers=auth_headers).get_json()['data']
assert all(row['printeditemid'] != item for row in listed)
kiosk = client.get('/api/printedparts/kiosk/item/3DP-9001')
kiosk = client.get('/api/printedparts/kiosk/item/3DP-9001', headers=auth_headers)
assert kiosk.status_code == 404
including = client.get('/api/printedparts/items?active=false',
@@ -270,7 +270,7 @@ def test_retire_hides_and_restore_returns(client, auth_headers, item):
assert client.post(f'/api/printedparts/items/{item}/restore',
headers=auth_headers).status_code == 200
assert client.get('/api/printedparts/kiosk/item/3DP-9001').status_code == 200
assert client.get('/api/printedparts/kiosk/item/3DP-9001', headers=auth_headers).status_code == 200
def test_file_revisions_append_and_download(client, auth_headers, item, tmp_path):
@@ -302,7 +302,7 @@ def test_file_revisions_append_and_download(client, auth_headers, item, tmp_path
assert [f['revision'] for f in listing] == [2, 1]
fileid = listing[1]['fileid']
download = client.get(f'/api/printedparts/files/{fileid}/download')
download = client.get(f'/api/printedparts/files/{fileid}/download', headers=auth_headers)
assert download.status_code == 200
assert download.data == b'solid part'
assert 'clip_v1.stl' in download.headers['Content-Disposition']
@@ -335,7 +335,7 @@ def test_alert_role_members_receive(client, auth_headers, app, item,
headers=auth_headers)
take = client.post('/api/printedparts/kiosk/take',
json={'itemcode': '3DP-9001',
'badge': directory_employee, 'quantity': 6})
'badge': directory_employee, 'quantity': 6}, headers=auth_headers)
assert take.status_code == 200
assert captured['to'] == ['crewone@site.test']
@@ -362,14 +362,14 @@ def test_gagelabtag_assigned_searched_and_kiosk_resolved(client, auth_headers):
headers=auth_headers).get_json()['data']
assert len(searched) == 1
by_tag = client.get('/api/printedparts/kiosk/item/WJRP0117')
by_tag = client.get('/api/printedparts/kiosk/item/WJRP0117', headers=auth_headers)
assert by_tag.status_code == 200
by_digits = client.get('/api/printedparts/kiosk/item/117')
by_digits = client.get('/api/printedparts/kiosk/item/117', headers=auth_headers)
assert by_digits.status_code == 200
assert by_digits.get_json()['data']['gagelabtag'] == 'WJRP0117'
def test_kiosk_take_records_revision_from_qr(client, app, db, directory_employee):
def test_kiosk_take_records_revision_from_qr(client, app, db, directory_employee, auth_headers):
"""A label QR payload 'TAG|rev' resolves the item by TAG and records the
revision on the take (traceability of which print-file rev was consumed)."""
with app.app_context():
@@ -380,7 +380,7 @@ def test_kiosk_take_records_revision_from_qr(client, app, db, directory_employee
iid = row.printeditemid
take = client.post('/api/printedparts/kiosk/take', json={
'itemcode': 'WJRP7000|3', 'badge': directory_employee, 'quantity': 1})
'itemcode': 'WJRP7000|3', 'badge': directory_employee, 'quantity': 1}, headers=auth_headers)
assert take.status_code == 200, take.get_json()
with app.app_context():
@@ -391,7 +391,7 @@ def test_kiosk_take_records_revision_from_qr(client, app, db, directory_employee
# Manual entry (no |rev) records a null revision, still resolves.
take2 = client.post('/api/printedparts/kiosk/take', json={
'itemcode': 'WJRP7000', 'badge': directory_employee, 'quantity': 1})
'itemcode': 'WJRP7000', 'badge': directory_employee, 'quantity': 1}, headers=auth_headers)
assert take2.status_code == 200
with app.app_context():
last = (PrintedItemTransaction.query
@@ -400,7 +400,7 @@ def test_kiosk_take_records_revision_from_qr(client, app, db, directory_employee
assert last.revision is None
def test_lowstock_routes_to_support_team_webhook(client, app, db, directory_employee):
def test_lowstock_routes_to_support_team_webhook(client, app, db, directory_employee, auth_headers):
"""A low-stock crossing posts to the selected support team's webhook URL."""
from unittest.mock import patch
from shopdb.core.models import SupportTeam, Setting
@@ -418,7 +418,7 @@ def test_lowstock_routes_to_support_team_webhook(client, app, db, directory_empl
with patch('shopdb.api.send_webhook') as mock_send:
# take 2 -> 6 crosses to 4 (<= threshold 5) -> alert fires
resp = client.post('/api/printedparts/kiosk/take', json={
'itemcode': 'WJRP7100', 'badge': directory_employee, 'quantity': 2})
'itemcode': 'WJRP7100', 'badge': directory_employee, 'quantity': 2}, headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert mock_send.called
assert mock_send.call_args.kwargs['url'] == 'https://teams.example/3dprint'

4
tests/test_plugins/test_supply_history.py
View File

@@ -299,10 +299,10 @@ def test_no_estimate_never_reaches_the_order_list():
assert orderlist([cartridge(None)]) == []
def test_forecast_endpoint_says_so_when_zabbix_is_absent(client, db):
def test_forecast_endpoint_says_so_when_zabbix_is_absent(client, db, auth_headers):
"""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')
response = client.get('/api/printers/supplies/forecast', headers=auth_headers)
assert response.status_code == 200
body = response.get_json()['data']

11
tests/test_plugins/test_toner_forecast_report.py
View File

@@ -14,6 +14,17 @@ from shopdb.core.models import Model, Vendor
from plugins.printers.models import ModelSupply, Printer
@pytest.fixture(autouse=True)
def _signed_in(client, auth_headers):
"""Reading needs a permission now, and these tests are about what the
payload says rather than about who may see it. Signing the client in once
beats threading headers through every helper in the module."""
client.environ_base['HTTP_AUTHORIZATION'] = auth_headers['Authorization']
yield
client.environ_base.pop('HTTP_AUTHORIZATION', None)
LEVELS = {
# itemid: (name, live level, history of daily readings)
# Black is a REAL zero, which is the only thing that means empty. Magenta is

278
tests/test_plugins/test_usb_kiosk.py Normal file
View File

@@ -0,0 +1,278 @@
"""The USB kiosk: an UNAUTHENTICATED write, so the guards are the feature.
The kiosk exists so somebody at the drive cabinet - with no ShopDB login - can
take a drive and put it back, with the action attributed to their badge instead
of to a signed-in operator. Everything here is about the bar that opens the door
only as far as it has to: it can move a device between checked-out and
checked-in and nothing else, it refuses a badge it cannot put a name to, and it
does not exist at all until a site turns it on.
Self-hosted mode. The external cmmc_usb path is the same contract against raw
SQL and needs that database to exercise.
"""
import pytest
from shopdb.core.models import Setting
@pytest.fixture
def kiosk(db):
"""Self-hosted USB, kiosk ON, one employee and one available drive."""
from plugins.employees.models import DirectoryEmployee
from plugins.usb.models import USBDevice
db.session.add(Setting(key='usb_directory_mode', value='selfhosted',
valuetype='string', category='site'))
db.session.add(Setting(key='usb_kiosk_enabled', value='true',
valuetype='boolean', category='usb'))
db.session.add(DirectoryEmployee(sso=270015376, firstname='Dana',
lastname='Reyes'))
db.session.add(DirectoryEmployee(sso=270099999, firstname='Sam',
lastname='Okafor'))
db.session.add(USBDevice(serialnumber='USB-1', label='Blue Kingston',
ischeckedout=False))
db.session.commit()
return db
def _setting(db, key, value):
row = Setting.query.filter_by(key=key).first()
if row is None:
db.session.add(Setting(key=key, value=value, valuetype='boolean',
category='usb'))
else:
row.value = value
db.session.commit()
def _out(client, badge='270015376', **extra):
return client.post('/api/usb/kiosk/checkout',
json=dict({'device_id': 'USB-1', 'badge': badge}, **extra))
def _in(client, badge='270015376', **extra):
return client.post('/api/usb/kiosk/checkin',
json=dict({'device_id': 'USB-1', 'badge': badge}, **extra))
# --- the door itself --------------------------------------------------------
def test_the_kiosk_does_not_exist_until_a_site_turns_it_on(client, kiosk):
"""Default off, and 404 rather than 403.
An unauthenticated write is a decision about the door. A site that has not
made that decision should not have these routes answering at all, and a
disabled kiosk should not advertise that there is a door here.
"""
_setting(kiosk, 'usb_kiosk_enabled', 'false')
assert client.get('/api/usb/kiosk/device/USB-1').status_code == 404
assert _out(client).status_code == 404
assert _in(client).status_code == 404
def test_it_ships_off(app):
"""The declared default is what a fresh site gets."""
from plugins.usb.plugin import USBPlugin
defaults = {s['key']: s for s in USBPlugin().get_settings_defaults()}
assert defaults['usb_kiosk_enabled']['value'] == 'false'
def test_no_login_is_required(client, kiosk):
"""The whole point: no Authorization header anywhere in this file."""
response = client.get('/api/usb/kiosk/device/USB-1')
assert response.status_code == 200
assert response.get_json()['data']['device_id'] == 'USB-1'
def test_the_kiosk_cannot_create_edit_or_delete(client, kiosk):
"""It moves a device between two states. It is not an admin surface."""
assert client.post('/api/usb/kiosk/device/USB-1').status_code in (404, 405)
assert client.delete('/api/usb/kiosk/device/USB-1').status_code in (404, 405)
# And the authenticated write routes are still gated.
assert client.post('/api/usb/USB-1/checkout',
json={'badge': '270015376'}).status_code == 401
# --- badges -----------------------------------------------------------------
def test_a_badge_it_cannot_name_is_refused(client, kiosk):
"""A checkout nobody can trace to a person defeats the tracking.
The signed-in route creates the user row for an unknown badge; the kiosk
must not, because there is no operator standing there to vouch for it.
"""
response = _out(client, badge='999999999')
assert response.status_code == 422
assert 'not recognized' in response.get_json()['data']['error']['message']
from plugins.usb.models import USBDevice
assert USBDevice.query.filter_by(serialnumber='USB-1').first().ischeckedout is False
def test_a_missing_badge_is_refused(client, kiosk):
assert client.post('/api/usb/kiosk/checkout',
json={'device_id': 'USB-1'}).status_code == 400
# --- taking one -------------------------------------------------------------
def test_taking_a_drive_records_who_took_it(client, kiosk):
from plugins.usb.models import USBDevice
response = _out(client)
assert response.status_code == 200, response.get_json()
device = USBDevice.query.filter_by(serialnumber='USB-1').first()
assert device.ischeckedout is True
assert device.currentuserid == '270015376'
assert device.currentusername == 'Dana Reyes'
def test_a_drive_already_out_cannot_be_taken_again(client, kiosk):
_out(client)
response = _out(client, badge='270099999')
assert response.status_code == 409
assert 'Dana Reyes' in response.get_json()['data']['error']['message']
def test_an_unknown_code_is_a_404(client, kiosk):
response = client.post('/api/usb/kiosk/checkout',
json={'device_id': 'NOPE', 'badge': '270015376'})
assert response.status_code == 404
# --- putting one back -------------------------------------------------------
def test_the_holder_returns_their_own_drive(client, kiosk):
from plugins.usb.models import USBDevice
_out(client)
assert _in(client).status_code == 200
device = USBDevice.query.filter_by(serialnumber='USB-1').first()
assert device.ischeckedout is False
assert device.currentuserid is None
def test_returning_somebody_elses_drive_asks_first(client, kiosk):
"""Not a silent success: the record is about to say they handed it back."""
from plugins.usb.models import USBDevice
_out(client)
response = _in(client, badge='270099999')
assert response.status_code == 409
details = response.get_json()['data']['error']['details']
assert details['needsconfirm'] is True
assert details['holdername'] == 'Dana Reyes'
assert USBDevice.query.filter_by(serialnumber='USB-1').first().ischeckedout is True
def test_confirming_returns_it_for_them(client, kiosk):
"""A drive left on a bench by somebody on leave still gets put back.
Refusing outright just moves the problem to a person with a login.
"""
from plugins.usb.models import USBDevice
_out(client)
response = _in(client, badge='270099999', forsomeoneelse=True)
assert response.status_code == 200, response.get_json()
assert USBDevice.query.filter_by(serialnumber='USB-1').first().ischeckedout is False
def test_a_site_can_refuse_returns_for_others_outright(client, kiosk):
_setting(kiosk, 'usb_kiosk_allow_return_for_others', 'false')
_out(client)
response = _in(client, badge='270099999', forsomeoneelse=True)
assert response.status_code == 403
def test_a_drive_that_is_not_out_cannot_be_returned(client, kiosk):
assert _in(client).status_code == 400
# --- what the screen is allowed to know -------------------------------------
def test_the_lookup_names_the_holder_but_not_their_badge(client, kiosk):
"""The confirm step has to show whose it is; it does not need their SSO."""
_out(client)
data = client.get('/api/usb/kiosk/device/USB-1').get_json()['data']
assert data['holdername'] == 'Dana Reyes'
assert '270015376' not in str(data)
# --- the picker -------------------------------------------------------------
# The drives carry no barcode, so the kiosk cannot ask somebody to present one -
# it has to offer a LIST. That is a real widening of what a logged-out screen
# exposes, so the list is not anonymous: it costs a badge that resolves to a
# real person, the same bar taking a drive already had to clear.
def _session(client, badge='270015376'):
return client.post('/api/usb/kiosk/session', json={'badge': badge})
def test_the_device_list_costs_a_valid_badge(client, kiosk):
assert client.post('/api/usb/kiosk/session', json={}).status_code == 400
assert _session(client, badge='999999999').status_code == 422
def test_the_list_is_not_reachable_when_the_kiosk_is_off(client, kiosk):
_setting(kiosk, 'usb_kiosk_enabled', 'false')
assert _session(client).status_code == 404
assert client.post('/api/usb/kiosk/others',
json={'badge': '270015376'}).status_code == 404
def test_the_session_separates_yours_from_available(client, kiosk):
from plugins.usb.models import USBDevice
kiosk.session.add(USBDevice(serialnumber='USB-2', label='Second',
ischeckedout=False))
kiosk.session.commit()
_out(client) # 270015376 takes USB-1
data = _session(client).get_json()['data']
assert data['name'] == 'Dana Reyes'
assert [d['device_id'] for d in data['yours']] == ['USB-1']
assert [d['device_id'] for d in data['available']] == ['USB-2']
assert data['otherscount'] == 0
def test_somebody_elses_drive_is_a_count_not_a_name(client, kiosk):
"""A passer-by should not read the cabinet's holders off a wall screen."""
_out(client) # Dana takes USB-1
data = _session(client, badge='270099999').get_json()['data']
assert data['yours'] == []
assert data['otherscount'] == 1
assert 'Dana' not in str(data)
def test_names_appear_only_on_the_deliberate_second_step(client, kiosk):
_out(client)
rows = client.post('/api/usb/kiosk/others',
json={'badge': '270099999'}).get_json()['data']
assert [r['device_id'] for r in rows] == ['USB-1']
assert rows[0]['holdername'] == 'Dana Reyes'
def test_the_others_list_is_gone_when_returns_for_others_are_refused(client, kiosk):
"""No point showing a list of drives the kiosk will not let anyone return."""
_setting(kiosk, 'usb_kiosk_allow_return_for_others', 'false')
_out(client)
assert client.post('/api/usb/kiosk/others',
json={'badge': '270099999'}).status_code == 404
def test_a_retired_drive_is_not_offered(client, kiosk):
from plugins.usb.models import USBDevice
device = USBDevice.query.filter_by(serialnumber='USB-1').first()
device.isactive = False
kiosk.session.commit()
data = _session(client).get_json()['data']
assert data['available'] == []

139
tests/test_plugins/test_usb_reports.py Normal file
View File

@@ -0,0 +1,139 @@
"""USB check-in / check-out audit report.
CMMC removable-media tracking is an audit obligation, and the question an audit
asks is "who had this drive, when" - which a total cannot answer. So this
reports EVENTS over a window, with the window named rather than expressed as a
day count nobody should have to work out.
"""
from datetime import datetime, timedelta, timezone
import pytest
from shopdb.core.models import Setting
def _now():
return datetime.now(timezone.utc).replace(tzinfo=None)
@pytest.fixture
def usbdata(db):
"""Self-hosted USB with a loan returned today and one from long ago."""
from plugins.usb.models import USBDevice, USBCheckout
db.session.add(Setting(key='usb_directory_mode', value='selfhosted',
valuetype='string', category='site'))
recent = USBDevice(serialnumber='USB-1', label='Blue Kingston')
old = USBDevice(serialnumber='USB-2', label='Red SanDisk')
db.session.add_all([recent, old])
db.session.flush()
# Taken and returned today: two events.
db.session.add(USBCheckout(usbdeviceid=recent.usbdeviceid, sso='270015376',
checkoutname='Dana Reyes',
checkouttime=_now() - timedelta(hours=3),
checkintime=_now() - timedelta(hours=1)))
# Taken 200 days ago and never returned: one event, outside most windows.
db.session.add(USBCheckout(usbdeviceid=old.usbdeviceid, sso='270099999',
checkoutname='Sam Okafor',
checkouttime=_now() - timedelta(days=200)))
db.session.commit()
return db
def _report(client, headers, **params):
query = '&'.join(f'{k}={v}' for k, v in params.items())
response = client.get(f'/api/usb/reports/audit?{query}', headers=headers)
assert response.status_code == 200, response.get_json()
return response.get_json()['data']
def test_a_loan_produces_both_a_take_and_a_return(client, usbdata, auth_headers):
"""One stored row holds both ends of a loan.
Reporting it once would make a returned drive look identical to one still
out, which is the opposite of what the log is for.
"""
data = _report(client, auth_headers, period='month')
actions = sorted(row['action'] for row in data['rows'])
assert actions == ['check-in', 'check-out']
assert data['summary']['checkouts'] == 1
assert data['summary']['checkins'] == 1
def test_the_window_actually_bounds_the_rows(client, usbdata, auth_headers):
month = _report(client, auth_headers, period='month')
year = _report(client, auth_headers, period='year')
assert {row['device_id'] for row in month['rows']} == {'USB-1'}
assert {row['device_id'] for row in year['rows']} == {'USB-1', 'USB-2'}
def test_all_time_keeps_everything(client, usbdata, auth_headers):
data = _report(client, auth_headers, period='all')
assert data['days'] == 0
assert data['summary']['events'] == 3
def test_today_is_a_window_too(client, usbdata, auth_headers):
"""A shift handover asks what happened today, not this month."""
data = _report(client, auth_headers, period='day')
assert data['days'] == 1
assert data['summary']['events'] == 2
@pytest.mark.parametrize('period,days', [('day', 1), ('month', 30),
('sixmonths', 182), ('year', 365),
('all', 0)])
def test_every_named_window_maps_to_a_day_count(client, usbdata, auth_headers,
period, days):
assert _report(client, auth_headers, period=period)['days'] == days
def test_an_unknown_period_falls_back_rather_than_failing(client, usbdata,
auth_headers):
"""A bad query string should not 500 a report somebody is mid-audit on."""
assert _report(client, auth_headers, period='fortnight')['days'] == 30
def test_rows_are_newest_first(client, usbdata, auth_headers):
rows = _report(client, auth_headers, period='all')['rows']
stamps = [row['timestamp'] for row in rows]
assert stamps == sorted(stamps, reverse=True)
def test_it_says_who_and_which_drive(client, usbdata, auth_headers):
rows = _report(client, auth_headers, period='month')['rows']
row = rows[0]
assert row['name'] == 'Dana Reyes'
assert row['badge'] == '270015376'
assert row['device_id'] == 'USB-1'
assert row['device_desc'] == 'Blue Kingston'
def test_it_exports_as_csv(client, usbdata, auth_headers):
response = client.get('/api/usb/reports/audit?period=all&format=csv',
headers=auth_headers)
assert response.status_code == 200
assert response.mimetype == 'text/csv'
body = response.get_data(as_text=True)
assert body.splitlines()[0].startswith('timestamp,action,device_id')
assert 'Dana Reyes' in body
def test_reading_the_audit_needs_permission(client, usbdata, member_headers):
"""A role-less member cannot read who had which drive."""
response = client.get('/api/usb/reports/audit?period=all',
headers=member_headers)
assert response.status_code == 403
def test_the_report_is_registered_so_it_appears_in_the_catalogue(client, usbdata,
auth_headers):
from plugins.usb.plugin import USBPlugin
report = USBPlugin().get_reports()[0]
assert report['id'] == 'usb-audit'
# The endpoint is what the reports page fetches; a wrong one renders blank.
assert report['endpoint'] == '/api/usb/reports/audit'