will
25592fa6c2
ci / test (push) Successful in 6m14s
History-free snapshot of the private development repository at 77a82eb. Code, docs and the public catalogue only: no plant or job knowledge layers, no client, job or producer names (tools/scrub.py clean against the widened term list; a second word-level grep clean). The intro deck is left out because its slide images cannot be checked for job drawings.
520 lines
22 KiB
Python
520 lines
22 KiB
Python
# -*- 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))
|