# -*- coding: utf-8 -*-
"""The live-model guard: what the bridge refuses to serve, and why.

This is the test the commission asked to be possible at the FUNCTION level
rather than as a source scan. `test_bridge_query_channel.py` had to parse
`startup.py`'s AST because that file does `from pyrevit import routes` at
module scope and cannot be imported off-Revit. The rule this suite exercises
was deliberately put in `vibe/bridgeguard.py` instead -- which imports nothing
that touches Revit and reads no files -- so the decision itself can be tested
directly, and `startup.py` is left holding only the plumbing.

THE CASE IT ALL TURNS ON
-------------------------
The snapshot copy carries the SAME `doc.Title` as the model it was copied
from. That is not sloppiness, it is `snapshotmeta.plan`'s default layout doing
its job: the job folder is keyed on the title, so the stamp goes in the folder
name and the title stays put. Which means a denylist of titles, on its own,
refuses the sandbox exactly as hard as it refuses the live model -- and a
bridge that refuses everything is a bridge nobody keeps.

So every test below is really about the same question: given two documents
with the same name, can the guard tell which one is the copy?
"""
import ntpath
import os

import pytest

from vibe import bridgeguard as G

# Built rather than spelled out, for the same reason `test_snapshotmeta.py`
# builds its filenames: a literal <word>.rvt in a tracked file reads as a
# model filename to `tools/scrub.py` and fails the leak audit.
#
# These are Windows model paths by definition (drive letter and all), so
# they are built with `ntpath` rather than `os.path` -- the host running
# this suite is not always Windows.
LIVE = 'sample_job'
SNAP_ROOT = ntpath.join('C:', ntpath.sep, 'snapshots')
SNAP_FILE = ntpath.join(SNAP_ROOT, LIVE + '_20260901-0930',
                         LIVE + G.MODEL_EXT)


# ------------------------------------------------------------ normalisation

@pytest.mark.parametrize('given', [
    LIVE, LIVE + G.MODEL_EXT, '  ' + LIVE + '  ', LIVE.upper(),
    LIVE.upper() + G.MODEL_EXT.upper(),
])
def test_a_title_compares_the_same_however_revit_spelled_it(given):
    """`doc.Title` comes back with and without the extension depending on
    version and on how the document was opened. Both have to match the same
    denylist entry or the guard protects one machine and not the next."""
    assert G.normalize_title(given) == G.normalize_title(LIVE)


def test_normalising_never_raises_on_rubbish():
    for value in (None, 0, [], object()):
        assert isinstance(G.normalize_title(value), str)


@pytest.mark.parametrize('raw,expect', [
    (None, []),
    ('', []),
    ([], []),
    ('one', ['one']),
    ('one;two', ['one', 'two']),
    (' one ; ; two ', ['one', 'two']),
    (['one', ' two '], ['one', 'two']),
])
def test_a_denylist_reads_the_same_from_yaml_or_from_the_environment(raw,
                                                                     expect):
    """A list is what the YAML layer produces and a `;` string is what an
    environment variable produces. A bare string in the YAML is a predictable
    mis-edit and is read as one entry, never as a list of characters."""
    assert G.split_titles(raw) == expect


# ------------------------------------------------------------------- paths

def test_a_path_inside_a_root_is_inside_it_whatever_the_slashes():
    assert G.is_under('C:/snapshots/a/b' + G.MODEL_EXT, SNAP_ROOT)
    assert G.is_under(SNAP_FILE, 'c:\\SNAPSHOTS')


def test_a_sibling_folder_with_the_same_prefix_is_not_inside():
    """The bug a bare `startswith` ships with: `C:\\snapshots-old` is not
    under `C:\\snapshots`, and a sandbox root that swallowed its neighbours
    would hand the guard's exemption to files nobody meant to exempt."""
    assert not G.is_under('C:\\snapshots-old\\x' + G.MODEL_EXT, SNAP_ROOT)


def test_no_path_is_under_nothing():
    assert not G.is_under('', SNAP_ROOT)
    assert not G.is_under(SNAP_FILE, '')
    assert not G.is_under(None, None)


# --------------------------------------------------------------- the verdict

def test_an_empty_denylist_protects_nothing_and_says_so_by_allowing():
    """The default, and it has to stay the default: the tool does not know
    which of a machine's models is somebody's live work, and inventing an
    answer would refuse a bridge on a machine that never asked for one."""
    assert G.refusal(LIVE, None, []) is None
    assert G.refusal(LIVE, None, None) is None


