File size: 9,448 Bytes
e4ab0d4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 | """Config-contract tests for the dedicated install flags.
Drives the real glob/manager_core config reader/writer through a
subprocess-isolated harness and pins: missing keys read False
(secure-by-default), only case-insensitive "true" is truthy, write
round-trips losslessly, edits need a restart (cached_config), the
exception-fallback path supplies False, no auto-migration seeds the
flags from a legacy security_level, and the get_bool missing->False
quirk the flags rely on stays frozen.
Harness: the child process injects a stub `folder_paths` (routing
import-time side effects into a tmpdir, and making has_system_user_api()
True so force_security_level_if_needed does not force 'strong'), prepends
`glob/` to ITS OWN sys.path (shadowing of stdlib `glob` confined to the
child), points manager_core.manager_config_path at a tmp config.ini,
resets cached_config, runs the scenario, and prints one JSON line for the
parent to assert.
"""
import json
import subprocess
import sys
import textwrap
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
_CHILD_PREAMBLE = textwrap.dedent(
"""
import sys, types, tempfile, os, json
tmp = tempfile.mkdtemp(prefix="cm_flags_cfg_")
stub = types.ModuleType("folder_paths")
stub.get_user_directory = lambda: tmp
stub.get_system_user_directory = lambda *a, **k: os.path.join(tmp, "sysuser")
sys.modules["folder_paths"] = stub
sys.path.insert(0, {glob_path!r})
import manager_core
CONFIG_PATH = os.path.join(tmp, "config.ini")
manager_core.manager_config_path = CONFIG_PATH
manager_core.cached_config = None
def write_ini(text):
with open(CONFIG_PATH, "w") as f:
f.write(text)
def fresh_read():
manager_core.cached_config = None
return manager_core.get_config()
def flag_view(cfg):
return {{
"git": cfg.get("allow_git_url_install", "<ABSENT>"),
"pip": cfg.get("allow_pip_install", "<ABSENT>"),
}}
"""
)
def _run_child(body):
"""Run a scenario body in the isolated child; return its JSON payload."""
script = _CHILD_PREAMBLE.format(glob_path=str(REPO_ROOT / "glob")) + textwrap.dedent(body)
proc = subprocess.run(
[sys.executable, "-c", script],
capture_output=True,
text=True,
timeout=180,
cwd=str(REPO_ROOT),
)
if proc.returncode != 0:
raise AssertionError(
"config-harness child failed (rc=%d). stderr tail:\n%s"
% (proc.returncode, "\n".join(proc.stderr.strip().splitlines()[-8:]))
)
lines = proc.stdout.strip().splitlines()
if not lines:
raise AssertionError(
"config-harness child exited 0 but produced no stdout. stderr tail:\n%s"
% "\n".join(proc.stderr.strip().splitlines()[-8:])
)
last_line = lines[-1]
try:
return json.loads(last_line)
except json.JSONDecodeError as e:
raise AssertionError(
"config-harness child emitted a non-JSON last line: %r\nfull stdout:\n%s"
% (last_line, proc.stdout)
) from e
class InstallFlagsConfigContractTest(unittest.TestCase):
def test_sc17_missing_keys_read_false(self):
"""Both keys absent from config.ini -> both flags read False
(secure-by-default)."""
payload = _run_child(
"""
write_ini("[default]\\nsecurity_level = normal\\n")
print(json.dumps(flag_view(fresh_read())))
"""
)
self.assertIs(payload["git"], False)
self.assertIs(payload["pip"], False)
def test_sc18_malformed_and_case_matrix(self):
"""Only case-insensitive "true" is truthy; malformed -> False."""
payload = _run_child(
"""
out = {}
for raw in ["1", "yes", "TRUE", "true ", "true"]:
write_ini("[default]\\nallow_git_url_install = %s\\nallow_pip_install = %s\\n" % (raw, raw))
cfg = fresh_read()
out[raw] = flag_view(cfg)
print(json.dumps(out))
"""
)
expected = {
"1": False, # malformed: numeric truthiness NOT honored
"yes": False, # malformed: yes/no NOT honored
"TRUE": True, # case-insensitive read (:1724)
"true ": True, # configparser strips surrounding whitespace
"true": True,
}
for raw, want in expected.items():
with self.subTest(value=raw):
self.assertIs(payload[raw]["git"], want)
self.assertIs(payload[raw]["pip"], want)
def test_sc19_write_round_trip(self):
"""write_config persists str(bool); round-trip is lossless."""
payload = _run_child(
"""
write_ini("[default]\\nsecurity_level = normal\\n")
cfg = fresh_read()
cfg["allow_git_url_install"] = True
cfg["allow_pip_install"] = False
manager_core.write_config()
raw = open(CONFIG_PATH).read()
reread = flag_view(fresh_read())
print(json.dumps({
"raw_has_git_true": "allow_git_url_install = True" in raw,
"raw_has_pip_false": "allow_pip_install = False" in raw,
"reread": reread,
}))
"""
)
self.assertTrue(
payload["raw_has_git_true"],
"write_config must persist allow_git_url_install = True in [default]",
)
self.assertTrue(
payload["raw_has_pip_false"],
"write_config must persist allow_pip_install = False in [default]",
)
self.assertIs(payload["reread"]["git"], True)
self.assertIs(payload["reread"]["pip"], False)
def test_sc20_restart_only_activation(self):
"""Editing config.ini without restart has NO effect (cache wins);
a reset (== restart) picks up the change."""
payload = _run_child(
"""
write_ini("[default]\\nallow_git_url_install = false\\n")
first = manager_core.get_config() # populates cached_config
before_edit = flag_view(first)
write_ini("[default]\\nallow_git_url_install = true\\n")
cached = flag_view(manager_core.get_config()) # NO reset: cache must win
after_restart = flag_view(fresh_read()) # reset == restart
print(json.dumps({
"before_edit": before_edit,
"cached_after_edit": cached,
"after_restart": after_restart,
}))
"""
)
self.assertIs(payload["before_edit"]["git"], False)
self.assertIs(
payload["cached_after_edit"]["git"],
False,
"cached_config must NOT hot-reload the edited flag",
)
self.assertIs(payload["after_restart"]["git"], True)
def test_sc21_exception_fallback_supplies_false(self):
"""Corrupted config.ini -> exception-fallback dict supplies flags False."""
payload = _run_child(
"""
# No [default] section header -> read_config raises inside try,
# lands in the exception-fallback dict.
write_ini("allow_git_url_install = true\\ngarbage without section\\n")
cfg = fresh_read()
print(json.dumps({
"flags": flag_view(cfg),
"fallback_marker_file_logging": cfg.get("file_logging"),
}))
"""
)
# file_logging True proves the FALLBACK dict was used (the parse
# path would yield False for a missing file_logging key).
self.assertIs(
payload["fallback_marker_file_logging"],
True,
"corrupted ini must route through the exception-fallback dict",
)
self.assertIs(payload["flags"]["git"], False)
self.assertIs(payload["flags"]["pip"], False)
def test_sc28_no_auto_migration_from_weak(self):
"""Legacy `security_level=weak` does NOT seed the flags (no auto-migration)."""
payload = _run_child(
"""
write_ini("[default]\\nsecurity_level = weak\\n")
cfg = fresh_read()
print(json.dumps({
"flags": flag_view(cfg),
"security_level": cfg.get("security_level"),
}))
"""
)
self.assertEqual(payload["security_level"], "weak")
self.assertIs(payload["flags"]["git"], False, "no auto-seed from weak")
self.assertIs(payload["flags"]["pip"], False, "no auto-seed from weak")
def test_sc42_get_bool_quirk_guard(self):
"""get_bool ignores its default param: missing `file_logging` reads
False despite a True default. The flags rely on this missing->False
quirk; this guard pins it."""
payload = _run_child(
"""
write_ini("[default]\\nsecurity_level = normal\\n")
cfg = fresh_read()
print(json.dumps({"file_logging": cfg.get("file_logging", "<ABSENT>")}))
"""
)
self.assertIs(
payload["file_logging"],
False,
"get_bool quirk changed: missing key no longer reads False — "
"new flags rely on missing->False",
)
if __name__ == "__main__":
unittest.main(verbosity=2)
|