|
14 | 14 | from tests.model_registry.model_catalog.utils import ( |
15 | 15 | get_models_from_catalog_api, |
16 | 16 | fetch_all_artifacts_with_dynamic_paging, |
| 17 | + validate_model_contains_search_term, |
17 | 18 | ) |
18 | 19 | from kubernetes.dynamic.exceptions import ResourceNotFoundError |
19 | 20 |
|
@@ -294,3 +295,150 @@ def test_multiple_artifact_type_filtering( |
294 | 295 | f"Filter returned {len(artifact_type_artifacts)} artifacts, " |
295 | 296 | f"but found {len(all_model_artifacts)} in complete list for {model_name}" |
296 | 297 | ) |
| 298 | + |
| 299 | + |
| 300 | +class TestSearchModelCatalogQParameter: |
| 301 | + """Test suite for the 'q' search parameter functionality (RHOAIENG-36911).""" |
| 302 | + |
| 303 | + @pytest.mark.parametrize("search_term", ["granite", "text", "deepseek", "red hat", "base"]) |
| 304 | + def test_q_parameter_basic_search( |
| 305 | + self: Self, search_term: str, model_catalog_rest_url: list[str], model_registry_rest_headers: dict[str, str] |
| 306 | + ): |
| 307 | + """Test basic search functionality with q parameter""" |
| 308 | + LOGGER.info(f"Testing search for term: {search_term}") |
| 309 | + |
| 310 | + response = get_models_from_catalog_api( |
| 311 | + model_catalog_rest_url=model_catalog_rest_url, |
| 312 | + model_registry_rest_headers=model_registry_rest_headers, |
| 313 | + q=search_term, |
| 314 | + ) |
| 315 | + |
| 316 | + assert "items" in response |
| 317 | + models = response.get("items", []) |
| 318 | + |
| 319 | + LOGGER.info(f"Found {len(models)} models for search term '{search_term}'") |
| 320 | + |
| 321 | + for model in models: |
| 322 | + assert validate_model_contains_search_term(model, search_term), ( |
| 323 | + f"Model '{model.get('name')}' doesn't contain search term '{search_term}' in any searchable field" |
| 324 | + ) |
| 325 | + |
| 326 | + @pytest.mark.parametrize( |
| 327 | + "search_term,case_variant", [("granite", "GRANITE"), ("text", "TEXT"), ("deepseek", "DeepSeek")] |
| 328 | + ) |
| 329 | + def test_q_parameter_case_insensitive( |
| 330 | + self: Self, |
| 331 | + search_term: str, |
| 332 | + case_variant: str, |
| 333 | + model_catalog_rest_url: list[str], |
| 334 | + model_registry_rest_headers: dict[str, str], |
| 335 | + ): |
| 336 | + """Test that search is case insensitive""" |
| 337 | + LOGGER.info(f"Testing case insensitivity: '{search_term}' vs '{case_variant}'") |
| 338 | + |
| 339 | + response1 = get_models_from_catalog_api( |
| 340 | + model_catalog_rest_url=model_catalog_rest_url, |
| 341 | + model_registry_rest_headers=model_registry_rest_headers, |
| 342 | + q=search_term, |
| 343 | + ) |
| 344 | + |
| 345 | + response2 = get_models_from_catalog_api( |
| 346 | + model_catalog_rest_url=model_catalog_rest_url, |
| 347 | + model_registry_rest_headers=model_registry_rest_headers, |
| 348 | + q=case_variant, |
| 349 | + ) |
| 350 | + |
| 351 | + models1 = response1.get("items", []) |
| 352 | + models2 = response2.get("items", []) |
| 353 | + |
| 354 | + model_ids1 = sorted([m.get("id") for m in models1]) |
| 355 | + model_ids2 = sorted([m.get("id") for m in models2]) |
| 356 | + |
| 357 | + assert model_ids1 == model_ids2, ( |
| 358 | + f"Case insensitive search failed:\n" |
| 359 | + f"'{search_term}' returned {len(models1)} models\n" |
| 360 | + f"'{case_variant}' returned {len(models2)} models" |
| 361 | + ) |
| 362 | + |
| 363 | + def test_q_parameter_no_results( |
| 364 | + self: Self, model_catalog_rest_url: list[str], model_registry_rest_headers: dict[str, str] |
| 365 | + ): |
| 366 | + """Test search with term that should return no results""" |
| 367 | + nonexistent_term = "nonexistent_search_term_12345_abcdef" |
| 368 | + LOGGER.info(f"Testing search for nonexistent term: {nonexistent_term}") |
| 369 | + |
| 370 | + response = get_models_from_catalog_api( |
| 371 | + model_catalog_rest_url=model_catalog_rest_url, |
| 372 | + model_registry_rest_headers=model_registry_rest_headers, |
| 373 | + q=nonexistent_term, |
| 374 | + ) |
| 375 | + |
| 376 | + models = response.get("items", []) |
| 377 | + assert len(models) == 0, f"Expected no results for '{nonexistent_term}', got {len(models)} models" |
| 378 | + |
| 379 | + @pytest.mark.parametrize("search_term", ["", None]) |
| 380 | + def test_q_parameter_empty_query( |
| 381 | + self: Self, search_term, model_catalog_rest_url: list[str], model_registry_rest_headers: dict[str, str] |
| 382 | + ): |
| 383 | + """Test behavior with empty or None q parameter""" |
| 384 | + LOGGER.info(f"Testing empty query: {repr(search_term)}") |
| 385 | + |
| 386 | + response = get_models_from_catalog_api( |
| 387 | + model_catalog_rest_url=model_catalog_rest_url, |
| 388 | + model_registry_rest_headers=model_registry_rest_headers, |
| 389 | + q=search_term, |
| 390 | + ) |
| 391 | + |
| 392 | + models = response.get("items", []) |
| 393 | + LOGGER.info(f"Empty/None query returned {len(models)} models") |
| 394 | + |
| 395 | + def test_q_parameter_with_source_label_filter( |
| 396 | + self: Self, model_catalog_rest_url: list[str], model_registry_rest_headers: dict[str, str] |
| 397 | + ): |
| 398 | + """Test q parameter combined with source_label filtering""" |
| 399 | + search_term = "granite" |
| 400 | + source_label = REDHAT_AI_FILTER |
| 401 | + |
| 402 | + LOGGER.info(f"Testing combined search: q='{search_term}' with sourceLabel='{source_label}'") |
| 403 | + |
| 404 | + response = get_models_from_catalog_api( |
| 405 | + model_catalog_rest_url=model_catalog_rest_url, |
| 406 | + model_registry_rest_headers=model_registry_rest_headers, |
| 407 | + q=search_term, |
| 408 | + source_label=source_label, |
| 409 | + ) |
| 410 | + |
| 411 | + models = response.get("items", []) |
| 412 | + LOGGER.info(f"Combined filter returned {len(models)} models") |
| 413 | + |
| 414 | + for model in models: |
| 415 | + assert validate_model_contains_search_term(model, search_term), ( |
| 416 | + f"Model '{model.get('name')}' doesn't contain search term '{search_term}'" |
| 417 | + ) |
| 418 | + |
| 419 | + @pytest.mark.parametrize( |
| 420 | + "special_query", |
| 421 | + [ |
| 422 | + "red hat", |
| 423 | + "granite-8b", |
| 424 | + "very_long_query_string" * 10, |
| 425 | + ], |
| 426 | + ) |
| 427 | + def test_q_parameter_edge_cases( |
| 428 | + self: Self, special_query: str, model_catalog_rest_url: list[str], model_registry_rest_headers: dict[str, str] |
| 429 | + ): |
| 430 | + """Test edge cases and special characters""" |
| 431 | + LOGGER.info(f"Testing edge case query: '{special_query[:50]}...'") |
| 432 | + |
| 433 | + response = get_models_from_catalog_api( |
| 434 | + model_catalog_rest_url=model_catalog_rest_url, |
| 435 | + model_registry_rest_headers=model_registry_rest_headers, |
| 436 | + q=special_query, |
| 437 | + ) |
| 438 | + |
| 439 | + models = response.get("items", []) |
| 440 | + if "very_long_query_string" in special_query: |
| 441 | + assert len(models) == 0, f"Expected no results for very long query string, got {len(models)} models" |
| 442 | + else: |
| 443 | + assert len(models) > 0, f"Expected results for edge case query, got {len(models)} models" |
| 444 | + LOGGER.info(f"Edge case query returned {len(models)} models") |
0 commit comments