user-engine/tests/test_application_access.py

import io
import json
import os
import tempfile
import unittest
from datetime import UTC, datetime, timedelta
from unittest.mock import patch
from urllib.error import HTTPError, URLError

from user_engine.adapters.application_access import ApplicationAccessReadbackHTTPAdapter
from user_engine.adapters.flex_auth import FlexAuthHTTPAdapter
from user_engine.application_access import (
    ApplicationAccessReader,
    ReadbackResult,
    ReadbackSystem,
    ReadbackUnavailable,
    Workload,
    render_rows,
    workload_catalogue,
)
from user_engine.domain import (
    Actor,
    Application,
    ApplicationBinding,
    AuthorizationRequest,
    PrincipalType,
)

ISSUER = "https://key-cape.example/realms/coulomb"
NOW = datetime(2026, 10, 10, 12, 0, tzinfo=UTC)


class _Response(io.BytesIO):
    def __enter__(self):
        return self

    def __exit__(self, *_args):
        self.close()


def _actor(subject="subject-1"):
    return Actor(
        issuer=ISSUER, subject=subject, tenant="tenant-a",
        principal_type=PrincipalType.HUMAN, audience=("user-engine",),
        roles=("member",), tenant_roles=("CUS",), groups=("tenant:a:users",),
        scopes=("openid",), assurance={"level": "aal2", "mfa": True},
    )


def _bound(app_id, system_id, client_id):
    return (
        Application(application_id=app_id, display_name=app_id.title(), owner="owner"),
        ApplicationBinding(application_id=app_id, oidc_client_id=client_id,
                           protected_system_id=system_id),
    )


def _system(system_id, client_id, issuer=ISSUER):
    return ReadbackSystem(system_id=system_id, name=system_id, readback=True,
                          client_issuer=issuer, client_id=client_id)


class WorkloadCatalogueTests(unittest.TestCase):
    def test_lists_only_workloads_bound_in_all_three_records(self):
        applications = [
            _bound("informed", "informed-decision", "informed-client"),
            _bound("unbound", None, "unbound-client"),
            _bound("wrongclient", "wrong-client", "client-a"),
            _bound("wrongissuer", "other-issuer", "client-b"),
            _bound("nobinding", "no-client-binding", "client-c"),
            _bound("unknown", "not-in-registry", "client-d"),
        ]
        systems = [
            _system("informed-decision", "informed-client"),
            _system("wrong-client", "client-z"),
            _system("other-issuer", "client-b", issuer="https://elsewhere"),
            ReadbackSystem("no-client-binding", "x", readback=True),
        ]
        workloads = workload_catalogue(applications, systems, issuer=ISSUER + "/")
        self.assertEqual([item.system_id for item in workloads], ["informed-decision"])


class RenderTests(unittest.TestCase):
    workloads = tuple(Workload(f"app-{n}", f"App {n}", f"sys-{n}") for n in range(5))

    def test_no_row_is_inferred_as_allow(self):
        rows = render_rows(self.workloads, [
            ReadbackResult("sys-0", "allow", NOW + timedelta(seconds=30)),
            ReadbackResult("sys-1", "allow", NOW - timedelta(seconds=1)),
            ReadbackResult("sys-2", "deny"),
            ReadbackResult("sys-3", "not_applicable"),
        ], now=NOW)
        self.assertEqual(
            [row.status for row in rows],
            ["available", "unchecked", "denied", "not_offered", "unchecked"],
        )
        self.assertEqual([row.available for row in rows], [True, False, False, False, False])

    def test_allow_without_expiry_is_not_available(self):
        rows = render_rows(self.workloads[:1], [ReadbackResult("sys-0", "allow")], now=NOW)
        self.assertEqual(rows[0].status, "unchecked")

    def test_failed_read_renders_every_row_unchecked(self):
        rows = render_rows(self.workloads, None, now=NOW)
        self.assertEqual({row.status for row in rows}, {"unchecked"})


class _FakeReadback:
    def __init__(self, results, systems=None, fail_read=False, fail_systems=False):
        self.results = results
        self._systems = systems or [_system("informed-decision", "informed-client")]
        self.fail_read = fail_read
        self.fail_systems = fail_systems
        self.reads = []

    def systems(self):
        if self.fail_systems:
            raise ReadbackUnavailable("down")
        return tuple(self._systems)

    def read(self, actor, systems):
        self.reads.append((actor.subject, tuple(systems)))
        if self.fail_read:
            raise ReadbackUnavailable("down")
        return tuple(self.results)


