Add tests for tiled check

Author: tpoliaw

tests/unit_tests/service/test_authorization.py

@@ -1,11 +1,13 @@
-from unittest.mock import MagicMock, patch
+from contextlib import AbstractContextManager, nullcontext
+from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
 from pydantic import HttpUrl
 
-from blueapi.config import OpaConfig
+from blueapi.config import OIDCConfig, OpaConfig, ServiceAccount
 from blueapi.service.authorization import (
     OpaClient,
+    validate_tiled_config,
 )
 
 # Reusable client patch decorator
@@ -20,9 +22,50 @@
 def opa_config() -> OpaConfig:
     return OpaConfig(
         root=HttpUrl("http://auth.example.com"),
+        tiled_service_account_check="/auth/tiled",
     )
 
 
+@patch_client_session
+@pytest.mark.parametrize(
+    "result,context",
+    [
+        (False, pytest.raises(ValueError, match="Tiled service account is not valid ")),
+        (True, nullcontext()),
+    ],
+)
+async def test_tiled_service_account(
+    session: MagicMock,
+    opa_config: OpaConfig,
+    result: bool,
+    context: AbstractContextManager,
+):
+    session.return_value.post = AsyncMock(
+        return_value=MagicMock(json=AsyncMock(return_value={"result": result}))
+    )
+
+    client = OpaClient(instrument="p99", config=opa_config)
+
+    session.assert_called_once_with(base_url="http://auth.example.com/")
+    with context:
+        await client.require_tiled_service_account(token="foo_bar")
+    session().post.assert_called_once_with(
+        "/auth/tiled",
+        json={"input": {"token": "foo_bar", "beamline": "p99", "audience": "account"}},
+    )
+
+
+@patch_client_session
+async def test_exception_raised_when_opa_fails(
+    session: MagicMock, opa_config: OpaConfig
+):
+    session.return_value.post = AsyncMock(side_effect=RuntimeError("Connection failed"))
+    async with OpaClient.for_config("p45", opa_config) as client:
+        assert client is not None
+        with pytest.raises(RuntimeError, match="Connection failed"):
+            await client.require_tiled_service_account(token="foo_bar")
+
+
 @patch_client_session
 async def test_session_closed(session: MagicMock, opa_config: OpaConfig):
     async with OpaClient.for_config("p45", opa_config):
@@ -46,3 +89,49 @@ async def test_opa_client_without_config(instrument: str | None):
 async def test_opa_fails_without_instrument(opa_config: OpaConfig):
     with pytest.raises(ValueError, match="Instrument name is required"):
         OpaClient.for_config(None, opa_config)
+
+
+async def test_validate_tiled_config():
+    opa = MagicMock(spec=OpaClient)
+    tiled = ServiceAccount()
+    oidc = Mock(spec=OIDCConfig)
+    oidc.token_endpoint = "token-endpoint"
+    with patch("blueapi.service.authorization.TiledAuth") as auth:
+        auth.return_value.get_access_token.return_value = "tiled-token"
+        await validate_tiled_config(tiled, oidc, opa)
+
+    auth.assert_called_once_with(tiled)
+    opa.require_tiled_service_account.assert_called_once_with("tiled-token")
+
+
+@pytest.mark.parametrize(
+    "tiled_auth,oidc,opa_client",
+    [
+        (None, None, MagicMock(spec=OpaClient)),
+        (
+            None,
+            OIDCConfig(well_known_url="http://example.com", client_id="test-client"),
+            MagicMock(spec=OpaClient),
+        ),
+        ("api_key", None, MagicMock(spec=OpaClient)),
+        (
+            "api_key",
+            OIDCConfig(well_known_url="http://example.com", client_id="test-client"),
+            MagicMock(spec=OpaClient),
+        ),
+        (ServiceAccount(), None, MagicMock(spec=OpaClient)),
+        (
+            ServiceAccount(),
+            OIDCConfig(well_known_url="http://example.com", client_id="test-client"),
+            None,
+        ),
+    ],
+)
+async def test_validate_tiled_config_with_missing_config(
+    tiled_auth: ServiceAccount | str | None,
+    oidc: OIDCConfig | None,
+    opa_client: MagicMock | None,
+):
+    assert await validate_tiled_config(tiled_auth, oidc, opa_client) is None
+    if opa_client is not None:
+        opa_client.require_tiled_service_account.assert_not_called()