# -*- coding: utf-8 -*- """Tests for tools/knowledge_drift.py. The checker is host CPython (it must be -- it parses IronPython source with `ast` and reads YAML with PyYAML), so it is loaded by path here rather than imported as a package, the same trick conftest.py uses for `vibe`. What is worth testing about a drift checker is not that it finds things. It is that it does not find things that are not there, and that it ADMITS what it could not look at -- a false finding and a silently-missed blind spot both end the same way, with somebody deciding the tool is not worth reading. """ import io 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) yaml = pytest.importorskip('yaml') knowledge_drift = pytest.importorskip('knowledge_drift') kd = knowledge_drift # ----------------------------------------------------------------- flattening def test_flatten_emits_containers_as_well_as_leaves(): # A container IS a read target: `strands.area_in2` is fetched whole and # walked, so both sides have to be able to name it. paths = kd.flatten({'a': {'b': 1, 'c': {'d': 2}}, 'e': 3}) assert set(paths) == {'a', 'a.b', 'a.c', 'a.c.d', 'e'} def test_flatten_ignores_non_string_keys(): assert kd.flatten({1: 'x', 'a': 2}) == ['a'] # ---------------------------------------------------------------- reader tiers @pytest.mark.parametrize('rel, tier', [ ('WillTools.extension/lib/vibe/ticketchecks.py', 'behaviour'), ('WillTools.extension/lib/vibe/configdump.py', 'surface'), ('WillTools.extension/lib/vibe/configmodel.py', 'surface'), # `tier_of` matches the `ProjectConfig.pushbutton` segment and nothing # around it, so this case survives the ribbon moving -- but the path is # written out as it really is (2026-08-30) rather than as it once was, # because a reader who cannot find the file stops trusting the test. ('WillTools.extension/WillTools.tab/Setup.panel/Config.pulldown/' 'ProjectConfig.pushbutton/script.py', 'surface'), ('WillTools.extension/tests/test_liftexport.py', 'test'), ('panellift/tests/conftest.py', 'test'), ('panellift/src/panellift/checks.py', 'behaviour'), ]) def test_tier_of(rel, tier): assert kd.tier_of(rel) == tier def test_tier_of_handles_windows_separators(): assert kd.tier_of(r'WillTools.extension\lib\vibe\configmodel.py') \ == 'surface' # --------------------------------------------------------------- the detector def _scan(source, rel='WillTools.extension/lib/vibe/thing.py'): import ast scan = kd.Scan(rel) scan.visit(ast.parse(source)) return scan def test_literal_get_is_a_reader(): scan = _scan("x = K.get(know, 'rails.strand_datum', 'left_rail')") assert [(r.path, r.default) for r in scan.readers] == \ [('rails.strand_datum', "'left_rail'")] assert scan.blind == [] def test_get_with_no_default_records_none_not_missing(): # `get(k, 'a')` and `get(k, 'a', None)` mean the same thing, and reporting # them as disagreeing fallbacks would be a finding about nothing. scan = _scan("K.get(know, 'a.b')\nK.get(know, 'a.b', None)") assert {r.default for r in scan.readers} == {'None'} def test_a_get_on_something_that_is_not_knowledge_is_ignored(): scan = _scan("d = {}\nd.get('a.b', 1)\nrow.get('handling.rig_mode')") assert scan.readers == [] def test_non_literal_path_becomes_a_blind_spot_not_a_reader(): scan = _scan("def f(know, path):\n return K.get(know, path, 1)\n") assert scan.readers == [] assert scan.blind and scan.blind[0][2] == 'non-literal path' def test_a_local_string_variable_is_resolved(): scan = _scan( "def f(know):\n" " key = 'handling.fast_lifter_box'\n" " return K.get(know, key, True)\n") assert [r.path for r in scan.readers] == ['handling.fast_lifter_box'] assert scan.blind == [] def test_a_ternary_between_two_literals_resolves_to_both(): # liftexport._fast_box. Refusing to follow it accused both keys of being # read only by the config screen. scan = _scan( "def f(know, batch):\n" " key = 'handling.a' if batch else 'handling.b'\n" " return K.get(know, key, not batch)\n") assert sorted(r.path for r in scan.readers) == ['handling.a', 'handling.b'] def test_a_variable_reassigned_to_something_opaque_is_not_resolved(): # Half-knowing is worse than not knowing: a confident finding about the # wrong key is unfalsifiable from the report. scan = _scan( "def f(know, other):\n" " key = 'handling.a'\n" " key = other\n" " return K.get(know, key, 1)\n") assert scan.readers == [] assert scan.blind def test_locals_do_not_leak_between_functions(): scan = _scan( "def a(know):\n" " key = 'handling.a'\n" " return K.get(know, key)\n" "def b(know, key):\n" " return K.get(know, key)\n") assert [r.path for r in scan.readers] == ['handling.a'] assert len(scan.blind) == 1 def test_concatenated_path_is_a_blind_prefix(): scan = _scan("def f(know, k):\n return K.get(know, 'nomenclature.' + k)") assert scan.blind[0][2] == 'concatenated path' assert scan.blind[0][3] == 'nomenclature.' def test_a_wrapper_body_does_not_blind_its_own_prefix(): # `_nomen` is resolved at its call sites, so treating the concat inside it # as a blind prefix would hide every nomenclature key behind the very # helper that makes them visible. scan = _scan( "def _nomen(know, key, default):\n" " return K.get(know, 'nomenclature.' + key, default)\n") assert scan.blind == [] def test_helper_call_sites_are_resolved(): scan = _scan("_nomen(know, 'design_folder', '')") assert [(r.path, r.default) for r in scan.readers] == \ [('nomenclature.design_folder', "''")] def test_param_names_call_sites_are_resolved(): scan = _scan("param_names(know, 'sheet.fc')") assert [r.path for r in scan.readers] == ['parameters.sheet.fc'] def test_helper_with_a_variable_key_is_a_blind_spot(): scan = _scan("def f(know, role):\n return param_names(know, role)") assert scan.readers == [] assert 'non-literal key into param_names' in scan.blind[0][2] def test_config_rows_are_non_covering_readers(): scan = _scan("_s('handling.rig_mode', 'Lift type', 's', 'choice', 'job')") assert len(scan.readers) == 1 assert scan.readers[0].path == 'handling.rig_mode' assert scan.readers[0].covering is False def test_bare_get_counts_only_inside_knowledge_py(): src = "def is_strand(k, n):\n return get(k, 'families.strand_include')" assert _scan(src).readers == [] inside = _scan(src, 'WillTools/lib/vibe/knowledge.py') assert [r.path for r in inside.readers] == ['families.strand_include'] def test_role_tuples_expand_into_parameter_readers(): import ast tree = ast.parse("PARAMETER_ROLES = ('panel.length', 'sheet.fc')") readers = kd._role_tuples(tree, 'x.py') assert sorted(r.path for r in readers) == \ ['parameters.panel.length', 'parameters.sheet.fc'] # ------------------------------------------------- unread / coverage semantics def _unread(declared, read, covering=None): covering = set(read) if covering is None else set(covering) out = [] kd._collect_unread(kd._paths_to_tree(sorted(declared)), '', set(read), covering, out) return sorted(out) def test_a_covering_read_of_a_branch_silences_everything_under_it(): # `strands.area_in2` is fetched whole and walked; every size in it is read. assert _unread(['s.area', 's.area.half', 's.area.three_eighths'], ['s.area']) == [] def test_an_unread_branch_is_reported_once_not_leaf_by_leaf(): # A 200-entry table nobody reads must be one line. A checker that prints # 200 gets ignored, and then the real findings go unread with them. assert _unread(['t.a', 't.a.x', 't.a.y', 't.a.z'], []) == ['t'] def test_the_tool_descends_to_the_exact_unread_sibling(): assert _unread(['n.a', 'n.b', 'n.c'], ['n.a']) == ['n.b', 'n.c'] def test_a_branch_only_NAMED_by_the_config_screen_is_not_split_into_leaves(): # One finding about the branch (surface-only) says the whole thing. assert _unread(['o', 'o.a', 'o.b'], ['o'], covering=set()) == [] def test_a_named_branch_with_one_read_leaf_still_reports_its_siblings(): assert _unread(['t', 't.a', 't.b'], ['t', 't.a'], covering={'t.a'}) \ == ['t.b'] # ------------------------------------------------------------------ allowlist def _allow(tmp_path, text, monkeypatch): path = tmp_path / 'allow.txt' path.write_text(text, encoding='utf-8') monkeypatch.setattr(kd, 'ALLOWLIST', str(path)) return kd.load_allowlist() def test_allowlist_reads_a_well_formed_entry(tmp_path, monkeypatch): entries, errors = _allow(tmp_path, '# a comment\n\nunread a.b -- derived, on purpose\n', monkeypatch) assert errors == [] assert entries == {('unread', 'a.b'): 'derived, on purpose'} def test_an_exception_with_no_reason_is_refused(tmp_path, monkeypatch): # An unexplained suppression is how a checker quietly stops checking. entries, errors = _allow(tmp_path, 'unread a.b\n', monkeypatch) assert entries == {} assert errors and 'needs a reason' in errors[0] def test_an_empty_reason_is_refused(tmp_path, monkeypatch): _entries, errors = _allow(tmp_path, 'unread a.b -- \n', monkeypatch) assert errors def test_an_unknown_category_is_refused(tmp_path, monkeypatch): _entries, errors = _allow(tmp_path, 'whoops a.b -- why\n', monkeypatch) assert errors and 'category must be one of' in errors[0] def test_a_missing_allowlist_is_not_an_error(tmp_path, monkeypatch): monkeypatch.setattr(kd, 'ALLOWLIST', str(tmp_path / 'nope.txt')) assert kd.load_allowlist() == ({}, []) def test_the_shipped_allowlist_parses(): entries, errors = kd.load_allowlist() assert errors == [], errors assert isinstance(entries, dict) # ---------------------------------------------------------------- end to end @pytest.fixture(scope='module') def result(): return kd.analyse() def test_it_runs_on_this_tree_and_finds_the_public_layers(result): assert 'knowledge.DEFAULTS' in result['layers'] assert 'catalogue.yaml' in result['layers'] assert result['counts']['declared'] > 100 assert result['counts']['read'] > 50 def test_no_python_file_failed_to_parse(result): # A file that will not parse is a file whose readers are invisible. It is # reported, never skipped quietly -- but on this tree there should be none. assert result['parse_fail'] == [] def test_the_families_tables_are_seen_as_read(result): # knowledge.is_strand calls its own `get` bare. Missing that reported the # most-read keys in the tool as read by nothing. unread = {row[0] for row in result['findings']['unread']} for path in ('families.strand_include', 'families.strand_exclude', 'families.lifter_include', 'catalogue.inserts', 'lifters.catalogue_alias', 'strands.area_in2'): assert path not in unread, path def test_nomenclature_keys_are_judged_not_hidden(result): # _nomen's own body must not blind the whole nomenclature subtree. assert 'nomenclature' not in result['dyn_prefixes'] def test_fixed_drift_does_not_come_back_as_a_finding(result): # Four of the seven known instances were fixed before this tool existed. # They are the regression test that its readers-side is not blind. everything = {row[0] for rows in result['findings'].values() for row in rows} for path in ('drawings.annotation_covered_by_parent', 'drawings.called_out_on_detail', 'rails.left_rail_is_fixed', 'catalogue.safety_factors.tension_fos'): assert path not in everything, path def test_a_test_only_reader_is_graded_as_such(): # weight_three_way's shape: production never touches it. readers = [kd.Reader('a.b', 'None', 'WillTools.extension/tests/t_x.py', 1, 'literal')] assert readers[0].tier == 'test' def test_the_report_always_states_what_it_could_not_see(result): buf = io.StringIO() kd.report(result, buf) text = buf.getvalue() assert 'WHAT THIS TOOL COULD NOT SEE' in text assert '"Could not judge" is not "judged and found nothing"' in text assert 'suppressed by' in text def test_the_blind_only_report_is_still_the_blind_report(result): buf = io.StringIO() kd.report(result, buf, blind_only=True) assert 'WHAT THIS TOOL COULD NOT SEE' in buf.getvalue() def test_the_report_never_names_a_private_file(result): # The plant and job layers are gitignored. Their KEYS may be reported; # their paths, filenames and values may not -- this output is meant to be # pasted into an issue. buf = io.StringIO() kd.report(result, buf, show_all=True) text = buf.getvalue() for banned in ('knowledge/plant', 'knowledge/job', 'knowledge\\plant', 'knowledge\\job', '.rvt', 'C:\\Users'): assert banned not in text, banned def test_an_allowlist_entry_suppresses_and_is_still_counted(tmp_path, monkeypatch): # The tool has to be able to reach a clean state on a codebase with real # exceptions -- and the run has to keep saying how many it is sitting on, # or a shrinking report reads as a shrinking problem. baseline = kd.analyse() unread = [row[0] for row in baseline['findings']['unread']] if not unread: pytest.skip('nothing unread on this tree to exempt') path = tmp_path / 'a.txt' path.write_text('unread %s -- exercised by the test suite\n' % unread[0], encoding='utf-8') monkeypatch.setattr(kd, 'ALLOWLIST', str(path)) after = kd.analyse() assert unread[0] not in [row[0] for row in after['findings']['unread']] assert after['suppressed']['unread'] == \ [(unread[0], 'exercised by the test suite')] buf = io.StringIO() kd.report(after, buf) assert '1 suppressed by' in buf.getvalue() def test_an_entry_that_matches_nothing_is_reported_stale(tmp_path, monkeypatch): # A key that was fixed or deleted must not leave a permanent blindfold. path = tmp_path / 'a.txt' path.write_text('unread no.such.key.anywhere -- gone\n', encoding='utf-8') monkeypatch.setattr(kd, 'ALLOWLIST', str(path)) res = kd.analyse() assert res['stale_allow'] == [('unread', 'no.such.key.anywhere')] buf = io.StringIO() kd.report(res, buf) assert 'STALE allowlist entries' in buf.getvalue() def test_a_malformed_allowlist_exits_two_not_one(tmp_path, monkeypatch, capsys): # Exit 1 means "drift found"; a broken exception file is a different # problem and must not be mistaken for one. path = tmp_path / 'a.txt' path.write_text('unread a.b\n', encoding='utf-8') monkeypatch.setattr(kd, 'ALLOWLIST', str(path)) assert kd.main([]) == 2 assert 'ALLOWLIST ERROR' in capsys.readouterr().out # ------------------------------------------------- honouring the env vars # # `machine-setup.md` in the private repo recommended pointing # WILLTOOLS_KNOWLEDGE / WILLTOOLS_JOBS at a private clone and said this tool # did not actually honour them -- "a recommendation, not implemented". These # pin the fix: the same env vars every other reader of these layers uses # (`vibe.plantpath`, `vibe.designspec.root_dir`) now decide where THIS tool # looks too, reusing `vibe.plantpath.plant_root()` rather than a second guess. def test_plant_root_defaults_to_the_checkout_folder_when_unset(monkeypatch): monkeypatch.delenv(kd.plantpath.PLANT_ENV, raising=False) assert kd._plant_root() == os.path.join(kd.ROOT, 'knowledge', 'plant') def test_plant_root_honours_willtools_knowledge_set_to_a_folder( tmp_path, monkeypatch): monkeypatch.setenv(kd.plantpath.PLANT_ENV, str(tmp_path)) assert kd._plant_root() == str(tmp_path) def test_plant_root_honours_willtools_knowledge_set_to_a_file( tmp_path, monkeypatch): # The single-file arrangement: the variable has always been allowed to # name the machine-wide file directly, and its DIRECTORY is the root -- # see `vibe.plantpath.plant_root`'s own docstring for why. plant_file = tmp_path / 'knowledge.yaml' monkeypatch.setenv(kd.plantpath.PLANT_ENV, str(plant_file)) assert kd._plant_root() == str(tmp_path) def test_jobs_root_defaults_to_the_checkout_folder_when_unset(monkeypatch): monkeypatch.delenv(kd.joblocate.JOBS_ENV, raising=False) assert kd._jobs_root() == os.path.join(kd.ROOT, 'knowledge', 'job') def test_jobs_root_honours_willtools_jobs_when_set(tmp_path, monkeypatch): monkeypatch.setenv(kd.joblocate.JOBS_ENV, str(tmp_path)) assert kd._jobs_root() == str(tmp_path) def test_plant_files_walks_the_env_pointed_root_not_the_checkout_one( tmp_path, monkeypatch): """The regression this exists for: an env var naming a folder OUTSIDE this checkout used to be silently ignored -- the walk never left `/knowledge/plant`, so a private clone living anywhere else was invisible to this tool no matter how the machine was set up.""" plant_dir = tmp_path / 'plants' / 'sample_plant' plant_dir.mkdir(parents=True) (plant_dir / 'knowledge.yaml').write_text( u'nomenclature:\n design_folder: X\n', encoding='utf-8') monkeypatch.setenv(kd.plantpath.PLANT_ENV, str(tmp_path / 'plants')) found = kd._plant_files() assert found == [str(plant_dir / 'knowledge.yaml')] def test_plant_files_skips_bak_files_from_the_env_pointed_root( tmp_path, monkeypatch): plant_dir = tmp_path / 'sample_plant' plant_dir.mkdir() (plant_dir / 'knowledge.yaml').write_text('x: 1\n', encoding='utf-8') (plant_dir / 'knowledge.yaml.bak-20260101').write_text( 'x: 1\n', encoding='utf-8') monkeypatch.setenv(kd.plantpath.PLANT_ENV, str(tmp_path)) found = kd._plant_files() assert found == [str(plant_dir / 'knowledge.yaml')] def test_job_files_walks_the_env_pointed_root_not_the_checkout_one( tmp_path, monkeypatch): job_dir = tmp_path / 'jobs' / 'sample_job' job_dir.mkdir(parents=True) (job_dir / 'knowledge.yaml').write_text( u'scope:\n declared_by: Will\n', encoding='utf-8') monkeypatch.setenv(kd.joblocate.JOBS_ENV, str(tmp_path / 'jobs')) found = kd._job_files() assert found == [str(job_dir / 'knowledge.yaml')] def test_plant_and_job_roots_on_a_fresh_machine_with_no_env_and_no_checkout( tmp_path, monkeypatch): """Neither the env var nor `/knowledge/plant` (or `/job`) existing is not an error here -- it is the ordinary state of a fresh clone, and `_plant_files` / `_job_files` must just come back empty.""" monkeypatch.delenv(kd.plantpath.PLANT_ENV, raising=False) monkeypatch.delenv(kd.joblocate.JOBS_ENV, raising=False) monkeypatch.setattr(kd, 'ROOT', str(tmp_path)) assert kd._plant_files() == [] assert kd._job_files() == [] def test_it_still_works_with_no_private_layers(monkeypatch): # A colleague cloning the repo has only the public files. The run must # still be useful, and must say which layers it actually saw. monkeypatch.setattr(kd, '_plant_files', lambda: []) monkeypatch.setattr(kd, '_job_files', lambda: []) res = kd.analyse() assert res['layers'] == ['PLANT_STARTER', 'STARTER', 'catalogue.yaml', 'knowledge.DEFAULTS'] buf = io.StringIO() kd.report(res, buf) assert 'No private plant or job layer in this checkout' in buf.getvalue() def test_json_mode_exits_the_same_way_as_the_text_report(capsys): import json code = kd.main(['--json']) payload = json.loads(capsys.readouterr().out) total = sum(len(v) for v in payload['findings'].values()) assert code == (1 if total else 0) assert set(payload['findings']) == set(kd.CATEGORIES)