def test_the_live_cloud_model_is_refused_and_the_message_names_it():
    """The whole point. A cloud model has no local `PathName`, so it can
    offer neither alibi, and the refusal is the outcome."""
    reason = G.refusal(LIVE, None, [LIVE])
    assert reason
    assert LIVE in reason
    assert 'REFUSED' in reason


def test_a_model_not_on_the_list_is_served():
    assert G.refusal('something_else', None, [LIVE]) is None


def test_a_snapshot_under_the_root_is_served_despite_the_identical_title():
    """The case the whole design turns on: same title, different file."""
    assert G.refusal(LIVE, SNAP_FILE, [LIVE], [SNAP_ROOT]) is None


def test_a_snapshot_elsewhere_is_refused_until_it_can_prove_itself():
    other = ntpath.join('C:', ntpath.sep, 'elsewhere', LIVE + G.MODEL_EXT)
    assert G.refusal(LIVE, other, [LIVE], [SNAP_ROOT])


def test_the_launched_model_is_served_on_the_environment_proof_alone():
    """`WILLTOOLS_SANDBOX_MODEL` is set by `tools/snapshot_cycle.py` on the
    Revit it starts, and by nothing else. A session started from the taskbar
    cannot have it, which is exactly the distinction being drawn."""
    other = ntpath.join('C:', ntpath.sep, 'elsewhere', LIVE + G.MODEL_EXT)
    assert G.refusal(LIVE, other, [LIVE], [], sandbox_model=other) is None


def test_the_environment_proof_has_to_name_THIS_file():
    other = ntpath.join('C:', ntpath.sep, 'elsewhere', LIVE + G.MODEL_EXT)
    assert G.refusal(LIVE, SNAP_FILE, [LIVE], [], sandbox_model=other)


def test_a_document_with_no_title_is_not_refused():
    """No document open is already useless in the ordinary way; there is
    nothing to protect and nothing to touch."""
    assert G.refusal(None, None, [LIVE]) is None
    assert G.refusal('', None, [LIVE]) is None


def test_could_not_read_the_denylist_is_refused_not_treated_as_empty():
    """CLAUDE.md's rule, at the one place it costs the most to get wrong:
    "could not judge" and "judged, found nothing" are different answers. A
    bridge that cannot find out whether it is in the live model has not
    established that it is not."""
    reason = G.refusal(LIVE, SNAP_FILE, [], [SNAP_ROOT],
                       config_error='knowledge.yaml is malformed')
    assert reason and 'REFUSED' in reason
    assert 'malformed' in reason
    assert G.ENV_TITLES in reason      # and it says how to answer it


def test_the_refusal_body_still_answers_which_session_you_reached():
    body = G.payload(LIVE, G.refusal(LIVE, None, [LIVE]))
    assert body['ok'] is False
    assert body['refused'] is True
    assert body['doc_title'] == LIVE
    assert 'REFUSED' in body['error']


# ------------------------------------------------------------ configuration

def _get(tree, dotted, default=None):
    node = tree
    for part in dotted.split('.'):
        if not isinstance(node, dict) or part not in node:
            return default
        node = node[part]
    return default if node is None else node


def test_the_snapshot_root_counts_as_a_sandbox_root_without_being_listed():
    """`snapshot.detached_dir` is where the copies go, so a copy that lives
    there needs no second declaration. Requiring one would be a setting whose
    only job is to repeat another setting."""
    know = {'snapshot': {'detached_dir': SNAP_ROOT}}
    titles, roots = G.config_from(know, _get, environ={})
    assert titles == []
    assert SNAP_ROOT in roots


def test_the_environment_adds_to_the_denylist_and_cannot_shrink_it():
    """A denylist you can empty by unsetting a variable is not a denylist."""
    know = {'bridge': {'refuse_titles': [LIVE]}}
    titles, _roots = G.config_from(know, _get,
                                   environ={G.ENV_TITLES: 'another'})
    assert LIVE in titles and 'another' in titles


def test_a_title_listed_in_both_places_is_listed_once():
    know = {'bridge': {'refuse_titles': [LIVE]}}
    titles, _roots = G.config_from(know, _get,
                                   environ={G.ENV_TITLES: LIVE.upper()})
    assert len(titles) == 1


def test_config_reading_survives_a_tree_that_is_not_a_tree():
    titles, roots = G.config_from(None, _get, environ={})
    assert titles == [] and roots == []


def test_readonly_titles_reads_the_same_from_yaml_or_from_the_environment():
    know = {'bridge': {'readonly_titles': [LIVE]}}
    titles = G.readonly_titles_from(know, _get, environ={})
    assert titles == [LIVE]


