diff --git a/shopdb/core/api/search.py b/shopdb/core/api/search.py index f02f944..9dfe497 100644 --- a/shopdb/core/api/search.py +++ b/shopdb/core/api/search.py @@ -23,6 +23,24 @@ logger = logging.getLogger(__name__) search_bp = Blueprint('search', __name__) +def _word_match(query, *columns): + """SQL clause matching rows that contain EVERY word of the query, each word + in any of the columns, in any order. + + A plain `col.ilike('%CSF Roles%')` needs the exact contiguous phrase, so a + multi-word search returned nothing. This splits the query into words and ANDs + them (OR across the columns per word): 'CSF Roles' matches a row with 'CSF' + and 'Roles' anywhere in the searched fields. + """ + words = [w for w in query.split() if w] + if not words: + words = [''] + return db.and_(*[ + db.or_(*[c.ilike(f'%{w}%') for c in columns]) + for w in words + ]) + + def _require_enabled(name): """Raise ImportError when the named plugin is disabled. @@ -194,10 +212,7 @@ def _search_applications(query, search_term): try: apps = Application.query.filter( Application.isactive == True, - db.or_( - Application.appname.ilike(search_term), - Application.appdescription.ilike(search_term) - ) + _word_match(query, Application.appname, Application.appdescription) ).limit(10).all() for app in apps: @@ -228,10 +243,8 @@ def _search_knowledgebase(query, search_term): from plugins.knowledgebase.models import KnowledgeBase kb_articles = KnowledgeBase.query.filter( KnowledgeBase.isactive == True, - db.or_( - KnowledgeBase.shortdescription.ilike(search_term), - KnowledgeBase.keywords.ilike(search_term) - ) + _word_match(query, KnowledgeBase.shortdescription, + KnowledgeBase.keywords) ).limit(20).all() for kb in kb_articles: @@ -255,7 +268,7 @@ def _search_knowledgebase(query, search_term): return results -def _employee_rows(search_term): +def _employee_rows(query): """Employee rows matching search_term, from the SELFHOSTED directory table or the external HR DB per the employee_directory_mode setting. Returns dicts shaped like the external query (SSO/First_Name/Last_Name/Team/Role).""" @@ -267,25 +280,30 @@ def _employee_rows(search_term): except ImportError: return [] rows = DirectoryEmployee.query.filter( - db.or_( - DirectoryEmployee.firstname.ilike(search_term), - DirectoryEmployee.lastname.ilike(search_term), - db.cast(DirectoryEmployee.sso, db.String).ilike(search_term), - ) + _word_match(query, DirectoryEmployee.firstname, + DirectoryEmployee.lastname, + db.cast(DirectoryEmployee.sso, db.String)) ).order_by(DirectoryEmployee.lastname, DirectoryEmployee.firstname).limit(10).all() return [{'SSO': e.sso, 'First_Name': e.firstname or '', 'Last_Name': e.lastname or '', 'Team': e.team, 'Role': e.role} for e in rows] # External HR DB: shared env-backed connection helper; never hardcode creds. + # Word-AND so "First Last" matches (each word in any name/SSO field), not just + # the exact contiguous phrase. Parameterized - one LIKE triple per word. from shopdb.utils.employee_db import employee_connection + words = [w for w in query.split() if w] or [''] + clauses, params = [], [] + for w in words: + clauses.append('(First_Name LIKE %s OR Last_Name LIKE %s ' + 'OR CAST(SSO AS CHAR) LIKE %s)') + like = f'%{w}%' + params += [like, like, like] + where = ' AND '.join(clauses) emp_conn = employee_connection() with emp_conn.cursor() as cur: - cur.execute(''' - SELECT SSO, First_Name, Last_Name, Team, Role - FROM employees - WHERE First_Name LIKE %s OR Last_Name LIKE %s OR CAST(SSO AS CHAR) LIKE %s - ORDER BY Last_Name, First_Name LIMIT 10 - ''', (search_term, search_term, search_term)) + cur.execute( + 'SELECT SSO, First_Name, Last_Name, Team, Role FROM employees ' + f'WHERE {where} ORDER BY Last_Name, First_Name LIMIT 10', params) rows = cur.fetchall() emp_conn.close() return rows @@ -295,7 +313,7 @@ def _search_employees(query, search_term): """Search employees (selfhosted directory or external HR DB, per mode).""" results = [] try: - for emp in _employee_rows(search_term): + for emp in _employee_rows(query): full_name = f"{emp['First_Name'].strip()} {emp['Last_Name'].strip()}" sso_str = str(emp['SSO']) @@ -329,12 +347,8 @@ def _search_assets(query, search_term): joinedload(Asset.location), ).filter( Asset.isactive == True, - db.or_( - Asset.assetnumber.ilike(search_term), - Asset.name.ilike(search_term), - Asset.serialnumber.ilike(search_term), - Asset.notes.ilike(search_term) - ) + _word_match(query, Asset.assetnumber, Asset.name, + Asset.serialnumber, Asset.notes) ).limit(15).all() for asset in assets: @@ -372,12 +386,8 @@ def _search_measuringtools(query, search_term): joinedload(Asset.location), ).filter( Asset.isactive == True, - db.or_( - Asset.assetnumber.ilike(search_term), - Asset.name.ilike(search_term), - Asset.serialnumber.ilike(search_term), - Asset.gaugelabreference.ilike(search_term), - ) + _word_match(query, Asset.assetnumber, Asset.name, + Asset.serialnumber, Asset.gaugelabreference) ).limit(15).all() for asset in assets: @@ -421,7 +431,7 @@ def _search_customfields(query, search_term): Asset.isactive == True, CustomField.searchable == True, CustomField.isactive == True, - CustomFieldValue.value.ilike(search_term), + _word_match(query, CustomFieldValue.value), ).limit(15).all() for asset, field in rows: @@ -438,7 +448,7 @@ def _search_by_ip(query, search_term): results = [] try: comms = Communication.query.filter( - Communication.ipaddress.ilike(search_term) + _word_match(query, Communication.ipaddress) ).options( joinedload(Communication.asset).joinedload(Asset.assettype), joinedload(Communication.asset).joinedload(Asset.location), @@ -498,7 +508,7 @@ def _search_hostnames(query, search_term): _require_enabled('computers') from plugins.computers.models import Computer computers = Computer.query.filter( - Computer.hostname.ilike(search_term) + _word_match(query, Computer.hostname) ).options( joinedload(Computer.asset).joinedload(Asset.assettype), joinedload(Computer.asset).joinedload(Asset.location), @@ -520,11 +530,8 @@ def _search_hostnames(query, search_term): _require_enabled('printers') from plugins.printers.models import Printer printers = Printer.query.filter( - db.or_( - Printer.hostname.ilike(search_term), - Printer.sharename.ilike(search_term), - Printer.windowsname.ilike(search_term), - ) + _word_match(query, Printer.hostname, Printer.sharename, + Printer.windowsname) ).options( joinedload(Printer.asset).joinedload(Asset.assettype), joinedload(Printer.asset).joinedload(Asset.location), @@ -547,7 +554,7 @@ def _search_hostnames(query, search_term): _require_enabled('network') from plugins.network.models import NetworkDevice devices = NetworkDevice.query.filter( - NetworkDevice.hostname.ilike(search_term) + _word_match(query, NetworkDevice.hostname) ).options( joinedload(NetworkDevice.asset).joinedload(Asset.assettype), joinedload(NetworkDevice.asset).joinedload(Asset.location), @@ -577,10 +584,8 @@ def _search_notifications(query, search_term): notifications = Notification.query.options( joinedload(Notification.notificationtype) ).filter( - db.or_( - Notification.notification.ilike(search_term), - Notification.ticketnumber.ilike(search_term) - ) + _word_match(query, Notification.notification, + Notification.ticketnumber) ).order_by(Notification.starttime.desc()).limit(15).all() now = datetime.now(timezone.utc).replace(tzinfo=None) @@ -635,11 +640,8 @@ def _search_vendor_model_type(query, search_term): joinedload(Asset.location), ).filter( Asset.isactive == True, - db.or_( - Vendor.vendor.ilike(search_term), - Model.modelnumber.ilike(search_term), - MachineType.machinetype.ilike(search_term) - ) + _word_match(query, Vendor.vendor, Model.modelnumber, + MachineType.machinetype) ).limit(10).all() for asset in machine_assets: @@ -666,11 +668,8 @@ def _search_vendor_model_type(query, search_term): joinedload(Asset.location), ).filter( Asset.isactive == True, - db.or_( - Vendor.vendor.ilike(search_term), - Model.modelnumber.ilike(search_term), - PrinterType.printertype.ilike(search_term) - ) + _word_match(query, Vendor.vendor, Model.modelnumber, + PrinterType.printertype) ).limit(10).all() for asset in printer_assets: @@ -695,10 +694,8 @@ def _search_vendor_model_type(query, search_term): joinedload(Asset.location), ).filter( Asset.isactive == True, - db.or_( - Vendor.vendor.ilike(search_term), - NetworkDeviceType.networkdevicetype.ilike(search_term) - ) + _word_match(query, Vendor.vendor, + NetworkDeviceType.networkdevicetype) ).limit(10).all() for asset in netdev_assets: diff --git a/tests/test_core/test_search_multiword.py b/tests/test_core/test_search_multiword.py new file mode 100644 index 0000000..ac77c1e --- /dev/null +++ b/tests/test_core/test_search_multiword.py @@ -0,0 +1,39 @@ +"""Global search matches multi-word queries by WORD (every word, any order or +field), not just the exact contiguous phrase. + +Regression: 'CSF Roles' returned nothing because search did a single +ilike('%CSF Roles%'). Now each word must match (ORed across fields), so a record +with both words - even reversed or split - is found; a record with only one is +not. +""" +from shopdb.core.models import AssetType, Asset + + +def _seed(client, db, auth_headers, assetnumber, name): + if not AssetType.query.filter_by(assettype='computer').first(): + db.session.add(AssetType(assettype='computer', pluginname='computer', + tablename='computer', description='c')) + db.session.commit() + r = client.post('/api/computers', json={'assetnumber': assetnumber}, + headers=auth_headers) + assert r.status_code == 201, r.get_json() + asset = Asset.query.filter_by(assetnumber=assetnumber).first() + asset.name = name + db.session.commit() + + +def _titles(client, auth_headers, term): + r = client.get(f'/api/search?q={term}', headers=auth_headers) + assert r.status_code == 200, r.get_json() + return {h['title'] for h in r.get_json()['data']['results']} + + +def test_multiword_matches_all_words_any_order(client, db, auth_headers): + _seed(client, db, auth_headers, 'MW-1', 'CSF Roles Alpha') # contiguous + _seed(client, db, auth_headers, 'MW-2', 'Roles for the CSF Beta') # split/reversed + _seed(client, db, auth_headers, 'MW-3', 'CSF only') # one word only + + titles = _titles(client, auth_headers, 'CSF Roles') + assert 'CSF Roles Alpha' in titles # exact phrase still works + assert 'Roles for the CSF Beta' in titles # words split/reversed now match + assert 'CSF only' not in titles # missing a word -> excluded