|
1 | 1 | """Unit tests for authentication/k8s module.""" |
2 | 2 |
|
3 | | -# pylint: disable=too-many-arguments,too-many-positional-arguments,too-few-public-methods,protected-access |
| 3 | +# pylint: disable=too-many-arguments,too-many-positional-arguments,too-few-public-methods,protected-access,too-many-lines |
4 | 4 |
|
5 | 5 | import os |
6 | 6 | from http import HTTPStatus |
@@ -481,6 +481,143 @@ async def test_auth_dependency_no_token_normal_endpoints( |
481 | 481 | assert detail["cause"] == "No Authorization header found" |
482 | 482 |
|
483 | 483 |
|
| 484 | +@pytest.mark.asyncio |
| 485 | +async def test_auth_dependency_no_token_metrics_endpoint_skip_enabled( |
| 486 | + mocker: MockerFixture, |
| 487 | +) -> None: |
| 488 | + """Test the auth dependency without a token for the metrics endpoint. |
| 489 | +
|
| 490 | + When skip_for_metrics is True, the /metrics endpoint should bypass auth. |
| 491 | + """ |
| 492 | + config_dict = { |
| 493 | + "name": "test", |
| 494 | + "service": { |
| 495 | + "host": "localhost", |
| 496 | + "port": 8080, |
| 497 | + "auth_enabled": False, |
| 498 | + "workers": 1, |
| 499 | + "color_log": True, |
| 500 | + "access_log": True, |
| 501 | + }, |
| 502 | + "llama_stack": { |
| 503 | + "api_key": "test-key", |
| 504 | + "url": "http://test.com:1234", |
| 505 | + "use_as_library_client": False, |
| 506 | + }, |
| 507 | + "authentication": { |
| 508 | + "module": "k8s", |
| 509 | + "skip_for_metrics": True, |
| 510 | + }, |
| 511 | + "user_data_collection": { |
| 512 | + "feedback_enabled": False, |
| 513 | + "feedback_storage": ".", |
| 514 | + "transcripts_enabled": False, |
| 515 | + "transcripts_storage": ".", |
| 516 | + }, |
| 517 | + } |
| 518 | + cfg = AppConfig() |
| 519 | + cfg.init_from_dict(config_dict) |
| 520 | + mocker.patch("authentication.k8s.configuration", cfg) |
| 521 | + |
| 522 | + dependency = K8SAuthDependency() |
| 523 | + |
| 524 | + # Mock the Kubernetes API calls |
| 525 | + mock_authn_api = mocker.patch("authentication.k8s.K8sClientSingleton.get_authn_api") |
| 526 | + mock_authz_api = mocker.patch("authentication.k8s.K8sClientSingleton.get_authz_api") |
| 527 | + |
| 528 | + # Setup mock responses for invalid token |
| 529 | + mock_authn_api.return_value.create_token_review.return_value = MockK8sResponse( |
| 530 | + authenticated=False |
| 531 | + ) |
| 532 | + mock_authz_api.return_value.create_subject_access_review.return_value = ( |
| 533 | + MockK8sResponse(allowed=False) |
| 534 | + ) |
| 535 | + |
| 536 | + request = Request( |
| 537 | + scope={ |
| 538 | + "type": "http", |
| 539 | + "headers": [], |
| 540 | + "path": "/metrics", |
| 541 | + } |
| 542 | + ) |
| 543 | + |
| 544 | + user_uid, username, skip_userid_check, token = await dependency(request) |
| 545 | + |
| 546 | + assert user_uid == "00000000-0000-0000-0000-000" |
| 547 | + assert username == "lightspeed-user" |
| 548 | + assert skip_userid_check is True |
| 549 | + assert token == "" |
| 550 | + |
| 551 | + |
| 552 | +@pytest.mark.asyncio |
| 553 | +async def test_auth_dependency_no_token_metrics_endpoint_skip_disabled( |
| 554 | + mocker: MockerFixture, |
| 555 | +) -> None: |
| 556 | + """Test the auth dependency without a token for the metrics endpoint. |
| 557 | +
|
| 558 | + When skip_for_metrics is False, the /metrics endpoint should require auth. |
| 559 | + """ |
| 560 | + config_dict = { |
| 561 | + "name": "test", |
| 562 | + "service": { |
| 563 | + "host": "localhost", |
| 564 | + "port": 8080, |
| 565 | + "auth_enabled": False, |
| 566 | + "workers": 1, |
| 567 | + "color_log": True, |
| 568 | + "access_log": True, |
| 569 | + }, |
| 570 | + "llama_stack": { |
| 571 | + "api_key": "test-key", |
| 572 | + "url": "http://test.com:1234", |
| 573 | + "use_as_library_client": False, |
| 574 | + }, |
| 575 | + "authentication": { |
| 576 | + "module": "k8s", |
| 577 | + "skip_for_metrics": False, |
| 578 | + }, |
| 579 | + "user_data_collection": { |
| 580 | + "feedback_enabled": False, |
| 581 | + "feedback_storage": ".", |
| 582 | + "transcripts_enabled": False, |
| 583 | + "transcripts_storage": ".", |
| 584 | + }, |
| 585 | + } |
| 586 | + cfg = AppConfig() |
| 587 | + cfg.init_from_dict(config_dict) |
| 588 | + mocker.patch("authentication.k8s.configuration", cfg) |
| 589 | + |
| 590 | + dependency = K8SAuthDependency() |
| 591 | + |
| 592 | + # Mock the Kubernetes API calls |
| 593 | + mock_authn_api = mocker.patch("authentication.k8s.K8sClientSingleton.get_authn_api") |
| 594 | + mock_authz_api = mocker.patch("authentication.k8s.K8sClientSingleton.get_authz_api") |
| 595 | + |
| 596 | + # Setup mock responses for invalid token |
| 597 | + mock_authn_api.return_value.create_token_review.return_value = MockK8sResponse( |
| 598 | + authenticated=False |
| 599 | + ) |
| 600 | + mock_authz_api.return_value.create_subject_access_review.return_value = ( |
| 601 | + MockK8sResponse(allowed=False) |
| 602 | + ) |
| 603 | + |
| 604 | + request = Request( |
| 605 | + scope={ |
| 606 | + "type": "http", |
| 607 | + "headers": [], |
| 608 | + "path": "/metrics", |
| 609 | + } |
| 610 | + ) |
| 611 | + |
| 612 | + with pytest.raises(HTTPException) as exc_info: |
| 613 | + await dependency(request) |
| 614 | + |
| 615 | + assert exc_info.value.status_code == 401 |
| 616 | + detail = cast(dict[str, str], exc_info.value.detail) |
| 617 | + assert detail["response"] == ("Missing or invalid credentials provided by client") |
| 618 | + assert detail["cause"] == "No Authorization header found" |
| 619 | + |
| 620 | + |
484 | 621 | @pytest.mark.asyncio |
485 | 622 | async def test_cluster_id_is_used_for_kube_admin(mocker: MockerFixture) -> None: |
486 | 623 | """Test the cluster id is used as user_id when user is kube:admin.""" |
|
0 commit comments