def test_readonly_environment_adds_to_the_list_and_cannot_shrink_it():
    know = {'bridge': {'readonly_titles': [LIVE]}}
    titles = G.readonly_titles_from(
        know, _get, environ={G.ENV_READONLY_TITLES: 'another'})
    assert LIVE in titles and 'another' in titles


def test_readonly_titles_reading_survives_a_tree_that_is_not_a_tree():
    assert G.readonly_titles_from(None, _get, environ={}) == []


# ============================================================ read-only /exec
#
# 2026-09-01, the owner reversed refuse_titles for the live model the same
# morning it shipped: "我要把revit桥给你。你直接来sample_job。铁律是在live里面只
# 读不写。" These exercise `write_signature` and `exec_refusal`, the pieces
# that make that rule mechanical rather than a memory again.

@pytest.mark.parametrize('snippet,expect_name_contains', [
    ('with tx("rename"):\n    pass', 'tx'),
    ('vibe.set_p(wall, "Comments", "x")', 'set_p'),
    ('t = Transaction(doc, "x"); t.Start()', 'Transaction'),
    ('prm.Set(5)', 'Set'),
    ('doc.Delete(el.Id)', 'Delete'),
    ('doc.Create.NewFamilyInstance(pt, sym, level, st)', 'New'),
    ('ElementTransformUtils.MoveElement(doc, id, XYZ(1,0,0))',
     'ElementTransformUtils'),
    ('wtype.Duplicate("copy")', 'Duplicate'),
    ('doc.Regenerate()', 'Regenerate'),
    ('doc.SaveAs(path)', 'SaveAs'),
    ('doc.Save()', 'Save'),
    ('doc.LoadFamily(path)', 'LoadFamily'),
    ('mgr.SetParameterValue(p, 5)', 'SetParameterValue'),
])
def test_write_signature_names_the_pattern_it_found(snippet,
                                                     expect_name_contains):
    hit = G.write_signature(snippet)
    assert hit is not None
    assert expect_name_contains in hit


@pytest.mark.parametrize('snippet', [
    'print(doc.Title)',
    'q("OST_Walls")',
    'p(el, "Comments")',
    'result = [e.Id.IntegerValue for e in selected()]',
    'result = doc.GetElement(ElementId(123)).Name',
    '',
    None,
])
def test_write_signature_is_none_for_a_plain_read(snippet):
    assert G.write_signature(snippet) is None


def test_exec_refusal_refuses_a_transaction_on_a_readonly_title():
    reason = G.exec_refusal(LIVE, None, [LIVE], tx_name='rename walls')
    assert reason
    assert 'REFUSED' in reason
    assert LIVE in reason
    assert 'rename walls' in reason


def test_exec_refusal_refuses_a_write_looking_snippet_and_names_the_pattern():
    reason = G.exec_refusal(LIVE, None, [LIVE], code='prm.Set(5)')
    assert reason
    assert 'REFUSED' in reason
    assert 'Set' in reason


def test_exec_refusal_allows_a_plain_read_on_a_readonly_title():
    assert G.exec_refusal(LIVE, None, [LIVE],
                          code='print(doc.Title)') is None
    assert G.exec_refusal(LIVE, None, [LIVE], code=None,
                          tx_name=None) is None


def test_exec_refusal_is_none_for_a_title_not_on_the_readonly_list():
    reason = G.exec_refusal('something_else', None, [LIVE],
                            tx_name='x', code='doc.Delete(id)')
    assert reason is None


def test_exec_refusal_exempts_a_proven_sandbox_even_with_a_write_snippet():
    """Same alibi as `refusal()`: a title match alone is not enough, because
    the snapshot carries the identical title on purpose."""
    assert G.exec_refusal(LIVE, SNAP_FILE, [LIVE], [SNAP_ROOT],
                          tx_name='x') is None
    other = ntpath.join('C:', ntpath.sep, 'elsewhere', LIVE + G.MODEL_EXT)
    assert G.exec_refusal(LIVE, other, [LIVE], [SNAP_ROOT],
                          code='doc.Delete(id)')


def test_refuse_titles_and_readonly_titles_are_never_layered():
    """A title on refuse_titles is refused by `refusal()` before anything in
    `exec_refusal()` runs -- see the AST test below for the wiring in
    `startup.py` that makes that literally true, not just true by
    convention. Here: `exec_refusal` on its own does not know about
    refuse_titles at all, so a title present on BOTH lists still gets
    stopped -- by whichever check the caller runs first."""
    assert G.refusal(LIVE, None, [LIVE]) is not None
    # exec_refusal, called in isolation, only ever looks at readonly_titles.
    assert G.exec_refusal(LIVE, None, [], tx_name='x') is None


