Initial import: shopdb-flask multi-site asset platform

Flask + Vue rewrite of the classic-ASP shopdb, built as a plugin
framework so sister GE Aerospace sites can adopt it. Product version
0.7.0-pre-0.8.0, plugin contract 0.11.0, 966 backend tests.

Key entry points:
- docs/INSTALL-WINDOWS-IIS.md - tested Windows + IIS install runbook
  (own-site and subpath deployment methods)
- docs/PILOT-DEPLOY.md - production pilot: stand up + import + verify
- docs/IMPORT-API.md - schema-agnostic legacy data import contract
- scripts/site_imports/wjf/ - West Jefferson reference import loader
- docs/adr/ - twelve accepted architecture decision records
This commit is contained in:
2026-07-14 07:58:51 -04:00
commit a9d0dfff4a
599 changed files with 88557 additions and 0 deletions

View File

View File

@@ -0,0 +1,419 @@
"""Personal API token tests.
Covers: create returns the secret once and stores only a hash; a PAT
authenticates a permission-gated write as its owner; a PAT is rejected when its
owner lacks the permission; expired and revoked tokens are rejected; lastusedat
updates on use; a non-owner member cannot revoke someone else's token; an admin
lists everyone's tokens with ?all=true; and import mode works over a PAT for an
admin.
"""
from datetime import datetime, timedelta, timezone
from werkzeug.security import generate_password_hash
from shopdb.core.models import ApiToken, Application, Vendor
from shopdb.extensions import db as _db
def _naive_utcnow():
return datetime.now(timezone.utc).replace(tzinfo=None)
def _create_token(client, headers, name='test token', expiresat=None, scopes=None):
body = {'name': name}
if expiresat is not None:
body['expiresat'] = expiresat
if scopes is not None:
body['scopes'] = scopes
response = client.post('/api/apitokens', json=body, headers=headers)
return response
def _pat_headers(secret):
return {'Authorization': f'Bearer {secret}'}
def _user_with_perms(client, db, username, permnames):
"""Create a non-admin user holding permnames, return (user, login headers).
Seeds the catalog so the named Permission rows exist, links them to a fresh
role, and logs the user in. Used to prove the apitokens.create gate and the
owner scope ceiling for non-admin owners.
"""
from shopdb.core.models import User, Role, Permission
Permission.seed()
role = Role(rolename=username + 'role')
db.session.add(role)
db.session.flush()
role.permissions = Permission.query.filter(
Permission.name.in_(permnames)).all()
user = User(username=username, email=username + '@test.local',
passwordhash=generate_password_hash('testpass'))
user.roles.append(role)
db.session.add(user)
db.session.commit()
login = client.post('/api/auth/login',
json={'username': username, 'password': 'testpass'})
headers = {'Authorization': f"Bearer {login.get_json()['data']['access_token']}"}
return user, headers
def test_create_returns_secret_once_and_stores_hash(client, db, auth_headers):
response = _create_token(client, auth_headers, name='import runner')
assert response.status_code == 201
data = response.get_json()['data']
secret = data['secret']
assert secret.startswith('shopdb_pat_')
assert 'warning' in data
# The stored row must not carry the raw secret; only its hash.
token = ApiToken.query.filter_by(tokenid=data['tokenid']).first()
assert token is not None
assert token.tokenhash == ApiToken.hash_secret(secret)
assert secret not in (token.tokenhash, token.tokenprefix or '')
assert token.tokenprefix and token.tokenprefix in secret
def test_pat_authenticates_permission_write_as_owner(client, db, admin_user,
auth_headers):
"""A PAT owned by an admin can create a vendor (admin-gated write)."""
secret = _create_token(client, auth_headers).get_json()['data']['secret']
response = client.post('/api/vendors', json={'vendor': 'PAT Vendor'},
headers=_pat_headers(secret))
assert response.status_code == 201
assert Vendor.query.filter_by(vendor='PAT Vendor').first() is not None
def test_pat_403_when_owner_lacks_permission(client, db):
"""A PAT owned by a non-admin (who can mint tokens but is not admin) is
forbidden from an admin-role write."""
_user, headers = _user_with_perms(client, db, 'tokenmaker',
['apitokens.create'])
secret = _create_token(client, headers).get_json()['data']['secret']
response = client.post('/api/vendors', json={'vendor': 'Nope'},
headers=_pat_headers(secret))
assert response.status_code == 403
assert Vendor.query.filter_by(vendor='Nope').first() is None
def test_expired_token_rejected(client, db, admin_user, auth_headers):
secret = _create_token(client, auth_headers).get_json()['data']['secret']
token = ApiToken.query.filter_by(
tokenhash=ApiToken.hash_secret(secret)).first()
token.expiresat = _naive_utcnow() - timedelta(days=1)
_db.session.commit()
response = client.post('/api/vendors', json={'vendor': 'Expired'},
headers=_pat_headers(secret))
assert response.status_code == 401
assert Vendor.query.filter_by(vendor='Expired').first() is None
def test_revoked_token_rejected(client, db, admin_user, auth_headers):
secret = _create_token(client, auth_headers).get_json()['data']['secret']
token = ApiToken.query.filter_by(
tokenhash=ApiToken.hash_secret(secret)).first()
tokenid = token.tokenid
revoke = client.delete(f'/api/apitokens/{tokenid}', headers=auth_headers)
assert revoke.status_code == 200
response = client.post('/api/vendors', json={'vendor': 'Revoked'},
headers=_pat_headers(secret))
assert response.status_code == 401
assert Vendor.query.filter_by(vendor='Revoked').first() is None
def test_lastusedat_updates_on_use(client, db, admin_user, auth_headers):
secret = _create_token(client, auth_headers).get_json()['data']['secret']
token = ApiToken.query.filter_by(
tokenhash=ApiToken.hash_secret(secret)).first()
assert token.lastusedat is None
client.get('/api/apitokens', headers=_pat_headers(secret))
_db.session.expire_all()
token = ApiToken.query.filter_by(
tokenhash=ApiToken.hash_secret(secret)).first()
assert token.lastusedat is not None
def test_member_cannot_revoke_other_users_token(client, db, admin_user,
auth_headers):
"""A token-capable non-admin cannot revoke a token owned by someone else."""
secret = _create_token(client, auth_headers).get_json()['data']['secret']
tokenid = ApiToken.query.filter_by(
tokenhash=ApiToken.hash_secret(secret)).first().tokenid
_user, headers = _user_with_perms(client, db, 'tokenmaker',
['apitokens.create'])
response = client.delete(f'/api/apitokens/{tokenid}', headers=headers)
assert response.status_code == 403
# Still active.
assert _db.session.get(ApiToken, tokenid).isactive is True
def test_roleless_member_cannot_create_token(client, db, member_user,
member_headers):
"""Creating a token now requires apitokens.create; a role-less member is
denied with 403."""
response = _create_token(client, member_headers, name='nope')
assert response.status_code == 403
def test_member_with_permission_can_manage_own_token(client, db):
"""A non-admin granted apitokens.create manages their OWN tokens."""
_user, headers = _user_with_perms(client, db, 'tokenmaker',
['apitokens.create'])
create = _create_token(client, headers, name='mine')
assert create.status_code == 201
tokenid = create.get_json()['data']['tokenid']
revoke = client.delete(f'/api/apitokens/{tokenid}', headers=headers)
assert revoke.status_code == 200
assert _db.session.get(ApiToken, tokenid).isactive is False
def test_admin_all_true_lists_everyone(client, db, admin_user, auth_headers):
_create_token(client, auth_headers, name='admin token')
member, member_headers = _user_with_perms(client, db, 'tokenmaker',
['apitokens.create'])
_create_token(client, member_headers, name='member token')
# Own-only (default) for admin: just the admin's token.
own = client.get('/api/apitokens', headers=auth_headers).get_json()['data']
assert all(t['userid'] == admin_user.userid for t in own)
# all=true: both users' tokens, with owner usernames.
everyone = client.get('/api/apitokens?all=true',
headers=auth_headers).get_json()['data']
userids = {t['userid'] for t in everyone}
assert admin_user.userid in userids and member.userid in userids
assert any(t.get('username') for t in everyone)
def test_member_all_true_ignored(client, db, admin_user, auth_headers):
"""A non-admin passing ?all=true still only sees their own tokens."""
_create_token(client, auth_headers, name='admin token')
member, member_headers = _user_with_perms(client, db, 'tokenmaker',
['apitokens.create'])
_create_token(client, member_headers, name='member token')
result = client.get('/api/apitokens?all=true',
headers=member_headers).get_json()['data']
assert all(t['userid'] == member.userid for t in result)
def test_import_mode_works_over_pat(client, db, admin_user, auth_headers):
"""An admin PAT plus X-Import-Mode backdates createddate on a write."""
secret = _create_token(client, auth_headers).get_json()['data']['secret']
headers = _pat_headers(secret)
headers['X-Import-Mode'] = 'true'
response = client.post(
'/api/vendors',
json={'vendor': 'Legacy Vendor', 'createddate': '2019-01-02 03:04:05'},
headers=headers)
assert response.status_code == 201
vendor = Vendor.query.filter_by(vendor='Legacy Vendor').first()
assert vendor is not None
assert vendor.createddate == datetime(2019, 1, 2, 3, 4, 5)
# --- Scoped tokens -------------------------------------------------------
def test_scoped_token_passes_listed_permission(client, db, admin_user,
auth_headers):
"""A token scoped to applications.create can create an application."""
secret = _create_token(client, auth_headers,
scopes=['applications.create']).get_json()['data']['secret']
response = client.post('/api/applications', json={'appname': 'Scoped App'},
headers=_pat_headers(secret))
assert response.status_code == 201
assert Application.query.filter_by(appname='Scoped App').first() is not None
def test_scoped_token_403_on_unlisted_permission(client, db, admin_user,
auth_headers):
"""A token scoped to only applications.create cannot delete (needs
applications.delete, which is not in scope)."""
app = Application(appname='Target App')
_db.session.add(app)
_db.session.commit()
appid = app.appid
secret = _create_token(client, auth_headers,
scopes=['applications.create']).get_json()['data']['secret']
response = client.delete(f'/api/applications/{appid}',
headers=_pat_headers(secret))
assert response.status_code == 403
assert _db.session.get(Application, appid) is not None
def test_scoped_admin_token_does_not_bypass(client, db, admin_user, auth_headers):
"""An admin owner's scoped token is genuinely limited: the admin permission
bypass is suspended, so an unlisted action is 403 even for an admin."""
secret = _create_token(client, auth_headers,
scopes=['applications.create']).get_json()['data']['secret']
# settings.edit is not scoped -> 403 despite the owner being admin.
response = client.put('/api/settings/site_name',
json={'value': 'Hacked'}, headers=_pat_headers(secret))
assert response.status_code == 403
def test_scoped_token_denied_on_require_role_route(client, db, admin_user,
auth_headers):
"""Scoped tokens cannot pass a role gate. Vendor writes are require_role
admin, so even an admin-owned scoped token is denied."""
secret = _create_token(client, auth_headers,
scopes=['applications.create']).get_json()['data']['secret']
response = client.post('/api/vendors', json={'vendor': 'RoleGated'},
headers=_pat_headers(secret))
assert response.status_code == 403
assert Vendor.query.filter_by(vendor='RoleGated').first() is None
def test_unscoped_token_unchanged(client, db, admin_user, auth_headers):
"""Regression: an unscoped admin token still acts with full authority,
including role-gated writes."""
secret = _create_token(client, auth_headers).get_json()['data']['secret']
response = client.post('/api/vendors', json={'vendor': 'Unscoped Vendor'},
headers=_pat_headers(secret))
assert response.status_code == 201
token = ApiToken.query.filter_by(
tokenhash=ApiToken.hash_secret(secret)).first()
assert token.scopelist is None
def test_unknown_scope_name_rejected_at_create(client, db, admin_user,
auth_headers):
"""An unknown permission name in scopes is rejected 400, listing it."""
response = _create_token(client, auth_headers, scopes=['not.a.permission'])
assert response.status_code == 400
assert 'not.a.permission' in response.get_json()['data']['error']['message']
def test_scope_ceiling_rejects_permission_owner_lacks(client, db):
"""The scope ceiling: a non-admin cannot mint a token scoped to a
permission they do not hold."""
_user, headers = _user_with_perms(client, db, 'tokenmaker',
['apitokens.create'])
response = _create_token(client, headers, scopes=['applications.create'])
assert response.status_code == 400
assert 'applications.create' in response.get_json()['data']['error']['message']
def test_non_admin_scoped_token_capped_at_own_permissions(client, db):
"""A non-admin holding applications.create + apitokens.create may mint a
token scoped to applications.create and use it."""
_user, headers = _user_with_perms(
client, db, 'tokenmaker', ['apitokens.create', 'applications.create'])
secret = _create_token(client, headers,
scopes=['applications.create']).get_json()['data']['secret']
response = client.post('/api/applications', json={'appname': 'Capped App'},
headers=_pat_headers(secret))
assert response.status_code == 201
def test_scoped_token_intersects_at_use_time(client, db):
"""Defense in depth: if the owner loses a scoped permission after the token
is minted, the token can no longer use it (use-time intersection)."""
from shopdb.core.models import Role
_user, headers = _user_with_perms(
client, db, 'tokenmaker', ['apitokens.create', 'applications.create'])
secret = _create_token(client, headers,
scopes=['applications.create']).get_json()['data']['secret']
# Strip applications.create from the owner's role.
role = Role.query.filter_by(rolename='tokenmakerrole').first()
role.permissions = [p for p in role.permissions
if p.name != 'applications.create']
_db.session.commit()
response = client.post('/api/applications', json={'appname': 'Gone'},
headers=_pat_headers(secret))
assert response.status_code == 403
assert Application.query.filter_by(appname='Gone').first() is None
def test_import_mode_denied_over_scoped_admin_token(client, db, admin_user,
auth_headers):
"""A scoped admin token does NOT get import mode (an admin-role capability):
the createddate in the body is ignored and 'now' is stamped instead."""
secret = _create_token(
client, auth_headers,
scopes=['applications.create']).get_json()['data']['secret']
headers = _pat_headers(secret)
headers['X-Import-Mode'] = 'true'
before = _naive_utcnow()
response = client.post(
'/api/applications',
json={'appname': 'No Backdate', 'createddate': '2019-01-02 03:04:05'},
headers=headers)
assert response.status_code == 201
app = Application.query.filter_by(appname='No Backdate').first()
assert app is not None
# Import mode was denied, so createddate is ~now, not the 2019 value.
assert app.createddate >= before
def test_scopes_update_round_trip(client, db, admin_user, auth_headers):
"""Scopes can be set, changed, and cleared via PUT, and surface in GET."""
create = _create_token(client, auth_headers, scopes=['applications.create'])
tokenid = create.get_json()['data']['tokenid']
got = client.get('/api/apitokens', headers=auth_headers).get_json()['data']
row = next(t for t in got if t['tokenid'] == tokenid)
assert row['scopes'] == ['applications.create']
# Change scopes.
updated = client.put(f'/api/apitokens/{tokenid}',
json={'scopes': ['applications.edit']},
headers=auth_headers)
assert updated.status_code == 200
assert updated.get_json()['data']['scopes'] == ['applications.edit']
# Clear scopes (back to unscoped/full).
cleared = client.put(f'/api/apitokens/{tokenid}', json={'scopes': None},
headers=auth_headers)
assert cleared.status_code == 200
assert cleared.get_json()['data']['scopes'] is None
def test_collector_scoped_token_denied_on_normal_endpoints(client, db,
admin_user,
auth_headers):
"""A collector-scoped token (collector.ingest only) is contained: the shim
still swaps it into a JWT, but collector.ingest gates no normal route, so it
is denied on a permission-gated create just like any other unlisted scope."""
secret = _create_token(
client, auth_headers,
scopes=['collector.ingest']).get_json()['data']['secret']
response = client.post('/api/applications', json={'appname': 'Nope'},
headers=_pat_headers(secret))
assert response.status_code == 403
assert Application.query.filter_by(appname='Nope').first() is None
def test_update_scopes_ceiling_enforced(client, db):
"""PUT scopes is also capped at the owner's permissions."""
_user, headers = _user_with_perms(client, db, 'tokenmaker',
['apitokens.create'])
tokenid = _create_token(client, headers).get_json()['data']['tokenid']
response = client.put(f'/api/apitokens/{tokenid}',
json={'scopes': ['applications.delete']},
headers=headers)
assert response.status_code == 400
assert 'applications.delete' in response.get_json()['data']['error']['message']

View File

@@ -0,0 +1,70 @@
"""Tests for IP-based login rate limiting.
The limiter is disabled in TestingConfig so login-heavy fixtures do not trip
it. These tests flip it on via app.config overrides and clear the shared cache
so counters do not bleed between cases.
"""
import pytest
from shopdb.extensions import cache
@pytest.fixture
def ratelimit_on(app):
"""Enable a tight login rate limit for the duration of one test."""
saved = {
'enabled': app.config.get('AUTH_RATELIMIT_ENABLED'),
'max': app.config.get('AUTH_RATELIMIT_MAX'),
'window': app.config.get('AUTH_RATELIMIT_WINDOW_SECONDS'),
}
app.config['AUTH_RATELIMIT_ENABLED'] = True
app.config['AUTH_RATELIMIT_MAX'] = 3
app.config['AUTH_RATELIMIT_WINDOW_SECONDS'] = 300
with app.app_context():
cache.clear()
yield
with app.app_context():
cache.clear()
app.config['AUTH_RATELIMIT_ENABLED'] = saved['enabled']
app.config['AUTH_RATELIMIT_MAX'] = saved['max']
app.config['AUTH_RATELIMIT_WINDOW_SECONDS'] = saved['window']
def _bad_login(client, ip):
"""Attempt a login for a non-existent user from a given source IP."""
return client.post(
'/api/auth/login',
json={'username': 'ghost', 'password': 'wrong'},
headers={'X-Forwarded-For': ip},
)
def test_login_flood_from_one_ip_is_rate_limited(client, db, ratelimit_on):
"""After the per-window budget is spent, further attempts get 429."""
ip = '203.0.113.10'
for _ in range(3):
assert _bad_login(client, ip).status_code == 401
# Fourth attempt in the same window is over budget.
blocked = _bad_login(client, ip)
assert blocked.status_code == 429
assert blocked.get_json()['data']['error']['code'] == 'RATE_LIMITED'
def test_different_ip_not_penalized(client, db, ratelimit_on):
"""One flooding IP does not lock out a different caller's IP."""
flooder = '203.0.113.20'
for _ in range(4):
_bad_login(client, flooder)
assert _bad_login(client, flooder).status_code == 429
# A separate IP still gets the normal 401, not a 429.
other = _bad_login(client, '203.0.113.21')
assert other.status_code == 401
def test_limiter_disabled_by_default_in_testing(client, db):
"""With TestingConfig defaults the limiter is off; repeated logins stay 401."""
for _ in range(10):
resp = _bad_login(client, '203.0.113.30')
assert resp.status_code == 401

