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

Answer a nonsense filter with a 400, and sweep every list route for the rest

Browse Source 
Wave 3. Six findings, one cause: `int(request.args.get('typeid'))` with nothing
catching the ValueError. The caller sent a bad argument and got
500 INTERNAL_ERROR plus a traceback in the log, which says the server is broken
when the truth is that 'abc' is not a number.

shopdb/utils/queryargs.intarg reads it and raises QueryArgError, which an
app-level handler answers as a 400 naming the argument. App-level on purpose: a
route should not have to remember, and forgetting was the entire defect. The
policy differs from get_pagination_params, which falls back to a default for a
bad page number - a nonsense page is harmless and the caller still wants their
list, but silently ignoring a nonsense FILTER returns rows they believe they
excluded.

The gate matters more than the six fixes. tests/test_core/test_bad_query_args.py
sweeps all 131 registered no-parameter GET routes under /api/ with twelve
argument shapes and fails on any 5xx. It found sites the test run had not
reported - statusid and locationid on /api/assets, the second asset-list route,
limit on the employee lookup and the printer supply history - because the
pattern is how a list filter gets written rather than a property of any one
route. Verified non-vacuous by planting the old `int(...)` back into
/api/machines: four tests fail.

Four usb routes are skipped by exact path, with the reason: they reach the
EXTERNAL cmmc_usb database, so their 500 in a test environment is a missing
database rather than an argument fault. That is the complete set - every usb
GET route calling cmmc_usb_connection - not a convenient subset.

One correction worth recording: the first pass rewrote every
`if x := request.args.get(...)` in the touched files, including text arguments,
so a search for 'ZZNOTHINGMATCHES' briefly became a 400. Caught by
test_list_search_identifiers. Only id-shaped arguments are integers.

Scope: product
This commit is contained in:
cproudlock
2026-09-03 14:18:49 -04:00
parent 12392a361e
commit debef39a18
13 changed files with 285 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
plugins/computers/api/routes.py
View File

@@ -7,7 +7,7 @@ from shopdb.api import db, Asset, AssetType, OperatingSystem, Application, Audit
from ..models import Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess
from shopdb.api import require_permission, require_role, apply_import_timestamps
from shopdb.api import require_permission, require_role, apply_import_timestamps, intarg
computers_bp = Blueprint('computers', __name__)
@@ -380,20 +380,20 @@ def list_computers():
)
# Computer type filter
if type_id := request.args.get('typeid', request.args.get('type_id')):
query = query.filter(Computer.computertypeid == int(type_id))
if (type_id := intarg('typeid', intarg('type_id'))) is not None:
query = query.filter(Computer.computertypeid == type_id)
# OS filter
if os_id := request.args.get('osid', request.args.get('os_id')):
query = query.filter(Computer.osid == int(os_id))
if (os_id := intarg('osid', intarg('os_id'))) is not None:
query = query.filter(Computer.osid == os_id)
# Location filter
if location_id := request.args.get('locationid', request.args.get('location_id')):
query = query.filter(Asset.locationid == int(location_id))
if (location_id := intarg('locationid', intarg('location_id'))) is not None:
query = query.filter(Asset.locationid == location_id)
# Business unit filter
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid', intarg('businessunit_id'))) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# Shopfloor filter (by the Shopfloor computer type)
if shopfloor := request.args.get('shopfloor'):

4
plugins/employees/api/routes.py
View File