class ReaderTests(unittest.TestCase):
    def setUp(self):
        self.now = NOW
        self.applications = [_bound("informed", "informed-decision", "informed-client")]

    def _reader(self, readback):
        return ApplicationAccessReader(
            readback, issuer=ISSUER, bound_applications=lambda: self.applications,
            clock=lambda: self.now,
        )

    def test_reuses_rows_until_allow_expiry_and_never_past_it(self):
        readback = _FakeReadback([
            ReadbackResult("informed-decision", "allow", NOW + timedelta(seconds=30), "decision:1"),
        ])
        reader = self._reader(readback)
        self.assertEqual(reader.rows(_actor(), "s1")[0].status, "available")
        self.now = NOW + timedelta(seconds=20)
        self.assertEqual(reader.rows(_actor(), "s1")[0].status, "available")
        self.assertEqual(len(readback.reads), 1)
        self.now = NOW + timedelta(seconds=30)
        reader.rows(_actor(), "s1")
        self.assertEqual(len(readback.reads), 2)

    def test_cache_is_per_subject_and_dropped_on_forget(self):
        readback = _FakeReadback([ReadbackResult("informed-decision", "deny")])
        reader = self._reader(readback)
        reader.rows(_actor("alice"), "s1")
        reader.rows(_actor("bob"), "s1")
        self.assertEqual([item[0] for item in readback.reads], ["alice", "bob"])
        reader.forget("s1")
        reader.rows(_actor("alice"), "s1")
        self.assertEqual(len(readback.reads), 3)

    def test_read_failure_is_unchecked_and_not_reused(self):
        readback = _FakeReadback([], fail_read=True)
        reader = self._reader(readback)
        self.assertEqual(reader.rows(_actor(), "s1")[0].status, "unchecked")
        reader.rows(_actor(), "s1")
        self.assertEqual(len(readback.reads), 2)

    def test_unreadable_registry_returns_none(self):
        reader = self._reader(_FakeReadback([], fail_systems=True))
        self.assertIsNone(reader.rows(_actor(), "s1"))

    def test_no_bound_workload_makes_no_decision_request(self):
        self.applications = []
        readback = _FakeReadback([])
        self.assertEqual(self._reader(readback).rows(_actor(), "s1"), ())
        self.assertEqual(readback.reads, [])


class HTTPAdapterTests(unittest.TestCase):
    def setUp(self):
        handle, self.token_file = tempfile.mkstemp()
        os.write(handle, b"readback-token\n")
        os.close(handle)
        self.addCleanup(os.unlink, self.token_file)
        self.adapter = ApplicationAccessReadbackHTTPAdapter(
            base_url="http://readback/", bearer_token_file=self.token_file,
        )

    def _patch(self, **kwargs):
        return patch("user_engine.adapters.application_access.urlopen", **kwargs)

    def test_read_sends_check_subject_shape_and_maps_results(self):
        body = _Response(json.dumps({
            "registry_snapshot_digest": "sha256:" + "0" * 64,
            "results": [
                {"system": "informed-decision", "effect": "allow", "decision_id": "decision:1",
                 "lifetime": {"kind": "ttl", "expires_at": "2026-10-10T12:00:30Z"}},
                {"system": "markitect", "effect": "not_applicable", "reason": "no rule"},
            ],
        }).encode())
        with self._patch(return_value=body) as call:
            results = self.adapter.read(_actor(), ["informed-decision", "markitect"])
        request = call.call_args.args[0]
        self.assertEqual(request.full_url, "http://readback/v1/readback/application_access")
        self.assertEqual(request.get_header("Authorization"), "Bearer readback-token")
        payload = json.loads(request.data)
        self.assertEqual(set(payload), {"tenant", "subject", "systems"})
        self.assertEqual(payload["subject"]["tenant"], "tenant-a")
        check = _check_payload(_actor())
        self.assertEqual(payload["subject"], check["subject"])
        self.assertEqual(results[0].effect, "allow")
        self.assertEqual(results[0].expires_at, NOW + timedelta(seconds=30))
        self.assertEqual(results[1].effect, "not_applicable")

    def test_mismatched_or_unknown_results_are_unavailable(self):
        bodies = [
            {"registry_snapshot_digest": "x", "results": [{"system": "other", "effect": "allow"}]},
            {"registry_snapshot_digest": "x", "results": [{"system": "a", "effect": "permit"}]},
            {"registry_snapshot_digest": "x", "results": []},
        ]
        for body in bodies:
            with self.subTest(body=body), self._patch(return_value=_Response(json.dumps(body).encode())):
                with self.assertRaises(ReadbackUnavailable):
                    self.adapter.read(_actor(), ["a"])

    def test_transport_failures_are_unavailable(self):
        for error in (URLError("down"), HTTPError("http://readback", 503, "x", {}, None), TimeoutError()):
            with self.subTest(error=error), self._patch(side_effect=error):
                with self.assertRaises(ReadbackUnavailable):
                    self.adapter.systems()

    def test_empty_token_never_calls(self):
        with open(self.token_file, "w", encoding="utf-8"):
            pass
        with self._patch() as call, self.assertRaises(ReadbackUnavailable):
            self.adapter.systems()
        call.assert_not_called()

    def test_systems_maps_client_binding(self):
        body = _Response(json.dumps({
            "registry_snapshot_digest": "sha256:" + "0" * 64,
            "systems": [
                {"id": "informed-decision", "name": "Informed Decision", "readback": True,
                 "client_binding": {"issuer": ISSUER, "client_id": "informed-client"}},
                {"id": "markitect", "readback": False},
            ],
        }).encode())
        with self._patch(return_value=body):
            systems = self.adapter.systems()
        self.assertEqual(systems[0].client_id, "informed-client")
        self.assertIsNone(systems[1].client_id)

    def test_request_size_is_bounded(self):
        with self.assertRaises(ValueError):
            self.adapter.read(_actor(), [f"s{n}" for n in range(65)])