View File

@@ -0,0 +1,342 @@
"""Authorization gating and login-lockout tests.
Covers the security fixes:
- write routes require a permission/role, not just authentication
- the admin role bypasses permission checks
- repeated bad logins lock the account
The parametrized sweep at the bottom is the standing CI guard: it walks the
whole URL map and proves every mutating route (POST/PUT/PATCH/DELETE) either
rejects a role-less member with 403 or is on the deliberate exemption list.
A future unguarded mutation therefore fails CI the moment it is added.
"""
import re
import pytest
from shopdb import create_app
# HTTP verbs that change state. GETs are intentionally left alone (kiosk pages
# rely on optional-jwt reads).
MUTATING_METHODS = {'POST', 'PUT', 'PATCH', 'DELETE'}
# Blueprints exempt by design (see task/ADR-006):
# auth - login/refresh/logout are the auth surface itself
# collector - agent ingest, authenticated by per-plugin API key (ADR-006)
# setup - first-run wizard, self-gated by the needs-admin state
EXEMPT_BLUEPRINTS = {'auth', 'collector', 'setup'}
# Individual endpoints exempt by design.
# knowledgebase.track_click - kiosk click-through counter. Optional-jwt
# telemetry: increments a click count on an already-active article and
# returns its link URL. Takes no caller-supplied data; gating it would
# break the unauthenticated kiosk redirect it exists to serve.
# users.update_user - admin-or-self (inline check, not a flat deny): a
# role-less member MAY edit their own record, so it does not fit the
# 403-for-every-member contract this sweep asserts. The other-user 403 is
# covered by test_member_cannot_update_other_user below.
# The apitokens create/update/revoke endpoints are NOT exempt: they now require
# the apitokens.create permission, so a role-less member gets the 403 this sweep
# asserts (ownership is still enforced inside the handler for non-admins).
# geenforce.post_report - GE-Enforce client ingest, authenticated by a
# geenforce.report managed service token (X-API-Key/Bearer), not JWT. Same
# shape as the exempt collector blueprint; the geenforce admin endpoints in
# the same blueprint are JWT+permission gated and ARE swept.
EXEMPT_ENDPOINTS = {'knowledgebase.track_click', 'users.update_user',
'geenforce.post_report'}
@pytest.fixture(autouse=True)
def _selfhosted_directory_modes(db):
# The usb/employees plugins default to EXTERNAL directory mode when no
# setting row exists, so exercising their handlers reaches for a MySQL
# server that exists on the dev box but not in CI. Pin selfhosted mode
# so the sweep is deterministic everywhere.
from shopdb.core.models import Setting
db.session.add(Setting(key='usb_directory_mode', value='selfhosted',
valuetype='string', category='site'))
db.session.add(Setting(key='employee_directory_mode', value='selfhosted',
valuetype='string', category='site'))
db.session.commit()
def _fill_url(rule):
"""Turn a Werkzeug rule into a concrete path (int/float params -> 1, else x)."""
def repl(match):
converter = match.group(1) or ''
return '1' if converter.startswith(('int', 'float')) else 'x'
# matches <name>, <int:name>, <path:name>, ...
return re.sub(r'<(?:([^:>]+):)?[^>]+>', repl, rule)
def _mutating_routes():
"""Every non-exempt mutating (endpoint, method, url) in the URL map.
Enumerated from a throwaway app: route strings are stable across app
instances, so the requests below run against the fixture client's app.
"""
app = create_app('testing')
routes = []
for rule in app.url_map.iter_rules():
methods = (rule.methods or set()) & MUTATING_METHODS
if not methods:
continue
blueprint = rule.endpoint.split('.')[0]
if blueprint in EXEMPT_BLUEPRINTS or rule.endpoint in EXEMPT_ENDPOINTS:
continue
for method in sorted(methods):
routes.append((rule.endpoint, method, _fill_url(rule.rule)))
return routes
# Collected once at import so each route is its own parametrized case (a
# newly-unguarded mutation shows up as a single named CI failure).
MUTATING_ROUTES = _mutating_routes()
ROUTE_IDS = [f'{ep}[{method}]' for ep, method, _ in MUTATING_ROUTES]
def test_member_cannot_write_settings(client, db, member_headers):
"""An authenticated user with no roles is forbidden from editing settings."""
from shopdb.core.models import Setting
setting = Setting(key='zabbix_enabled', value='false', valuetype='boolean',
category='integrations')
db.session.add(setting)
db.session.commit()
response = client.put('/api/settings/zabbix_enabled',
json={'value': True}, headers=member_headers)
assert response.status_code == 403
assert response.get_json()['data']['error']['code'] == 'FORBIDDEN'
def test_admin_can_write_settings(client, db, auth_headers):
"""The admin role bypasses the permission check and may edit settings."""
from shopdb.core.models import Setting
setting = Setting(key='zabbix_enabled', value='false', valuetype='boolean',
category='integrations')
db.session.add(setting)
db.session.commit()
response = client.put('/api/settings/zabbix_enabled',
json={'value': True}, headers=auth_headers)
assert response.status_code == 200
def test_member_cannot_delete_asset(client, db, member_headers):
"""A role-less user cannot delete assets (was previously allowed)."""
response = client.delete('/api/assets/1', headers=member_headers)
# Forbidden by authz, not a 404 - the gate runs before the lookup.
assert response.status_code == 403
def test_unauthenticated_write_is_rejected(client, db):
"""No token at all still cannot reach a write route."""
response = client.delete('/api/assets/1')
assert response.status_code in (401, 422) # missing/!invalid JWT
def test_member_cannot_create_business_unit(client, db, member_headers):
"""Reference-data writes are admin-only."""
response = client.post('/api/businessunits',
json={'businessunit': 'Test BU'}, headers=member_headers)
assert response.status_code == 403
def test_member_cannot_list_users(client, db, member_headers):
"""User listing is admin-only (converted from inline check to decorator)."""
response = client.get('/api/users', headers=member_headers)
assert response.status_code == 403
assert response.get_json()['data']['error']['code'] == 'FORBIDDEN'
def test_member_cannot_create_user(client, db, member_headers):
"""User creation is admin-only."""
response = client.post('/api/users',
json={'username': 'x', 'email': 'x@test.local',
'password': 'secret'},
headers=member_headers)
assert response.status_code == 403
def test_member_cannot_delete_role(client, db, member_headers):
"""Role deletion is admin-only."""
response = client.delete('/api/users/roles/1', headers=member_headers)
assert response.status_code == 403
def test_admin_can_list_users(client, db, auth_headers):
"""The admin role passes the require_role gate on user listing."""
response = client.get('/api/users', headers=auth_headers)
assert response.status_code == 200
def test_member_can_read_own_user(client, db, member_user, member_headers):
"""Admin-or-self: a role-less user may read their OWN record."""
response = client.get(f'/api/users/{member_user.userid}',
headers=member_headers)
assert response.status_code == 200
assert response.get_json()['data']['username'] == 'testmember'
def test_member_cannot_read_other_user(client, db, member_user, member_headers):
"""Admin-or-self: a role-less user may not read a DIFFERENT record."""
response = client.get(f'/api/users/{member_user.userid + 999}',
headers=member_headers)
assert response.status_code == 403
def test_account_locks_after_repeated_bad_logins(client, db, admin_user):
"""Five bad passwords lock the account; a correct password is then refused."""
for _ in range(5):
bad = client.post('/api/auth/login',
json={'username': 'testadmin', 'password': 'wrong'})
assert bad.status_code == 401
# Account is now locked - even the correct password is refused with 403.
locked = client.post('/api/auth/login',
json={'username': 'testadmin', 'password': 'testpass'})
assert locked.status_code == 403
assert 'locked' in locked.get_json()['data']['error']['message'].lower()
def test_lockout_auto_unlocks_after_expiry(client, db, admin_user):
"""Once lockeduntil is in the past the account unlocks itself: a correct
password logs in and clears the lock state."""
from datetime import datetime, timezone, timedelta
# Simulate an expired lock (naive-UTC to match the column comparisons).
admin_user.lockeduntil = (datetime.now(timezone.utc).replace(tzinfo=None)
- timedelta(minutes=1))
admin_user.failedlogins = 5
db.session.commit()
ok = client.post('/api/auth/login',
json={'username': 'testadmin', 'password': 'testpass'})
assert ok.status_code == 200, ok.get_json()
db.session.refresh(admin_user)
assert admin_user.lockeduntil is None
assert admin_user.failedlogins == 0
def test_successful_login_resets_failed_counter(client, db, admin_user):
"""A good login before the threshold clears the failure count."""
for _ in range(3):
client.post('/api/auth/login',
json={'username': 'testadmin', 'password': 'wrong'})
ok = client.post('/api/auth/login',
json={'username': 'testadmin', 'password': 'testpass'})
assert ok.status_code == 200
from shopdb.core.models import User
user = User.query.filter_by(username='testadmin').first()
assert user.failedlogins == 0
assert user.lockeduntil is None
# ---------------------------------------------------------------------------
# Standing CI guard: sweep every mutating route.
# ---------------------------------------------------------------------------
def test_member_cannot_update_other_user(client, db, member_user, member_headers):
"""Admin-or-self write gate: a role-less user cannot edit a DIFFERENT user."""
response = client.put(f'/api/users/{member_user.userid + 999}',
json={'firstname': 'x'}, headers=member_headers)
assert response.status_code == 403
def test_route_map_has_mutating_routes():
"""Sanity: the sweep actually found routes (catches a broken collector)."""
assert len(MUTATING_ROUTES) > 50, 'route enumeration looks empty/broken'
@pytest.mark.parametrize('endpoint,method,url', MUTATING_ROUTES, ids=ROUTE_IDS)
def test_mutation_rejects_roleless_member(client, db, member_headers,
endpoint, method, url):
"""Every non-exempt mutating route must 403 a logged-in, role-less user.
A 200/201/400/404/409 here means the route ran its handler for an
unprivileged caller - i.e. it is not permission-gated. That is the exact
regression this guard exists to catch.
"""
response = client.open(url, method=method, headers=member_headers, json={})
assert response.status_code == 403, (
f'{method} {endpoint} ({url}) returned {response.status_code}, '
f'expected 403 - route is not permission-gated. If it is intended to '
f'be public/kiosk, add it to EXEMPT_ENDPOINTS with a reason.'
)
assert response.get_json()['data']['error']['code'] == 'FORBIDDEN'
@pytest.mark.parametrize('endpoint,method,url', MUTATING_ROUTES, ids=ROUTE_IDS)
def test_mutation_admin_passes_authz(client, db, auth_headers,
endpoint, method, url):
"""Admin must clear the gate on every mutating route (never 401/403).
Downstream the handler may 400/404/409/422 on the empty body - that is
validation, not authorization, and is fine. We only assert the authz gate
let the admin through.
"""
response = client.open(url, method=method, headers=auth_headers, json={})
assert response.status_code not in (401, 403), (
f'{method} {endpoint} ({url}) returned {response.status_code} for '
f'admin - the admin role should bypass every permission check.'
)
# --- Settings exposure allowlist (unauthenticated recon guard) ---------------
def _seed_settings(db):
from shopdb.core.models import Setting
db.session.add_all([
Setting(key='facility_name', value='WJ', valuetype='string', category='site'),
Setting(key='brand_primary_color', value='#123', valuetype='string', category='branding'),
Setting(key='smtp_host', value='mail.internal', valuetype='string', category='communication'),
Setting(key='employee_db_host', value='hr.internal', valuetype='string', category='integrations'),
Setting(key='zabbix_token', value='supersecret', valuetype='string', category='integrations'),
])
db.session.commit()
def test_unauthenticated_settings_list_is_allowlisted(client, db):
"""No token: only the public allowlist (branding + bootstrap keys) is
visible; internal infra config is not enumerable."""
_seed_settings(db)
keys = {s['key'] for s in client.get('/api/settings').get_json()['data']}
assert 'facility_name' in keys and 'brand_primary_color' in keys
assert 'smtp_host' not in keys
assert 'employee_db_host' not in keys
assert 'zabbix_token' not in keys
def test_authenticated_settings_list_full_but_masked(client, db, member_headers):
"""Any authed principal sees the full table; secret values stay masked."""
_seed_settings(db)
rows = {s['key']: s['value']
for s in client.get('/api/settings', headers=member_headers).get_json()['data']}
assert 'smtp_host' in rows and 'employee_db_host' in rows
assert rows['zabbix_token'] == '********' # secret masked even when visible
def test_unauthenticated_private_setting_get_is_404(client, db):
"""A single private key is not confirmable without auth; a public one is."""
_seed_settings(db)
assert client.get('/api/settings/smtp_host').status_code == 404
assert client.get('/api/settings/facility_name').status_code == 200
# --- audit.view gating -------------------------------------------------------
def test_member_cannot_read_auditlogs(client, db, member_headers):
"""Audit-read routes now require audit.view; a role-less member is 403."""
for url in ('/api/auditlogs', '/api/auditlogs/entity/Setting/1', '/api/auditlogs/stats'):
assert client.get(url, headers=member_headers).status_code == 403, url
def test_admin_can_read_auditlogs(client, db, auth_headers):
"""Admin holds audit.view and reads the audit trail."""
assert client.get('/api/auditlogs', headers=auth_headers).status_code == 200

View File

