user-engine/tests/test_application_access.py

import io
import json
import os
import tempfile
import unittest
from dataclasses import replace
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_tenant_switch_cannot_reuse_another_tenants_allow(self):
        readback = _FakeReadback([
            ReadbackResult("informed-decision", "allow", NOW + timedelta(seconds=30), "decision:1"),
        ])
        reader = self._reader(readback)
        actor = _actor()
        self.assertEqual(reader.rows(actor, "s1")[0].status, "available")
        readback.results = [ReadbackResult("informed-decision", "deny")]
        other_tenant = replace(actor, tenant="tenant-b")
        self.assertEqual(reader.rows(other_tenant, "s1")[0].status, "denied")
        self.assertEqual(len(readback.reads), 2)
        reader.forget("s1")
        reader.rows(actor, "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())