# ==================================================== the wiring in startup.py
#
# `startup.py` does `from pyrevit import routes` at module scope and cannot be
# imported on host CPython (see tests/conftest.py's docstring), so this half is
# AST over its source -- the same approach `test_bridge_query_channel.py` uses
# on the same file. The DECISION is tested above, by calling it; what is left
# to check here is only that every route actually asks.

import ast                                                       # noqa: E402

_EXT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
STARTUP = os.path.join(_EXT, 'startup.py')
RVT_PY = os.path.join(os.path.dirname(_EXT), 'tools', 'rvt.py')


def _source(path):
    with open(path, 'r', encoding='utf-8') as fh:
        return fh.read()


def _routes(tree):
    """{function name: route path} for every `@api.route(...)` handler."""
    found = {}
    for node in ast.walk(tree):
        if not isinstance(node, ast.FunctionDef):
            continue
        for dec in node.decorator_list:
            if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute) \
                    and dec.func.attr == 'route':
                path = dec.args[0].value if dec.args and \
                    isinstance(dec.args[0], ast.Constant) else None
                found[node.name] = path
    return found


def test_every_single_route_asks_the_guard_first():
    """Not most of them. A route that skips the guard is a route an agent can
    read the live model through, and the whole point of putting this in code
    was to stop it depending on somebody remembering."""
    tree = ast.parse(_source(STARTUP))
    handlers = _routes(tree)
    assert handlers, 'no @api.route handlers found in startup.py at all'
    funcs = {n.name: n for n in ast.walk(tree)
             if isinstance(n, ast.FunctionDef)}
    missing = []
    for name in handlers:
        calls = [n for n in ast.walk(funcs[name])
                 if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
                 and n.func.id == '_guard']
        if not calls:
            missing.append('{0} ({1})'.format(name, handlers[name]))
    assert not missing, (
        'these routes never call _guard(), so they would serve the live '
        'model: {0}'.format(', '.join(missing)))


def test_status_refuses_too_because_that_is_the_only_port_race_signal():
    """A refused `/status` is what tells `tools/snapshot_cycle.py` "the live
    Revit is holding this port" apart from "the sandbox is not up yet". Left
    unguarded, `/status` would answer ok with the right title -- the snapshot
    carries the same title -- and the pipeline would sweep the live model."""
    tree = ast.parse(_source(STARTUP))
    handlers = _routes(tree)
    status = [name for name, path in handlers.items() if path == '/status']
    assert status, '/status route is gone from startup.py'
    funcs = {n.name: n for n in ast.walk(tree)
             if isinstance(n, ast.FunctionDef)}
    assert '_guard' in ast.dump(funcs[status[0]])


def test_the_guard_has_no_off_switch():
    """There is deliberately no environment variable that disables it. The
    escape hatch for a machine whose knowledge layers will not load is
    WILLTOOLS_BRIDGE_REFUSE_TITLES, which STATES the answer -- an off switch
    would be the hole this exists to close."""
    text = _source(STARTUP) + _source(
        os.path.join(_EXT, 'lib', 'vibe', 'bridgeguard.py'))
    for word in ('BRIDGE_GUARD_OFF', 'DISABLE_BRIDGE_GUARD',
                 'WILLTOOLS_BRIDGE_GUARD'):
        assert word not in text, (
            '{0} looks like a way to switch the live-model guard off'
            .format(word))


def test_a_missing_guard_module_refuses_rather_than_serving():
    """A broken install is not a licence to run in the live model. The one
    fallback path in `_guard` has to be a refusal, not a `return None`."""
    tree = ast.parse(_source(STARTUP))
    funcs = {n.name: n for n in ast.walk(tree)
             if isinstance(n, ast.FunctionDef)}
    assert '_guard' in funcs
    handlers = [n for n in ast.walk(funcs['_guard'])
                if isinstance(n, ast.ExceptHandler)]
    assert handlers, '_guard no longer handles a failed import at all'
    first = handlers[0]
    returns = [n for n in ast.walk(first) if isinstance(n, ast.Return)]
    assert returns, 'the import-failure branch returns nothing'
    assert 'refused' in ast.dump(returns[0]), (
        'the import-failure branch does not return a refusal -- a bridge '
        'that cannot load its guard must not serve requests')