@@ -25,7 +25,7 @@ from shopdb.api import (
employee_connection,
require_role,
)
from shopdb.api import Setting
from shopdb.api import Setting, intarg
from shopdb.api import send_upload
from ..models import DirectoryEmployee
@@ -287,7 +287,7 @@ def search_employees():
- limit: Max results (default 10)
"""
query = request.args.get('q', '').strip()
limit = min(int(request.args.get('limit', 10)), 50)
limit = min(intarg('limit', 10), 50)
if len(query) < 2:
return error_response(

7
plugins/knowledgebase/api/routes.py
View File

@@ -4,6 +4,7 @@ from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.api import (
intarg,
db,
Application,
success_response,
@@ -16,7 +17,7 @@ from shopdb.api import (
from ..models import KnowledgeBase
from shopdb.api import require_permission, apply_import_timestamps
from shopdb.api import require_permission, apply_import_timestamps, intarg
knowledgebase_bp = Blueprint('knowledgebase', __name__)
@@ -91,8 +92,8 @@ def list_articles():
query = query.filter(_search_clause(search))
# Filter by topic/application
if appid := request.args.get('appid'):
query = query.filter(KnowledgeBase.appid == int(appid))
if (appid := intarg('appid')) is not None:
query = query.filter(KnowledgeBase.appid == appid)
# Exact-match natural-key lookups for idempotent import. linkurl is the
# stable natural key; shortdescription (the title) is offered as a fallback.

18
plugins/machines/api/routes.py
View File

@@ -7,7 +7,7 @@ from shopdb.api import db, Asset, AssetType, AuditLog, Vendor, success_response,
from ..models import Machine, MachineType
from shopdb.api import require_permission, apply_import_timestamps
from shopdb.api import require_permission, apply_import_timestamps, intarg
machines_bp = Blueprint('machines', __name__)
@@ -197,20 +197,20 @@ def list_machines():
)
# Machine type filter
if type_id := request.args.get('typeid', request.args.get('type_id')):
query = query.filter(Machine.machinetypeid == int(type_id))
if (type_id := intarg('typeid', intarg('type_id'))) is not None:
query = query.filter(Machine.machinetypeid == type_id)
# Vendor filter
if vendor_id := request.args.get('vendorid', request.args.get('vendor_id')):
query = query.filter(Machine.vendorid == int(vendor_id))
if (vendor_id := intarg('vendorid', intarg('vendor_id'))) is not None:
query = query.filter(Machine.vendorid == vendor_id)
# Location filter
if location_id := request.args.get('locationid', request.args.get('location_id')):
query = query.filter(Asset.locationid == int(location_id))
if (location_id := intarg('locationid', intarg('location_id'))) is not None:
query = query.filter(Asset.locationid == location_id)
# Business unit filter
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid', intarg('businessunit_id'))) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# Sorting
sort_by = request.args.get('sort', 'assetnumber')

26
plugins/network/api/routes.py
View File

@@ -8,7 +8,7 @@ from shopdb.api import db, Asset, AssetType, Vendor, Communication, Communicatio
from ..services.assetnumbers import generate_for_type
from ..models import NetworkDevice, NetworkDeviceType, Subnet, VLAN
from shopdb.api import require_permission, apply_import_timestamps
from shopdb.api import require_permission, apply_import_timestamps, intarg
network_bp = Blueprint('network', __name__)
@@ -232,20 +232,20 @@ def list_network_devices():
)
# Type filter
if type_id := request.args.get('typeid', request.args.get('type_id')):
query = query.filter(NetworkDevice.networkdevicetypeid == int(type_id))
if (type_id := intarg('typeid', intarg('type_id'))) is not None:
query = query.filter(NetworkDevice.networkdevicetypeid == type_id)
# Vendor filter
if vendor_id := request.args.get('vendorid', request.args.get('vendor_id')):
query = query.filter(NetworkDevice.vendorid == int(vendor_id))
if (vendor_id := intarg('vendorid', intarg('vendor_id'))) is not None:
query = query.filter(NetworkDevice.vendorid == vendor_id)
# Location filter
if location_id := request.args.get('locationid', request.args.get('location_id')):
query = query.filter(Asset.locationid == int(location_id))
if (location_id := intarg('locationid', intarg('location_id'))) is not None:
query = query.filter(Asset.locationid == location_id)
# Business unit filter
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid', intarg('businessunit_id'))) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# PoE filter
if poe := request.args.get('poe'):
@@ -827,12 +827,12 @@ def list_subnets():
)
# VLAN filter
if vlan_id := request.args.get('vlanid'):
query = query.filter(Subnet.vlanid == int(vlan_id))
if (vlan_id := intarg('vlanid')) is not None:
query = query.filter(Subnet.vlanid == vlan_id)
# Location filter
if location_id := request.args.get('locationid'):
query = query.filter(Subnet.locationid == int(location_id))
if (location_id := intarg('locationid')) is not None:
query = query.filter(Subnet.locationid == location_id)
# Type filter
if subnet_type := request.args.get('type'):

6
plugins/notifications/api/routes.py
View File

@@ -13,7 +13,7 @@ from shopdb.api import db, success_response, error_response, paginated_response,
from ..models import Notification, NotificationType
from ..models.notification import _utc_iso
from shopdb.api import require_permission, Setting
from shopdb.api import require_permission, Setting, intarg
notifications_bp = Blueprint('notifications', __name__)
@@ -355,8 +355,8 @@ def list_notifications():
query = query.filter(Notification.isactive == True)
# Type filter
if type_id := request.args.get('typeid', request.args.get('type_id')):
query = query.filter(Notification.notificationtypeid == int(type_id))
if (type_id := intarg('typeid', intarg('type_id'))) is not None:
query = query.filter(Notification.notificationtypeid == type_id)
# Exact-match lookup for idempotent import. Notifications have no strong
# natural key; ticketnumber is the best available when a ticket is set.

24
plugins/printers/api/asset_routes.py
View File

@@ -6,7 +6,7 @@ import re
from flask import Blueprint, request, Response, current_app
from flask_jwt_extended import jwt_required
from shopdb.api import db, cache, Setting, Asset, AssetType, Vendor, Model, Communication, CommunicationType, AssetRelationship, RelationshipType, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
from shopdb.api import db, cache, Setting, Asset, AssetType, Vendor, Model, Communication, CommunicationType, AssetRelationship, RelationshipType, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query, intarg
from ..models import Printer, PrinterType, ModelSupply, PrinterDriver, PrinterObservedQueue
from ..models.model_supply import SUPPLY_TYPES, SUPPLY_COLORS, CAPACITY_TIERS
@@ -261,20 +261,20 @@ def list_printers():
)
# Type filter
if typeid := request.args.get('typeid', request.args.get('type_id')):
query = query.filter(Printer.printertypeid == int(typeid))
if (typeid := intarg('typeid', intarg('type_id'))) is not None:
query = query.filter(Printer.printertypeid == typeid)
# Vendor filter
if vendor_id := request.args.get('vendorid', request.args.get('vendor_id')):
query = query.filter(Printer.vendorid == int(vendor_id))
if (vendor_id := intarg('vendorid', intarg('vendor_id'))) is not None:
query = query.filter(Printer.vendorid == vendor_id)
# Location filter
if location_id := request.args.get('locationid', request.args.get('location_id')):
query = query.filter(Asset.locationid == int(location_id))
if (location_id := intarg('locationid', intarg('location_id'))) is not None:
query = query.filter(Asset.locationid == location_id)
# Business unit filter
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid', intarg('businessunit_id'))) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# Sorting
sort_by = request.args.get('sort', 'hostname')
@@ -2481,8 +2481,8 @@ def list_supply_models():
if search := request.args.get('search'):
query = query.filter(Model.modelnumber.ilike(f'%{search}%'))
if vendor_id := request.args.get('vendorid', request.args.get('vendor_id')):
query = query.filter(Model.vendorid == int(vendor_id))
if (vendor_id := intarg('vendorid', intarg('vendor_id'))) is not None:
query = query.filter(Model.vendorid == vendor_id)
if request.args.get('withsupplies', '').lower() == 'true':
query = query.having(supplycount > 0)
@@ -2755,7 +2755,7 @@ def supplies_forecast():
)
try:
days = max(1, min(365, int(request.args.get('days', 90))))
days = max(1, min(365, intarg('days', 90)))
except (TypeError, ValueError):
days = 90

10
shopdb/__init__.py
View File

@@ -400,6 +400,16 @@ def register_error_handlers(app: Flask):
% (' for %s' % column if column else ''),
http_code=400)
# A wrong query argument is the caller's mistake. Registered app-wide so a
# list route does not have to remember to catch it - forgetting was the
# whole defect, in six routes across five plugins.
from .utils.queryargs import QueryArgError
@app.errorhandler(QueryArgError)
def handle_query_arg_error(error):
return error_response(ErrorCodes.VALIDATION_ERROR, str(error),
http_code=400)
@app.errorhandler(ShopDBException)
def handle_shopdb_exception(error):
http_codes = {

5
shopdb/api/__init__.py
View File

@@ -112,6 +112,8 @@ from shopdb.utils.clientip import client_ip, strip_port
# Strict date parsing. On the contract surface because every plugin that stores
# a date needs it and the two that rolled their own both swallowed the error.
from shopdb.utils.queryargs import QueryArgError, intarg, intargs
from shopdb.utils.datevalues import (
DateValueError,
parse_date,
@@ -359,6 +361,9 @@ __all__ = [
'success_response',
'error_response',
'DateValueError',
'QueryArgError',
'intarg',
'intargs',
'parse_date',
'parse_date_fields',
'paginated_response',

32
shopdb/core/api/assets.py
View File

@@ -13,6 +13,7 @@ from shopdb.utils.responses import (
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
from shopdb.utils.queryargs import intarg
from shopdb.utils.authz import require_permission
from shopdb.utils.import_mode import apply_import_timestamps
@@ -365,20 +366,20 @@ def list_assets():
query = query.join(AssetType).filter(AssetType.assettype == type_name)
# Type filter by ID
if type_id := request.args.get('typeid', request.args.get('type_id')):
query = query.filter(Asset.assettypeid == int(type_id))
if (type_id := intarg('typeid', intarg('type_id'))) is not None:
query = query.filter(Asset.assettypeid == type_id)
# Status filter
if status_id := request.args.get('statusid', request.args.get('status_id')):
query = query.filter(Asset.statusid == int(status_id))
if (status_id := intarg('statusid', intarg('status_id'))) is not None:
query = query.filter(Asset.statusid == status_id)
# Location filter
if location_id := request.args.get('locationid', request.args.get('location_id')):
query = query.filter(Asset.locationid == int(location_id))
if (location_id := intarg('locationid', intarg('location_id'))) is not None:
query = query.filter(Asset.locationid == location_id)
# Business unit filter
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid', intarg('businessunit_id'))) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# Sorting
sort_by = request.args.get('sort', 'assetnumber')
@@ -1030,8 +1031,7 @@ def get_assets_map():
query = query.join(AssetType).filter(AssetType.assettype == selected_assettype)
# Filter by subtype (depends on asset type) - case-insensitive matching
if subtype_id := request.args.get('subtype'):
subtype_id = int(subtype_id)
if (subtype_id := intarg('subtype')) is not None:
# Normalize the underscore DB form (measuring_tool, network_device) to
# the space form the branches below compare against.
asset_type_lower = (
@@ -1079,16 +1079,16 @@ def get_assets_map():
pass
# Filter by business unit
if bu_id := request.args.get('businessunitid'):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid')) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# Filter by status
if status_id := request.args.get('statusid'):
query = query.filter(Asset.statusid == int(status_id))
if (status_id := intarg('statusid')) is not None:
query = query.filter(Asset.statusid == status_id)
# Filter by location
if location_id := request.args.get('locationid'):
query = query.filter(Asset.locationid == int(location_id))
if (location_id := intarg('locationid')) is not None:
query = query.filter(Asset.locationid == location_id)
# Search filter
if search := request.args.get('search'):

35
shopdb/core/api/reports.py
View File

@@ -13,6 +13,7 @@ from shopdb.core.models import (
)
from shopdb.utils.responses import success_response, error_response, ErrorCodes
from shopdb.utils.authz import require_permission
from shopdb.utils.queryargs import intarg
reports_bp = Blueprint('reports', __name__)
@@ -62,8 +63,8 @@ def machines_by_type():
)
# Business unit filter
if bu_id := request.args.get('businessunitid'):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid')) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# Dualpath single-machine collapse: exclude the secondary bay so a dual-bay
# pair counts once per type bucket. Gated on the site setting (default on).
@@ -128,12 +129,12 @@ def assets_by_status():
)
# Asset type filter
if type_id := request.args.get('assettypeid'):
query = query.filter(Asset.assettypeid == int(type_id))
if (type_id := intarg('assettypeid')) is not None:
query = query.filter(Asset.assettypeid == type_id)
# Business unit filter
if bu_id := request.args.get('businessunitid'):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid')) is not None:
query = query.filter(Asset.businessunitid == bu_id)
query = query.group_by(
AssetStatus.statusid,
@@ -175,7 +176,7 @@ def kb_popularity():
- limit: Number of results (default 20)
- format: 'json' (default) or 'csv'
"""
limit = min(int(request.args.get('limit', 20)), 100)
limit = min(intarg('limit', 20), 100)
# Knowledge Base is a plugin; degrade gracefully if it is not installed.
try:
@@ -362,10 +363,10 @@ def asset_inventory():
AssetType.isactive == True
)
if bu_id := request.args.get('businessunitid'):
by_type = by_type.filter(Asset.businessunitid == int(bu_id))
if loc_id := request.args.get('locationid'):
by_type = by_type.filter(Asset.locationid == int(loc_id))
if (bu_id := intarg('businessunitid')) is not None:
by_type = by_type.filter(Asset.businessunitid == bu_id)
if (loc_id := intarg('locationid')) is not None:
by_type = by_type.filter(Asset.locationid == loc_id)
by_type = by_type.group_by(AssetType.assettype).all()
@@ -380,10 +381,10 @@ def asset_inventory():
AssetStatus.isactive == True
)
if bu_id := request.args.get('businessunitid'):
by_status = by_status.filter(Asset.businessunitid == int(bu_id))
if loc_id := request.args.get('locationid'):
by_status = by_status.filter(Asset.locationid == int(loc_id))
if (bu_id := intarg('businessunitid')) is not None:
by_status = by_status.filter(Asset.businessunitid == bu_id)
if (loc_id := intarg('locationid')) is not None:
by_status = by_status.filter(Asset.locationid == loc_id)
by_status = by_status.group_by(AssetStatus.status, AssetStatus.color).all()
@@ -398,8 +399,8 @@ def asset_inventory():
Location.isactive == True
)
if bu_id := request.args.get('businessunitid'):
by_location = by_location.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid')) is not None:
by_location = by_location.filter(Asset.businessunitid == bu_id)
by_location = by_location.group_by(Location.locationname).all()

