Coverage for src/lpip/config.py: 75%

366 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-09-17 20:19 +0000

1"""Load effective lpip configuration and persist set/unset with tomlkit.""" 

2 

3from __future__ import annotations 

4 

5import os 

6import tomllib 

7from collections.abc import Mapping, Sequence 

8from dataclasses import dataclass 

9from pathlib import Path 

10from typing import Any 

11 

12import tomlkit 

13from platformdirs import PlatformDirs 

14from tomlkit.exceptions import TOMLKitError 

15 

16from lpip.errors import ( 

17 ConfigAlreadyExistsError, 

18 ConfigSchemaError, 

19 InvalidSourceNameError, 

20 SourceAlreadyExistsError, 

21 SourceNotFoundError, 

22 TomlParseError, 

23 UnknownTomlKeyError, 

24) 

25from lpip.models import ( 

26 EffectiveConfig, 

27 PolicyConfig, 

28 RefreshMode, 

29 RuntimeConfig, 

30 SourceConfig, 

31 SourceType, 

32) 

33from lpip.validation import ( 

34 CONFIG_TOP_KEYS, 

35 POLICY_KEYS, 

36 RUNTIME_KEYS, 

37 SOURCE_ITEM_KEYS, 

38 SOURCE_KEYS, 

39 SOURCE_NAME_RE, 

40 first_unknown_key, 

41 optional_str, 

42) 

43 

44APPNAME = "lpip" 

45APPAUTHOR = "libesys" 

46WORKSPACE_CONFIG = Path(".tools") / "libesys" / "lpip" / "config.toml" 

47 

48# Architecture template (ADR-0013). Static write; tomlkit mutations are batch-15. 

49CONFIG_INIT_TEMPLATE = """\ 

50# lpip persistent configuration (v1). 

51# Edit values. Unknown keys fail validation. 

52 

53[source] 

54type = "git" # "git" or "file" 

55url = "ssh://git.example/lpip-catalog.git" 

56revision = "main" 

57path = "lpip-sources.toml" 

58refresh = "auto" # always | auto | never 

59 

60# Local catalog instead of Git: 

61# type = "file" 

62# path = "~/company/lpip-sources.toml" 

63# (omit url, revision, and refresh when type is file) 

64# 

65# More than one catalog: lpip source add / list / remove ([[sources]]). 

66# Do not keep both [source] and [[sources]] in the same file. 

67 

68[policy] 

69allowed_schemes = ["git+ssh", "git+https"] 

70require_pinned_revision = false 

71 

72[runtime] 

73python = "python3" 

74""" 

75 

76_ENV_CONFIG = "LPIP_CONFIG" 

77_ENV_TYPE = "LPIP_SOURCE_TYPE" 

78_ENV_PATH = "LPIP_SOURCE_PATH" 

79_ENV_URL = "LPIP_SOURCE_URL" 

80_ENV_REVISION = "LPIP_SOURCE_REVISION" 

81 

82 

83@dataclass(frozen=True) 

84class ConfigOverrides: 

85 """CLI-layer overlays (applied last).""" 

86 

87 config_path: Path | None = None 

88 source_type: str | None = None 

89 source_path: str | None = None 

90 source_url: str | None = None 

91 source_revision: str | None = None 

92 

93 

94def platform_dirs() -> PlatformDirs: 

95 return PlatformDirs(appname=APPNAME, appauthor=APPAUTHOR) 

96 

97 

98def default_user_config_path() -> Path: 

99 return Path(platform_dirs().user_config_dir) / "config.toml" 

100 

101 

102def default_user_cache_dir() -> Path: 

103 return Path(platform_dirs().user_cache_dir) 

104 

105 

106def find_git_root(start: Path) -> Path | None: 

107 current = start.resolve() 

108 for directory in (current, *current.parents): 

109 marker = directory / ".git" 

110 if marker.is_dir() or marker.is_file(): 

111 return directory 

112 return None 

113 

114 

115def workspace_config_path(git_root: Path) -> Path: 

116 return git_root / WORKSPACE_CONFIG 

117 

118 

