# -*- 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()