"""The can()/scope() equivalence property (04-architecture: 'can/scope
equivalence property test') over the seeded mini-graph: for EVERY actor
and EVERY row, the row-level Decision and the query-side filter agree
exactly. Also pins T-PRM-07/08 (importer scoping; no-batch rows
invisible to non-owners; owner/super-admin tenant-wide) and the
imports:read metadata rules incl. the view-as narrowing (B48 parity).
"""

import pytest
from sqlalchemy import select

from app.models import ImportBatch, TimeEntry
from app.services.authz.policy import EntryRef, can
from app.services.authz.scope import scope
from tests.authz.conftest import actor_for

ACTOR_SPECS = [
    ("rian", None),  # super admin + owner
    ("adi", None),  # manager
    ("gary", None),  # worker (no imports of his own)
    ("mallory", None),  # no role
    ("stranger", None),  # no users row at all
    ("rian", {"with_view_as": "adi"}),  # view-as manager
    ("rian", {"with_view_as": "gary"}),  # view-as worker
]


def _all_entry_refs(graph):
    rows = graph.session.execute(
        select(
            TimeEntry.id, TimeEntry.tenant_id, ImportBatch.imported_by
        ).outerjoin(ImportBatch, ImportBatch.id == TimeEntry.import_id)
    ).all()
    return {
        row.id: EntryRef(tenant_id=row.tenant_id, import_imported_by=row.imported_by)
        for row in rows
    }


@pytest.mark.parametrize(("username", "cookies"), ACTOR_SPECS)
def test_can_scope_equivalence_entries(graph, username, cookies):
    actor = actor_for(graph, username, cookies)
    refs = _all_entry_refs(graph)

    visible_ids = set(
        graph.session.scalars(
            scope(actor, "entries:read", select(TimeEntry.id))
        ).all()
    )
    for entry_id, ref in refs.items():
        decision = can(actor, "entries:read", ref)
        assert decision.allowed == (entry_id in visible_ids), (
            f"{username} cookies={cookies} entry={entry_id}: can()="
            f"{decision.allowed} scope={entry_id in visible_ids}\n"
            f"reasons: {decision.reason_chain}"
        )


@pytest.mark.parametrize(("username", "cookies"), ACTOR_SPECS)
def test_can_scope_equivalence_imports(graph, username, cookies):
    actor = actor_for(graph, username, cookies)
    visible = set(
        graph.session.scalars(scope(actor, "imports:read", select(ImportBatch.id))).all()
    )
    for batch in graph.session.scalars(select(ImportBatch)).all():
        decision = can(actor, "imports:read", batch)
        assert decision.allowed == (batch.id in visible), (
            f"{username} cookies={cookies} batch={batch.name}: "
            f"{decision.reason_chain}"
        )


def test_t_prm_07_manager_sees_only_own_imports(graph):
    actor = actor_for(graph, "adi")
    descriptions = set(
        graph.session.scalars(
            scope(actor, "entries:read", select(TimeEntry.description))
        ).all()
    )
    assert descriptions == {"adi row 1", "adi row 2"}
    # no-batch and orphan-batch rows are invisible (rpcs §3.7 branch 3)
    assert "manual row (no batch)" not in descriptions
    assert "orphan-batch row" not in descriptions


def test_t_prm_08_owner_and_super_admin_see_all_tenant_rows(graph):
    for username in ("rian",):
        actor = actor_for(graph, username)
        rows = graph.session.scalars(
            scope(actor, "entries:read", select(TimeEntry.description))
        ).all()
        assert len(rows) == 5  # everything in BW — but NOT the Tingang row
        assert "tingang row" not in rows


def test_worker_with_no_imports_sees_nothing(graph):
    actor = actor_for(graph, "gary")
    assert (
        graph.session.scalars(
            scope(actor, "entries:read", select(TimeEntry.id))
        ).all()
        == []
    )


def test_view_as_manager_scopes_to_target_imports(graph):
    actor = actor_for(graph, "rian", cookies={"with_view_as": "adi"})
    descriptions = set(
        graph.session.scalars(
            scope(actor, "entries:read", select(TimeEntry.description))
        ).all()
    )
    assert descriptions == {"adi row 1", "adi row 2"}


def test_view_as_narrows_batch_metadata(graph):
    """B48 parity: a REAL manager sees all batch metadata, but view-as
    of that manager narrows the batch list to their own."""
    real = actor_for(graph, "adi")
    assert (
        len(graph.session.scalars(scope(real, "imports:read", select(ImportBatch.id))).all())
        == 3
    )
    impersonated = actor_for(graph, "rian", cookies={"with_view_as": "adi"})
    names = graph.session.scalars(
        scope(impersonated, "imports:read", select(ImportBatch.name))
    ).all()
    assert names == ["adi batch"]


def test_decisions_carry_reason_chains(graph):
    """T-AZ-003: reasons name the granting engagement."""
    adi = actor_for(graph, "adi")
    ref = _all_entry_refs(graph)
    some_own = next(
        r for r in ref.values() if r.import_imported_by == graph.users["adi"].id
    )
    decision = can(adi, "entries:read", some_own)
    joined = " | ".join(decision.reason_chain)
    assert decision.allowed
    assert "subcontract#" in joined and "imported_by" in joined

    capability = can(adi, "capability:can_manage_imports")
    assert capability.allowed
    assert "subcontract#" in " | ".join(capability.reason_chain)