119def load_effective_config( 

120 *, 

121 cwd: Path | None = None, 

122 environ: Mapping[str, str] | None = None, 

123 overrides: ConfigOverrides | None = None, 

124 user_config_path: Path | None = None, 

125) -> EffectiveConfig: 

126 """Merge CLI, env, workspace/user file, and defaults. Does not create files.""" 

127 cwd = (cwd or Path.cwd()).resolve() 

128 env = environ if environ is not None else os.environ 

129 overrides = overrides or ConfigOverrides() 

130 user_path = user_config_path if user_config_path is not None else default_user_config_path() 

131 

132 persistent = _select_persistent_file(cwd, env, overrides, user_path) 

133 raw = _read_config_file(persistent) if persistent is not None else {} 

134 merged = _apply_env_and_cli(raw, env, overrides) 

135 return _build_effective(merged, persistent) 

136 

137 

138def resolve_init_target( 

139 *, 

140 cwd: Path | None = None, 

141 environ: Mapping[str, str] | None = None, 

142 user_config_path: Path | None = None, 

143) -> Path: 

144 """Path ``config init`` would write (ADR-0013). Does not create files.""" 

145 cwd = (cwd or Path.cwd()).resolve() 

146 env = environ if environ is not None else os.environ 

147 user_path = user_config_path if user_config_path is not None else default_user_config_path() 

148 git_root = find_git_root(cwd) 

149 if git_root is not None: 

150 return workspace_config_path(git_root) 

151 explicit = env.get(_ENV_CONFIG) 

152 if explicit: 152 ↛ 154line 152 didn't jump to line 154 because the condition on line 152 was always true

153 return Path(explicit).expanduser() 

154 return user_path 

155 

156 

157def write_init_template(path: Path) -> Path: 

158 """Write the v1 commented template. Fails if the path already exists.""" 

159 if path.exists(): 

160 raise ConfigAlreadyExistsError(path) 

161 path.parent.mkdir(parents=True, exist_ok=True) 

162 path.write_text(CONFIG_INIT_TEMPLATE, encoding="utf-8") 

163 return path 

164 

165 

166_SECTION_KEYS = { 

167 "source": SOURCE_KEYS, 

168 "policy": POLICY_KEYS, 

169 "runtime": RUNTIME_KEYS, 

170} 

171 

172 

173def _writable_config_path( 

174 *, 

175 cwd: Path | None = None, 

176 environ: Mapping[str, str] | None = None, 

177 user_config_path: Path | None = None, 

178) -> Path: 

179 path = resolve_init_target(cwd=cwd, environ=environ, user_config_path=user_config_path) 

180 if not path.is_file(): 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true

181 raise ConfigSchemaError( 

182 f"Config file was not found: {path}", 

183 path=path, 

184 hint="Create the file with lpip config init before set or unset.", 

185 ) 

186 return path 

187 

188 

189def _split_config_key(key: str, *, path: Path) -> tuple[str, str]: 

190 parts = key.split(".") 

191 if len(parts) != 2 or not parts[0] or not parts[1]: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true

192 raise ConfigSchemaError( 

193 f"Config key {key!r} must be a dotted v1 key such as source.path.", 

194 path=path, 

195 ) 

196 table, field = parts 

197 allowed = _SECTION_KEYS.get(table) 

198 if allowed is None: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true

199 raise UnknownTomlKeyError(table, path=path, kind="config") 

200 if field not in allowed: 

201 raise UnknownTomlKeyError(f"{table}.{field}", path=path, kind="config") 

202 return table, field 

203 

204 

205def _parse_set_value(field: str, raw: str, *, path: Path) -> object: 

206 if field == "require_pinned_revision": 206 ↛ 214line 206 didn't jump to line 214 because the condition on line 206 was always true

207 low = raw.strip().lower() 

208 if low in {"true", "false"}: 208 ↛ 210line 208 didn't jump to line 210 because the condition on line 208 was always true

209 return low == "true" 

210 raise ConfigSchemaError( 

211 "policy.require_pinned_revision must be true or false.", 

212 path=path, 

213 ) 