@@ -0,0 +1,434 @@
"""Tests for the generic plugin collector contract (ADR-006).
Covers the auto-dispatched /api/collector/<plugin> endpoint, per-plugin API
key auth, idempotent upsert by identity field, and the JWT-protected
/api/collector/_schemas listing.
"""
import pytest
KEY = 'testcollectorkey'
@pytest.fixture
def collector_key(app):
"""Set the shared collector API key for the test."""
old = app.config.get('COLLECTOR_API_KEY')
app.config['COLLECTOR_API_KEY'] = KEY
yield KEY
app.config['COLLECTOR_API_KEY'] = old
@pytest.fixture
def computer_assettype(db):
"""Seed the computer asset type needed for create-on-ingest."""
from shopdb.core.models import AssetType
at = AssetType(assettype='computer', pluginname='computers',
tablename='computers', description='PCs')
db.session.add(at)
db.session.commit()
return at
def test_schemas_requires_jwt(client, db):
"""The schema listing rejects unauthenticated callers."""
response = client.get('/api/collector/_schemas')
assert response.status_code == 401
def test_schemas_lists_computers(client, db, auth_headers):
"""Computers plugin exposes a collector schema keyed by hostname."""
response = client.get('/api/collector/_schemas', headers=auth_headers)
assert response.status_code == 200
schemas = response.get_json()['data']['schemas']
assert 'computers' in schemas
assert schemas['computers']['identityfield'] == 'hostname'
def test_missing_key_rejected(client, db, collector_key, computer_assettype):
"""No API key -> 401."""
response = client.post('/api/collector/computers',
json={'hostname': 'WJPC001'})
assert response.status_code == 401
def test_unknown_plugin_404(client, db, collector_key):
"""A plugin with no collector schema returns 404."""
response = client.post('/api/collector/nosuchplugin',
json={'hostname': 'x'},
headers={'X-API-Key': KEY})
assert response.status_code == 404
def test_missing_identity_rejected(client, db, collector_key, computer_assettype):
"""Missing the identity field is a validation error."""
response = client.post('/api/collector/computers',
json={'currentuser': 'someone'},
headers={'X-API-Key': KEY})
assert response.status_code == 400
def test_create_then_idempotent_update(client, db, collector_key,
computer_assettype):
"""First post creates; second post with same hostname updates (no dup)."""
from plugins.computers.models import Computer
payload = {'hostname': 'WJPC100', 'currentuser': 'alice',
'serialnumber': 'SN-100'}
first = client.post('/api/collector/computers', json=payload,
headers={'X-API-Key': KEY})
assert first.status_code == 200, first.get_json()
data = first.get_json()['data']
assert data['action'] == 'created'
assert data['assetid'] is not None
assert data['identityvalue'] == 'WJPC100'
payload['currentuser'] = 'bob'
second = client.post('/api/collector/computers', json=payload,
headers={'X-API-Key': KEY})
assert second.status_code == 200
assert second.get_json()['data']['action'] == 'updated'
with client.application.app_context():
comps = Computer.query.filter(Computer.hostname.ilike('WJPC100')).all()
assert len(comps) == 1
assert comps[0].loggedinuser == 'bob'
def test_per_plugin_key_overrides_shared(client, db, app, computer_assettype):
"""COLLECTOR_API_KEY_<PLUGIN> takes precedence over the shared key."""
app.config['COLLECTOR_API_KEY'] = 'sharedkey'
app.config['COLLECTOR_API_KEY_COMPUTERS'] = 'computerskey'
try:
# Shared key now rejected for this plugin.
rejected = client.post('/api/collector/computers',
json={'hostname': 'WJPC200'},
headers={'X-API-Key': 'sharedkey'})
assert rejected.status_code == 401
accepted = client.post('/api/collector/computers',
json={'hostname': 'WJPC200'},
headers={'X-API-Key': 'computerskey'})
assert accepted.status_code == 200
finally:
app.config.pop('COLLECTOR_API_KEY_COMPUTERS', None)
app.config['COLLECTOR_API_KEY'] = None
def test_complete_asset_payload_maps_enrollment_data(client, db, collector_key,
computer_assettype):
"""A GE-Enforce-shaped payload maps machinenumber/pctype/vendor/model/os."""
from shopdb.core.models import OperatingSystem
from plugins.computers.models import Computer, ComputerType
# Seed the controlled-vocab rows the collector looks up.
db.session.add(ComputerType(computertype='Shopfloor PC'))
db.session.add(OperatingSystem(osname='Windows 11'))
db.session.commit()
payload = {
'hostname': 'WJSF1234',
'machinenumber': '0615',
'pctype': 'gea-shopfloor-collections',
'serialnumber': 'SN-ENROLL',
'vendorname': 'Dell',
'modelnumber': 'OptiPlex 7090',
'osname': 'Windows 11',
}
resp = client.post('/api/collector/computers', json=payload,
headers={'X-API-Key': KEY})
assert resp.status_code == 200, resp.get_json()
assert resp.get_json()['data']['action'] == 'created'
with client.application.app_context():
comp = Computer.query.filter(Computer.hostname.ilike('WJSF1234')).first()
assert comp is not None
assert comp.asset.assetnumber == '0615' # machinenumber -> assetnumber
assert comp.asset.serialnumber == 'SN-ENROLL'
assert comp.computertype.computertype == 'Shopfloor PC' # pctype mapped
assert comp.vendor.vendor == 'Dell' # created
assert comp.model.modelnumber == 'OptiPlex 7090' # created
assert comp.operatingsystem.osname == 'Windows 11'
def test_placeholder_machinenumber_9999_falls_back_to_hostname(client, db,
collector_key,
computer_assettype):
"""The 9999 imaging placeholder is ignored; assetnumber falls back to host."""
resp = client.post('/api/collector/computers',
json={'hostname': 'WJSF9999', 'machinenumber': '9999'},
headers={'X-API-Key': KEY})
assert resp.status_code == 200
from plugins.computers.models import Computer
with client.application.app_context():
comp = Computer.query.filter(Computer.hostname.ilike('WJSF9999')).first()
assert comp.asset.assetnumber == 'WJSF9999'
def test_internal_error_message_is_generic(client, app, db, collector_key):
"""An unexpected upsert failure returns a generic 500, not the exception text."""
pm = app.extensions['plugin_manager']
plugin = pm.get_all_plugins()['computers']
original = plugin.apply_collector_payload
def boom(payload):
raise RuntimeError('secret internal dsn leaked here')
plugin.apply_collector_payload = boom
try:
resp = client.post('/api/collector/computers',
json={'hostname': 'WJPC500'},
headers={'X-API-Key': KEY})
finally:
plugin.apply_collector_payload = original
assert resp.status_code == 500
message = resp.get_json()['data']['error']['message']
assert message == 'Internal error processing collector payload'
assert 'secret internal dsn' not in message
def test_generic_querystring_api_key_rejected(client, db, collector_key,
computer_assettype):
"""The dropped ?api_key= querystring fallback no longer authenticates."""
resp = client.post('/api/collector/computers?api_key=' + KEY,
json={'hostname': 'WJPCQS'})
assert resp.status_code == 401
def test_legacy_querystring_api_key_rejected(client, db, collector_key):
"""Header-only auth on the legacy endpoints too: querystring key is rejected."""
resp = client.get('/api/collector/status?api_key=' + KEY)
assert resp.status_code == 401
def test_legacy_header_api_key_accepted(client, db, collector_key):
"""The X-API-Key header still authenticates the legacy endpoints."""
resp = client.get('/api/collector/status', headers={'X-API-Key': KEY})
assert resp.status_code == 200
# --- Managed collector service tokens ------------------------------------
def _mint_collector_token(client, auth_headers, scopes):
"""Mint a managed token with the given scopes as the logged-in admin."""
resp = client.post('/api/apitokens',
json={'name': 'collector svc', 'scopes': scopes},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
return resp.get_json()['data']['secret']
def test_collector_token_via_x_api_key(client, db, auth_headers,
computer_assettype):
"""A collector-scoped managed token authorizes ingest via X-API-Key, with
no env key set."""
secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
resp = client.post('/api/collector/computers',
json={'hostname': 'WJTOK01'},
headers={'X-API-Key': secret})
assert resp.status_code == 200, resp.get_json()
assert resp.get_json()['data']['action'] == 'created'
def test_collector_token_via_bearer(client, db, auth_headers,
computer_assettype):
"""The same token also works on the standard Bearer PAT transport."""
secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
resp = client.post('/api/collector/computers',
json={'hostname': 'WJTOK02'},
headers={'Authorization': f'Bearer {secret}'})
assert resp.status_code == 200, resp.get_json()
assert resp.get_json()['data']['action'] == 'created'
def test_collector_token_stamps_lastusedat(client, db, auth_headers,
computer_assettype):
"""Using a collector token stamps lastusedat."""
from shopdb.core.models import ApiToken
secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
token = ApiToken.query.filter_by(
tokenhash=ApiToken.hash_secret(secret)).first()
assert token.lastusedat is None
client.post('/api/collector/computers', json={'hostname': 'WJTOK03'},
headers={'X-API-Key': secret})
db.session.expire_all()
token = ApiToken.query.filter_by(
tokenhash=ApiToken.hash_secret(secret)).first()
assert token.lastusedat is not None
def test_unscoped_token_rejected_by_collector(client, db, auth_headers,
collector_key, computer_assettype):
"""An unscoped PAT is NOT a collector token: it lacks the collector.ingest
scope, so the collector API rejects it (env-key semantics unchanged)."""
secret = client.post('/api/apitokens', json={'name': 'full'},
headers=auth_headers).get_json()['data']['secret']
resp = client.post('/api/collector/computers', json={'hostname': 'WJTOK04'},
headers={'X-API-Key': secret})
assert resp.status_code == 401
def test_wrong_scoped_token_rejected_by_collector(client, db, auth_headers,
collector_key,
computer_assettype):
"""A PAT scoped to some OTHER permission is not a collector token."""
secret = _mint_collector_token(client, auth_headers, ['applications.create'])
resp = client.post('/api/collector/computers', json={'hostname': 'WJTOK05'},
headers={'X-API-Key': secret})
assert resp.status_code == 401
def test_revoked_collector_token_rejected(client, db, auth_headers,
collector_key, computer_assettype):
"""A revoked collector token is rejected on both transports."""
from shopdb.core.models import ApiToken
secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
tokenid = ApiToken.query.filter_by(
tokenhash=ApiToken.hash_secret(secret)).first().tokenid
assert client.delete(f'/api/apitokens/{tokenid}',
headers=auth_headers).status_code == 200
viakey = client.post('/api/collector/computers', json={'hostname': 'WJTOK06'},
headers={'X-API-Key': secret})
assert viakey.status_code == 401
viabearer = client.post('/api/collector/computers',
json={'hostname': 'WJTOK06'},
headers={'Authorization': f'Bearer {secret}'})
assert viabearer.status_code == 401
def test_collector_token_contained_to_collector_api(client, db, admin_user,
auth_headers):
"""Containment: a collector-scoped token is denied on a permission-gated
write (applications.create) AND a role-gated write (vendors, require_role)."""
from shopdb.core.models import Application, Vendor
secret = _mint_collector_token(client, auth_headers, ['collector.ingest'])
perm_gated = client.post('/api/applications', json={'appname': 'Nope'},
headers={'Authorization': f'Bearer {secret}'})
assert perm_gated.status_code == 403
assert Application.query.filter_by(appname='Nope').first() is None
role_gated = client.post('/api/vendors', json={'vendor': 'Nope'},
headers={'Authorization': f'Bearer {secret}'})
assert role_gated.status_code == 403
assert Vendor.query.filter_by(vendor='Nope').first() is None
def test_env_key_still_works_alongside_tokens(client, db, collector_key,
computer_assettype):
"""Regression: the env key path is unchanged when tokens are in play."""
resp = client.post('/api/collector/computers', json={'hostname': 'WJENV01'},
headers={'X-API-Key': KEY})
assert resp.status_code == 200
assert resp.get_json()['data']['action'] == 'created'
# -- metrology measuring-tool auto-link ------------------------------------
# A CMM/Keyence/Genspect/wax-trace PC stays a shopfloor PC, but the collector
# creates the attached measuring instrument as a MeasuringTool asset and links
# the PC to it with a directional 'controls' relationship.
@pytest.fixture
def measuringtool_env(db):
"""Seed the measuring_tool asset type + 'controls' relationship type the
metrology collector path links through."""
from shopdb.core.models import AssetType, RelationshipType
db.session.add(AssetType(assettype='measuring_tool',
pluginname='measuringtools',
tablename='measuringtools',
description='Metrology instruments'))
db.session.add(RelationshipType(relationshiptype='controls',
description='Operational authority',
isdirectional=True))
db.session.commit()
def test_metrology_pctype_creates_and_links_measuringtool(
client, db, collector_key, computer_assettype, measuringtool_env):
"""A CMM PC gets a MeasuringTool asset + controls link; the CMM tool type
is auto-created (it is not in the measuringtools starter seed)."""
from plugins.computers.models import Computer
from plugins.measuringtools.models import MeasuringTool, MeasuringToolType
from shopdb.core.models import AssetRelationship
payload = {'hostname': 'WJCMM01', 'pctype': 'gea-shopfloor-cmm'}
resp = client.post('/api/collector/computers', json=payload,
headers={'X-API-Key': KEY})
assert resp.status_code == 200, resp.get_json()
data = resp.get_json()['data']
assert data['measuringtoollinkcount'] == 1
link = data['measuringtoollinks'][0]
assert link['relationshiptype'] == 'controls'
assert link['measuringtooltype'] == 'CMM'
with client.application.app_context():
tools = MeasuringTool.query.all()
assert len(tools) == 1
assert MeasuringToolType.query.filter_by(name='CMM').first() is not None
pc = Computer.query.filter(Computer.hostname.ilike('WJCMM01')).first()
rels = AssetRelationship.query.filter_by(
sourceassetid=pc.assetid, label='collector:measuringtool').all()
assert len(rels) == 1
assert rels[0].targetassetid == tools[0].assetid
assert rels[0].isactive
def test_metrology_link_idempotent(client, db, collector_key,
computer_assettype, measuringtool_env):
"""Re-pushing the same metrology PC does not duplicate the tool asset."""
from plugins.measuringtools.models import MeasuringTool
payload = {'hostname': 'WJCMM02', 'pctype': 'gea-shopfloor-cmm'}
for _ in range(2):
resp = client.post('/api/collector/computers', json=payload,
headers={'X-API-Key': KEY})
assert resp.status_code == 200
with client.application.app_context():
assert MeasuringTool.query.count() == 1
def test_nonmetrology_pctype_makes_no_tool(client, db, collector_key,
computer_assettype, measuringtool_env):
"""A plain shopfloor PC gets no measuring-tool asset."""
from plugins.measuringtools.models import MeasuringTool
payload = {'hostname': 'WJPC900', 'pctype': 'gea-shopfloor-collections'}
resp = client.post('/api/collector/computers', json=payload,
headers={'X-API-Key': KEY})
assert resp.status_code == 200
assert resp.get_json()['data']['measuringtoollinkcount'] == 0
with client.application.app_context():
assert MeasuringTool.query.count() == 0
def test_repurposed_pc_archives_tool_link(client, db, collector_key,
computer_assettype, measuringtool_env):
"""A metrology PC re-imaged to a non-metrology type has its collector tool
link archived (tool asset kept, relationship deactivated)."""
from plugins.measuringtools.models import MeasuringTool
from shopdb.core.models import AssetRelationship
host = {'hostname': 'WJCMM03'}
client.post('/api/collector/computers',
json={**host, 'pctype': 'gea-shopfloor-cmm'},
headers={'X-API-Key': KEY})
resp = client.post('/api/collector/computers',
json={**host, 'pctype': 'gea-shopfloor-collections'},
headers={'X-API-Key': KEY})
assert resp.status_code == 200
assert resp.get_json()['data']['measuringtoollinkcount'] == 0
with client.application.app_context():
# asset preserved (may carry calibration history), link deactivated
assert MeasuringTool.query.count() == 1
rels = AssetRelationship.query.filter_by(
label='collector:measuringtool').all()
assert len(rels) == 1
assert rels[0].isactive is False

View File

@@ -0,0 +1,199 @@
"""Tests for the computers collector PC->printer relationship sync.
A collector payload carrying defaultprinter/printers creates AssetRelationship
rows linking the PC asset to the resolved printer assets ('defaultprinter' for
the default, 'connectedto' for the rest). Stale collector links archive on
re-report; manual links survive; unresolved identifiers warn without failing.
"""
import pytest
from shopdb.core.models import (
Asset, AssetType, AssetRelationship, RelationshipType, CommunicationType,
Communication)
from plugins.printers.models import Printer
KEY = 'testcollectorkey'
@pytest.fixture
def collector_key(app):
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):
"""Computer + printer asset types, the two relationship types, and three
printer assets identified by windowsname / hostname / IP."""
pc_type = AssetType(assettype='computer', pluginname='computers',
tablename='computers')
pr_type = AssetType(assettype='printer', pluginname='printers',
tablename='printers')
dp_type = RelationshipType(relationshiptype='defaultprinter',
description='PC to default printer',
isdirectional=True)
ct_type = RelationshipType(relationshiptype='connectedto',
description='Network link', isdirectional=False)
ip_comtype = CommunicationType(comtype='IP')
db.session.add_all([pc_type, pr_type, dp_type, ct_type, ip_comtype])
db.session.flush()
# Printer A: matched by windowsname. Printer B: by hostname. Printer C: by IP.
a_asset = Asset(assetnumber='PRN-A', name='Materials HP',
assettypeid=pr_type.assettypeid, isactive=True)
b_asset = Asset(assetnumber='PRN-B', name='Front Office',
assettypeid=pr_type.assettypeid, isactive=True)
c_asset = Asset(assetnumber='PRN-C', name='Shipping Label',
assettypeid=pr_type.assettypeid, isactive=True)
db.session.add_all([a_asset, b_asset, c_asset])
db.session.flush()
db.session.add_all([
Printer(assetid=a_asset.assetid, windowsname='HP-Materials'),
Printer(assetid=b_asset.assetid, hostname='prn-front.wjs.local'),
Printer(assetid=c_asset.assetid, windowsname='Zebra-Ship'),
])
db.session.add(Communication(assetid=c_asset.assetid,
comtypeid=ip_comtype.comtypeid,
ipaddress='10.1.2.3', isprimary=True))
db.session.commit()
return {'pc_type': pc_type, 'dp_type': dp_type, 'ct_type': ct_type,
'a': a_asset, 'b': b_asset, 'c': c_asset}
def _post(client, payload):
return client.post('/api/collector/computers', json=payload,
headers={'X-API-Key': KEY})
def _pc_asset(db):
from plugins.computers.models import Computer
comp = Computer.query.first()
return comp.asset
def _active_links(db, pcid):
return AssetRelationship.query.filter_by(
sourceassetid=pcid, isactive=True).all()
def test_creates_default_and_connected_links(client, db, collector_key, scene):
"""Default -> defaultprinter, others -> connectedto, resolved by three id forms."""
resp = _post(client, {
'hostname': 'WJPC-P1',
'defaultprinter': 'HP-Materials', # windowsname
'printers': ['prn-front.wjs.local', '10.1.2.3'], # hostname, IP
})
assert resp.status_code == 200, resp.get_json()
data = resp.get_json()['data']
assert data['printerlinkcount'] == 3
assert not data['warnings']
pc = _pc_asset(db)
links = {(r.targetassetid, r.relationshiptype.relationshiptype)
for r in _active_links(db, pc.assetid)}
assert (scene['a'].assetid, 'defaultprinter') in links
assert (scene['b'].assetid, 'connectedto') in links
assert (scene['c'].assetid, 'connectedto') in links
def test_unresolved_identifier_warns_not_fails(client, db, collector_key, scene):
"""An unknown printer identifier is a warning, the rest still link."""
resp = _post(client, {
'hostname': 'WJPC-P2',
'defaultprinter': 'HP-Materials',
'printers': ['does-not-exist'],
})
assert resp.status_code == 200
data = resp.get_json()['data']
assert any('unresolved printer: does-not-exist' in w
for w in data['warnings'])
pc = _pc_asset(db)
assert data['printerlinkcount'] == 1
assert len(_active_links(db, pc.assetid)) == 1
def test_idempotent_no_duplicates(client, db, collector_key, scene):
"""Re-report the same set: no new rows, still one link per printer."""
payload = {'hostname': 'WJPC-P3', 'defaultprinter': 'HP-Materials',
'printers': ['Zebra-Ship']}
assert _post(client, payload).status_code == 200
assert _post(client, payload).status_code == 200
pc = _pc_asset(db)
all_rows = AssetRelationship.query.filter_by(sourceassetid=pc.assetid).all()
assert len(all_rows) == 2 # one default + one connected, no dupes
assert all(r.isactive for r in all_rows)
def test_dropped_printer_is_archived(client, db, collector_key, scene):
"""A printer that stops being reported loses its collector link."""
_post(client, {'hostname': 'WJPC-P4', 'defaultprinter': 'HP-Materials',
'printers': ['Zebra-Ship']})
pc = _pc_asset(db)
assert len(_active_links(db, pc.assetid)) == 2
# Re-report without Zebra-Ship.
_post(client, {'hostname': 'WJPC-P4', 'defaultprinter': 'HP-Materials',
'printers': []})
active = _active_links(db, pc.assetid)
assert len(active) == 1
assert active[0].targetassetid == scene['a'].assetid
# the dropped row still exists but is inactive
zebra = AssetRelationship.query.filter_by(
sourceassetid=pc.assetid, targetassetid=scene['c'].assetid).first()
assert zebra is not None and zebra.isactive is False
def test_default_change_archives_old_default(client, db, collector_key, scene):
"""Changing the default printer archives the previous defaultprinter link."""
_post(client, {'hostname': 'WJPC-P5', 'defaultprinter': 'HP-Materials'})
pc = _pc_asset(db)
_post(client, {'hostname': 'WJPC-P5', 'defaultprinter': 'prn-front.wjs.local'})
active = _active_links(db, pc.assetid)
assert len(active) == 1
assert active[0].targetassetid == scene['b'].assetid
assert active[0].relationshiptype.relationshiptype == 'defaultprinter'
def test_manual_link_survives_collector_sync(client, db, collector_key, scene):
"""A manually-created link (no collector label) is never archived."""
# PC created first via a bare report.
_post(client, {'hostname': 'WJPC-P6', 'defaultprinter': 'HP-Materials',
'printers': ['Zebra-Ship']})
pc = _pc_asset(db)
# Manual connectedto link PC -> printer B (label stays NULL).
manual = AssetRelationship(
sourceassetid=pc.assetid, targetassetid=scene['b'].assetid,
relationshiptypeid=scene['ct_type'].relationshiptypeid, notes='by hand')
db.session.add(manual)
db.session.commit()
manual_id = manual.relationshipid
# Re-report dropping Zebra-Ship and never mentioning printer B.
_post(client, {'hostname': 'WJPC-P6', 'defaultprinter': 'HP-Materials',
'printers': []})
surviving = db.session.get(AssetRelationship, manual_id)
assert surviving.isactive is True # manual link untouched
# Zebra (collector) archived.
zebra = AssetRelationship.query.filter_by(
sourceassetid=pc.assetid, targetassetid=scene['c'].assetid).first()
assert zebra.isactive is False
def test_no_printer_keys_leaves_links_untouched(client, db, collector_key, scene):
"""A report with no printer keys does not archive existing links."""
_post(client, {'hostname': 'WJPC-P7', 'defaultprinter': 'HP-Materials'})
pc = _pc_asset(db)
assert len(_active_links(db, pc.assetid)) == 1
# A plain update (no defaultprinter, no printers key) must not archive.
_post(client, {'hostname': 'WJPC-P7', 'currentuser': 'alice'})
assert len(_active_links(db, pc.assetid)) == 1

View File

