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.
226 lines
8.6 KiB
Python
226 lines
8.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""`tools/rvt.py`'s bridge DISCOVERY and the clean-Hot-Reload exemption.
|
|
|
|
Two additive surfaces, both host-CPython testable with no Revit:
|
|
|
|
* `discover_bridges` / `cmd_bridges` / `cmd_use` -- a bare, read-only probe
|
|
of ports base..base+span-1 that answers "which live session is which",
|
|
exercised through the `_probe` injection seam (for `discover_bridges`) and
|
|
by monkeypatching `rvt._probe_status` (for the `cmd_*` wrappers). No socket
|
|
is opened: the probe is faked.
|
|
|
|
* `check_bridge_safe` -- the reload landmine guard, and its one exemption. A
|
|
second load of the same Revit process is normally the crash (a raw
|
|
`pyRevit > Reload` leaves the handler on a dead engine). A last line stamped
|
|
`hotreload` is the stop-first Hot Reload, which re-registered a fresh server
|
|
against a fresh engine and is SAFE. Both cases are pinned here against a log
|
|
written into `tmp_path`, never the repo tree.
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
_EXT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
TOOLS = os.path.join(os.path.dirname(_EXT), 'tools')
|
|
if TOOLS not in sys.path:
|
|
sys.path.insert(0, TOOLS)
|
|
|
|
import rvt # noqa: E402
|
|
|
|
BASE = rvt.DEFAULT_PORT
|
|
|
|
|
|
class Args(object):
|
|
"""A stand-in for an argparse Namespace with the fields the cmds read."""
|
|
|
|
def __init__(self, **kw):
|
|
self.span = kw.get('span', 8)
|
|
self.timeout = kw.get('timeout', 3)
|
|
self.json = kw.get('json', False)
|
|
self.export = kw.get('export', False)
|
|
self.target = kw.get('target')
|
|
|
|
|
|
def _probe_map(mapping):
|
|
"""A `_probe(port, timeout)` that returns `mapping[port]`, else None."""
|
|
def probe(port, timeout):
|
|
return mapping.get(port)
|
|
return probe
|
|
|
|
|
|
# --------------------------------------------------------- discover_bridges
|
|
|
|
def test_two_live_bridges_come_back_with_their_own_titles():
|
|
probe = _probe_map({
|
|
BASE: {'ok': True, 'doc_title': 'sample_job', 'pid': 111},
|
|
BASE + 1: {'ok': True, 'doc_title': 'sample_job_snap', 'pid': 222},
|
|
})
|
|
found = rvt.discover_bridges(_probe=probe)
|
|
assert [(e['port'], e['model']) for e in found] == [
|
|
(BASE, 'sample_job'), (BASE + 1, 'sample_job_snap')]
|
|
assert all(e['ok'] and not e['refused'] for e in found)
|
|
|
|
|
|
def test_a_gap_port_that_will_not_answer_contributes_nothing():
|
|
"""A dead port refuses the connect, the probe returns None, and the scan
|
|
steps over it rather than stopping -- the live session two ports up is
|
|
still found."""
|
|
probe = _probe_map({
|
|
BASE: {'ok': True, 'doc_title': 'sample_job'},
|
|
# BASE + 1 is a gap: no entry -> probe returns None.
|
|
BASE + 2: {'ok': True, 'doc_title': 'other_job'},
|
|
})
|
|
found = rvt.discover_bridges(_probe=probe)
|
|
assert [e['port'] for e in found] == [BASE, BASE + 2]
|
|
|
|
|
|
def test_no_bridge_up_anywhere_is_an_empty_list_not_an_error():
|
|
found = rvt.discover_bridges(_probe=lambda port, timeout: None)
|
|
assert found == []
|
|
|
|
|
|
def test_a_refused_bridge_still_counts_as_a_live_session():
|
|
"""A `refused` bridge DOES answer HTTP, so it comes back as a dict -- that
|
|
is how `bridges` shows a live-but-protected session apart from a dead
|
|
port."""
|
|
probe = _probe_map({
|
|
BASE: {'ok': False, 'refused': True, 'doc_title': 'sample_job'},
|
|
})
|
|
found = rvt.discover_bridges(_probe=probe)
|
|
assert len(found) == 1
|
|
assert found[0]['refused'] is True and found[0]['model'] == 'sample_job'
|
|
|
|
|
|
# ------------------------------------------------------------------ cmd_use
|
|
|
|
def _patch_probe(monkeypatch, mapping):
|
|
monkeypatch.setattr(rvt, '_probe_status', _probe_map(mapping))
|
|
|
|
|
|
def test_use_resolves_a_single_substring_match_to_its_port(monkeypatch, capsys):
|
|
_patch_probe(monkeypatch, {
|
|
BASE: {'ok': True, 'doc_title': 'alpha_job'},
|
|
BASE + 1: {'ok': True, 'doc_title': 'beta_job'},
|
|
})
|
|
assert rvt.cmd_use(Args(target='beta')) == 0
|
|
out = capsys.readouterr().out.strip()
|
|
assert out == str(BASE + 1)
|
|
|
|
|
|
def test_use_is_ambiguous_when_more_than_one_title_matches(monkeypatch):
|
|
_patch_probe(monkeypatch, {
|
|
BASE: {'ok': True, 'doc_title': 'sample_job_live'},
|
|
BASE + 1: {'ok': True, 'doc_title': 'sample_job_snap'},
|
|
})
|
|
with pytest.raises(SystemExit):
|
|
rvt.cmd_use(Args(target='sample_job'))
|
|
|
|
|
|
def test_use_refuses_when_no_title_matches(monkeypatch):
|
|
_patch_probe(monkeypatch, {
|
|
BASE: {'ok': True, 'doc_title': 'alpha_job'},
|
|
})
|
|
with pytest.raises(SystemExit):
|
|
rvt.cmd_use(Args(target='nothing_like_this'))
|
|
|
|
|
|
def test_use_resolves_an_explicit_port_number(monkeypatch, capsys):
|
|
_patch_probe(monkeypatch, {
|
|
BASE: {'ok': True, 'doc_title': 'alpha_job'},
|
|
BASE + 1: {'ok': True, 'doc_title': 'beta_job'},
|
|
})
|
|
assert rvt.cmd_use(Args(target=str(BASE + 1))) == 0
|
|
assert capsys.readouterr().out.strip() == str(BASE + 1)
|
|
|
|
|
|
def test_use_by_a_port_that_is_not_live_is_refused(monkeypatch):
|
|
_patch_probe(monkeypatch, {BASE: {'ok': True, 'doc_title': 'alpha_job'}})
|
|
with pytest.raises(SystemExit):
|
|
rvt.cmd_use(Args(target=str(BASE + 5)))
|
|
|
|
|
|
def test_use_export_form_prints_a_shell_assignment(monkeypatch, capsys):
|
|
_patch_probe(monkeypatch, {BASE: {'ok': True, 'doc_title': 'alpha_job'}})
|
|
assert rvt.cmd_use(Args(target='alpha', export=True)) == 0
|
|
assert capsys.readouterr().out.strip() == \
|
|
'export {0}={1}'.format(rvt.PORT_ENV, BASE)
|
|
|
|
|
|
# --------------------------------------------------------------- cmd_bridges
|
|
|
|
def test_bridges_lists_one_line_per_session_and_flags_the_refused(monkeypatch,
|
|
capsys):
|
|
_patch_probe(monkeypatch, {
|
|
BASE: {'ok': True, 'doc_title': 'alpha_job'},
|
|
BASE + 1: {'ok': False, 'refused': True, 'doc_title': 'sample_job'},
|
|
})
|
|
assert rvt.cmd_bridges(Args()) == 0
|
|
lines = capsys.readouterr().out.strip().splitlines()
|
|
assert lines[0] == '{0} alpha_job'.format(BASE)
|
|
assert lines[1].startswith('{0} sample_job'.format(BASE + 1))
|
|
assert 'REFUSED' in lines[1]
|
|
|
|
|
|
def test_bridges_with_nothing_up_returns_one_and_says_none(monkeypatch,
|
|
capsys):
|
|
_patch_probe(monkeypatch, {})
|
|
assert rvt.cmd_bridges(Args()) == 1
|
|
assert capsys.readouterr().out.strip() == '(none)'
|
|
|
|
|
|
# --------------------------------------------------- check_bridge_safe / reload
|
|
|
|
def _write_log(path, lines):
|
|
path.write_text('\n'.join(lines) + '\n', encoding='utf-8')
|
|
|
|
|
|
def test_a_clean_hot_reload_is_safe_and_does_not_refuse(tmp_path, monkeypatch):
|
|
"""Two loads of one Revit process, the LAST stamped `hotreload`: the
|
|
stop-first Hot Reload re-registered a fresh server against a fresh engine,
|
|
so the second load is safe and the guard returns without raising."""
|
|
log = tmp_path / 'sessions.log'
|
|
_write_log(log, [
|
|
'2026-09-03T09:00:00|4242|2026-09-03T08:00:00.0000000',
|
|
'2026-09-03T09:05:00|4242|2026-09-03T08:00:00.0000000|hotreload',
|
|
])
|
|
monkeypatch.setattr(rvt, 'SESSION_LOG', log)
|
|
assert rvt.check_bridge_safe() is None
|
|
|
|
|
|
def test_a_raw_second_reload_with_no_flag_still_refuses(tmp_path, monkeypatch):
|
|
"""The landmine: two 3-field loads of the same process, no `hotreload`
|
|
stamp. That is a raw `pyRevit > Reload` into a live bridge, and the guard
|
|
must refuse rather than let the next request crash Revit."""
|
|
log = tmp_path / 'sessions.log'
|
|
_write_log(log, [
|
|
'2026-09-03T09:00:00|4242|2026-09-03T08:00:00.0000000',
|
|
'2026-09-03T09:05:00|4242|2026-09-03T08:00:00.0000000',
|
|
])
|
|
monkeypatch.setattr(rvt, 'SESSION_LOG', log)
|
|
with pytest.raises(SystemExit):
|
|
rvt.check_bridge_safe()
|
|
|
|
|
|
def test_a_single_load_is_safe(tmp_path, monkeypatch):
|
|
log = tmp_path / 'sessions.log'
|
|
_write_log(log, ['2026-09-03T09:00:00|4242|2026-09-03T08:00:00.0000000'])
|
|
monkeypatch.setattr(rvt, 'SESSION_LOG', log)
|
|
assert rvt.check_bridge_safe() is None
|
|
|
|
|
|
def test_a_hotreload_in_a_DIFFERENT_process_does_not_excuse_this_one(tmp_path,
|
|
monkeypatch):
|
|
"""The exemption is keyed on the LAST line's own (pid, proc_start). A clean
|
|
hot reload in an earlier, unrelated process must not clear the refusal for
|
|
a raw double-load in the current one."""
|
|
log = tmp_path / 'sessions.log'
|
|
_write_log(log, [
|
|
'2026-09-03T08:00:00|1111|2026-09-03T07:00:00.0000000|hotreload',
|
|
'2026-09-03T09:00:00|4242|2026-09-03T08:00:00.0000000',
|
|
'2026-09-03T09:05:00|4242|2026-09-03T08:00:00.0000000',
|
|
])
|
|
monkeypatch.setattr(rvt, 'SESSION_LOG', log)
|
|
with pytest.raises(SystemExit):
|
|
rvt.check_bridge_safe()
|