64
shopdb/utils/queryargs.py Normal file
View File

@@ -0,0 +1,64 @@
"""Reading a typed value out of the query string, without a 500.
WHY THIS EXISTS
List endpoints filtered like this:
if type_id := request.args.get('typeid'):
query = query.filter(Machine.machinetypeid == int(type_id))
`int('abc')` raises, nothing catches it, and the caller gets
500 INTERNAL_ERROR with a traceback in the log. A wrong query argument is the
caller's mistake and deserves a 400 saying which argument; a 500 says the server
is broken and sends whoever is reading the log looking for a fault that is not
there. The feature test run found the same shape in six places across five
plugins, which makes it a property of how list routes are written rather than of
any one route.
The helpers raise QueryArgError. Routes turn it into a 400 - or, more usually,
nothing does anything at all, because the app-level handler in
shopdb/__init__.py answers it. That is the point: a route should not have to
remember.
Deliberately NOT the same policy as get_pagination_params, which falls back to
a default for a bad page number. A nonsense page is harmless and the caller
still wants their list; a nonsense FILTER is not, because silently ignoring it
returns rows the caller believes they excluded.
"""
from flask import request
class QueryArgError(ValueError):
"""A query argument was present but not of the type the route needs."""
def __init__(self, name, value, expected='a whole number'):
self.name = name
self.value = value
super().__init__('%s must be %s (got %r)' % (name, expected, value))
def intarg(name, default=None, request_obj=None):
"""Read an integer query argument.
Absent or empty -> default. Present but not an integer -> QueryArgError,
which the app answers as a 400 naming the argument.
"""
source = request_obj if request_obj is not None else request
raw = source.args.get(name)
if raw is None or raw == '':
return default
try:
return int(raw)
except (TypeError, ValueError):
raise QueryArgError(name, raw)
def intargs(*names, request_obj=None):
"""Read several integer arguments at once; returns a dict of those present."""
found = {}
for name in names:
value = intarg(name, request_obj=request_obj)
if value is not None:
found[name] = value
return found