def _check_payload(actor):
    body = _Response(json.dumps({"id": "decision:1", "effect": "deny"}).encode())
    with patch("user_engine.adapters.flex_auth.urlopen", return_value=body) as call:
        FlexAuthHTTPAdapter(base_url="http://flex-auth").check(AuthorizationRequest(
            actor=actor, resource_type="user-engine:user", resource_id="u",
            action="user.read", tenant=actor.tenant, correlation_id="c",
        ))
    return json.loads(call.call_args.args[0].data)


if __name__ == "__main__":
    unittest.main()


class PortalApplicationAccessTests(unittest.TestCase):
    def setUp(self):
        import test_portal_navigation

        test_portal_navigation.PortalNavigationTests.setUp(self)
        store = self.app.service.store
        for app_id, system_id, client_id in [
            ("informed", "informed-decision", "informed-client"),
            ("markitect", "markitect", "markitect-client"),
        ]:
            application, binding = _bound(app_id, system_id, client_id)
            store.save_application(application)
            store.save_binding(binding)
        self.readback = _FakeReadback(
            [
                ReadbackResult("informed-decision", "allow",
                               datetime.now(UTC) + timedelta(seconds=30), "decision:1"),
                ReadbackResult("markitect", "not_applicable"),
            ],
            systems=[_system("informed-decision", "informed-client"),
                     _system("markitect", "markitect-client")],
        )
        self.app.application_access = ApplicationAccessReader(
            self.readback, issuer=ISSUER,
            bound_applications=lambda: [
                (store.application(binding.application_id), binding)
                for binding in store.all_bindings()
            ],
        )

    def _onboarding(self):
        from test_web import invoke

        return invoke(self.app, "/onboarding", cookie="ue_session=member")[1]

    def test_account_page_renders_readback_rows(self):
        body = self._onboarding()
        self.assertIn(b'<li data-access="available">Informed - Available</li>', body)
        self.assertIn(b'<li data-access="not_offered">Markitect - Not available: not offered here</li>', body)
        self.assertIn(b"is not a complete inventory", body)

    def test_unreadable_registry_says_could_not_be_checked(self):
        self.readback.fail_systems = True
        self.assertIn(b"Application access could not be checked.", self._onboarding())

    def test_logout_drops_the_cached_rows(self):
        from test_web import invoke

        self._onboarding()
        self._onboarding()
        self.assertEqual(len(self.readback.reads), 1)
        invoke(self.app, "/logout", method="POST", cookie="ue_session=member", form={})
        self._onboarding()
        self.assertEqual(len(self.readback.reads), 2)

    def test_section_absent_without_readback(self):
        self.app.application_access = None
        self.assertNotIn(b"application-access-list", self._onboarding())