def test_the_guard_is_imported_lazily_not_at_module_scope():
    """`vibe/__init__.py` pulls in the Revit API. Importing it at the top of
    startup.py would tie the bridge's registration to it loading."""
    tree = ast.parse(_source(STARTUP))
    top = [n for n in tree.body if isinstance(n, (ast.Import, ast.ImportFrom))]
    names = []
    for node in top:
        if isinstance(node, ast.ImportFrom) and (node.module or '') == 'vibe':
            names.extend(a.name for a in node.names)
    assert 'bridgeguard' not in names


def test_vibe_exec_calls_the_readonly_guard_after_the_live_model_guard():
    """`_guard_exec` must run, and it must run AFTER `_guard` -- if it ran
    first or on its own, a title on refuse_titles could reach the readonly
    check instead of being refused outright. Checked by source position:
    `_guard(` has to appear before `_guard_exec(` in `vibe_exec`'s body."""
    tree = ast.parse(_source(STARTUP))
    funcs = {n.name: n for n in ast.walk(tree)
             if isinstance(n, ast.FunctionDef)}
    assert 'vibe_exec' in funcs
    assert '_guard_exec' in funcs, \
        '_guard_exec is gone -- /exec would no longer refuse writes on a ' \
        'readonly_titles document'
    calls = [n for n in ast.walk(funcs['vibe_exec'])
             if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
             and n.func.id in ('_guard', '_guard_exec')]
    names = [c.func.id for c in sorted(calls, key=lambda c: c.lineno)]
    assert '_guard' in names and '_guard_exec' in names
    assert names.index('_guard') < names.index('_guard_exec'), (
        'vibe_exec calls _guard_exec before (or without regard to) _guard, '
        'so a refuse_titles document could reach the readonly check instead '
        'of being refused outright')


def test_only_exec_calls_the_readonly_guard():
    """`/status`, `/collect`, `/params`, `/selection` are reads by
    construction -- no transaction, no arbitrary code -- so none of them
    should ever call `_guard_exec`; it would be dead weight at best and a
    sign the read/write split has blurred at worst."""
    tree = ast.parse(_source(STARTUP))
    handlers = _routes(tree)
    funcs = {n.name: n for n in ast.walk(tree)
             if isinstance(n, ast.FunctionDef)}
    for name, path in handlers.items():
        if path == '/exec':
            continue
        calls = [n for n in ast.walk(funcs[name])
                 if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
                 and n.func.id == '_guard_exec']
        assert not calls, '{0} ({1}) calls _guard_exec but is not /exec' \
            .format(name, path)


def test_the_readonly_guard_reads_config_through_the_same_cached_path():
    """`_guard_exec` must not open a NEW knowledge read of its own -- it has
    to go through `_refuse_config`, the same memoised function `_guard`
    already calls, or a readonly_titles document pays for a second
    `knowledge.load()` on every single `/exec` call."""
    tree = ast.parse(_source(STARTUP))
    funcs = {n.name: n for n in ast.walk(tree)
             if isinstance(n, ast.FunctionDef)}
    assert '_guard_exec' in funcs
    calls = [n for n in ast.walk(funcs['_guard_exec'])
             if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
             and n.func.id == '_refuse_config']
    assert calls, '_guard_exec does not call _refuse_config -- it must be ' \
        'reading configuration some other, uncached way'


def test_the_readonly_guard_calls_exec_refusal():
    tree = ast.parse(_source(STARTUP))
    funcs = {n.name: n for n in ast.walk(tree)
             if isinstance(n, ast.FunctionDef)}
    assert '_guard_exec' in funcs
    assert 'exec_refusal' in ast.dump(funcs['_guard_exec'])


def test_the_client_lets_the_pipeline_pick_the_port():
    """pyRevit walks the port upward per running Revit, so a hard-coded 48884
    is "whichever Revit started first" rather than a session anybody chose."""
    source = _source(RVT_PY)
    assert 'WILLTOOLS_BRIDGE_PORT' in source
    tree = ast.parse(source)
    funcs = {n.name: n for n in ast.walk(tree)
             if isinstance(n, ast.FunctionDef)}
    assert 'base_url' in funcs, 'rvt.py lost its per-call base URL'
    # `_send` is the one function that opens the connection; `call` wraps it
    # with the correlation check (see tests/test_bridge_correlation.py). The
    # invariant is unchanged: whichever of them builds the URL must do it per
    # call, not off a module constant frozen at import.
    sender = '_send' if '_send' in funcs else 'call'
    assert 'base_url' in ast.dump(funcs[sender]), (
        '{0}() no longer builds its URL per call, so a port set after '
        'import would be ignored'.format(sender))