@@ -0,0 +1,38 @@
"""Tests for the dashboard-widgets hook consumer (/api/dashboard/widgets).
Pins the wiring added for the BasePlugin.get_dashboard_widgets hook: the
endpoint aggregates enabled plugins' widgets and skips disabled ones.
Plugin enabled-state is monkeypatched (not persisted) so these tests do not
mutate the shared instance/plugins.json registry file.
"""
def _widget_plugins(client, headers):
response = client.get('/api/dashboard/widgets', headers=headers)
assert response.status_code == 200, response.get_json()
widgets = response.get_json()['data']
assert isinstance(widgets, list)
return widgets, {w.get('plugin') for w in widgets}
def test_widgets_endpoint_aggregates_enabled_plugins(app, client, auth_headers,
monkeypatch):
"""An enabled plugin that implements the hook contributes a widget."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
widgets, plugins = _widget_plugins(client, auth_headers)
assert 'computers' in plugins # computers implements get_dashboard_widgets
positions = [w.get('position', 99) for w in widgets]
assert positions == sorted(positions)
def test_widgets_endpoint_skips_disabled_plugin(app, client, auth_headers,
monkeypatch):
"""A disabled plugin's widgets drop out of the aggregate."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: name != 'computers')
_, plugins = _widget_plugins(client, auth_headers)
assert 'computers' not in plugins

View File

@@ -0,0 +1,99 @@
"""Tests for dashboard defaults (visitor-IP -> business-unit resolution)."""
import pytest
@pytest.fixture
def businessunit(db):
from shopdb.core.models import BusinessUnit
bu = BusinessUnit(businessunit='Materials')
db.session.add(bu)
db.session.commit()
return bu
def test_create_and_resolve_visitor_location(client, db, auth_headers, businessunit):
"""A mapped IP resolves to its business unit; explicit ipaddress param works."""
created = client.post('/api/dashboarddefaults', json={
'ipaddress': '10.20.30.40',
'businessunitid': businessunit.businessunitid,
'description': 'Materials lobby kiosk',
}, headers=auth_headers)
assert created.status_code == 201, created.get_json()
resolved = client.get('/api/dashboarddefaults/visitor-location?ipaddress=10.20.30.40')
assert resolved.status_code == 200
data = resolved.get_json()['data']
assert data['businessunitid'] == businessunit.businessunitid
assert data['businessunit'] == 'Materials'
def test_unmapped_ip_resolves_to_null(client, db):
"""An unmapped IP returns a null business unit, not an error."""
resolved = client.get('/api/dashboarddefaults/visitor-location?ipaddress=1.2.3.4')
assert resolved.status_code == 200
assert resolved.get_json()['data']['businessunitid'] is None
def test_duplicate_ip_rejected(client, db, auth_headers, businessunit):
"""Mapping the same IP twice is a conflict."""
payload = {'ipaddress': '10.0.0.9', 'businessunitid': businessunit.businessunitid}
first = client.post('/api/dashboarddefaults', json=payload, headers=auth_headers)
assert first.status_code == 201
dup = client.post('/api/dashboarddefaults', json=payload, headers=auth_headers)
assert dup.status_code == 409
def test_non_admin_cannot_create(client, db, member_headers, businessunit):
"""A logged-in non-admin is forbidden from creating a mapping (RBAC)."""
resp = client.post('/api/dashboarddefaults', json={
'ipaddress': '10.20.30.50',
'businessunitid': businessunit.businessunitid,
}, headers=member_headers)
assert resp.status_code == 403
def test_unauthenticated_cannot_create(client, db, businessunit):
"""No token -> 401 on create."""
resp = client.post('/api/dashboarddefaults', json={
'ipaddress': '10.20.30.51',
'businessunitid': businessunit.businessunitid,
})
assert resp.status_code == 401
def test_admin_can_update_and_delete(client, db, auth_headers, businessunit):
"""Admin can PUT and DELETE a mapping."""
created = client.post('/api/dashboarddefaults', json={
'ipaddress': '10.20.30.60',
'businessunitid': businessunit.businessunitid,
}, headers=auth_headers)
assert created.status_code == 201
default_id = created.get_json()['data']['dashboarddefaultid']
updated = client.put(f'/api/dashboarddefaults/{default_id}',
json={'description': 'moved'}, headers=auth_headers)
assert updated.status_code == 200
deleted = client.delete(f'/api/dashboarddefaults/{default_id}',
headers=auth_headers)
assert deleted.status_code == 200
def test_non_admin_cannot_update_or_delete(client, db, auth_headers,
member_headers, businessunit):
"""A non-admin is forbidden from PUT and DELETE."""
created = client.post('/api/dashboarddefaults', json={
'ipaddress': '10.20.30.70',
'businessunitid': businessunit.businessunitid,
}, headers=auth_headers)
assert created.status_code == 201
default_id = created.get_json()['data']['dashboarddefaultid']
put_resp = client.put(f'/api/dashboarddefaults/{default_id}',
json={'description': 'nope'}, headers=member_headers)
assert put_resp.status_code == 403
del_resp = client.delete(f'/api/dashboarddefaults/{default_id}',
headers=member_headers)
assert del_resp.status_code == 403

View File

@@ -0,0 +1,270 @@
"""Tests for the email service, forced password change, welcome email, and
on-demand report/alert delivery.
smtplib is always patched so tests never touch the network.
"""
import os
from unittest.mock import patch, MagicMock
import pytest
from werkzeug.security import generate_password_hash
@pytest.fixture(autouse=True)
def _strip_smtp_env(monkeypatch):
"""Remove any SMTP_* env so env-fallback cannot enable email unexpectedly."""
for key in list(os.environ):
if key.startswith('SMTP_'):
monkeypatch.delenv(key, raising=False)
def _enable_smtp(db):
"""Seed the minimum settings to make SMTP 'configured'."""
from shopdb.core.models import Setting
from shopdb.core.api.settings import invalidate_settings_cache
rows = {
'smtp_enabled': ('true', 'boolean'),
'smtp_host': ('smtp.test.local', 'string'),
'smtp_port': ('587', 'integer'),
'smtp_use_tls': ('true', 'boolean'),
'smtp_from_address': ('noreply@test.local', 'string'),
'smtp_from_name': ('ShopDB', 'string'),
'alert_recipients': ('ops@test.local', 'string'),
}
for key, (value, valuetype) in rows.items():
db.session.add(Setting(key=key, value=value, valuetype=valuetype,
category='email'))
db.session.commit()
invalidate_settings_cache()
def _make_user(db, username, password='temppass1', mustchange=False,
failedlogins=0):
from shopdb.core.models import User
user = User(
username=username,
email=f'{username}@test.local',
passwordhash=generate_password_hash(password),
mustchangepassword=mustchange,
failedlogins=failedlogins,
)
db.session.add(user)
db.session.commit()
return user
def _login(client, username, password):
response = client.post('/api/auth/login',
json={'username': username, 'password': password})
return response
# ---------------------------------------------------------------------------
# Mailer no-op behavior
# ---------------------------------------------------------------------------
def test_mailer_noop_when_disabled(app, db):
"""send_email is a no-op returning False when SMTP is disabled; no connect."""
from shopdb.utils import mailer
with app.app_context():
with patch('shopdb.utils.mailer.smtplib.SMTP') as smtp:
result = mailer.send_email('a@test.local', 'Hi', '<p>Hi</p>')
assert result is False
smtp.assert_not_called()
def test_mailer_sends_when_configured(app, db):
"""send_email connects and sends when SMTP is configured (patched)."""
from shopdb.utils import mailer
with app.app_context():
_enable_smtp(db)
with patch('shopdb.utils.mailer.smtplib.SMTP') as smtp:
instance = MagicMock()
smtp.return_value = instance
result = mailer.send_email('a@test.local', 'Hi', '<p>Hi</p>')
assert result is True
instance.sendmail.assert_called_once()
def test_mailer_never_logs_password(app, db):
"""A send failure carrying the password is scrubbed in the returned error."""
from shopdb.utils import mailer
with app.app_context():
from shopdb.core.models import Setting
from shopdb.core.api.settings import invalidate_settings_cache
_enable_smtp(db)
db.session.add(Setting(key='smtp_password', value='s3cret',
valuetype='string', category='email'))
db.session.commit()
invalidate_settings_cache()
with patch('shopdb.utils.mailer.smtplib.SMTP') as smtp:
instance = MagicMock()
instance.sendmail.side_effect = RuntimeError('auth failed s3cret')
smtp.return_value = instance
ok, error = mailer.try_send('a@test.local', 'Hi', '<p>Hi</p>')
assert ok is False
assert 's3cret' not in error
# ---------------------------------------------------------------------------
# Test-email endpoint
# ---------------------------------------------------------------------------
def test_test_email_requires_admin(client, db, member_headers):
response = client.post('/api/settings/test-email',
json={'to': 'x@test.local'}, headers=member_headers)
assert response.status_code == 403
def test_test_email_graceful_when_unconfigured(client, db, auth_headers):
response = client.post('/api/settings/test-email',
json={'to': 'x@test.local'}, headers=auth_headers)
assert response.status_code == 200
assert response.get_json()['data']['sent'] is False
def test_test_email_sends_when_configured(client, db, auth_headers):
_enable_smtp(db)
with patch('shopdb.utils.mailer.smtplib.SMTP') as smtp:
smtp.return_value = MagicMock()
response = client.post('/api/settings/test-email',
json={'to': 'x@test.local'}, headers=auth_headers)
assert response.status_code == 200
assert response.get_json()['data']['sent'] is True
# ---------------------------------------------------------------------------
# Forced password change
# ---------------------------------------------------------------------------
def test_login_surfaces_mustchangepassword(client, db):
_make_user(db, 'forceduser', password='temppass1', mustchange=True)
response = _login(client, 'forceduser', 'temppass1')
assert response.status_code == 200
assert response.get_json()['data']['user']['mustchangepassword'] is True
def test_forced_change_clears_flag_and_lockout(client, db):
user = _make_user(db, 'forceduser', password='temppass1', mustchange=True,
failedlogins=3)
token = _login(client, 'forceduser', 'temppass1').get_json()['data']['access_token']
headers = {'Authorization': f'Bearer {token}'}
# Forced case: new_password only, no current required.
response = client.post('/api/auth/change-password',
json={'new_password': 'brandnew123'}, headers=headers)
assert response.status_code == 200
from shopdb.core.models import User
refreshed = db.session.get(User, user.userid)
assert refreshed.mustchangepassword is False
assert refreshed.failedlogins == 0
assert refreshed.lockeduntil is None
# New password works.
assert _login(client, 'forceduser', 'brandnew123').status_code == 200
def test_selfservice_change_rejects_wrong_current(client, db):
_make_user(db, 'normaluser', password='rightpass1', mustchange=False)
token = _login(client, 'normaluser', 'rightpass1').get_json()['data']['access_token']
headers = {'Authorization': f'Bearer {token}'}
response = client.post('/api/auth/change-password',
json={'current_password': 'wrongpass',
'new_password': 'brandnew123'}, headers=headers)
assert response.status_code == 401
def test_selfservice_change_succeeds_with_correct_current(client, db):
_make_user(db, 'normaluser', password='rightpass1', mustchange=False)
token = _login(client, 'normaluser', 'rightpass1').get_json()['data']['access_token']
headers = {'Authorization': f'Bearer {token}'}
response = client.post('/api/auth/change-password',
json={'current_password': 'rightpass1',
'new_password': 'brandnew123'}, headers=headers)
assert response.status_code == 200
assert _login(client, 'normaluser', 'brandnew123').status_code == 200
def test_change_password_rejects_short_password(client, db):
_make_user(db, 'normaluser', password='rightpass1', mustchange=False)
token = _login(client, 'normaluser', 'rightpass1').get_json()['data']['access_token']
headers = {'Authorization': f'Bearer {token}'}
response = client.post('/api/auth/change-password',
json={'current_password': 'rightpass1',
'new_password': 'short'}, headers=headers)
assert response.status_code == 400
# ---------------------------------------------------------------------------
# New-user welcome
# ---------------------------------------------------------------------------
def test_create_user_sets_mustchangepassword_and_sends(client, db, auth_headers):
_enable_smtp(db)
with patch('shopdb.utils.mailer.smtplib.SMTP') as smtp:
instance = MagicMock()
smtp.return_value = instance
response = client.post('/api/users',
json={'username': 'newhire',
'email': 'newhire@test.local',
'password': 'temppass1'},
headers=auth_headers)
assert response.status_code == 201
assert response.get_json()['data']['mustchangepassword'] is True
instance.sendmail.assert_called_once()
def test_create_user_survives_mail_failure(client, db, auth_headers):
"""User is still created (with a warning) when the welcome email fails."""
_enable_smtp(db)
with patch('shopdb.utils.mailer.smtplib.SMTP') as smtp:
instance = MagicMock()
instance.sendmail.side_effect = RuntimeError('relay down')
smtp.return_value = instance
response = client.post('/api/users',
json={'username': 'newhire2',
'email': 'newhire2@test.local',
'password': 'temppass1'},
headers=auth_headers)
assert response.status_code == 201
assert 'warning' in response.get_json()['data']
# ---------------------------------------------------------------------------
# On-demand report/alert delivery
# ---------------------------------------------------------------------------
def test_email_report_requires_permission(client, db, member_headers):
response = client.post('/api/reports/email',
json={'subject': 'X', 'columns': [], 'rows': []},
headers=member_headers)
assert response.status_code == 403
def test_email_report_renders_and_sends(client, db, auth_headers):
_enable_smtp(db)
with patch('shopdb.utils.mailer.smtplib.SMTP') as smtp:
instance = MagicMock()
smtp.return_value = instance
response = client.post('/api/reports/email',
json={'subject': 'Warranty Report',
'columns': [{'key': 'vendor', 'label': 'Vendor'}],
'rows': [{'vendor': 'Dell'}],
'to': 'boss@test.local'},
headers=auth_headers)
assert response.status_code == 200
assert response.get_json()['data']['sent'] is True
instance.sendmail.assert_called_once()
def test_email_report_defaults_to_alert_recipients(client, db, auth_headers):
_enable_smtp(db)
with patch('shopdb.utils.mailer.smtplib.SMTP') as smtp:
instance = MagicMock()
smtp.return_value = instance
response = client.post('/api/reports/email',
json={'subject': 'Toner', 'columns': [], 'rows': []},
headers=auth_headers)
assert response.status_code == 200
assert response.get_json()['data']['sent'] is True

View File

@@ -0,0 +1,26 @@
"""Characterization tests for the employee directory endpoints.
Written before extracting employees into a plugin. The wjf_employees DB is not
available under test, so these assert the DB-independent behaviors: validation
(400) and graceful failure (500 envelope, not an unhandled crash) when the
directory is unreachable. Behavior must be identical as a plugin blueprint.
"""
def test_search_requires_min_query(client, db):
"""A too-short query is rejected before touching the directory DB."""
resp = client.get('/api/employees/search?q=a')
assert resp.status_code == 400
def test_lookup_requires_numeric_sso(client, db):
"""Non-numeric SSO is rejected without touching the directory DB."""
resp = client.get('/api/employees/lookup/abc')
assert resp.status_code == 400
def test_search_degrades_when_directory_unreachable(client, db):
"""With the directory DB unavailable, search returns a clean 500 envelope."""
resp = client.get('/api/employees/search?q=smith')
assert resp.status_code == 500
assert resp.get_json()['status'] == 'error'

View File

@@ -0,0 +1,77 @@
"""Tests for per-asset-type optional identifiers (gauge/maintenance refs).
Pins two behaviors that shipped without coverage:
1. gaugelabreference + maintenancereference round-trip through each plugin's
asset create AND update endpoints (computer, printer, network).
2. The per-type identifier seed produces one setting per
(identifier x asset type) pair.
"""
import pytest
from shopdb.core.api.settings import IDENTIFIER_LABELS, IDENTIFIER_ASSETTYPES
@pytest.fixture
def asset_types(db):
"""Seed the asset types the create endpoints look up by name."""
from shopdb.core.models import AssetType
for name in ('computer', 'printer', 'network_device'):
db.session.add(AssetType(assettype=name, pluginname=name,
tablename=name, description=name))
db.session.commit()
# (endpoint prefix, extension key in the response, extra create fields)
PLUGIN_CASES = [
('/api/computers', 'computer', {}),
('/api/printers', 'printer', {}),
('/api/network', 'networkdevice', {}),
]
@pytest.mark.parametrize('prefix,extkey,extra', PLUGIN_CASES)
def test_gauge_maintenance_roundtrip(client, db, auth_headers, asset_types,
prefix, extkey, extra):
"""Gauge/maintenance refs persist on create and update for each type."""
payload = {
'assetnumber': f'AST-{extkey}',
'gaugelabreference': 'GL-1',
'maintenancereference': 'MNT-1',
**extra,
}
created = client.post(prefix, json=payload, headers=auth_headers)
assert created.status_code == 201, created.get_json()
body = created.get_json()['data']
assert body['gaugelabreference'] == 'GL-1'
assert body['maintenancereference'] == 'MNT-1'
extid = body[extkey][f'{extkey}id'] if extkey != 'networkdevice' \
else body[extkey]['networkdeviceid']
updated = client.put(f'{prefix}/{extid}',
json={'gaugelabreference': 'GL-2',
'maintenancereference': 'MNT-2'},
headers=auth_headers)
assert updated.status_code == 200, updated.get_json()
fetched = client.get(f'{prefix}/{extid}', headers=auth_headers).get_json()['data']
assert fetched['gaugelabreference'] == 'GL-2'
assert fetched['maintenancereference'] == 'MNT-2'
def test_per_type_identifier_seed_count(client, db, auth_headers):
"""Seeding produces one boolean setting per identifier x asset type."""
response = client.post('/api/settings/seed', headers=auth_headers)
assert response.status_code == 200
from shopdb.core.models import Setting
keys = {s.key for s in Setting.query.filter_by(category='identifiers').all()}
expected = {
f'identifier_{name}_{assettype}_enabled'
for name in IDENTIFIER_LABELS
for assettype in IDENTIFIER_ASSETTYPES
}
assert expected <= keys
assert len(expected) == len(IDENTIFIER_LABELS) * len(IDENTIFIER_ASSETTYPES)

View File