214 if field == "allowed_schemes": 

215 try: 

216 parsed = tomlkit.parse(f"x = {raw}")["x"] 

217 except TOMLKitError as exc: 

218 raise ConfigSchemaError( 

219 "policy.allowed_schemes must be a TOML array of strings.", 

220 path=path, 

221 hint='Example: \'["git+ssh", "git+https"]\'', 

222 ) from exc 

223 if not isinstance(parsed, list) or not all(isinstance(item, str) for item in parsed): 

224 raise ConfigSchemaError( 

225 "policy.allowed_schemes must be a TOML array of strings.", 

226 path=path, 

227 ) 

228 return parsed 

229 if field == "type": 

230 if raw not in {item.value for item in SourceType}: 

231 raise ConfigSchemaError( 

232 f"source.type must be 'file' or 'git', got {raw!r}.", 

233 path=path, 

234 ) 

235 return raw 

236 if field == "refresh": 

237 if raw not in {item.value for item in RefreshMode}: 

238 raise ConfigSchemaError( 

239 f"source.refresh must be always, auto, or never, got {raw!r}.", 

240 path=path, 

241 ) 

242 return raw 

243 return raw 

244 

245 

246def _load_tomlkit(path: Path) -> tomlkit.TOMLDocument: 

247 try: 

248 return tomlkit.parse(path.read_text(encoding="utf-8")) 

249 except TOMLKitError as exc: 

250 raise TomlParseError(path, detail=str(exc)) from exc 

251 

252 

253def _dump_tomlkit(path: Path, document: tomlkit.TOMLDocument) -> None: 

254 path.write_text(tomlkit.dumps(document), encoding="utf-8") 

255 

256 

257def _toml_table(document: tomlkit.TOMLDocument, key: str) -> Any: 

258 return document[key] 

259 

260 

261def set_config_key( 

262 key: str, 

263 value: str, 

264 *, 

265 cwd: Path | None = None, 

266 environ: Mapping[str, str] | None = None, 

267 user_config_path: Path | None = None, 

268) -> Path: 

269 """Set a dotted v1 key in the init-target config file (ADR-0014).""" 

270 path = _writable_config_path(cwd=cwd, environ=environ, user_config_path=user_config_path) 

271 table, field = _split_config_key(key, path=path) 

272 parsed = _parse_set_value(field, value, path=path) 

273 document = _load_tomlkit(path) 

274 if table not in document: 274 ↛ 275line 274 didn't jump to line 275 because the condition on line 274 was never true

275 document[table] = tomlkit.table() 

276 _toml_table(document, table)[field] = parsed 

277 _dump_tomlkit(path, document) 

278 return path 

279 

280 

281def unset_config_key( 

282 key: str, 

283 *, 

284 cwd: Path | None = None, 

285 environ: Mapping[str, str] | None = None, 

286 user_config_path: Path | None = None, 

287) -> Path: 

288 """Remove a dotted v1 key from the init-target config file (ADR-0014).""" 

289 path = _writable_config_path(cwd=cwd, environ=environ, user_config_path=user_config_path) 

290 table, field = _split_config_key(key, path=path) 

291 document = _load_tomlkit(path) 

292 section = document.get(table) 

293 if isinstance(section, dict) and field in section: 293 ↛ 295line 293 didn't jump to line 295 because the condition on line 293 was always true

294 del section[field] 

295 _dump_tomlkit(path, document) 

296 return path 

297 

298 

299def format_config_value(cfg: EffectiveConfig, key: str) -> str: 

300 """Format one dotted v1 key from effective config (``config get``).""" 

301 values: dict[str, str] = { 

302 "source.type": cfg.source.type.value, 

303 "source.path": cfg.source.path, 

304 "source.url": cfg.source.url or "", 

305 "source.revision": cfg.source.revision or "", 

306 "source.refresh": cfg.source.refresh.value if cfg.source.refresh is not None else "", 

307 "policy.require_pinned_revision": "true" if cfg.policy.require_pinned_revision else "false", 

308 "policy.allowed_schemes": "[" 

309 + ", ".join(f'"{item}"' for item in cfg.policy.allowed_schemes) 

310 + "]", 

311 "runtime.python": cfg.runtime.python, 

312 } 