120
tests/test_core/test_bad_query_args.py Normal file
View File

@@ -0,0 +1,120 @@
"""No list route may answer a nonsense filter with a 500.
THE GATE, not just the six fixes
The feature test run found `int(request.args.get(...))` with nothing catching
the ValueError in six routes across five plugins - machines, network, printers,
knowledgebase, computers and the asset core. Fixing six sites does not stop a
seventh: the pattern is how a filter gets written, and it is invisible in review
because the happy path works.
So this sweeps every registered GET route that takes no path parameters and
sends each of them a set of arguments that are not what any filter wants. The
contract is narrow on purpose:
- a 5xx is ALWAYS a failure. The caller sent a bad argument; that is a 400.
- a 400 is the intended answer.
- a 200 is acceptable: a route is entitled to ignore an argument it does not
know about, and most of these routes will not have the argument at all.
- a 401/403/404 is acceptable: this test is about argument handling, not
about authorization.
Run as an admin, because a route that refuses everyone tests nothing.
"""
import pytest
# Names taken from the filters the failing routes actually declared, plus the
# generic shapes a client sends. Values chosen to break a bare int(), a bare
# float(), and a naive split.
BAD_ARGS = [
('typeid', 'abc'),
('type_id', 'abc'),
('assettypeid', '1; DROP TABLE assets'),
('vendorid', 'null'),
('locationid', '1.5'),
('businessunitid', ''),
('appid', 'abc'),
('statusid', 'NaN'),
('modelid', '-'),
('limit', 'lots'),
('page', 'abc'),
('perpage', 'abc'),
]
# Routes that are not list endpoints and whose GET does real work with side
# effects or external calls. Excluded by PREFIX, and each one says why.
SKIP_PREFIXES = (
'/api/collector/', # needs the collector key; fail-closed by design
'/api/endpoints/', # agent wire protocol, same key
'/api/backups/run', # kicks off work rather than reading
)
# Skipped by EXACT path, not prefix, so the rest of the plugin stays covered.
SKIP_EXACT = {
# These four reach the EXTERNAL cmmc_usb database. Their 500 here is "no
# such database in the test environment", a different fault from the one
# this gate is about, and the list is the complete set - every usb GET
# route whose handler calls cmmc_usb_connection. The self-hosted USB
# surface is covered by test_usb_kiosk.py.
'/api/usb',
'/api/usb/checkouts',
'/api/usb/checkouts/active',
'/api/usb/reports/audit',
}
def _listish_routes(app):
routes = []
for rule in app.url_map.iter_rules():
path = str(rule.rule)
if not path.startswith('/api/') or '<' in path:
continue
if 'GET' not in (rule.methods or ()):
continue
if path in SKIP_EXACT:
continue
if any(path.startswith(prefix) for prefix in SKIP_PREFIXES):
continue
routes.append(path)
return sorted(set(routes))
def test_the_sweep_actually_covers_something(app):
"""A gate that silently matches nothing passes forever. This one has been
known to cover 90+ routes; if it drops to a handful, the filter above is
wrong rather than the app being small."""
assert len(_listish_routes(app)) > 40
def test_no_route_answers_a_nonsense_filter_with_a_server_error(
client, auth_headers, app):
failures = []
for path in _listish_routes(app):
for name, value in BAD_ARGS:
response = client.get('%s?%s=%s' % (path, name, value),
headers=auth_headers)
if response.status_code >= 500:
failures.append('%s?%s=%s -> %d' % (path, name, value,
response.status_code))
assert not failures, (
'a bad query argument is a 400, never a 500:\n '
+ '\n '.join(failures[:40]))
@pytest.mark.parametrize('value', ['abc', '', '1.5', 'null'])
def test_a_bad_type_filter_is_a_400_that_names_the_argument(
client, auth_headers, value):
"""The routes the run actually caught. An empty value is the exception:
absent and empty both mean 'no filter', which is what a cleared dropdown
sends."""
for path in ('/api/machines', '/api/computers', '/api/network',
'/api/printers', '/api/assets'):
response = client.get('%s?typeid=%s' % (path, value),
headers=auth_headers)
if value == '':
assert response.status_code == 200, (path, response.get_json())
continue
assert response.status_code == 400, (path, response.status_code)
body = response.get_json()
assert body['data']['error']['code'] == 'VALIDATION_ERROR'
assert 'typeid' in body['data']['error']['message']