@@ -0,0 +1,209 @@
"""Import-mode behavior: admin-gated legacy timestamp + event backdating.
Covers deliverables of the import-API work:
- createddate/modifieddate passthrough on a reference entity (vendor) and an
asset plugin (computer), gated on admin + X-Import-Mode header.
- both negative cases: no header, and header sent by a non-admin.
- a natural-key exact-match lookup filter added for idempotent import.
- backdated USB checkout time in selfhosted mode.
"""
import pytest
from werkzeug.security import generate_password_hash
# Import the usb models at collection time so their tables register on the
# metadata before the per-test db fixture calls create_all. The usb plugin is
# disabled in the dev registry, so its tables would otherwise never be created.
from plugins.usb.models import USBDevice, USBCheckout # noqa: F401
IMPORT_HEADER = {'X-Import-Mode': 'true'}
LEGACY_CREATED = '2020-01-05 08:30:00'
LEGACY_MODIFIED = '2021-06-07T14:15:16'
def _headers(auth_headers, importmode=True):
merged = dict(auth_headers)
if importmode:
merged.update(IMPORT_HEADER)
return merged
@pytest.fixture
def asset_types(db):
"""Seed the asset types the plugin create endpoints look up by name."""
from shopdb.core.models import AssetType
for name in ('computer',):
db.session.add(AssetType(assettype=name, pluginname=name,
tablename=name, description=name))
db.session.commit()
@pytest.fixture
def creator_headers(client, db):
"""A NON-admin user granted only computers.create/edit, logged in.
Proves import mode requires admin: this user carries the write permission
but must NOT get timestamp passthrough.
"""
from shopdb.core.models import User, Role, Permission
perms = []
for name in ('computers.create', 'computers.edit'):
p = Permission(name=name, description=name, category='computers')
db.session.add(p)
perms.append(p)
role = Role(rolename='pcoperator', description='PC operator')
role.permissions.extend(perms)
db.session.add(role)
db.session.flush()
user = User(username='pcop', email='pcop@test.local',
passwordhash=generate_password_hash('testpass'))
user.roles.append(role)
db.session.add(user)
db.session.commit()
response = client.post('/api/auth/login',
json={'username': 'pcop', 'password': 'testpass'})
assert response.status_code == 200, response.get_json()
token = response.get_json()['data']['access_token']
return {'Authorization': f'Bearer {token}'}
# ---------------------------------------------------------------------------
# Timestamp passthrough (positive)
# ---------------------------------------------------------------------------
def test_admin_import_mode_preserves_vendor_timestamps(client, db, auth_headers):
"""Admin + header: vendor keeps the legacy createddate/modifieddate."""
payload = {'vendor': 'LegacyVendor',
'createddate': LEGACY_CREATED,
'modifieddate': LEGACY_MODIFIED}
response = client.post('/api/vendors', json=payload,
headers=_headers(auth_headers))
assert response.status_code == 201, response.get_json()
from shopdb.core.models import Vendor
v = Vendor.query.filter_by(vendor='LegacyVendor').first()
assert v.createddate.year == 2020 and v.createddate.month == 1
assert v.createddate.day == 5 and v.createddate.hour == 8
assert v.modifieddate.year == 2021 and v.modifieddate.month == 6
def test_admin_import_mode_preserves_computer_asset_timestamps(
client, db, auth_headers, asset_types):
"""Admin + header: the created asset row keeps its legacy createddate."""
payload = {'assetnumber': 'PC-IMPORT-1',
'hostname': 'legacy-pc',
'createddate': LEGACY_CREATED,
'modifieddate': LEGACY_MODIFIED}
response = client.post('/api/computers', json=payload,
headers=_headers(auth_headers))
assert response.status_code == 201, response.get_json()
from shopdb.core.models import Asset
asset = Asset.query.filter_by(assetnumber='PC-IMPORT-1').first()
assert asset.createddate.year == 2020
assert asset.modifieddate.year == 2021
# ---------------------------------------------------------------------------
# Timestamp passthrough (negative): ignored without header, or for non-admin
# ---------------------------------------------------------------------------
def test_no_header_ignores_timestamps(client, db, auth_headers):
"""Admin but NO X-Import-Mode header: createddate falls back to now."""
payload = {'vendor': 'NoHeaderVendor', 'createddate': LEGACY_CREATED}
response = client.post('/api/vendors', json=payload,
headers=_headers(auth_headers, importmode=False))
assert response.status_code == 201, response.get_json()
from shopdb.core.models import Vendor
v = Vendor.query.filter_by(vendor='NoHeaderVendor').first()
assert v.createddate.year >= 2024 # server-stamped now, not the 2020 payload
def test_non_admin_ignores_timestamps(client, db, creator_headers, asset_types):
"""Non-admin WITH the write permission + header: timestamps still ignored."""
payload = {'assetnumber': 'PC-NONADMIN-1',
'createddate': LEGACY_CREATED}
response = client.post('/api/computers', json=payload,
headers=_headers(creator_headers))
assert response.status_code == 201, response.get_json()
from shopdb.core.models import Asset
asset = Asset.query.filter_by(assetnumber='PC-NONADMIN-1').first()
assert asset.createddate.year >= 2024 # not backdated: caller is not admin
# ---------------------------------------------------------------------------
# Natural-key lookup filter (idempotency recipe support)
# ---------------------------------------------------------------------------
def test_vendor_exact_lookup_filter(client, db, auth_headers):
"""GET /api/vendors?vendor=<name> exact-matches one vendor for lookup."""
for name in ('Acme', 'AcmeTools', 'Beta'):
resp = client.post('/api/vendors', json={'vendor': name},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
listed = client.get('/api/vendors?vendor=Acme', headers=auth_headers)
assert listed.status_code == 200
rows = listed.get_json()['data']
assert len(rows) == 1
assert rows[0]['vendor'] == 'Acme'
# ---------------------------------------------------------------------------
# Backdated USB checkout (selfhosted mode)
#
# The usb plugin is disabled in the dev registry, so its blueprint may not be
# registered. Exercise the selfhosted layer directly inside a request context
# carrying the admin JWT + import header, which is what the route delegates to.
# ---------------------------------------------------------------------------
def _seed_usb_device(db, serialnumber):
from plugins.usb.models import USBDevice
device = USBDevice(serialnumber=serialnumber, ischeckedout=False, isactive=True)
db.session.add(device)
db.session.commit()
return device
def test_usb_backdated_checkout(app, db, auth_headers):
"""Admin + header: a USB checkout records the historical checkouttime."""
from plugins.usb.api import selfhosted
from plugins.usb.models import USBDevice, USBCheckout
device = _seed_usb_device(db, 'USB-IMPORT-1')
headers = _headers(auth_headers)
with app.test_request_context('/api/usb/USB-IMPORT-1/checkout',
method='POST', headers=headers):
resp = selfhosted.checkout_device(
'USB-IMPORT-1',
{'badge': '570005354', 'checkouttime': '2020-03-04 09:10:11'})
assert resp.status_code == 201, resp.get_json()
row = USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid).first()
assert row.checkouttime.year == 2020 and row.checkouttime.month == 3
reloaded = db.session.get(USBDevice, device.usbdeviceid)
assert reloaded.currentcheckoutdate.year == 2020
def test_usb_checkout_ignores_backdate_without_header(app, db, auth_headers):
"""No import header: the checkouttime override is ignored, uses now."""
from plugins.usb.api import selfhosted
from plugins.usb.models import USBDevice, USBCheckout
device = _seed_usb_device(db, 'USB-IMPORT-2')
with app.test_request_context('/api/usb/USB-IMPORT-2/checkout',
method='POST', headers=auth_headers):
resp = selfhosted.checkout_device(
'USB-IMPORT-2',
{'badge': '570005354', 'checkouttime': '2020-03-04 09:10:11'})
assert resp.status_code == 201, resp.get_json()
row = USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid).first()
assert row.checkouttime.year >= 2024 # server now, not the 2020 override

View File

@@ -0,0 +1,34 @@
"""Characterization tests for the Knowledge Base API.
Written before extracting Knowledge Base into a plugin (pinning-flask-behavior):
the endpoints must behave identically whether KB is a core blueprint or a plugin
blueprint, since both register at /api/knowledgebase.
"""
def test_create_list_get_article(client, db, auth_headers):
"""Create an article, see it in the list, fetch it by id."""
created = client.post('/api/knowledgebase',
json={'shortdescription': 'How to reset a printer',
'linkurl': 'https://kb.example/printer-reset',
'keywords': 'printer reset'},
headers=auth_headers)
assert created.status_code == 201, created.get_json()
linkid = created.get_json()['data']['linkid']
listing = client.get('/api/knowledgebase', headers=auth_headers)
assert listing.status_code == 200
ids = [a['linkid'] for a in listing.get_json()['data']]
assert linkid in ids
fetched = client.get(f'/api/knowledgebase/{linkid}', headers=auth_headers)
assert fetched.status_code == 200
assert fetched.get_json()['data']['shortdescription'] == 'How to reset a printer'
def test_create_requires_shortdescription(client, db, auth_headers):
"""shortdescription is required."""
resp = client.post('/api/knowledgebase',
json={'linkurl': 'https://kb.example/x'},
headers=auth_headers)
assert resp.status_code == 400

View File

@@ -0,0 +1,27 @@
"""Unit tests for the location operation-code derivation.
Locations carry no dedicated code column, so the operation code is the leading
whitespace-delimited token of the location name (e.g. '0615 Blisk Inspection'
-> '0615'). Printed asset labels for measuring tools encode this so every tool
at one operation shares one code.
"""
from shopdb.core.models.location import derive_locationcode
def test_derive_leading_numeric_token():
assert derive_locationcode('0615 Blisk Inspection') == '0615'
def test_derive_single_token_name():
assert derive_locationcode('Lab') == 'Lab'
def test_derive_strips_surrounding_whitespace():
assert derive_locationcode(' 0700 Final ') == '0700'
def test_derive_none_when_blank():
assert derive_locationcode('') is None
assert derive_locationcode(' ') is None
assert derive_locationcode(None) is None

View File

@@ -0,0 +1,149 @@
"""Tests for vendor-model photo upload, replace, delete, and serve.
Covers the core model-image endpoints: admin-gated upload/delete, the public
serve route, one-image-per-model replace semantics, and the guarantee that
deleting a model whose imageurl is an external URL clears the field without
touching the filesystem.
"""
import io
import os
from shopdb.core.models import Model
from shopdb.core.api.models import MODEL_IMAGE_URL_PREFIX
def _make_model(db, imageurl=None):
m = Model(modelnumber='TESTMODEL-1', imageurl=imageurl)
db.session.add(m)
db.session.commit()
return m
def test_upload_forbidden_for_non_admin(client, db, member_headers):
"""A role-less authenticated user cannot upload a model image."""
m = _make_model(db)
data = {'file': (io.BytesIO(b'<svg/>'), 'photo.svg')}
resp = client.post(f'/api/models/{m.modelnumberid}/image', data=data,
content_type='multipart/form-data', headers=member_headers)
assert resp.status_code == 403
assert resp.get_json()['data']['error']['code'] == 'FORBIDDEN'
def test_upload_sets_imageurl_and_file_exists(client, db, auth_headers, app):
"""Admin upload writes the served URL and the file lands in the instance dir."""
m = _make_model(db)
payload = b'\x89PNG\r\n\x1a\nfake-png-bytes'
data = {'file': (io.BytesIO(payload), 'photo.png')}
resp = client.post(f'/api/models/{m.modelnumberid}/image', data=data,
content_type='multipart/form-data', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
expected_url = f'{MODEL_IMAGE_URL_PREFIX}model-{m.modelnumberid}.png'
assert resp.get_json()['data']['imageurl'] == expected_url
refreshed = db.session.get(Model, m.modelnumberid)
assert refreshed.imageurl == expected_url
path = os.path.join(app.instance_path, 'modelimages', f'model-{m.modelnumberid}.png')
assert os.path.exists(path)
with open(path, 'rb') as handle:
assert handle.read() == payload
def test_upload_rejects_invalid_extension(client, db, auth_headers):
"""A disallowed file extension is rejected."""
m = _make_model(db)
data = {'file': (io.BytesIO(b'MZ...'), 'photo.exe')}
resp = client.post(f'/api/models/{m.modelnumberid}/image', data=data,
content_type='multipart/form-data', headers=auth_headers)
assert resp.status_code == 400
assert resp.get_json()['data']['error']['code'] == 'VALIDATION_ERROR'
def test_upload_missing_model_is_404(client, db, auth_headers):
"""Uploading to a nonexistent model id is a 404."""
data = {'file': (io.BytesIO(b'<svg/>'), 'photo.svg')}
resp = client.post('/api/models/999999/image', data=data,
content_type='multipart/form-data', headers=auth_headers)
assert resp.status_code == 404
def test_reupload_replaces_and_removes_old_extension(client, db, auth_headers, app):
"""Re-upload with a different extension deletes the prior file."""
m = _make_model(db)
first = {'file': (io.BytesIO(b'first'), 'photo.png')}
resp = client.post(f'/api/models/{m.modelnumberid}/image', data=first,
content_type='multipart/form-data', headers=auth_headers)
assert resp.status_code == 200
oldpath = os.path.join(app.instance_path, 'modelimages', f'model-{m.modelnumberid}.png')
assert os.path.exists(oldpath)
second = {'file': (io.BytesIO(b'second'), 'photo.jpg')}
resp = client.post(f'/api/models/{m.modelnumberid}/image', data=second,
content_type='multipart/form-data', headers=auth_headers)
assert resp.status_code == 200
# Old .png is gone; new .jpg exists and imageurl points at it.
assert not os.path.exists(oldpath)
newpath = os.path.join(app.instance_path, 'modelimages', f'model-{m.modelnumberid}.jpg')
assert os.path.exists(newpath)
refreshed = db.session.get(Model, m.modelnumberid)
assert refreshed.imageurl == f'{MODEL_IMAGE_URL_PREFIX}model-{m.modelnumberid}.jpg'
def test_serve_returns_bytes(client, db, auth_headers):
"""The public serve route returns the uploaded bytes without auth."""
m = _make_model(db)
payload = b'\x89PNG\r\n\x1a\nserved-bytes'
data = {'file': (io.BytesIO(payload), 'photo.png')}
client.post(f'/api/models/{m.modelnumberid}/image', data=data,
content_type='multipart/form-data', headers=auth_headers)
served = client.get(f'{MODEL_IMAGE_URL_PREFIX}model-{m.modelnumberid}.png')
assert served.status_code == 200
assert served.get_data() == payload
def test_delete_clears_field_and_removes_file(client, db, auth_headers, app):
"""Delete clears imageurl and removes the owned file."""
m = _make_model(db)
data = {'file': (io.BytesIO(b'bytes'), 'photo.png')}
client.post(f'/api/models/{m.modelnumberid}/image', data=data,
content_type='multipart/form-data', headers=auth_headers)
path = os.path.join(app.instance_path, 'modelimages', f'model-{m.modelnumberid}.png')
assert os.path.exists(path)
resp = client.delete(f'/api/models/{m.modelnumberid}/image', headers=auth_headers)
assert resp.status_code == 200
assert resp.get_json()['data']['imageurl'] is None
assert not os.path.exists(path)
refreshed = db.session.get(Model, m.modelnumberid)
assert refreshed.imageurl is None
def test_delete_external_url_clears_field_without_filesystem_error(client, db, auth_headers):
"""Delete on a model whose imageurl is an external URL just clears the field."""
m = _make_model(db, imageurl='https://example.com/product.png')
resp = client.delete(f'/api/models/{m.modelnumberid}/image', headers=auth_headers)
assert resp.status_code == 200
assert resp.get_json()['data']['imageurl'] is None
refreshed = db.session.get(Model, m.modelnumberid)
assert refreshed.imageurl is None
def test_delete_legacy_path_clears_field_without_filesystem_error(client, db, auth_headers):
"""Delete on a shipped /images/models/* path clears the field, touches no disk."""
m = _make_model(db, imageurl='/images/models/machines/legacy.png')
resp = client.delete(f'/api/models/{m.modelnumberid}/image', headers=auth_headers)
assert resp.status_code == 200
assert resp.get_json()['data']['imageurl'] is None
def test_delete_forbidden_for_non_admin(client, db, member_headers):
"""A role-less authenticated user cannot delete a model image."""
m = _make_model(db, imageurl='https://example.com/product.png')
resp = client.delete(f'/api/models/{m.modelnumberid}/image', headers=member_headers)
assert resp.status_code == 403

View File

@@ -0,0 +1,90 @@
"""Tests for the get_permissions hook consumer chain (contract 0.10.0).
Pins full_permission_catalog() and its consumers: the core catalog holds no
plugin permissions, an enabled plugin's permissions merge in and drop out when
disabled, seeding is idempotent, API-token scope validation follows the enabled
set, and the role grid groups plugin categories.
Plugin enabled-state is monkeypatched (not persisted) so these tests do not
mutate the shared instance/plugins.json registry file.
"""
from shopdb.core.models import Permission, full_permission_catalog
from shopdb.core.models.apitoken import ApiToken
# Permission-owning plugin domains that must NOT live in the core catalog.
PLUGIN_PREFIXES = ('machines.', 'computers.', 'printers.', 'network.', 'kb.',
'notifications.', 'usb.', 'warranty.', 'measuringtools.')
def test_core_catalog_excludes_plugin_permissions():
"""Permission.CORE_PERMISSIONS holds only genuinely core permissions."""
names = {n for n, _d, _c in Permission.CORE_PERMISSIONS}
for name in names:
assert not name.startswith(PLUGIN_PREFIXES), (
f'{name} is a plugin permission and must move to its plugin hook')
# Core sets stay put.
assert 'assets.edit' in names
assert 'settings.edit' in names
assert 'collector.ingest' in names
def test_enabled_plugin_permissions_merge(app, monkeypatch):
"""An enabled plugin's permissions appear in full_permission_catalog()."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
with app.app_context():
names = {n for n, _d, _c in full_permission_catalog()}
assert 'machines.edit' in names
assert 'printers.edit' in names
# core still present alongside plugin permissions
assert 'assets.edit' in names
def test_disabled_plugin_permissions_drop_out(app, monkeypatch):
"""A disabled plugin's permissions disappear from the catalog."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled',
lambda name: name != 'machines')
with app.app_context():
names = {n for n, _d, _c in full_permission_catalog()}
assert 'machines.edit' not in names
assert 'printers.edit' in names # other plugins still enabled
def test_seed_is_idempotent(app, db, monkeypatch):
"""Seeding the full catalog twice creates each permission once."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
with app.app_context():
first = Permission.seed()
db.session.commit()
assert first > len(Permission.CORE_PERMISSIONS) # plugins contributed
second = Permission.seed()
db.session.commit()
assert second == 0
def test_token_scope_validation_follows_enabled_set(app, monkeypatch):
"""A plugin permission is an accepted scope only while the plugin is enabled."""
pm = app.extensions['plugin_manager']
with app.app_context():
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
assert ApiToken.unknown_scope_names(['machines.edit']) == []
assert ApiToken.unknown_scope_names(['assets.edit']) == []
monkeypatch.setattr(pm.registry, 'is_enabled',
lambda name: name != 'machines')
assert ApiToken.unknown_scope_names(['machines.edit']) == ['machines.edit']
def test_list_permissions_endpoint_groups_plugin_category(
app, client, auth_headers, monkeypatch):
"""GET /api/users/permissions surfaces enabled plugins' categories."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
response = client.get('/api/users/permissions', headers=auth_headers)
assert response.status_code == 200, response.get_json()
grouped = response.get_json()['data']['grouped']
assert 'assets' in grouped # core category always present
assert 'machines' in grouped # enabled plugin category merged in

View File