313 if key not in values: 

314 raise ConfigSchemaError( 

315 f"Unknown config key {key!r}.", 

316 hint="Use a dotted v1 key such as source.path or runtime.python.", 

317 ) 

318 return values[key] 

319 

320 

321def format_effective_config(cfg: EffectiveConfig) -> str: 

322 """Human-readable effective config (``config show``).""" 

323 path = str(cfg.config_path) if cfg.config_path is not None else "" 

324 keys = ( 

325 "source.type", 

326 "source.path", 

327 "source.url", 

328 "source.revision", 

329 "source.refresh", 

330 "policy.require_pinned_revision", 

331 "policy.allowed_schemes", 

332 "runtime.python", 

333 ) 

334 lines = [f"config_path: {path}"] 

335 if len(cfg.sources) > 1: 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true

336 lines.append(format_sources_table(cfg.sources)) 

337 extra = [key for key in keys if not key.startswith("source.")] 

338 else: 

339 extra = list(keys) 

340 lines.extend(f"{key}: {format_config_value(cfg, key)}" for key in extra) 

341 return "\n".join(lines) 

342 

343 

344def _source_location(src: SourceConfig) -> str: 

345 if src.type is SourceType.GIT: 345 ↛ 346line 345 didn't jump to line 346 because the condition on line 345 was never true

346 loc = src.url or "" 

347 if src.revision: 

348 loc = f"{loc}@{src.revision}" 

349 return f"{loc} {src.path}".strip() 

350 return src.path 

351 

352 

353def format_sources_table(sources: Sequence[SourceConfig]) -> str: 

354 """Table for ``lpip source list`` (ADR-0021).""" 

355 headers = ("name", "type", "location") 

356 rows = [(src.name, src.type.value, _source_location(src)) for src in sources] 

357 widths = [len(h) for h in headers] 

358 for row in rows: 

359 widths = [max(width, len(cell)) for width, cell in zip(widths, row, strict=True)] 

360 

361 def _line(cells: tuple[str, ...]) -> str: 

362 return " ".join(cell.ljust(width) for cell, width in zip(cells, widths, strict=True)) 

363 

364 lines = [_line(headers)] 

365 lines.extend(_line(row) for row in rows) 

366 return "\n".join(lines) 

367 

368 

369def _validate_source_name(name: str) -> str: 

370 cleaned = name.strip() 

371 if not SOURCE_NAME_RE.fullmatch(cleaned): 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true

372 raise InvalidSourceNameError(name) 

373 return cleaned 

374 

375 

376def _source_table( 

377 *, 

378 name: str, 

379 source_type: SourceType, 

380 path: str, 

381 url: str | None, 

382 revision: str | None, 

383 refresh: str | None, 

384) -> Any: 

385 item = tomlkit.table() 

386 item["name"] = name 

387 item["type"] = source_type.value 

388 item["path"] = path 

389 if source_type is SourceType.GIT: 389 ↛ 390line 389 didn't jump to line 390 because the condition on line 389 was never true

390 if url: 

391 item["url"] = url 

392 if revision: 

393 item["revision"] = revision 

394 item["refresh"] = refresh or RefreshMode.AUTO.value 

395 return item 

396 

397 

398def _migrate_legacy_source(document: tomlkit.TOMLDocument) -> None: 

399 legacy = document.get("source") 

400 if not isinstance(legacy, dict): 

401 return 

402 item = tomlkit.table() 

403 item["name"] = "default" 

404 for key in ("type", "path", "url", "revision", "refresh"): 

405 if key in legacy: 

406 item[key] = legacy[key] 

407 del document["source"] 

408 array = tomlkit.aot() 

409 array.append(item) 

410 document["sources"] = array 

411 

412 

413def add_configured_source( 

414 name: str, 

415 *, 

416 source_type: SourceType, 

417 path: str, 

418 url: str | None = None, 

419 revision: str | None = None, 

420 refresh: str | None = None, 

421 cwd: Path | None = None, 

422 environ: Mapping[str, str] | None = None, 

423 user_config_path: Path | None = None, 

424) -> Path: 

