# -*- coding: utf-8 -*- """Tests for `tools/snapshot_cycle.py` -- everything that needs no Revit. The pipeline's five steps split into two kinds of code. One kind launches and kills processes and cannot be tested here at all. The other kind DECIDES: which pid may be closed, what a `/status` answer means, which port is the sandbox, what the settings resolve to, what the request file says. Those are pure functions with their inputs injected, and they are the ones where a mistake is invisible until it is expensive -- `may_kill` most of all, because the cost of getting it wrong is the owner's Revit session. Loaded by path like `test_snapshot_defer.py`: host CPython, in a repository whose extension is IronPython 2.7. """ import json import os import sys import pytest ROOT = os.path.dirname(os.path.dirname(os.path.dirname( os.path.abspath(__file__)))) TOOLS = os.path.join(ROOT, 'tools') if TOOLS not in sys.path: sys.path.insert(0, TOOLS) from vibe import snapshotmeta as S # noqa: E402 (registers the vibe stub) snapshot_cycle = pytest.importorskip('snapshot_cycle') C = snapshot_cycle LIVE = 'sample_job' SNAP_ROOT = os.path.join('C:', os.sep, 'snapshots') SNAP_FILE = os.path.join(SNAP_ROOT, LIVE + '_20260901-0930', LIVE + S.MODEL_EXT) @pytest.fixture(autouse=True) def _no_ambient_job_env(monkeypatch): """No test here may depend on which job folders the machine running it happens to have under a real WILLTOOLS_JOBS -- this checkout's own `knowledge/job/` holds three on the machine this suite was written on, which is exactly the "ambiguous" case `resolve_job_layer` is meant to refuse on, and a test suite that refuses depending on who runs it is not a test suite. Every test that wants WILLTOOLS_JOBS or WILLTOOLS_JOB_KNOWLEDGE set states so itself, on top of this.""" monkeypatch.delenv(C.JL.JOBS_ENV, raising=False) monkeypatch.delenv('WILLTOOLS_JOB_KNOWLEDGE', raising=False) class Args(object): """The argparse namespace, with every default this module reads.""" def __init__(self, **over): self.fresh_detach = False self.use = None self.run = None self.request = None self.source = None self.base = None self.sandbox_pid = 0 self.knowledge = None self.job = None self.detached_dir = None self.revit = None self.revit_exe = None self.pyrevit = None self.log = None self.port = None self.wait_minutes = None self.keep_open = False self.dry_run = False for key, value in over.items(): setattr(self, key, value) # --------------------------------------------------------- what may be killed # # The most important block in this file. Every allowance is a claim that this # pid is not the model the owner is working in, and the refusals are what # stand between a mis-set flag and a closed live session. def test_a_pid_this_run_launched_may_be_closed(): assert C.may_kill(4242, launched=[4242]) is None def test_a_pid_nobody_can_vouch_for_is_refused_and_says_why(): reason = C.may_kill(4242) assert reason and 'REFUS' in reason assert '4242' in reason assert '--sandbox-pid' in reason def test_no_pid_at_all_is_a_refusal_not_a_silent_pass(): for value in (0, None, '', 'nonsense'): assert C.may_kill(value) def test_an_explicit_pid_that_is_a_revit_may_be_closed(): assert C.may_kill(4242, explicit=4242, image='Revit.exe') is None assert C.may_kill(4242, explicit=4242, image='REVIT.EXE') is None def test_an_explicit_pid_that_is_not_a_revit_is_refused(): reason = C.may_kill(4242, explicit=4242, image='notepad.exe') assert reason and 'notepad.exe' in reason def test_an_explicit_pid_whose_image_cannot_be_read_is_refused(): """Unknown is not a licence. Being wrong this way costs a message; being wrong the other way ends somebody's Revit session.""" reason = C.may_kill(4242, explicit=4242, image=None) assert reason and 'could not be read' in reason def test_a_pid_named_explicitly_is_STILL_refused_if_a_bridge_calls_it_live(): """The flag is a human remembering a number; the refusal is the document itself, over its own bridge. The document wins.""" reason = C.may_kill(4242, explicit=4242, image='Revit.exe', live_pids=[4242]) assert reason and 'live' in reason.lower() def test_a_launched_pid_is_still_refused_if_a_bridge_calls_it_live(): """Belt and braces: if the pid we think we started answers with a live refusal, something is wrong enough that killing is not the response.""" assert C.may_kill(4242, launched=[4242], live_pids=[4242]) def test_a_pid_the_bridge_proved_is_a_sandbox_may_be_closed(): """This is what lets the button start the pipeline without knowing a pid -- and therefore what stops a stale sandbox becoming the third Revit.""" assert C.may_kill(4242, proven=[4242]) is None # --------------------------------------------------- proving it is a sandbox def _seen(**over): row = {'port': 48884, 'verdict': 'up', 'why': 'ready', 'pid': 4242, 'doc_path': SNAP_FILE, 'doc_title': LIVE} row.update(over) return row def test_a_session_whose_model_is_under_a_snapshot_root_is_proven(): assert C.sandbox_pids_from([_seen()], [SNAP_ROOT]) == [4242] def test_a_session_with_no_local_file_can_never_be_proven(): """Which is precisely the live cloud model: it reports no path at all.""" assert C.sandbox_pids_from([_seen(doc_path=None)], [SNAP_ROOT]) == [] def test_a_refused_session_is_never_proven_whatever_it_reports(): assert C.sandbox_pids_from( [_seen(verdict='refused')], [SNAP_ROOT]) == [] def test_a_session_on_a_model_somewhere_else_is_not_proven(): other = os.path.join('C:', os.sep, 'elsewhere', LIVE + S.MODEL_EXT) assert C.sandbox_pids_from([_seen(doc_path=other)], [SNAP_ROOT]) == [] def test_no_roots_configured_proves_nothing(): assert C.sandbox_pids_from([_seen()], []) == [] def test_the_refused_sessions_are_collected_as_live_pids(): rows = [_seen(port=48884, verdict='refused', pid=111), _seen(port=48885, pid=222)] assert C.live_pids_from(rows) == [111] # ------------------------------------------------------ reading /status back def test_a_matching_session_is_up(): body = {'ok': True, 'doc_title': LIVE} assert C.status_verdict(body, LIVE)[0] == 'up' def test_the_extension_on_the_title_does_not_change_the_answer(): body = {'ok': True, 'doc_title': LIVE + S.MODEL_EXT} assert C.status_verdict(body, LIVE)[0] == 'up' def test_a_refusal_is_its_own_verdict_and_not_a_retry(): """The distinction the whole port race turns on. "The live Revit is holding this port" and "the sandbox has not started yet" look identical from outside and need opposite responses -- one is waited out, the other never resolves.""" body = {'ok': False, 'refused': True, 'doc_title': LIVE, 'error': 'REFUSED: ...'} verdict, why = C.status_verdict(body, LIVE) assert verdict == 'refused' assert 'REFUSED' in why def test_a_session_with_the_wrong_model_is_not_the_sandbox(): body = {'ok': True, 'doc_title': 'something_else'} verdict, why = C.status_verdict(body, LIVE) assert verdict == 'other-model' assert 'something_else' in why def test_nothing_answering_is_silent_rather_than_an_exception(): assert C.status_verdict(None, LIVE)[0] == 'silent' assert C.status_verdict({'ok': False, 'error': 'boom'}, LIVE)[0] == 'silent' def test_no_wanted_title_accepts_any_healthy_session(): """The pre-flight survey asks "who is out there", not "is it mine".""" assert C.status_verdict({'ok': True, 'doc_title': 'anything'}, None)[0] == 'up' # ------------------------------------------------------------- the port scan def test_the_scan_walks_up_past_the_live_session_to_the_sandbox(): """pyRevit hands 48884 to whichever Revit started first, so the sandbox is routinely NOT on the configured port. Finding it is the whole job.""" answers = { 48884: {'ok': False, 'refused': True, 'doc_title': LIVE, 'pid': 111}, 48885: {'ok': True, 'doc_title': LIVE, 'doc_path': SNAP_FILE, 'pid': 222}, } port, seen = C.find_sandbox(48884, 4, LIVE, SNAP_FILE, ask=lambda p: answers.get(p)) assert port == 48885 assert seen[0]['verdict'] == 'refused' assert C.live_pids_from(seen) == [111] def test_the_scan_reports_every_port_it_saw_when_it_finds_nothing(): port, seen = C.find_sandbox(48884, 3, LIVE, ask=lambda p: None) assert port is None assert [row['port'] for row in seen] == [48884, 48885, 48886] assert all(row['verdict'] == 'silent' for row in seen) def test_a_right_title_on_the_wrong_file_is_not_accepted(): """The trap the identical titles create: the sandbox and the live model are both called the same thing, so the file has to be checked too.""" other = os.path.join('C:', os.sep, 'elsewhere', LIVE + S.MODEL_EXT) answers = {48884: {'ok': True, 'doc_title': LIVE, 'doc_path': other, 'pid': 1}} port, seen = C.find_sandbox(48884, 1, LIVE, SNAP_FILE, ask=lambda p: answers.get(p)) assert port is None assert seen[0]['verdict'] == 'other-model' def test_the_survey_visits_every_port_even_after_a_match(): """`stop_at_first=False` exists because the pre-flight scan's real job is to find the LIVE session's pid, and that session can be sitting above a port that already answered.""" answers = { 48884: {'ok': True, 'doc_title': 'scratch', 'pid': 1}, 48885: {'ok': False, 'refused': True, 'doc_title': LIVE, 'pid': 2}, } port, seen = C.find_sandbox(48884, 2, None, ask=lambda p: answers.get(p), stop_at_first=False) assert port == 48884 assert C.live_pids_from(seen) == [2] # ------------------------------------------------------------------ settings def test_flags_beat_the_knowledge_layer_which_beats_the_defaults(): know, error = C.read_knowledge(None) assert error is None settings = C.resolve(Args(detached_dir=SNAP_ROOT, run=['echo one']), know) assert settings['detached_dir'] == SNAP_ROOT assert settings['batch'] == ['echo one'] assert settings['layout'] == S.FOLDER # straight from DEFAULTS def test_the_default_batch_is_the_sweep(): """The commission's default: open the sandbox, re-export, get a verdict.""" know, _ = C.read_knowledge(None) settings = C.resolve(Args(), know) assert settings['batch'] == ['python tools/sweep.py'] def test_the_detached_dir_is_a_sandbox_root_even_when_it_came_from_a_flag(): """Otherwise a `--detached-dir` run cannot recognise its own snapshots and refuses to close the sandbox it opened last time.""" know, _ = C.read_knowledge(None) settings = C.resolve(Args(detached_dir=SNAP_ROOT), know) assert any(C.G.is_under(SNAP_FILE, root) for root in settings['sandbox_roots']) def test_the_defaults_protect_no_model_at_all(): """Empty is the only honest default: the tool does not know which of a machine's models is somebody's live work.""" know, _ = C.read_knowledge(None) assert C.resolve(Args(), know)['refuse_titles'] == [] def test_a_knowledge_file_supplies_the_denylist_and_the_roots(tmp_path): path = tmp_path / 'knowledge.yaml' path.write_text( 'bridge:\n' ' refuse_titles: [{0}]\n' ' allow_roots: [D:/other]\n' 'snapshot:\n' ' detached_dir: {1}\n'.format(LIVE, SNAP_ROOT.replace('\\', '/')), encoding='utf-8') know, error = C.read_knowledge(str(path)) assert error is None settings = C.resolve(Args(), know) assert settings['refuse_titles'] == [LIVE] assert 'D:/other' in settings['sandbox_roots'] def test_a_knowledge_file_that_is_not_there_is_an_error_not_a_shrug(): """Same rule as `knowledge._config_missing`: a pointer to a file that is gone must never fall back quietly, because the run would then use a configuration nobody chose and look completely normal.""" know, error = C.read_knowledge(os.path.join(str(SNAP_ROOT), 'nope.yaml')) assert error and 'no such knowledge file' in error def test_a_knowledge_file_that_will_not_parse_is_reported(tmp_path): path = tmp_path / 'knowledge.yaml' path.write_text('just a sentence, not a mapping\n', encoding='utf-8') _know, error = C.read_knowledge(str(path)) assert error # ------------------------------------------------------------- the blockers def test_it_names_every_missing_setting_at_once(): """A pipeline that reports one missing setting per attempt is a pipeline configured over four runs.""" know, _ = C.read_knowledge(None) args = Args(fresh_detach=True, run=[]) settings = C.resolve(args, know) settings['detached_dir'] = None settings['pyrevit_exe'] = None settings['revit_exe'] = None settings['batch'] = [] stops = C.blockers(settings, args) text = ' | '.join(stops) assert 'detached_dir' in text assert 'pyRevit CLI' in text assert 'Revit.exe' in text assert '--run' in text assert 'WHICH model' in text def test_neither_fresh_nor_use_is_a_blocker(): know, _ = C.read_knowledge(None) args = Args() stops = C.blockers(C.resolve(args, know), args) assert any('--fresh-detach' in line for line in stops) def test_use_that_names_no_file_is_a_blocker(tmp_path): know, _ = C.read_knowledge(None) args = Args(use=str(tmp_path / ('gone' + S.MODEL_EXT))) stops = C.blockers(C.resolve(args, know), args) assert any('names no file' in line for line in stops) def _job_dir(tmp_path, name, text): folder = tmp_path / name folder.mkdir() (folder / 'knowledge.yaml').write_text(text, encoding='utf-8') return folder def test_a_blocker_says_the_layer_was_read_and_did_not_set_the_key(tmp_path): """The distinction the disease report is named for: "not found" and "found but does not set this" are different sentences.""" job = _job_dir(tmp_path, 'onlyjob', u'scope:\n declared_by: Will\n') job_result = C.JL.JobLocateResult(path=str(job / 'knowledge.yaml'), how='single') know, _ = C.read_knowledge(str(job / 'knowledge.yaml')) args = Args(fresh_detach=True, run=[]) settings = C.resolve(args, know) settings['detached_dir'] = None stops = C.blockers(settings, args, job_result=job_result) text = ' | '.join(stops) assert str(job / 'knowledge.yaml') in text assert 'does not set this' in text def test_a_blocker_says_no_layer_was_read_when_none_was(tmp_path): job_result = C.JL.JobLocateResult( reason='WILLTOOLS_JOBS is not set, so there is no folder to look ' 'for a job under.') know, _ = C.read_knowledge(None) args = Args(fresh_detach=True, run=[]) settings = C.resolve(args, know) settings['detached_dir'] = None stops = C.blockers(settings, args, job_result=job_result) text = ' | '.join(stops) assert 'none was read' in text assert 'WILLTOOLS_JOBS is not set' in text def test_a_blocker_with_no_job_result_at_all_still_says_something_sane(): """Every existing caller before this feature passes no `job_result` -- it must read as "no information", not raise.""" know, _ = C.read_knowledge(None) args = Args(fresh_detach=True, run=[]) settings = C.resolve(args, know) settings['detached_dir'] = None stops = C.blockers(settings, args) assert any('detached_dir' in line for line in stops) # --------------------------------------------------- resolving the job layer # # Off-Revit there is no doc.Title to pick a job folder by, so WILLTOOLS_JOBS # is asked directly -- see `vibe.joblocate`. These exercise the plumbing in # `resolve_job_layer` specifically (env reading, wiring --job through); # `test_joblocate.py` is where the resolution RULE itself is tested. def test_resolve_job_layer_prefers_an_explicit_knowledge_flag(tmp_path, monkeypatch): monkeypatch.setenv(C.JL.JOBS_ENV, str(tmp_path)) _job_dir(tmp_path, 'alpha', u'x: 1\n') _job_dir(tmp_path, 'beta', u'x: 1\n') args = Args(knowledge='explicit.yaml') result = C.resolve_job_layer(args) assert result.path == 'explicit.yaml' assert result.how == 'explicit' def test_resolve_job_layer_finds_the_single_job_under_willtools_jobs( tmp_path, monkeypatch): job = _job_dir(tmp_path, 'sample_job', u'x: 1\n') monkeypatch.setenv(C.JL.JOBS_ENV, str(tmp_path)) result = C.resolve_job_layer(Args()) assert result.path == str(job / 'knowledge.yaml') assert result.blocking is False def test_resolve_job_layer_refuses_when_willtools_jobs_is_ambiguous( tmp_path, monkeypatch): _job_dir(tmp_path, 'sample_job', u'x: 1\n') _job_dir(tmp_path, 'SampleJob', u'x: 1\n') monkeypatch.setenv(C.JL.JOBS_ENV, str(tmp_path)) result = C.resolve_job_layer(Args()) assert result.path is None assert result.blocking is True assert 'sample_job' in result.reason assert 'SampleJob' in result.reason def test_resolve_job_layer_honours_the_job_flag_among_several(tmp_path, monkeypatch): _job_dir(tmp_path, 'sample_job', u'x: 1\n') samplejob = _job_dir(tmp_path, 'SampleJob', u'x: 1\n') monkeypatch.setenv(C.JL.JOBS_ENV, str(tmp_path)) result = C.resolve_job_layer(Args(job='SampleJob')) assert result.path == str(samplejob / 'knowledge.yaml') def test_resolve_job_layer_with_no_willtools_jobs_is_not_blocking( monkeypatch): monkeypatch.delenv(C.JL.JOBS_ENV, raising=False) result = C.resolve_job_layer(Args()) assert result.path is None assert result.blocking is False # ------------------------------------------------- end to end, via main() def test_dry_run_auto_finds_a_single_job_layer_under_willtools_jobs( tmp_path, monkeypatch, capsys): """The regression this whole feature exists for: no --knowledge flag, the job layer states snapshot.detached_dir, and it gets used.""" jobs_root = tmp_path / 'jobs' jobs_root.mkdir() job = _job_dir(jobs_root, 'sample_job', u'snapshot:\n' u' detached_dir: {0}\n'.format( str(tmp_path / 'snaps').replace('\\', '/'))) monkeypatch.setenv(C.JL.JOBS_ENV, str(jobs_root)) fake_revit = tmp_path / 'Revit.exe' fake_revit.write_text('x', encoding='utf-8') fake_pyrevit = tmp_path / 'pyrevit.exe' fake_pyrevit.write_text('x', encoding='utf-8') code = C.main(['--fresh-detach', '--source', str(tmp_path / (LIVE + S.MODEL_EXT)), '--revit-exe', str(fake_revit), '--pyrevit', str(fake_pyrevit), '--run', 'echo one', '--dry-run']) out = capsys.readouterr().out assert code == 0 assert str(job / 'knowledge.yaml') in out assert 'snaps' in out # snapshot.detached_dir, read from the job layer def test_main_refuses_rather_than_guesses_when_willtools_jobs_is_ambiguous( tmp_path, monkeypatch, capsys): _job_dir(tmp_path, 'sample_job', u'x: 1\n') _job_dir(tmp_path, 'sample_job_detached', u'x: 1\n') monkeypatch.setenv(C.JL.JOBS_ENV, str(tmp_path)) code = C.main(['--fresh-detach', '--dry-run']) err = capsys.readouterr().err assert code == 2 assert 'cannot start' in err assert 'sample_job' in err assert '--job' in err # --------------------------------------------------------------- the handoff def test_the_request_file_carries_the_plan_the_runner_will_write(tmp_path): know, _ = C.read_knowledge(None) settings = C.resolve(Args(detached_dir=str(tmp_path)), know) plan = S.plan(str(tmp_path), LIVE, '20260901-0930') source = {'path': None, 'title': LIVE, 'cloud': {'region': 'US', 'project_guid': 'p', 'model_guid': 'm'}} path = C.write_detach_request(settings, source, plan, None, 'c1') with open(path, 'r', encoding='utf-8') as handle: request = json.loads(handle.read()) assert request['plan']['path'] == plan['path'] assert request['source']['cloud']['model_guid'] == 'm' assert request['run'] == 'c1' # and the schema is the one the button writes, so one runner reads both for key in ('run', 'base', 'layout', 'worksets', 'plan', 'log', 'source'): assert key in request def test_the_bridge_port_travels_in_the_environment(): """Batch commands are the owner's own strings and call `rvt.py` themselves; a flag would have to be threaded through every one of them.""" env = C.bridge_env(48885) assert env[C.PORT_ENV] == '48885' # ------------------------------------------------------------- waiting for it class Clock(object): def __init__(self): self.t = 1000.0 self.slept = [] def now(self): return self.t def sleep(self, seconds): self.slept.append(seconds) self.t += seconds def test_waiting_gives_up_on_the_clock_rather_than_spinning_forever(): know, _ = C.read_knowledge(None) settings = C.resolve(Args(wait_minutes=1.0), know) clock = Clock() port, seen = C.wait_for_sandbox(settings, LIVE, SNAP_FILE, 0, sleep=clock.sleep, clock=clock.now, ask=lambda p: None) assert port is None assert seen assert clock.t - 1000.0 >= 60 @pytest.mark.skipif(sys.platform != 'win32', reason='snapshot_defer.process_alive is Windows-only ' '(OpenProcess / tasklist) and by design reports ' 'ALIVE, never dead, when it cannot tell -- which ' 'is every call off Windows') def test_waiting_stops_early_when_the_revit_it_waits_for_has_died(): """Polling a dead Revit for fifteen minutes tells nobody anything.""" know, _ = C.read_knowledge(None) settings = C.resolve(Args(wait_minutes=30.0), know) clock = Clock() dead = 0xFFFFFFF0 # cannot be a live Windows pid port, _seen = C.wait_for_sandbox(settings, LIVE, SNAP_FILE, dead, sleep=clock.sleep, clock=clock.now, ask=lambda p: None) assert port is None assert clock.t - 1000.0 < 30 * 60 def test_waiting_returns_the_port_as_soon_as_the_sandbox_answers(): know, _ = C.read_knowledge(None) settings = C.resolve(Args(), know) clock = Clock() answers = {48884: {'ok': True, 'doc_title': LIVE, 'doc_path': SNAP_FILE, 'pid': 9}} port, _seen = C.wait_for_sandbox(settings, LIVE, SNAP_FILE, 0, sleep=clock.sleep, clock=clock.now, ask=lambda p: answers.get(p)) assert port == 48884 assert clock.slept == [] # --------------------------------------------------------------- the dry run def test_the_dry_run_prints_the_plan_and_launches_nothing(tmp_path, capsys): root = tmp_path / 'snapshots' root.mkdir() fake_revit = tmp_path / 'Revit.exe' fake_revit.write_text('x', encoding='utf-8') fake_pyrevit = tmp_path / 'pyrevit.exe' fake_pyrevit.write_text('x', encoding='utf-8') code = C.main(['--fresh-detach', '--source', str(tmp_path / (LIVE + S.MODEL_EXT)), '--detached-dir', str(root), '--revit-exe', str(fake_revit), '--pyrevit', str(fake_pyrevit), '--run', 'echo one', '--dry-run']) out = capsys.readouterr().out assert code == 0 assert str(root) in out assert 'echo one' in out assert 'Nothing was launched' in out # nothing was created: not the snapshot folder, not a request file assert os.listdir(str(root)) == [] def test_the_dry_run_says_which_titles_are_protected(tmp_path, capsys): know_file = tmp_path / 'knowledge.yaml' know_file.write_text('bridge:\n refuse_titles: [{0}]\n'.format(LIVE), encoding='utf-8') fake_revit = tmp_path / 'Revit.exe' fake_revit.write_text('x', encoding='utf-8') model = tmp_path / (LIVE + S.MODEL_EXT) model.write_text('x', encoding='utf-8') C.main(['--use', str(model), '--revit-exe', str(fake_revit), '--knowledge', str(know_file), '--run', 'echo one', '--dry-run']) out = capsys.readouterr().out assert LIVE in out assert 'refuse titles' in out def test_a_run_that_cannot_start_says_so_and_starts_nothing(capsys): code = C.main(['--fresh-detach', '--detached-dir', '', '--dry-run']) assert code == 2 assert 'cannot start' in capsys.readouterr().err # --------------------------------------------------- the 2026-09-02 incident # # See the matching block in `test_snapshot_defer.py`: the button-launched run # that failed on 2026-09-01 printed its refusal to a stderr the button never # redirects anywhere (`_spawn` uses `CreateNoWindow = True` with no output # capture), so a failure that only printed was, in practice, a failure that # vanished. These lock in that a `--detached-dir` (which the button always # passes) now gets a journal line for the pre-`resolve()` refusals too, and # that `snapshot.revit_exe`'s own path is used to recover the Revit year # when `snapshot.revit_year` was never separately configured -- the exact # job-layer shape (`revit_exe` set, `revit_year` not) that let the actual # incident's `pyrevit run` omit `--revit=` entirely. def test_revit_year_is_recovered_from_the_revit_exe_path_when_unset(): know, _ = C.read_knowledge(None) args = Args(revit_exe=r'C:\Program Files\Autodesk\Revit 2024\Revit.exe') settings = C.resolve(args, know) assert settings['revit_year'] == '2024' def test_an_explicit_revit_year_is_never_overridden_by_the_exe_guess(): know, _ = C.read_knowledge(None) args = Args(revit=2025, revit_exe=r'C:\Program Files\Autodesk\Revit 2024\Revit.exe') settings = C.resolve(args, know) assert settings['revit_year'] == 2025 def test_a_run_that_cannot_start_because_of_a_bad_job_layer_journals_too( tmp_path, monkeypatch): """Before 2026-09-02 this branch (an ambiguous WILLTOOLS_JOBS) only ever printed to stderr -- exactly the channel a button-launched run cannot be read from. `--detached-dir` is what the button always passes, so it is what recovers a log path before `settings` exists at all.""" monkeypatch.setenv(C.JL.JOBS_ENV, str(tmp_path)) _job_dir(tmp_path, 'sample_job', u'x: 1\n') _job_dir(tmp_path, 'SampleJob', u'x: 1\n') detached_dir = tmp_path / 'snaps' code = C.main(['--fresh-detach', '--detached-dir', str(detached_dir), '--dry-run']) assert code == 2 rows = S.read(str(detached_dir / S.LOG_NAME)) assert rows and rows[-1]['event'] == S.CYCLE_FAILED assert 'sample_job' in rows[-1]['error'] def test_a_blocker_after_resolve_journals_to_the_resolved_log(tmp_path): """Once `settings` exists it names the real, resolved log path -- which can differ from the `--detached-dir`-only guess when `--log` was also given.""" detached_dir = tmp_path / 'snaps' log_path = tmp_path / 'elsewhere.jsonl' code = C.main(['--fresh-detach', '--detached-dir', str(detached_dir), '--log', str(log_path), '--dry-run']) assert code == 2 rows = S.read(str(log_path)) assert rows and rows[-1]['event'] == S.CYCLE_FAILED # --------------------------------------------------- where the identity comes from # # A cloud model is found by its region and two GUIDs, and nothing on this # machine knows those except a Revit session with the model open. So the # identity has to be written down once and reused, or every run needs the # owner to press a button first -- which is exactly the thing he said was too # slow to be part of the loop. CLOUD = {'region': 'US', 'project_guid': 'p-guid', 'model_guid': 'm-guid'} def test_a_request_file_is_the_most_direct_answer(tmp_path): req = tmp_path / 'req.json' req.write_text(json.dumps({'source': {'cloud': CLOUD, 'title': LIVE}}), encoding='utf-8') source, request = C.source_from(Args(request=str(req)), {}) assert source['cloud'] == CLOUD assert request is not None def test_a_stated_path_needs_no_revit_and_no_cloud(tmp_path): model = str(tmp_path / (LIVE + S.MODEL_EXT)) source, _ = C.source_from(Args(source=model), {}) assert source['path'] == model assert source['cloud'] is None assert source['title'] == LIVE def test_the_job_layer_can_state_the_cloud_identity_outright(): source, _ = C.source_from(Args(), {'source_cloud': CLOUD, 'source_title': LIVE}) assert source['cloud'] == CLOUD def test_the_cache_makes_the_next_run_need_no_revit_at_all(tmp_path): """One button press is the whole bootstrap: it writes the identity, and every run after it starts from a terminal.""" written = C.remember_source(str(tmp_path), {'cloud': CLOUD, 'path': None, 'title': LIVE}) assert written and os.path.isfile(written) source, _ = C.source_from(Args(), {'detached_dir': str(tmp_path)}) assert source['cloud'] == CLOUD def test_a_source_with_nothing_in_it_is_not_cached(tmp_path): """A cache file saying "no model" is worse than none: the next run would read it and think the question was answered.""" assert C.remember_source(str(tmp_path), {'path': None, 'cloud': None}) is None assert C.remember_source(str(tmp_path), {}) is None assert not os.path.isfile(C.source_cache_path(str(tmp_path))) def test_caching_never_raises_on_a_folder_it_cannot_write(): assert C.remember_source('', {'cloud': CLOUD}) is None assert C.remember_source(None, {'cloud': CLOUD}) is None def test_a_corrupt_cache_reads_as_no_answer_rather_than_a_crash(tmp_path): cache = tmp_path / C.SOURCE_CACHE cache.write_text('{not json', encoding='utf-8') source, _ = C.source_from(Args(), {'detached_dir': str(tmp_path)}) assert not C.has_source(source) def test_a_fresh_detach_with_no_identity_anywhere_refuses_and_explains(): """And the explanation has to name the bootstrap, because "press the button once" is not a thing anybody guesses from "no source".""" know, _ = C.read_knowledge(None) args = Args(fresh_detach=True) stops = C.blockers(C.resolve(args, know), args, {}) text = ' | '.join(stops) assert 'WHICH model' in text assert 'Snapshot button' in text def test_a_session_with_the_right_title_and_NO_file_is_never_the_sandbox(): """The live cloud session reports no local `PathName` and carries the same title as the snapshot. Treating "no path" as "nothing to compare" accepted it as the sandbox and would have pointed the whole batch at the model this feature exists to protect. No file means no proof, and no proof is a no -- the same asymmetry `bridgeguard` uses from the other end of the wire.""" answers = {48884: {'ok': True, 'doc_title': LIVE, 'doc_path': None, 'pid': 1}, 48885: {'ok': True, 'doc_title': LIVE, 'doc_path': SNAP_FILE, 'pid': 2}} port, seen = C.find_sandbox(48884, 2, LIVE, SNAP_FILE, ask=lambda p: answers.get(p)) assert port == 48885 assert seen[0]['verdict'] == 'other-model' assert 'cloud session' in seen[0]['why'] def test_an_empty_string_path_is_treated_the_same_as_none(): """Which is how the API actually reports it for a cloud document.""" answers = {48884: {'ok': True, 'doc_title': LIVE, 'doc_path': '', 'pid': 1}} port, _seen = C.find_sandbox(48884, 1, LIVE, SNAP_FILE, ask=lambda p: answers.get(p)) assert port is None