diff --git a/plugins/computers/api/routes.py b/plugins/computers/api/routes.py index 9adb608a..d90d1c36 100644 --- a/plugins/computers/api/routes.py +++ b/plugins/computers/api/routes.py @@ -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'): diff --git a/plugins/employees/api/routes.py b/plugins/employees/api/routes.py index 4e9788c1..07124b25 100644 --- a/plugins/employees/api/routes.py +++ b/plugins/employees/api/routes.py @@ -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( diff --git a/plugins/knowledgebase/api/routes.py b/plugins/knowledgebase/api/routes.py index a3dd9800..f782b6bb 100644 --- a/plugins/knowledgebase/api/routes.py +++ b/plugins/knowledgebase/api/routes.py @@ -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. diff --git a/plugins/machines/api/routes.py b/plugins/machines/api/routes.py index 267d84dd..f8726a7f 100644 --- a/plugins/machines/api/routes.py +++ b/plugins/machines/api/routes.py @@ -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') diff --git a/plugins/network/api/routes.py b/plugins/network/api/routes.py index 83b44ce8..93e9c8a0 100644 --- a/plugins/network/api/routes.py +++ b/plugins/network/api/routes.py @@ -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'): diff --git a/plugins/notifications/api/routes.py b/plugins/notifications/api/routes.py index a2db5d9e..b1058325 100644 --- a/plugins/notifications/api/routes.py +++ b/plugins/notifications/api/routes.py @@ -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. diff --git a/plugins/printers/api/asset_routes.py b/plugins/printers/api/asset_routes.py index fa14ac5e..f52fa508 100644 --- a/plugins/printers/api/asset_routes.py +++ b/plugins/printers/api/asset_routes.py @@ -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 diff --git a/shopdb/__init__.py b/shopdb/__init__.py index d7dd2884..6df790d5 100644 --- a/shopdb/__init__.py +++ b/shopdb/__init__.py @@ -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 = { diff --git a/shopdb/api/__init__.py b/shopdb/api/__init__.py index d26a932e..c1c4e9f7 100644 --- a/shopdb/api/__init__.py +++ b/shopdb/api/__init__.py @@ -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', diff --git a/shopdb/core/api/assets.py b/shopdb/core/api/assets.py index 006c16d5..2ad24b1e 100644 --- a/shopdb/core/api/assets.py +++ b/shopdb/core/api/assets.py @@ -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'): diff --git a/shopdb/core/api/reports.py b/shopdb/core/api/reports.py index 0b523580..3504a8a4 100644 --- a/shopdb/core/api/reports.py +++ b/shopdb/core/api/reports.py @@ -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() diff --git a/shopdb/utils/queryargs.py b/shopdb/utils/queryargs.py new file mode 100644 index 00000000..848e5f34 --- /dev/null +++ b/shopdb/utils/queryargs.py @@ -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 diff --git a/tests/test_core/test_bad_query_args.py b/tests/test_core/test_bad_query_args.py new file mode 100644 index 00000000..7fb58793 --- /dev/null +++ b/tests/test_core/test_bad_query_args.py @@ -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']