425 """Append [[sources]] in the init-target config (ADR-0021).""" 

426 name = _validate_source_name(name) 

427 if source_type is SourceType.GIT and not (url or "").strip(): 427 ↛ 428line 427 didn't jump to line 428 because the condition on line 427 was never true

428 raise ConfigSchemaError( 

429 "source add --type git requires --url.", 

430 hint="Pass --url ssh://git.example/lpip-catalog.git", 

431 ) 

432 if source_type is SourceType.FILE and not path.strip(): 432 ↛ 433line 432 didn't jump to line 433 because the condition on line 432 was never true

433 raise ConfigSchemaError("source add --type file requires --path.") 

434 dest = _writable_config_path(cwd=cwd, environ=environ, user_config_path=user_config_path) 

435 document = _load_tomlkit(dest) 

436 if "source" in document and "sources" not in document: 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true

437 _migrate_legacy_source(document) 

438 existing = document.get("sources") 

439 names: list[str] = [] 

440 if existing is not None: 

441 for row in existing: 

442 row_name = str(row.get("name") or "") 

443 names.append(row_name) 

444 if name in names: 444 ↛ 445line 444 didn't jump to line 445 because the condition on line 444 was never true

445 raise SourceAlreadyExistsError(name, path=dest) 

446 if "sources" not in document: 

447 document["sources"] = tomlkit.aot() 

448 _toml_table(document, "sources").append( 

449 _source_table( 

450 name=name, 

451 source_type=source_type, 

452 path=path or "lpip-sources.toml", 

453 url=url, 

454 revision=revision, 

455 refresh=refresh, 

456 ) 

457 ) 

458 _dump_tomlkit(dest, document) 

459 return dest 

460 

461 

462def remove_configured_source( 

463 name: str, 

464 *, 

465 cwd: Path | None = None, 

466 environ: Mapping[str, str] | None = None, 

467 user_config_path: Path | None = None, 

468) -> Path: 

469 """Remove a [[sources]] row (ADR-0021).""" 

470 name = _validate_source_name(name) 

471 dest = _writable_config_path(cwd=cwd, environ=environ, user_config_path=user_config_path) 

472 document = _load_tomlkit(dest) 

473 if "source" in document and "sources" not in document: 473 ↛ 474line 473 didn't jump to line 474 because the condition on line 473 was never true

474 _migrate_legacy_source(document) 

475 existing = document.get("sources") 

476 if existing is None: 476 ↛ 477line 476 didn't jump to line 477 because the condition on line 476 was never true

477 raise SourceNotFoundError(name) 

478 kept = tomlkit.aot() 

479 found = False 

480 for row in existing: 

481 if str(row.get("name") or "") == name: 

482 found = True 

483 continue 

484 kept.append(row) 

485 if not found: 485 ↛ 486line 485 didn't jump to line 486 because the condition on line 485 was never true

486 raise SourceNotFoundError(name) 

487 if len(kept) == 0: 487 ↛ 488line 487 didn't jump to line 488 because the condition on line 487 was never true

488 del document["sources"] 

489 else: 

490 document["sources"] = kept 

491 _dump_tomlkit(dest, document) 

492 return dest 

493 

494 

495def _select_persistent_file( 

496 cwd: Path, 

497 env: Mapping[str, str], 

498 overrides: ConfigOverrides, 

499 user_path: Path, 

500) -> Path | None: 

501 explicit = overrides.config_path 

502 if explicit is None: 502 ↛ 506line 502 didn't jump to line 506 because the condition on line 502 was always true

503 cfg = env.get(_ENV_CONFIG) 

504 if cfg: 

505 explicit = Path(cfg) 

506 if explicit is not None: 

507 path = explicit.expanduser() 

508 if not path.is_file(): 508 ↛ 509line 508 didn't jump to line 509 because the condition on line 508 was never true

509 raise ConfigSchemaError( 

510 f"Config file was not found: {path}", 

511 path=path, 

512 hint="Create the file with lpip config init, or fix LPIP_CONFIG.", 

513 ) 