@@ -0,0 +1,29 @@
"""GET /api/plugins/enabled: the flat enabled-plugin-name list that drives
frontend route gating (ADR-009).
Pins two properties: the endpoint is reachable anonymously and returns a
plain array of names, and a disabled plugin's name drops out of that array
(so its frontend routes gate off), mirroring the search-disabled pattern.
"""
def test_enabled_anonymous_returns_array(client):
"""No auth required (jwt-optional): returns a flat array of name strings."""
resp = client.get('/api/plugins/enabled')
assert resp.status_code == 200, resp.get_json()
data = resp.get_json()['data']
assert isinstance(data, list)
assert all(isinstance(name, str) for name in data)
def test_disabled_plugin_absent(app, client, monkeypatch):
"""A disabled plugin's name is missing; enabled ones remain present."""
pm = app.extensions['plugin_manager']
# Pretend printers is enabled but usb is not.
monkeypatch.setattr(pm.registry, 'get_enabled_plugins',
lambda: ['printers', 'computers'])
resp = client.get('/api/plugins/enabled')
assert resp.status_code == 200, resp.get_json()
data = resp.get_json()['data']
assert 'printers' in data
assert 'usb' not in data

View File

@@ -0,0 +1,144 @@
"""Tests for the ADR-010 frontend-contribution consumers (/api/pluginui/*).
Pins the wiring added for the four data-only presentation hooks
(get_settings_cards, get_asset_panels, get_map_overlays,
get_asset_presentation): each endpoint aggregates enabled plugins'
contributions, tags them with the plugin name, and drops disabled plugins.
Also asserts the pilot plugins (warranty, measuringtools) contribute.
Plugin enabled-state is monkeypatched (not persisted) so these tests do not
mutate the shared instance/plugins.json registry file.
"""
def _data(client, path, headers):
response = client.get(path, headers=headers)
assert response.status_code == 200, response.get_json()
payload = response.get_json()['data']
assert isinstance(payload, list)
return payload
# =============================================================================
# settings-cards
# =============================================================================
def test_settings_cards_aggregate_enabled_plugins(app, client, auth_headers, monkeypatch):
"""measuringtools contributes its settings card, tagged + position-sorted."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
cards = _data(client, '/api/pluginui/settings-cards', auth_headers)
plugins = {c.get('plugin') for c in cards}
assert 'measuringtools' in plugins
card = next(c for c in cards if c['plugin'] == 'measuringtools')
assert card['group'] == 'Measuring Tools'
assert card['to'] == '/settings/measuringtooltypes'
positions = [c.get('position', 99) for c in cards]
assert positions == sorted(positions)
def test_settings_cards_skip_disabled_plugin(app, client, auth_headers, monkeypatch):
"""A disabled measuringtools drops its settings card."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: name != 'measuringtools')
cards = _data(client, '/api/pluginui/settings-cards', auth_headers)
assert 'measuringtools' not in {c.get('plugin') for c in cards}
# =============================================================================
# asset-panels
# =============================================================================
def _make_asset(db):
from shopdb.core.models import Asset, AssetType
atype = AssetType.query.filter_by(assettype='computer').first()
if not atype:
atype = AssetType(assettype='computer', pluginname='computer',
tablename='computer', description='c')
db.session.add(atype)
db.session.commit()
asset = Asset(assetnumber='AST-PANEL01', assettypeid=atype.assettypeid,
isactive=True)
db.session.add(asset)
db.session.commit()
return asset.assetid
def test_asset_panels_require_assetid(client, auth_headers):
"""asset-panels without assetid is a 400."""
response = client.get('/api/pluginui/asset-panels', headers=auth_headers)
assert response.status_code == 400
def test_asset_panels_match_asset_type(app, client, db, auth_headers, monkeypatch):
"""Warranty's ['*'] panel matches any asset; tagged with its plugin."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
assetid = _make_asset(db)
panels = _data(client, f'/api/pluginui/asset-panels?assetid={assetid}', auth_headers)
warranty = next((p for p in panels if p.get('plugin') == 'warranty'), None)
assert warranty is not None, 'warranty asset panel missing'
assert warranty['id'] == 'warranty'
assert warranty['endpoint'] == '/api/warranty/asset/{assetid}'
assert warranty['render'] in ('keyvalue', 'table', 'badge')
def test_asset_panels_skip_disabled_plugin(app, client, db, auth_headers, monkeypatch):
"""A disabled warranty drops its asset panel."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: name != 'warranty')
assetid = _make_asset(db)
panels = _data(client, f'/api/pluginui/asset-panels?assetid={assetid}', auth_headers)
assert 'warranty' not in {p.get('plugin') for p in panels}
# =============================================================================
# map-overlays
# =============================================================================
def test_map_overlays_aggregate_enabled_plugins(app, client, auth_headers, monkeypatch):
"""measuringtools contributes its calibration-due overlay."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
overlays = _data(client, '/api/pluginui/map-overlays', auth_headers)
overlay = next((o for o in overlays if o.get('plugin') == 'measuringtools'), None)
assert overlay is not None
assert overlay['id'] == 'calibration-due'
assert overlay['legend'] is True
def test_map_overlays_skip_disabled_plugin(app, client, auth_headers, monkeypatch):
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: name != 'measuringtools')
overlays = _data(client, '/api/pluginui/map-overlays', auth_headers)
assert 'measuringtools' not in {o.get('plugin') for o in overlays}
# =============================================================================
# asset-presentation
# =============================================================================
def test_asset_presentation_aggregate_enabled_plugins(app, client, auth_headers, monkeypatch):
"""measuringtools declares presentation for its measuring_tool type."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
entries = _data(client, '/api/pluginui/asset-presentation', auth_headers)
entry = next((e for e in entries if e.get('assettype') == 'measuring_tool'), None)
assert entry is not None
assert entry['plugin'] == 'measuringtools'
assert entry['route'] == '/measuringtools/by-asset/{assetid}'
def test_asset_presentation_skip_disabled_plugin(app, client, auth_headers, monkeypatch):
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: name != 'measuringtools')
entries = _data(client, '/api/pluginui/asset-presentation', auth_headers)
assert 'measuring_tool' not in {e.get('assettype') for e in entries}

View File

@@ -0,0 +1,106 @@
"""Relationship-type directionality: CRUD carries isdirectional and the
per-asset relationships endpoint tags each row with its type directionality.
The frontend uses isdirectional to collapse symmetric-type rows (Dualpath and
other physical links) into one direction-blind card entry, and to render
directional types (controls, partof...) with source->target phrasing. That
dedupe is view logic verified live; here we lock the backend surface it needs.
"""
from shopdb.extensions import db
from shopdb.core.models import Asset, AssetType
from shopdb.core.models.relationship import RelationshipType, AssetRelationship
def _make_asset(assetnumber, typeid):
asset = Asset(assetnumber=assetnumber, assettypeid=typeid)
db.session.add(asset)
db.session.flush()
return asset
# ---------------------------------------------------------------------------
# Relationship type CRUD carries isdirectional
# ---------------------------------------------------------------------------
def test_create_type_defaults_directional(client, db, auth_headers):
resp = client.post('/api/assets/relationshiptypes',
json={'relationshiptype': 'Backup For'},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
assert resp.get_json()['data']['isdirectional'] is True
def test_create_symmetric_type(client, db, auth_headers):
resp = client.post('/api/assets/relationshiptypes',
json={'relationshiptype': 'Dualpath',
'isdirectional': False},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
assert resp.get_json()['data']['isdirectional'] is False
def test_update_toggles_directional(client, db, auth_headers):
typeid = client.post('/api/assets/relationshiptypes',
json={'relationshiptype': 'USB'},
headers=auth_headers).get_json()['data']['relationshiptypeid']
resp = client.put(f'/api/assets/relationshiptypes/{typeid}',
json={'isdirectional': False}, headers=auth_headers)
assert resp.status_code == 200
assert resp.get_json()['data']['isdirectional'] is False
def test_list_includes_directional(client, db, auth_headers):
client.post('/api/assets/relationshiptypes',
json={'relationshiptype': 'WiFi', 'isdirectional': False},
headers=auth_headers)
resp = client.get('/api/assets/relationshiptypes')
assert resp.status_code == 200
row = next(t for t in resp.get_json()['data']
if t['relationshiptype'] == 'WiFi')
assert row['isdirectional'] is False
# ---------------------------------------------------------------------------
# Per-asset relationships endpoint tags each row with isdirectional
# ---------------------------------------------------------------------------
def test_relationships_row_carries_isdirectional(client, db, auth_headers):
atype = AssetType(assettype='machine')
db.session.add(atype)
db.session.flush()
symmetric = RelationshipType(relationshiptype='Dualpath', isdirectional=False)
directional = RelationshipType(relationshiptype='controls', isdirectional=True)
db.session.add_all([symmetric, directional])
db.session.flush()
a = _make_asset('M-A', atype.assettypeid)
b = _make_asset('M-B', atype.assettypeid)
# symmetric link stored in both directions (legacy shape)
db.session.add(AssetRelationship(sourceassetid=a.assetid, targetassetid=b.assetid,
relationshiptypeid=symmetric.relationshiptypeid))
db.session.add(AssetRelationship(sourceassetid=b.assetid, targetassetid=a.assetid,
relationshiptypeid=symmetric.relationshiptypeid))
# directional link a controls b
db.session.add(AssetRelationship(sourceassetid=a.assetid, targetassetid=b.assetid,
relationshiptypeid=directional.relationshiptypeid))
db.session.commit()
resp = client.get(f'/api/assets/{a.assetid}/relationships')
assert resp.status_code == 200
data = resp.get_json()['data']
for row in data['outgoing']:
assert 'isdirectional' in row
for row in data['incoming']:
assert 'isdirectional' in row
# symmetric type reports False both directions; directional reports True
sym_rows = [r for r in data['outgoing'] + data['incoming']
if r['relationshiptypename'] == 'Dualpath']
assert sym_rows and all(r['isdirectional'] is False for r in sym_rows)
ctrl_rows = [r for r in data['outgoing']
if r['relationshiptypename'] == 'controls']
assert ctrl_rows and all(r['isdirectional'] is True for r in ctrl_rows)

View File

@@ -0,0 +1,295 @@
"""Relationship propagation across symmetric rails (ADR-001, wired).
A Dualpath pair is one physical dual-bay machine with a single controller PC
and a bay-selector switch, so BOTH bay records must carry the controls link to
that PC. Creating PC controls bayA where bayA has a Dualpath partner bayB
auto-creates PC controls bayB. The propagation set is M:N
(relationshiptypepropagations): controls propagates through Dualpath
(symmetric, consumed) and through partof (directional, declared but inert).
"""
from shopdb.extensions import db
from shopdb.core.models import Asset, AssetType
from shopdb.core.models.relationship import (
RelationshipType,
RelationshipTypePropagation,
AssetRelationship,
)
def _make_asset(assetnumber, typeid):
asset = Asset(assetnumber=assetnumber, assettypeid=typeid)
db.session.add(asset)
db.session.flush()
return asset
def _setup_types():
atype = AssetType(assettype='machine')
db.session.add(atype)
db.session.flush()
controls = RelationshipType(relationshiptype='controls', isdirectional=True)
dualpath = RelationshipType(relationshiptype='Dualpath', isdirectional=False)
partof = RelationshipType(relationshiptype='partof', isdirectional=True)
db.session.add_all([controls, dualpath, partof])
db.session.flush()
return atype, controls, dualpath, partof
def _rel_exists(sourceid, targetid, typeid):
return AssetRelationship.query.filter_by(
sourceassetid=sourceid, targetassetid=targetid, relationshiptypeid=typeid
).first() is not None
# ---------------------------------------------------------------------------
# Symmetric through-type propagates (controls fans across a Dualpath pair)
# ---------------------------------------------------------------------------
def test_controls_propagates_across_dualpath_pair(client, db, auth_headers):
atype, controls, dualpath, partof = _setup_types()
db.session.add(RelationshipTypePropagation(
relationshiptypeid=controls.relationshiptypeid,
throughtypeid=dualpath.relationshiptypeid))
pc = _make_asset('PC-1', atype.assettypeid)
baya = _make_asset('BAY-A', atype.assettypeid)
bayb = _make_asset('BAY-B', atype.assettypeid)
# the two bays are Dualpath partners (one physical machine)
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
relationshiptypeid=dualpath.relationshiptypeid))
db.session.commit()
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
'sourceassetid': pc.assetid,
'targetassetid': baya.assetid,
'relationshiptypeid': controls.relationshiptypeid,
})
assert resp.status_code == 201, resp.get_json()
assert resp.get_json()['data']['propagatedcount'] == 1
# PC now controls BOTH bays
assert _rel_exists(pc.assetid, baya.assetid, controls.relationshiptypeid)
assert _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
def test_propagation_finds_partner_stored_either_direction(client, db, auth_headers):
# Dualpath row stored partner->target; propagation must still catch it.
atype, controls, dualpath, partof = _setup_types()
db.session.add(RelationshipTypePropagation(
relationshiptypeid=controls.relationshiptypeid,
throughtypeid=dualpath.relationshiptypeid))
pc = _make_asset('PC-2', atype.assettypeid)
baya = _make_asset('BAY-A2', atype.assettypeid)
bayb = _make_asset('BAY-B2', atype.assettypeid)
# note: bayb is the SOURCE of the Dualpath row, target is baya
db.session.add(AssetRelationship(sourceassetid=bayb.assetid, targetassetid=baya.assetid,
relationshiptypeid=dualpath.relationshiptypeid))
db.session.commit()
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
'sourceassetid': pc.assetid,
'targetassetid': baya.assetid,
'relationshiptypeid': controls.relationshiptypeid,
})
assert resp.status_code == 201
assert _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
def test_controls_propagates_when_bay_is_source(client, db, auth_headers):
# Matches real dev data: controls is stored bay -> PC (source is the bay).
# The bay's Dualpath partner must also get a controls -> PC row.
atype, controls, dualpath, partof = _setup_types()
db.session.add(RelationshipTypePropagation(
relationshiptypeid=controls.relationshiptypeid,
throughtypeid=dualpath.relationshiptypeid))
pc = _make_asset('PC-S', atype.assettypeid)
baya = _make_asset('BAY-AS', atype.assettypeid)
bayb = _make_asset('BAY-BS', atype.assettypeid)
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
relationshiptypeid=dualpath.relationshiptypeid))
db.session.commit()
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
'sourceassetid': baya.assetid,
'targetassetid': pc.assetid,
'relationshiptypeid': controls.relationshiptypeid,
})
assert resp.status_code == 201, resp.get_json()
assert resp.get_json()['data']['propagatedcount'] == 1
# partner bay now also controls the PC, same direction (bay -> PC)
assert _rel_exists(bayb.assetid, pc.assetid, controls.relationshiptypeid)
# ---------------------------------------------------------------------------
# Idempotent: pre-existing propagated row is not duplicated
# ---------------------------------------------------------------------------
def test_propagation_idempotent(client, db, auth_headers):
atype, controls, dualpath, partof = _setup_types()
db.session.add(RelationshipTypePropagation(
relationshiptypeid=controls.relationshiptypeid,
throughtypeid=dualpath.relationshiptypeid))
pc = _make_asset('PC-3', atype.assettypeid)
baya = _make_asset('BAY-A3', atype.assettypeid)
bayb = _make_asset('BAY-B3', atype.assettypeid)
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
relationshiptypeid=dualpath.relationshiptypeid))
# the mirrored row already exists
db.session.add(AssetRelationship(sourceassetid=pc.assetid, targetassetid=bayb.assetid,
relationshiptypeid=controls.relationshiptypeid))
db.session.commit()
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
'sourceassetid': pc.assetid,
'targetassetid': baya.assetid,
'relationshiptypeid': controls.relationshiptypeid,
})
assert resp.status_code == 201
assert resp.get_json()['data']['propagatedcount'] == 0
count = AssetRelationship.query.filter_by(
sourceassetid=pc.assetid, targetassetid=bayb.assetid,
relationshiptypeid=controls.relationshiptypeid).count()
assert count == 1
# ---------------------------------------------------------------------------
# Non-propagating type creates no extras
# ---------------------------------------------------------------------------
def test_non_propagating_type_no_extras(client, db, auth_headers):
atype, controls, dualpath, partof = _setup_types()
# controls has NO propagation rows here
pc = _make_asset('PC-4', atype.assettypeid)
baya = _make_asset('BAY-A4', atype.assettypeid)
bayb = _make_asset('BAY-B4', atype.assettypeid)
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
relationshiptypeid=dualpath.relationshiptypeid))
db.session.commit()
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
'sourceassetid': pc.assetid,
'targetassetid': baya.assetid,
'relationshiptypeid': controls.relationshiptypeid,
})
assert resp.status_code == 201
assert resp.get_json()['data']['propagatedcount'] == 0
assert not _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
# ---------------------------------------------------------------------------
# Directional through-type (partof) does NOT propagate (deferred)
# ---------------------------------------------------------------------------
def test_directional_through_type_does_not_propagate(client, db, auth_headers):
atype, controls, dualpath, partof = _setup_types()
# controls propagates only through partof, which is directional -> inert
db.session.add(RelationshipTypePropagation(
relationshiptypeid=controls.relationshiptypeid,
throughtypeid=partof.relationshiptypeid))
pc = _make_asset('PC-5', atype.assettypeid)
parent = _make_asset('PARENT-5', atype.assettypeid)
child = _make_asset('CHILD-5', atype.assettypeid)
db.session.add(AssetRelationship(sourceassetid=child.assetid, targetassetid=parent.assetid,
relationshiptypeid=partof.relationshiptypeid))
db.session.commit()
resp = client.post('/api/assets/relationships', headers=auth_headers, json={
'sourceassetid': pc.assetid,
'targetassetid': parent.assetid,
'relationshiptypeid': controls.relationshiptypeid,
})
assert resp.status_code == 201
assert resp.get_json()['data']['propagatedcount'] == 0
assert not _rel_exists(pc.assetid, child.assetid, controls.relationshiptypeid)
# ---------------------------------------------------------------------------
# CLI backfill creates missing rows and is idempotent run twice
# ---------------------------------------------------------------------------
def test_cli_backfill_propagates_and_is_idempotent(client, db, auth_headers, runner):
atype, controls, dualpath, partof = _setup_types()
db.session.add(RelationshipTypePropagation(
relationshiptypeid=controls.relationshiptypeid,
throughtypeid=dualpath.relationshiptypeid))
pc = _make_asset('PC-6', atype.assettypeid)
baya = _make_asset('BAY-A6', atype.assettypeid)
bayb = _make_asset('BAY-B6', atype.assettypeid)
db.session.add(AssetRelationship(sourceassetid=baya.assetid, targetassetid=bayb.assetid,
relationshiptypeid=dualpath.relationshiptypeid))
# legacy-import shape: controls exists only on the primary bay
db.session.add(AssetRelationship(sourceassetid=pc.assetid, targetassetid=baya.assetid,
relationshiptypeid=controls.relationshiptypeid))
db.session.commit()
assert not _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
result = runner.invoke(args=['relationships', 'propagate'])
assert result.exit_code == 0, result.output
assert 'Propagated 1' in result.output
assert _rel_exists(pc.assetid, bayb.assetid, controls.relationshiptypeid)
# second run creates nothing more
result2 = runner.invoke(args=['relationships', 'propagate'])
assert result2.exit_code == 0, result2.output
assert 'Propagated 0' in result2.output
count = AssetRelationship.query.filter_by(
sourceassetid=pc.assetid, targetassetid=bayb.assetid,
relationshiptypeid=controls.relationshiptypeid).count()
assert count == 1
# ---------------------------------------------------------------------------
# CLI fix-controls-direction flips reversed legacy rows to PC -> machine
# ---------------------------------------------------------------------------
def test_cli_fix_controls_direction(client, db, auth_headers, runner):
atype, controls, dualpath, partof = _setup_types()
ctype = AssetType(assettype='computer')
db.session.add(ctype)
db.session.flush()
pc = _make_asset('PC-7', ctype.assettypeid)
pc2 = _make_asset('PC-8', ctype.assettypeid)
bay = _make_asset('BAY-7', atype.assettypeid)
bay2 = _make_asset('BAY-8', atype.assettypeid)
# reversed legacy row: machine -> PC
db.session.add(AssetRelationship(sourceassetid=bay.assetid, targetassetid=pc.assetid,
relationshiptypeid=controls.relationshiptypeid))
# reversed row whose flip already exists -> reversed one gets deactivated
db.session.add(AssetRelationship(sourceassetid=bay2.assetid, targetassetid=pc2.assetid,
relationshiptypeid=controls.relationshiptypeid))
db.session.add(AssetRelationship(sourceassetid=pc2.assetid, targetassetid=bay2.assetid,
relationshiptypeid=controls.relationshiptypeid))
db.session.commit()
result = runner.invoke(args=['relationships', 'fix-controls-direction'])
assert result.exit_code == 0, result.output
assert 'Flipped 1' in result.output
assert '1 reversed duplicate(s) deactivated' in result.output
# first row now reads PC -> machine
assert _rel_exists(pc.assetid, bay.assetid, controls.relationshiptypeid)
assert not _rel_exists(bay.assetid, pc.assetid, controls.relationshiptypeid)
# duplicate pair: canonical row stays active, reversed one deactivated
rev = AssetRelationship.query.filter_by(
sourceassetid=bay2.assetid, targetassetid=pc2.assetid,
relationshiptypeid=controls.relationshiptypeid).first()
assert rev is not None and rev.isactive is False
# idempotent: second run finds nothing
result2 = runner.invoke(args=['relationships', 'fix-controls-direction'])
assert result2.exit_code == 0, result2.output
assert 'Flipped 0' in result2.output
assert '0 reversed duplicate(s) deactivated' in result2.output

