190 lines
9.5 KiB
Python
190 lines
9.5 KiB
Python
"""Load the permanent source-resolution contract."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from common import ROOT, ProbeError, safe_relative
|
|
import source_audit
|
|
|
|
|
|
_KEYS = {'schema', 'adaptations', 'source_paths', 'library_paths'}
|
|
_SCHEMA = 'kam-source-resolution/1'
|
|
_SOURCE_SUFFIXES = {'.pas', '.pp', '.inc', '.inl'}
|
|
_UNIT_DECLARATION = re.compile(r'^\s*unit\s+([\w.]+)\s*;', re.I | re.M)
|
|
|
|
|
|
def _unit_name(path: Path) -> str | None:
|
|
match = _UNIT_DECLARATION.search(path.read_text(encoding='utf-8-sig', errors='replace'))
|
|
return match[1] if match else None
|
|
|
|
|
|
def _inside(path: Path, root: Path) -> bool:
|
|
return path == root or path.is_relative_to(root)
|
|
|
|
|
|
def validate_source_contract(
|
|
contract: dict,
|
|
vendor_root: Path,
|
|
vendor_receipt: dict,
|
|
vendor_inventory: dict,
|
|
) -> dict:
|
|
"""Validate sparse overrides and generic source paths before compilation."""
|
|
sparse_root = ROOT / 'port-source/kam_remake'
|
|
if sparse_root.exists() and sparse_root.is_symlink():
|
|
raise ProbeError(f'Sparse source root is linked: {sparse_root}')
|
|
try:
|
|
sparse_resolved = sparse_root.resolve()
|
|
vendor_resolved = vendor_root.resolve()
|
|
except (OSError, ValueError, RuntimeError) as error:
|
|
raise ProbeError(f'Cannot resolve source roots: {error}') from error
|
|
|
|
receipt_paths = {entry.get('path') for entry in vendor_receipt.get('files', [])
|
|
if isinstance(entry, dict) and isinstance(entry.get('path'), str)}
|
|
targets = contract.get('adaptation_targets', ())
|
|
registered = set()
|
|
adaptations = []
|
|
for target in targets:
|
|
if not isinstance(target, str) or target.casefold() in registered:
|
|
raise ProbeError(f'Duplicate or invalid adaptation target: {target!r}')
|
|
registered.add(target.casefold())
|
|
try:
|
|
target = safe_relative(target)
|
|
except ProbeError:
|
|
raise
|
|
if Path(target).suffix.casefold() not in _SOURCE_SUFFIXES:
|
|
raise ProbeError(f'Unsupported adaptation target: {target}')
|
|
project_file, vendor_file = sparse_root / target, vendor_root / target
|
|
if target not in receipt_paths:
|
|
raise ProbeError(f'Adaptation is not an exact vendor receipt path: {target}')
|
|
if not project_file.is_file() or project_file.is_symlink():
|
|
raise ProbeError(f'Registered adaptation missing or linked: {project_file}')
|
|
for parent in (project_file.parent, *project_file.parents):
|
|
if parent == sparse_root.parent:
|
|
break
|
|
if parent.exists() and parent.is_symlink():
|
|
raise ProbeError(f'Linked component in adaptation path: {parent}')
|
|
try:
|
|
if not _inside(project_file.resolve(), sparse_resolved):
|
|
raise ProbeError(f'Adaptation escapes sparse root: {project_file}')
|
|
except (OSError, ValueError, RuntimeError) as error:
|
|
raise ProbeError(f'Cannot resolve adaptation: {project_file}') from error
|
|
if not vendor_file.is_file():
|
|
raise ProbeError(f'Vendor adaptation missing: {vendor_file}')
|
|
unit_name = None
|
|
if project_file.suffix.casefold() in {'.pas', '.pp'}:
|
|
unit_name, vendor_unit = _unit_name(project_file), _unit_name(vendor_file)
|
|
if not unit_name or not vendor_unit or unit_name.casefold() != vendor_unit.casefold():
|
|
raise ProbeError(f'Adaptation unit mismatch: {target}')
|
|
if len(vendor_inventory.get('units', {}).get(unit_name.casefold(), [])) > 1:
|
|
raise ProbeError(f'Ambiguous vendor unit adaptation: {unit_name}')
|
|
adaptations.append({'target': target, 'project_file': project_file,
|
|
'vendor_file': vendor_file, 'unit_name': unit_name})
|
|
|
|
if sparse_root.exists():
|
|
for path in sparse_root.rglob('*'):
|
|
if path.suffix.casefold() not in _SOURCE_SUFFIXES:
|
|
continue
|
|
if path.is_symlink() or not path.is_file():
|
|
raise ProbeError(f'Linked or non-regular sparse source: {path}')
|
|
if path.relative_to(sparse_root).as_posix() not in {item['target'] for item in adaptations}:
|
|
raise ProbeError(f'Unregistered sparse source: {path}')
|
|
|
|
vendor_units = vendor_inventory.get('units', {})
|
|
source_paths = tuple(contract.get('source_paths', ()))
|
|
for source_path in source_paths:
|
|
if not source_path.is_dir():
|
|
raise ProbeError(f'Generic source path is not a directory: {source_path}')
|
|
resolved = source_path.resolve()
|
|
if _inside(resolved, vendor_resolved) or _inside(resolved, sparse_resolved):
|
|
raise ProbeError(f'Generic source path is protected: {source_path}')
|
|
found = source_audit.inventory(source_path)
|
|
if set(found['units']) & set(vendor_units):
|
|
raise ProbeError(f'Generic source path shadows vendor unit: {source_path}')
|
|
for path in source_path.rglob('*'):
|
|
if path.is_file() and path.suffix.casefold() in {'.pas', '.pp'} and path.stem.casefold() in vendor_units:
|
|
raise ProbeError(f'Generic source filename shadows vendor unit: {path}')
|
|
|
|
library_paths = tuple(contract.get('library_paths', ()))
|
|
if any(not path.is_dir() for path in library_paths):
|
|
raise ProbeError('Library path is not a directory')
|
|
return {'adaptations': tuple(adaptations), 'vendor_unit_keys': frozenset(vendor_units),
|
|
'source_paths': source_paths, 'library_paths': library_paths}
|
|
|
|
|
|
def build_source_resolution(
|
|
validated: dict,
|
|
vendor_root: Path,
|
|
application_source: Path,
|
|
) -> dict:
|
|
"""Project validated source inputs into deterministic compiler precedence."""
|
|
try:
|
|
vendor_root, application_source = vendor_root.resolve(), application_source.resolve()
|
|
if not vendor_root.is_dir() or not application_source.is_dir():
|
|
raise ProbeError('Vendor root and application source must exist as directories')
|
|
|
|
vendor_unit_keys = validated['vendor_unit_keys']
|
|
if set(source_audit.inventory(application_source)['units']) & vendor_unit_keys:
|
|
raise ProbeError('Application source cannot replace vendor units; put replacements under '
|
|
'port-source/kam_remake/<target> and register them in adaptations')
|
|
for path in application_source.rglob('*'):
|
|
if path.is_file() and path.suffix.casefold() in {'.pas', '.pp'} and path.stem.casefold() in vendor_unit_keys:
|
|
raise ProbeError('Application source cannot replace vendor units; put replacements under '
|
|
'port-source/kam_remake/<target> and register them in adaptations')
|
|
|
|
adaptation_dirs = tuple(dict.fromkeys(
|
|
record['project_file'].parent.resolve()
|
|
for record in sorted(validated['adaptations'], key=lambda record: (record['target'].casefold(), record['target']))
|
|
))
|
|
generic_source_paths = tuple(path.resolve() for path in validated['source_paths'])
|
|
vendor_fallback_paths = tuple(path.resolve() for path in source_audit.search_paths(vendor_root))
|
|
compiler_source_paths = tuple(dict.fromkeys((application_source, *adaptation_dirs,
|
|
*generic_source_paths, *vendor_fallback_paths)))
|
|
library_paths = tuple(path.resolve() for path in validated['library_paths'])
|
|
except ProbeError:
|
|
raise
|
|
except (AttributeError, KeyError, OSError, RuntimeError, TypeError, ValueError) as error:
|
|
raise ProbeError(f'Cannot construct compiler source order: {error}') from error
|
|
return {'application_source': application_source, 'adaptation_dirs': adaptation_dirs,
|
|
'generic_source_paths': generic_source_paths, 'vendor_fallback_paths': vendor_fallback_paths,
|
|
'compiler_source_paths': compiler_source_paths, 'library_paths': library_paths}
|
|
|
|
|
|
def load_source_contract(config_path: Path | None = None) -> dict:
|
|
"""Return validated, immutable source-resolution inputs."""
|
|
path = config_path or ROOT / 'config/extra-paths.json'
|
|
try:
|
|
value = json.loads(path.read_text(encoding='utf-8'))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as error:
|
|
raise ProbeError(f'Cannot read source-resolution contract: {error}') from error
|
|
if not isinstance(value, dict) or set(value) != _KEYS:
|
|
raise ProbeError('Invalid source-resolution contract fields')
|
|
if value['schema'] != _SCHEMA:
|
|
raise ProbeError('Invalid source-resolution schema')
|
|
for key in _KEYS - {'schema'}:
|
|
if not isinstance(value[key], list) or not all(isinstance(item, str) for item in value[key]):
|
|
raise ProbeError(f'{key} must be an array of strings')
|
|
|
|
try:
|
|
adaptations = tuple(safe_relative(target) for target in value['adaptations'])
|
|
except ProbeError:
|
|
raise
|
|
|
|
def paths(key: str) -> tuple[Path, ...]:
|
|
result = []
|
|
for item in value[key]:
|
|
if not item:
|
|
raise ProbeError(f'{key} cannot contain an empty path')
|
|
try:
|
|
candidate = Path(item)
|
|
resolved = (candidate if candidate.is_absolute() else ROOT / candidate).resolve()
|
|
except (OSError, ValueError, RuntimeError) as error:
|
|
raise ProbeError(f'Invalid {key} path: {item!r}') from error
|
|
if resolved not in result:
|
|
result.append(resolved)
|
|
return tuple(result)
|
|
|
|
return {'schema': _SCHEMA, 'adaptation_targets': adaptations,
|
|
'source_paths': paths('source_paths'), 'library_paths': paths('library_paths')}
|