514 return path 

515 

516 git_root = find_git_root(cwd) 

517 if git_root is not None: 

518 workspace = workspace_config_path(git_root) 

519 if workspace.is_file(): 519 ↛ 522line 519 didn't jump to line 522 because the condition on line 519 was always true

520 return workspace 

521 

522 if user_path.is_file(): 

523 return user_path 

524 return None 

525 

526 

527def _read_config_file(path: Path) -> dict[str, Any]: 

528 try: 

529 with path.open("rb") as fh: 

530 data = tomllib.load(fh) 

531 except tomllib.TOMLDecodeError as exc: 

532 raise TomlParseError(path, detail=str(exc)) from exc 

533 

534 unknown = first_unknown_key(data, CONFIG_TOP_KEYS) 

535 if unknown is not None: 

536 raise UnknownTomlKeyError(unknown, path=path, kind="config") 

537 result: dict[str, Any] = {} 

538 if "source" in data and "sources" in data: 538 ↛ 539line 538 didn't jump to line 539 because the condition on line 538 was never true

539 raise ConfigSchemaError( 

540 "Config has both [source] and [[sources]].", 

541 path=path, 

542 hint="Use one [source] table, or only [[sources]] (lpip source add).", 

543 ) 

544 if "source" in data: 

545 result["source"] = _section(data["source"], SOURCE_KEYS, path, "source") 

546 if "sources" in data: 

547 raw_sources = data["sources"] 

548 if not isinstance(raw_sources, list): 548 ↛ 549line 548 didn't jump to line 549 because the condition on line 548 was never true

549 raise ConfigSchemaError("[[sources]] must be an array of tables.", path=path) 

550 result["sources"] = [ 

551 _section(item, SOURCE_ITEM_KEYS, path, "sources") for item in raw_sources 

552 ] 

553 if "policy" in data: 

554 result["policy"] = _section(data["policy"], POLICY_KEYS, path, "policy") 

555 if "runtime" in data: 

556 result["runtime"] = _section(data["runtime"], RUNTIME_KEYS, path, "runtime") 

557 return result 

558 

559 

560def _section(body: object, allowed: frozenset[str], path: Path, name: str) -> dict[str, Any]: 

561 if not isinstance(body, dict): 561 ↛ 562line 561 didn't jump to line 562 because the condition on line 561 was never true

562 raise ConfigSchemaError(f"[{name}] must be a table.", path=path) 

563 unknown = first_unknown_key(body, allowed) 

564 if unknown is not None: 

565 raise UnknownTomlKeyError(f"{name}.{unknown}", path=path, kind="config") 

566 return dict(body) 

567 

568 

569def _apply_env_and_cli( 

570 raw: dict[str, Any], 

571 env: Mapping[str, str], 

572 overrides: ConfigOverrides, 

573) -> dict[str, Any]: 

574 source = dict(raw.get("source") or {}) 

575 sources = [dict(item) for item in raw.get("sources") or []] 

576 policy = dict(raw.get("policy") or {}) 

577 runtime = dict(raw.get("runtime") or {}) 

578 

579 def overlay(target: dict[str, Any], key: str, value: str | None) -> None: 

580 if value is not None and value != "": 

581 target[key] = value 

582 

583 first = sources[0] if sources else source 

584 overlay(first, "type", env.get(_ENV_TYPE)) 

585 overlay(first, "path", env.get(_ENV_PATH)) 

586 overlay(first, "url", env.get(_ENV_URL)) 

587 overlay(first, "revision", env.get(_ENV_REVISION)) 

588 overlay(first, "type", overrides.source_type) 

589 overlay(first, "path", overrides.source_path) 

590 overlay(first, "url", overrides.source_url) 

591 overlay(first, "revision", overrides.source_revision) 

592 if sources: 

593 sources[0] = first 

594 return {"sources": sources, "policy": policy, "runtime": runtime} 

595 return {"source": source, "policy": policy, "runtime": runtime} 

596 

597 

598def _opt(value: object, *, path: Path | None) -> str | None: 