View File

@@ -0,0 +1,55 @@
"""Tests for the get_reports hook consumer (GET /api/reports).
Pins the wiring added for BasePlugin.get_reports: the reports list merges
enabled plugins' report cards after the static core reports and skips disabled
ones. Also guards the removed legacy warranty-status core report.
Plugin enabled-state is monkeypatched (not persisted) so these tests do not
mutate the shared instance/plugins.json registry file.
"""
def _report_ids(client, headers):
response = client.get('/api/reports', headers=headers)
assert response.status_code == 200, response.get_json()
reports = response.get_json()['data']['reports']
assert isinstance(reports, list)
return reports, {r['id'] for r in reports}
def test_reports_include_core_reports(client, auth_headers):
"""The static core reports are always listed."""
_, ids = _report_ids(client, auth_headers)
assert 'machines-by-type' in ids
assert 'pc-relationships' in ids
def test_legacy_warranty_status_report_absent(client, auth_headers):
"""The dead warranty-status core report was removed."""
_, ids = _report_ids(client, auth_headers)
assert 'warranty-status' not in ids
def test_enabled_plugin_reports_are_merged(app, client, auth_headers, monkeypatch):
"""Enabled plugins implementing get_reports contribute their cards."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
reports, ids = _report_ids(client, auth_headers)
# warranty + printers implement get_reports
assert 'warranty' in ids
assert 'toner' in ids
# merged cards carry their originating plugin name
warranty_card = next(r for r in reports if r['id'] == 'warranty')
assert warranty_card['plugin'] == 'warranty'
assert warranty_card['route'] == '/reports/warranty'
def test_disabled_plugin_reports_drop_out(app, client, auth_headers, monkeypatch):
"""A disabled plugin's report cards disappear from the list."""
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: name != 'warranty')
_, ids = _report_ids(client, auth_headers)
assert 'warranty' not in ids
assert 'toner' in ids # printers still enabled

View File