599 try: 

600 return optional_str(value) 

601 except TypeError as exc: 

602 raise ConfigSchemaError(str(exc), path=path) from exc 

603 

604 

605def _parse_source_config(src_raw: Mapping[str, Any], config_path: Path | None) -> SourceConfig: 

606 type_s = _opt(src_raw.get("type"), path=config_path) or SourceType.FILE.value 

607 try: 

608 source_type = SourceType(type_s) 

609 except ValueError as exc: 

610 raise ConfigSchemaError( 

611 f"source.type must be 'file' or 'git', got {type_s!r}.", 

612 path=config_path, 

613 ) from exc 

614 

615 path = _opt(src_raw.get("path"), path=config_path) or "lpip-sources.toml" 

616 url = _opt(src_raw.get("url"), path=config_path) 

617 revision = _opt(src_raw.get("revision"), path=config_path) 

618 refresh_s = _opt(src_raw.get("refresh"), path=config_path) 

619 if refresh_s is None: 

620 refresh = RefreshMode.AUTO if source_type is SourceType.GIT else None 

621 else: 

622 try: 

623 refresh = RefreshMode(refresh_s) 

624 except ValueError as exc: 

625 raise ConfigSchemaError( 

626 f"source.refresh must be always, auto, or never, got {refresh_s!r}.", 

627 path=config_path, 

628 ) from exc 

629 

630 if source_type is SourceType.GIT and not url: 

631 raise ConfigSchemaError( 

632 "source.url is required when source.type is git.", 

633 path=config_path, 

634 hint="Set source.url in config.toml or LPIP_SOURCE_URL.", 

635 ) 

636 

637 name = _opt(src_raw.get("name"), path=config_path) or "default" 

638 if not SOURCE_NAME_RE.fullmatch(name): 638 ↛ 639line 638 didn't jump to line 639 because the condition on line 638 was never true

639 raise InvalidSourceNameError(name) 

640 return SourceConfig( 

641 type=source_type, 

642 path=path, 

643 url=url, 

644 revision=revision, 

645 refresh=refresh, 

646 name=name, 

647 ) 

648 

649 

650def _build_effective(merged: dict[str, Any], config_path: Path | None) -> EffectiveConfig: 

651 pol_raw = merged.get("policy") or {} 

652 run_raw = merged.get("runtime") or {} 

653 raw_sources = merged.get("sources") 

654 if raw_sources: 

655 sources = tuple(_parse_source_config(item, config_path) for item in raw_sources) 

656 names = [item.name for item in sources] 

657 if len(names) != len(set(names)): 657 ↛ 658line 657 didn't jump to line 658 because the condition on line 657 was never true

658 raise ConfigSchemaError( 

659 "[[sources]] names must be unique.", 

660 path=config_path, 

661 ) 

662 else: 

663 sources = (_parse_source_config(merged.get("source") or {}, config_path),) 

664 

665 schemes = pol_raw.get("allowed_schemes") 

666 if schemes is None: 

667 allowed = ("git+ssh", "git+https") 

668 elif isinstance(schemes, list) and all(isinstance(item, str) for item in schemes): 668 ↛ 671line 668 didn't jump to line 671 because the condition on line 668 was always true

669 allowed = tuple(schemes) 

670 else: 

671 raise ConfigSchemaError( 

672 "policy.allowed_schemes must be an array of strings.", 

673 path=config_path, 

674 ) 

675 

676 pin = pol_raw.get("require_pinned_revision", False) 

677 if not isinstance(pin, bool): 677 ↛ 678line 677 didn't jump to line 678 because the condition on line 677 was never true

678 raise ConfigSchemaError( 

679 "policy.require_pinned_revision must be a boolean.", 

680 path=config_path, 

681 ) 

682 

683 python = _opt(run_raw.get("python"), path=config_path) or "python3" 

684 

685 return EffectiveConfig( 

686 sources=sources, 

687 policy=PolicyConfig(allowed_schemes=allowed, require_pinned_revision=pin), 

688 runtime=RuntimeConfig(python=python), 

689 config_path=config_path, 

690 )