@@ -0,0 +1,114 @@
"""Global search matches custom-field VALUES for fields flagged searchable.
Pins: a searchable custom field's value routes to its asset; a non-searchable
field is NOT matched; disabling the asset's search domain excludes the hit; an
asset matched by BOTH its assetnumber and a custom field appears once; an
inactive field is excluded.
Search by a distinctive value string so only the custom-field domain can match
(the value is not any built-in identifier).
"""
from shopdb.core.api.settings import invalidate_settings_cache
def _seed_computer(client, db, auth_headers, assetnumber):
"""Create a computer asset and return its assetid."""
from shopdb.core.models import AssetType, Asset
if not AssetType.query.filter_by(assettype='computer').first():
db.session.add(AssetType(assettype='computer', pluginname='computer',
tablename='computer', description='c'))
db.session.commit()
resp = client.post('/api/computers',
json={'assetnumber': assetnumber},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
asset = Asset.query.filter_by(assetnumber=assetnumber).first()
return asset.assetid
def _seed_field(db, label, searchable=True, isactive=True):
"""Create a computer-scoped custom field and return its fieldid."""
from shopdb.core.models import AssetType, CustomField
assettype = AssetType.query.filter_by(assettype='computer').first()
field = CustomField(assettypeid=assettype.assettypeid,
fieldkey=label.replace(' ', '').lower(),
label=label, datatype='text',
searchable=searchable, isactive=isactive)
db.session.add(field)
db.session.commit()
return field.fieldid
def _set_value(db, fieldid, assetid, value):
from shopdb.core.models import CustomFieldValue
db.session.add(CustomFieldValue(fieldid=fieldid, assetid=assetid, value=value))
db.session.commit()
def _hits(client, auth_headers, term):
resp = client.get(f'/api/search?q={term}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
return resp.get_json()['data']['results']
def test_searchable_field_value_matches_and_routes(client, db, auth_headers):
"""A searchable custom-field value matches and routes to the asset page."""
assetid = _seed_computer(client, db, auth_headers, 'AST-CF01')
fieldid = _seed_field(db, 'Tool Crib Number', searchable=True)
_set_value(db, fieldid, assetid, 'cribval01')
hits = [r for r in _hits(client, auth_headers, 'cribval01')
if r['type'] == 'computer']
assert len(hits) == 1
assert hits[0]['url'].startswith('/pcs/')
assert hits[0]['subtitle'] == 'Tool Crib Number'
def test_non_searchable_field_value_does_not_match(client, db, auth_headers):
"""A value on a NON-searchable field is not matched by global search."""
assetid = _seed_computer(client, db, auth_headers, 'AST-CF02')
fieldid = _seed_field(db, 'Internal Note', searchable=False)
_set_value(db, fieldid, assetid, 'cribval02')
assert _hits(client, auth_headers, 'cribval02') == []
def test_disabled_domain_excludes_customfield_hit(client, db, auth_headers):
"""search_computer_enabled=false hides a custom-field hit on a computer."""
from shopdb.core.models import Setting
assetid = _seed_computer(client, db, auth_headers, 'AST-CF03')
fieldid = _seed_field(db, 'Bench Tag', searchable=True)
_set_value(db, fieldid, assetid, 'cribval03')
# Sanity: enabled by default.
assert [r for r in _hits(client, auth_headers, 'cribval03')
if r['type'] == 'computer']
Setting.set('search_computer_enabled', False, valuetype='boolean',
category='search')
invalidate_settings_cache()
assert [r for r in _hits(client, auth_headers, 'cribval03')
if r['type'] == 'computer'] == []
def test_dedupe_assetnumber_and_customfield(client, db, auth_headers):
"""An asset matched by both assetnumber and a custom field appears once."""
assetid = _seed_computer(client, db, auth_headers, 'cribval04')
fieldid = _seed_field(db, 'Crib Ref', searchable=True)
_set_value(db, fieldid, assetid, 'cribval04')
hits = [r for r in _hits(client, auth_headers, 'cribval04')
if r['type'] == 'computer']
assert len(hits) == 1
# The higher-relevance built-in assetnumber match wins the dedupe.
assert hits[0]['relevance'] == 100
def test_inactive_field_excluded(client, db, auth_headers):
"""A value on an inactive (searchable) field is not matched."""
assetid = _seed_computer(client, db, auth_headers, 'AST-CF05')
fieldid = _seed_field(db, 'Retired Tag', searchable=True, isactive=False)
_set_value(db, fieldid, assetid, 'cribval05')
assert _hits(client, auth_headers, 'cribval05') == []

View File

@@ -0,0 +1,43 @@
"""Global search must exclude rows from disabled plugins (search.py _require_enabled).
Pins the data-exposure property: disabling a plugin removes its plugin-specific
matches (e.g. hostname) from /api/search, mirroring the plugin being absent.
Search by HOSTNAME (distinct from assetnumber) so only the gated hostname domain
can match - the core asset-number domain is intentionally always-on.
"""
def _seed_computer(client, db, auth_headers, assetnumber, hostname):
from shopdb.core.models import AssetType
if not AssetType.query.filter_by(assettype='computer').first():
db.session.add(AssetType(assettype='computer', pluginname='computer',
tablename='computer', description='c'))
db.session.commit()
resp = client.post('/api/computers',
json={'assetnumber': assetnumber, 'hostname': hostname},
headers=auth_headers)
assert resp.status_code == 201, resp.get_json()
def _computer_hits(client, auth_headers, term):
resp = client.get(f'/api/search?q={term}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
results = resp.get_json()['data']['results']
return [r for r in results if r.get('type') == 'computer']
def test_search_includes_enabled_plugin(app, client, db, auth_headers, monkeypatch):
"""An enabled plugin's hostname matches appear in search."""
_seed_computer(client, db, auth_headers, 'AST-SD01', 'hostsearch01')
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: True)
assert _computer_hits(client, auth_headers, 'hostsearch01')
def test_search_excludes_disabled_plugin(app, client, db, auth_headers, monkeypatch):
"""A disabled plugin's hostname matches drop out of search results."""
_seed_computer(client, db, auth_headers, 'AST-SD02', 'hostsearch02')
pm = app.extensions['plugin_manager']
monkeypatch.setattr(pm.registry, 'is_enabled', lambda name: name != 'computers')
assert _computer_hits(client, auth_headers, 'hostsearch02') == []

View File

@@ -0,0 +1,105 @@
"""Settings-driven search integrations (ServiceNow + employee-ID pattern).
The ServiceNow ticket prefixes, the ServiceNow search URL/enabled flag, and the
employee-ID recognition pattern all ship as GE defaults but are configurable
per-site via Settings. These tests pin that the search endpoint honors those
settings and that a bad employeeid_pattern regex falls back rather than 500ing.
Settings are cached, so every case seeds its rows and calls
invalidate_settings_cache() before exercising search.
"""
from shopdb.core.models import Setting
from shopdb.core.api.settings import invalidate_settings_cache
from shopdb.core.api.search import _get_search_integrations, _classify_query
def _redirect(client, auth_headers, term):
resp = client.get(f'/api/search?q={term}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
return resp.get_json()['data'].get('redirect')
def test_default_prefix_classifies_servicenow(client, db, auth_headers):
"""With no override, a GE-prefixed ticket redirects to ServiceNow."""
invalidate_settings_cache()
redirect = _redirect(client, auth_headers, 'GEINC123')
assert redirect is not None
assert redirect['type'] == 'servicenow'
def test_custom_prefix_classifies_servicenow(client, db, auth_headers):
"""A site-configured prefix classifies its tickets as ServiceNow."""
Setting.set('servicenow_ticket_prefixes', 'ACMEINC', valuetype='string',
category='integrations')
invalidate_settings_cache()
redirect = _redirect(client, auth_headers, 'ACMEINC123')
assert redirect is not None
assert redirect['type'] == 'servicenow'
# Ticket is url-encoded into the {ticket} placeholder.
assert 'ACMEINC123' in redirect['url']
def test_custom_prefix_ignores_default_prefix(client, db, auth_headers):
"""Overriding the prefixes drops the built-in GE prefixes."""
Setting.set('servicenow_ticket_prefixes', 'ACMEINC', valuetype='string',
category='integrations')
invalidate_settings_cache()
redirect = _redirect(client, auth_headers, 'GEINC123')
# GEINC is no longer a configured prefix, so no ServiceNow redirect.
assert redirect is None or redirect.get('type') != 'servicenow'
def test_disabled_servicenow_yields_no_redirect(client, db, auth_headers):
"""servicenow_enabled=false suppresses the ServiceNow redirect entirely."""
Setting.set('servicenow_enabled', False, valuetype='boolean',
category='integrations')
invalidate_settings_cache()
redirect = _redirect(client, auth_headers, 'GEINC123')
assert redirect is None or redirect.get('type') != 'servicenow'
def test_blank_search_url_yields_no_redirect(client, db, auth_headers):
"""An empty servicenow_search_url disables the redirect."""
Setting.set('servicenow_search_url', '', valuetype='string',
category='integrations')
invalidate_settings_cache()
redirect = _redirect(client, auth_headers, 'GEINC123')
assert redirect is None or redirect.get('type') != 'servicenow'
def test_custom_employeeid_pattern_recognizes_six_digits(client, db, auth_headers):
"""A ^\\d{6}$ pattern makes six-digit IDs classify as employee queries."""
Setting.set('employeeid_pattern', r'^\d{6}$', valuetype='string',
category='site')
invalidate_settings_cache()
integrations = _get_search_integrations()
assert _classify_query('123456', integrations)['is_sso'] is True
# Anchored pattern rejects the old nine-digit shape.
assert _classify_query('123456789', integrations)['is_sso'] is False
def test_default_employeeid_pattern_recognizes_nine_digits(client, db, auth_headers):
"""The shipped default recognizes nine-digit SSO IDs."""
invalidate_settings_cache()
integrations = _get_search_integrations()
assert _classify_query('123456789', integrations)['is_sso'] is True
assert _classify_query('123456', integrations)['is_sso'] is False
def test_invalid_employeeid_pattern_falls_back(client, db, auth_headers):
"""A malformed regex falls back to nine-digit matching without erroring."""
Setting.set('employeeid_pattern', '[', valuetype='string', category='site')
invalidate_settings_cache()
# Must not raise on compile of the bad pattern.
integrations = _get_search_integrations()
assert _classify_query('123456789', integrations)['is_sso'] is True
assert _classify_query('12345', integrations)['is_sso'] is False
def test_invalid_pattern_search_does_not_500(client, db, auth_headers):
"""End-to-end: a bad employeeid_pattern must not break the search endpoint."""
Setting.set('employeeid_pattern', '(', valuetype='string', category='site')
invalidate_settings_cache()
resp = client.get('/api/search?q=hello', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()

View File

@@ -0,0 +1,44 @@
"""Per-domain global-search toggles (search_<type>_enabled).
A content type can be hidden from search independently of whether its plugin is
enabled, by setting search_<type>_enabled = false.
"""
from shopdb.core.api.settings import SEARCH_DOMAINS
def _kb_types(client, auth_headers, term):
resp = client.get(f'/api/search?q={term}', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
return [r for r in resp.get_json()['data']['results']
if r.get('type') == 'knowledgebase']
def test_domain_included_by_default(client, db, auth_headers):
"""With no toggle set, the knowledgebase domain appears in search."""
client.post('/api/knowledgebase',
json={'shortdescription': 'searchtoggle alpha',
'linkurl': 'https://kb.example/a'},
headers=auth_headers)
assert _kb_types(client, auth_headers, 'searchtoggle')
def test_disabled_domain_excluded(client, db, auth_headers):
"""search_knowledgebase_enabled=false hides KB results."""
from shopdb.core.models import Setting
client.post('/api/knowledgebase',
json={'shortdescription': 'searchtoggle beta',
'linkurl': 'https://kb.example/b'},
headers=auth_headers)
Setting.set('search_knowledgebase_enabled', False, valuetype='boolean',
category='search')
assert _kb_types(client, auth_headers, 'searchtoggle') == []
def test_search_seed_creates_domain_keys(client, db, auth_headers):
"""Seeding produces one boolean setting per search domain."""
resp = client.post('/api/settings/seed', headers=auth_headers)
assert resp.status_code == 200
from shopdb.core.models import Setting
keys = {s.key for s in Setting.query.filter_by(category='search').all()}
assert {f'search_{k}_enabled' for k in SEARCH_DOMAINS} <= keys

View File

@@ -0,0 +1,56 @@
"""Tests for the Setting key-value model, incl. the create-race recovery."""
from sqlalchemy.exc import IntegrityError
from shopdb.core.models import Setting
from shopdb.extensions import db as _db
def test_set_creates_then_updates(db):
"""set() creates a missing key, then updates the existing row in place."""
created = Setting.set('greeting', 'hello', category='site')
assert created.settingid is not None
assert Setting.get('greeting') == 'hello'
Setting.set('greeting', 'goodbye')
assert Setting.get('greeting') == 'goodbye'
# Only one row for the key (update, not a second insert).
assert Setting.query.filter_by(key='greeting').count() == 1
def test_set_boolean_roundtrip(db):
"""Booleans store as canonical strings and read back typed."""
Setting.set('flag', True, valuetype='boolean')
assert Setting.get('flag') is True
Setting.set('flag', False)
assert Setting.get('flag') is False
def test_set_recovers_from_create_race(db, monkeypatch):
"""A concurrent insert of the same key makes our commit raise IntegrityError;
set() rolls back, re-fetches the winner's row, and applies our value."""
real_commit = _db.session.commit
calls = {'n': 0}
def flaky_commit():
calls['n'] += 1
if calls['n'] == 1:
# Drop our own pending insert, then persist the "winner" row exactly
# as a racing transaction would have, and simulate the unique-key
# violation our losing INSERT would raise.
_db.session.rollback()
winner = Setting(key='racekey', value='winner', valuetype='string')
_db.session.add(winner)
real_commit()
raise IntegrityError('INSERT INTO settings', {}, Exception('duplicate key'))
return real_commit()
monkeypatch.setattr(_db.session, 'commit', flaky_commit)
result = Setting.set('racekey', 'mine')
# Recovery path ran: the row exists once and carries our value.
assert result is not None
assert Setting.get('racekey') == 'mine'
assert Setting.query.filter_by(key='racekey').count() == 1

View File

@@ -0,0 +1,192 @@
"""Tests for branding-logo upload and settings-secret masking.
Covers WP1 of the multi-site distribution work: the branding-logo upload
endpoint (admin-only, kind/extension validation, round-trip), the public
serve route, the secret-masking guarantee on the settings list, and the
canonical new-settings defaults.
"""
import io
from shopdb.core.api.settings import (
build_default_settings,
BRANDING_KIND_SETTINGS,
)
def _defaults_by_key():
return {d['key']: d for d in build_default_settings()}
def test_anon_settings_list_excludes_nonpublic(client, db):
"""Anonymous GET /api/settings returns only the public allowlist - secret
AND non-secret internal keys are absent, so plaintext can never leak."""
from shopdb.core.models import Setting
Setting.set('smtp_password', 'supersecret', valuetype='string', category='email')
Setting.set('zabbix_token', 'tok-abc-123', valuetype='string', category='integrations')
Setting.set('warranty_dell_clientsecret', 'shh', valuetype='string', category='integrations')
# A public key stays visible for contrast.
Setting.set('facility_name', 'Test Plant', valuetype='string', category='site')
resp = client.get('/api/settings')
assert resp.status_code == 200, resp.get_json()
by_key = {s['key']: s['value'] for s in resp.get_json()['data']}
# Non-public keys are not enumerable at all (stronger than masking).
assert 'smtp_password' not in by_key
assert 'zabbix_token' not in by_key
assert 'warranty_dell_clientsecret' not in by_key
# Plaintext secrets must not leak anywhere in the response body.
assert 'supersecret' not in resp.get_data(as_text=True)
assert 'tok-abc-123' not in resp.get_data(as_text=True)
# A public key passes through untouched.
assert by_key['facility_name'] == 'Test Plant'
def test_branding_upload_forbidden_for_non_admin(client, db, member_headers):
"""A role-less authenticated user cannot upload a branding logo."""
data = {
'kind': 'site',
'file': (io.BytesIO(b'<svg/>'), 'logo.svg'),
}
resp = client.post('/api/settings/branding-logo', data=data,
content_type='multipart/form-data', headers=member_headers)
assert resp.status_code == 403
assert resp.get_json()['data']['error']['code'] == 'FORBIDDEN'
def test_branding_upload_requires_auth(client, db):
"""No token at all cannot reach the upload route."""
data = {'kind': 'site', 'file': (io.BytesIO(b'<svg/>'), 'logo.svg')}
resp = client.post('/api/settings/branding-logo', data=data,
content_type='multipart/form-data')
assert resp.status_code in (401, 422)
def test_branding_upload_round_trip(client, db, auth_headers):
"""Admin upload writes the matching setting and the file is then served."""
payload = b'<svg xmlns="http://www.w3.org/2000/svg"></svg>'
data = {'kind': 'site', 'file': (io.BytesIO(payload), 'mylogo.svg')}
resp = client.post('/api/settings/branding-logo', data=data,
content_type='multipart/form-data', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
body = resp.get_json()['data']
assert body['key'] == 'site_logo'
assert body['value'] == '/api/settings/branding/logo-site.svg'
# Setting row now points at the served URL.
from shopdb.core.models import Setting
setting = Setting.query.filter_by(key='site_logo').first()
assert setting is not None
assert setting.value == '/api/settings/branding/logo-site.svg'
# Public serve route returns the uploaded bytes without auth.
served = client.get('/api/settings/branding/logo-site.svg')
assert served.status_code == 200
assert served.get_data() == payload
def test_branding_upload_favicon_allows_ico(client, db, auth_headers):
"""The favicon kind accepts .ico (extra extension beyond map images)."""
data = {'kind': 'favicon', 'file': (io.BytesIO(b'icodata'), 'fav.ico')}
resp = client.post('/api/settings/branding-logo', data=data,
content_type='multipart/form-data', headers=auth_headers)
assert resp.status_code == 200, resp.get_json()
assert resp.get_json()['data']['key'] == 'site_favicon'
def test_branding_upload_rejects_invalid_kind(client, db, auth_headers):
"""An unknown kind is a validation error."""
data = {'kind': 'banner', 'file': (io.BytesIO(b'<svg/>'), 'logo.svg')}
resp = client.post('/api/settings/branding-logo', data=data,
content_type='multipart/form-data', headers=auth_headers)
assert resp.status_code == 400
assert resp.get_json()['data']['error']['code'] == 'VALIDATION_ERROR'
def test_branding_upload_rejects_invalid_extension(client, db, auth_headers):
"""A disallowed file extension is rejected."""
data = {'kind': 'site', 'file': (io.BytesIO(b'MZ...'), 'logo.exe')}
resp = client.post('/api/settings/branding-logo', data=data,
content_type='multipart/form-data', headers=auth_headers)
assert resp.status_code == 400
assert resp.get_json()['data']['error']['code'] == 'VALIDATION_ERROR'
def test_kind_map_covers_expected_kinds():
"""The kind->setting map matches the documented branding kinds."""
assert BRANDING_KIND_SETTINGS == {
'site': 'site_logo',
'qr': 'qr_logo',
'badge': 'badge_logo',
'favicon': 'site_favicon',
}
def test_defaults_contain_new_branding_keys():
"""build_default_settings seeds every branding key with its default."""
by_key = _defaults_by_key()
expected = {
'site_logo': '/ge-aerospace-logo.svg',
'qr_logo': '/ge-monogram.svg',
'badge_logo': '/ge-aerospace-logo.svg',
'site_favicon': '',
'brand_primary_color': '',
}
for key, value in expected.items():
assert key in by_key, f'missing default {key}'
assert by_key[key]['value'] == value
assert by_key[key]['category'] == 'branding'
def test_defaults_contain_new_integration_keys():
"""build_default_settings seeds the ServiceNow integration keys."""
by_key = _defaults_by_key()
assert by_key['servicenow_enabled']['value'] == 'true'
assert by_key['servicenow_enabled']['category'] == 'integrations'
assert by_key['servicenow_ticket_prefixes']['value'] == 'GEINC,GECHG,GERIT,GESCT'
assert '{ticket}' in by_key['servicenow_search_url']['value']
assert '{ticket}' in by_key['servicenow_incident_url']['value']
assert '{ticket}' in by_key['servicenow_change_url']['value']
def test_defaults_contain_new_site_keys():
"""build_default_settings seeds the employee-id and printer-hostname keys."""
by_key = _defaults_by_key()
assert by_key['employeeid_pattern']['value'] == r'^\d{9}$'
assert by_key['printer_hostname_template']['value'] == 'Printer-{ip}.printer.geaerospace.net'
assert by_key['contact_email_domain']['value'] == 'geaerospace.com'
assert by_key['contact_email_domain']['category'] == 'site'
def test_defaults_contain_label_target_keys():
"""build_default_settings seeds the per-type label targets + label defaults."""
by_key = _defaults_by_key()
for key in ('qr_target_machine', 'qr_target_computer',
'qr_target_network_device', 'qr_target_measuring_tool'):
assert key in by_key, f'missing default {key}'
assert by_key[key]['value'] == ''
assert by_key[key]['category'] == 'printing'
assert '{locationcode}' in by_key['qr_target_measuring_tool']['description']
assert by_key['label_default_style']['value'] == 'card'
assert by_key['label_default_codetype']['value'] == 'qr'
assert by_key['label_default_style']['category'] == 'printing'
assert by_key['label_default_codetype']['category'] == 'printing'
# Per-type default encode mode: machines -> their machine number,
# measuring tools -> inspection location, the rest -> asset page.
assert by_key['label_default_encodes_machine']['value'] == 'assetnumber'
assert by_key['label_default_encodes_measuring_tool']['value'] == 'location'
for assettype in ('computer', 'printer', 'network_device'):
assert by_key[f'label_default_encodes_{assettype}']['value'] == 'assetpage'
assert by_key[f'label_default_encodes_{assettype}']['category'] == 'printing'
def test_defaults_changed_facility_and_map():
"""facility_name default is now blank; map blueprints point at the placeholder."""
by_key = _defaults_by_key()
assert by_key['facility_name']['value'] == ''
assert by_key['map_blueprint_light']['value'] == '/static/images/floorplan-placeholder.svg'
assert by_key['map_blueprint_dark']['value'] == '/static/images/floorplan-placeholder.svg'
# No leftover West Jefferson sitemap references.
assert 'sitemap2025' not in by_key['map_blueprint_light']['value']

View File

@@ -0,0 +1,25 @@
"""Characterization test for the slides (TV slideshow) plugin feed.
The slide manager rework moved the public playlist to /api/slides/feed with a
flat (non-enveloped) response shape so the screensaver's existing parser needs
no change. Pin that contract here.
"""
def test_slides_feed_shape(client, db):
"""GET /api/slides/feed returns a flat playlist for a surface (no auth)."""
resp = client.get('/api/slides/feed?surface=lobby')
assert resp.status_code == 200
data = resp.get_json()
assert data['success'] is True
assert data['surface'] == 'lobby'
assert isinstance(data['slides'], list)
assert data['basepath'] == '/api/slides/img/lobby/'
assert 'interval' in data
def test_slides_feed_unknown_surface_falls_back(client, db):
"""An invalid surface name falls back to lobby instead of erroring."""
resp = client.get('/api/slides/feed?surface=../etc')
assert resp.status_code == 200
assert resp.get_json()['surface'] == 'lobby'

View File

@@ -0,0 +1,234 @@
"""Support teams + contacts: team/contact CRUD, lookup, delete-guard, import.
The application/support-team wiring: a team owns ordered contacts, an
application carries one team, and its payload flattens the team name/url +
active contacts so the frontend needs a single call.
"""
IMPORT_HEADER = {'X-Import-Mode': 'true'}
LEGACY_CREATED = '2020-01-05 08:30:00'
LEGACY_MODIFIED = '2021-06-07T14:15:16'
def _import_headers(auth_headers):
merged = dict(auth_headers)
merged.update(IMPORT_HEADER)
return merged
def _create_team(client, auth_headers, teamname='Controls', teamurl=None):
return client.post('/api/supportteams',
json={'teamname': teamname, 'teamurl': teamurl},
headers=auth_headers)
# ---------------------------------------------------------------------------
# Team CRUD
# ---------------------------------------------------------------------------
def test_create_and_get_team(client, db, auth_headers):
"""Create a team, then fetch it back with an (empty) contacts list."""
resp = _create_team(client, auth_headers, 'Controls',
'https://servicenow.example/group/controls')
assert resp.status_code == 201, resp.get_json()
teamid = resp.get_json()['data']['supportteamid']
got = client.get(f'/api/supportteams/{teamid}', headers=auth_headers)
assert got.status_code == 200
data = got.get_json()['data']
assert data['teamname'] == 'Controls'
assert data['teamurl'] == 'https://servicenow.example/group/controls'
assert data['contacts'] == []
def test_create_team_requires_teamname(client, db, auth_headers):
resp = client.post('/api/supportteams', json={}, headers=auth_headers)
assert resp.status_code == 400
def test_create_team_duplicate_conflict(client, db, auth_headers):
_create_team(client, auth_headers, 'Controls')
dup = _create_team(client, auth_headers, 'Controls')
assert dup.status_code == 409
def test_update_team(client, db, auth_headers):
teamid = _create_team(client, auth_headers, 'Controls').get_json()['data']['supportteamid']
resp = client.put(f'/api/supportteams/{teamid}',
json={'teamname': 'Controls Renamed',
'teamurl': 'https://x.example'},
headers=auth_headers)
assert resp.status_code == 200
assert resp.get_json()['data']['teamname'] == 'Controls Renamed'
def test_delete_team(client, db, auth_headers):
teamid = _create_team(client, auth_headers, 'ToDelete').get_json()['data']['supportteamid']
resp = client.delete(f'/api/supportteams/{teamid}', headers=auth_headers)
assert resp.status_code == 200
gone = client.get(f'/api/supportteams/{teamid}', headers=auth_headers)
assert gone.status_code == 404
# ---------------------------------------------------------------------------
# teamname exact-match lookup (import recipe) + active filter
# ---------------------------------------------------------------------------
def test_teamname_lookup_filter(client, db, auth_headers):
for name in ('Controls', 'ControlsB', 'Networking'):
_create_team(client, auth_headers, name)
listed = client.get('/api/supportteams?teamname=Controls', headers=auth_headers)
assert listed.status_code == 200
rows = listed.get_json()['data']
assert len(rows) == 1
assert rows[0]['teamname'] == 'Controls'
def test_list_active_filter_hides_inactive(client, db, auth_headers):
teamid = _create_team(client, auth_headers, 'Retired').get_json()['data']['supportteamid']
client.put(f'/api/supportteams/{teamid}', json={'isactive': False},
headers=auth_headers)
active = client.get('/api/supportteams', headers=auth_headers)
assert all(t['teamname'] != 'Retired' for t in active.get_json()['data'])
allteams = client.get('/api/supportteams?active=false', headers=auth_headers)
assert any(t['teamname'] == 'Retired' for t in allteams.get_json()['data'])
# ---------------------------------------------------------------------------
# Delete-with-applications 409
# ---------------------------------------------------------------------------
def test_delete_team_with_applications_conflicts(client, db, auth_headers):
teamid = _create_team(client, auth_headers, 'InUse').get_json()['data']['supportteamid']
appresp = client.post('/api/applications',
json={'appname': 'DependentApp',
'supportteamid': teamid},
headers=auth_headers)
assert appresp.status_code == 201, appresp.get_json()
conflict = client.delete(f'/api/supportteams/{teamid}', headers=auth_headers)
assert conflict.status_code == 409
# Still there.
assert client.get(f'/api/supportteams/{teamid}',
headers=auth_headers).status_code == 200
# ---------------------------------------------------------------------------
# Nested contact CRUD + ordering
# ---------------------------------------------------------------------------
def test_contact_crud_and_ordering(client, db, auth_headers):
teamid = _create_team(client, auth_headers, 'Controls').get_json()['data']['supportteamid']
# Add two contacts out of sort order.
c2 = client.post(f'/api/supportteams/{teamid}/contacts',
json={'name': 'Second', 'sso': '222', 'sortorder': 2},
headers=auth_headers)
assert c2.status_code == 201, c2.get_json()
c1 = client.post(f'/api/supportteams/{teamid}/contacts',
json={'name': 'First', 'sso': '111', 'sortorder': 1},
headers=auth_headers)
assert c1.status_code == 201
# Team now nests active contacts in sortorder.
team = client.get(f'/api/supportteams/{teamid}', headers=auth_headers).get_json()['data']
names = [c['name'] for c in team['contacts']]
assert names == ['First', 'Second']
# Update one contact.
contactid = c1.get_json()['data']['contactid']
upd = client.put(f'/api/supportteams/{teamid}/contacts/{contactid}',
json={'name': 'First Updated'}, headers=auth_headers)
assert upd.status_code == 200
assert upd.get_json()['data']['name'] == 'First Updated'
# Delete the other contact.
otherid = c2.get_json()['data']['contactid']
dele = client.delete(f'/api/supportteams/{teamid}/contacts/{otherid}',
headers=auth_headers)
assert dele.status_code == 200
team = client.get(f'/api/supportteams/{teamid}', headers=auth_headers).get_json()['data']
assert [c['name'] for c in team['contacts']] == ['First Updated']
def test_contact_requires_name(client, db, auth_headers):
teamid = _create_team(client, auth_headers, 'Controls').get_json()['data']['supportteamid']
resp = client.post(f'/api/supportteams/{teamid}/contacts', json={},
headers=auth_headers)
assert resp.status_code == 400
def test_delete_team_cascades_contacts(client, db, auth_headers):
from shopdb.core.models import SupportTeamContact
teamid = _create_team(client, auth_headers, 'Controls').get_json()['data']['supportteamid']
client.post(f'/api/supportteams/{teamid}/contacts',
json={'name': 'Someone'}, headers=auth_headers)
client.delete(f'/api/supportteams/{teamid}', headers=auth_headers)
assert SupportTeamContact.query.filter_by(supportteamid=teamid).count() == 0
# ---------------------------------------------------------------------------
# Application carries the team + flattened contacts
# ---------------------------------------------------------------------------
def test_application_carries_supportteam_payload(client, db, auth_headers):
teamid = _create_team(client, auth_headers, 'Controls',
'https://sn.example/controls').get_json()['data']['supportteamid']
client.post(f'/api/supportteams/{teamid}/contacts',
json={'name': 'Alice', 'sso': 'a01', 'sortorder': 0},
headers=auth_headers)
appresp = client.post('/api/applications',
json={'appname': 'TeamApp', 'supportteamid': teamid},
headers=auth_headers)
appid = appresp.get_json()['data']['appid']
got = client.get(f'/api/applications/{appid}', headers=auth_headers)
data = got.get_json()['data']
assert data['supportteamid'] == teamid
assert data['supportteamname'] == 'Controls'
assert data['teamurl'] == 'https://sn.example/controls'
assert data['contacts'] == [{'name': 'Alice', 'sso': 'a01'}]
# ---------------------------------------------------------------------------
# Import-mode timestamps
# ---------------------------------------------------------------------------
def test_import_mode_preserves_team_timestamps(client, db, auth_headers):
resp = client.post('/api/supportteams',
json={'teamname': 'Legacy',
'createddate': LEGACY_CREATED,
'modifieddate': LEGACY_MODIFIED},
headers=_import_headers(auth_headers))
assert resp.status_code == 201, resp.get_json()
from shopdb.core.models import SupportTeam
team = SupportTeam.query.filter_by(teamname='Legacy').first()
assert team.createddate.year == 2020 and team.createddate.month == 1
assert team.modifieddate.year == 2021
def test_import_mode_preserves_contact_timestamps(client, db, auth_headers):
teamid = _create_team(client, auth_headers, 'Controls').get_json()['data']['supportteamid']
resp = client.post(f'/api/supportteams/{teamid}/contacts',
json={'name': 'LegacyOwner',
'createddate': LEGACY_CREATED,
'modifieddate': LEGACY_MODIFIED},
headers=_import_headers(auth_headers))
assert resp.status_code == 201, resp.get_json()
from shopdb.core.models import SupportTeamContact
contact = SupportTeamContact.query.filter_by(name='LegacyOwner').first()
assert contact.createddate.year == 2020
assert contact.modifieddate.year == 2021
# ---------------------------------------------------------------------------
# Authz (belt-and-suspenders; the sweep in test_authz also covers these)
# ---------------------------------------------------------------------------
def test_member_cannot_create_team(client, db, member_headers):
resp = client.post('/api/supportteams', json={'teamname': 'X'},
headers=member_headers)
assert resp.status_code == 403