diff --git a/interscript-py/README.md b/interscript-py/README.md index ff0a4240..adc4249a 100644 --- a/interscript-py/README.md +++ b/interscript-py/README.md @@ -28,9 +28,13 @@ License: BSD-3-Clause. ``` This package **executes the compiled JSON-IR maps** — the same artifact -the TypeScript runtime consumes. It does NOT parse `.isc` sources; the -compiler is the maps repo's CI (single source of truth, shared by every -runtime, which is what makes cross-runtime byte-identity checkable). +the TypeScript runtime consumes, which is what makes cross-runtime +byte-identity checkable. It also ships `interscript.isc`, a +recursive-descent parser for the `.isc` DSL (a port of the TypeScript +`src/isc/parser.ts`) that implements the ISC-WF-1 workflow extension: +named `ruleset`s and `use`/`override` expand at parse time, while +`languages`/`when` guards flow into the IR and resolve at run time +(`transliterate(map, text, language="kur")`). ### Map execution (now) @@ -149,6 +153,8 @@ make lint # ruff if available src/interscript/ __init__.py public API: add_load_path, map_list, transliterate, Runtime _ir_types.py typed mirror of the JSON-IR wire format (strict decode) + isc.py .isc DSL parser (port of ts src/isc/parser.ts) + WF-1 + parse-time ruleset/use/override expansion + IR lowering loader.py artifact dirs, SHA256SUMS verify, tarball extraction context.py execution context: alias + dependency resolution items.py Item -> regex/literal compilation (port of compile-item.ts) diff --git a/interscript-py/src/interscript/__init__.py b/interscript-py/src/interscript/__init__.py index 5c796dee..49218f03 100644 --- a/interscript-py/src/interscript/__init__.py +++ b/interscript-py/src/interscript/__init__.py @@ -26,10 +26,12 @@ from .errors import ( ArtifactIntegrityError, AsyncFunctionError, + IscParseError, MapFormatError, MapLogicError, MapNotFoundError, ) +from .isc import IscDocument, isc_to_compiled_map, parse_isc from .loader import ( MapLoader, default_load_paths, @@ -72,11 +74,26 @@ def map_exist(self, system_code: str) -> bool: def load_map(self, system_code: str): return self.loader.load(system_code) - def transliterate(self, system_code: str, text: str, stage: str | None = None) -> str: + def transliterate( + self, + system_code: str, + text: str, + stage: str | None = None, + language: str | None = None, + ) -> str: + """Apply a transliteration system to `text` (default stage: "main"). + + `language` is the ISC-WF-1 W3 caller selection: when it names one + of the map's declared `languages`, `when`-guarded rules for it + apply (and beat unguarded same-from rules); any other value makes + every guard inert (§5.1). + """ from .executor import execute_stage map_ = self.loader.load(system_code) - return execute_stage(map_, stage or self.default_stage, text, self.loader) + return execute_stage( + map_, stage or self.default_stage, text, self.loader, language + ) def _default_runtime() -> Runtime: @@ -109,15 +126,26 @@ def load_map(system_code: str): return _default_runtime().load_map(system_code) -def transliterate(system_code: str, text: str, stage: str | None = None) -> str: - """Apply a transliteration system to `text` (default stage: "main").""" - return _default_runtime().transliterate(system_code, text, stage) +def transliterate( + system_code: str, + text: str, + stage: str | None = None, + language: str | None = None, +) -> str: + """Apply a transliteration system to `text` (default stage: "main"). + + `language` selects one of the map's declared languages (ISC-WF-1 W3, + `--language kur` in the CLI contract); guards are inert otherwise. + """ + return _default_runtime().transliterate(system_code, text, stage, language) __all__ = [ "DEFAULT_STAGE", "ArtifactIntegrityError", "AsyncFunctionError", + "IscDocument", + "IscParseError", "MapFormatError", "MapLoader", "MapLogicError", @@ -126,10 +154,12 @@ def transliterate(system_code: str, text: str, stage: str | None = None) -> str: "add_load_path", "default_load_paths", "extract_release_tarball", + "isc_to_compiled_map", "load_manifest", "load_map", "map_exist", "map_list", + "parse_isc", "transliterate", "verify_artifact", ] diff --git a/interscript-py/src/interscript/_ir_types.py b/interscript-py/src/interscript/_ir_types.py index 36a8fed8..0d6f04c5 100644 --- a/interscript-py/src/interscript/_ir_types.py +++ b/interscript-py/src/interscript/_ir_types.py @@ -15,12 +15,14 @@ systemCode: str dependencies: [str] metadata?: {str: ...} # omitted when empty + languages?: [str] # ISC-WF-1 W3; omitted when undeclared stages: [Stage] aliases: {name: Item} functions: {name: {...}} # always {} today } Stage = {kind: "stage", name: str, rules: [Rule]} Rule = SubRule | RunRule | FuncallRule | ParallelRule | SequentialRule + SubRule carries optional `when: [str]` (ISC-WF-1 W3 guard). Item = StringItem | CaptureGroupItem | CaptureRefItem | AliasItem | AnyItem | AnyCharClassItem | GroupItem | RepeatItem | StageItem """ @@ -131,6 +133,7 @@ class SubRule: not_before: Item | None = None not_after: Item | None = None priority: int | None = None + when: tuple[str, ...] | None = None # ISC-WF-1 W3 guard languages kind: str = "sub" @@ -189,6 +192,7 @@ class CompiledMap: aliases: dict[str, Item] functions: dict[str, FunctionDef] metadata: dict[str, Any] | None = None + languages: tuple[str, ...] | None = None # ISC-WF-1 W3 declared languages def stage(self, name: str) -> Stage | None: for s in self.stages: @@ -254,6 +258,7 @@ def rule_from_json(obj: dict[str, Any]) -> Rule: not_before=item_from_json(obj["notBefore"]) if obj.get("notBefore") else None, not_after=item_from_json(obj["notAfter"]) if obj.get("notAfter") else None, priority=int(obj["priority"]) if obj.get("priority") is not None else None, + when=tuple(str(w) for w in obj["when"]) if obj.get("when") else None, ) if kind == "run": return RunRule(stage=obj["stage"], doc_name=obj.get("docName")) @@ -287,4 +292,5 @@ def map_from_json(obj: dict[str, Any]) -> CompiledMap: for name, fd in obj.get("functions", {}).items() }, metadata=dict(obj["metadata"]) if obj.get("metadata") else None, + languages=tuple(str(x) for x in obj["languages"]) if obj.get("languages") else None, ) diff --git a/interscript-py/src/interscript/context.py b/interscript-py/src/interscript/context.py index c0d3165c..aca637ef 100644 --- a/interscript-py/src/interscript/context.py +++ b/interscript-py/src/interscript/context.py @@ -21,19 +21,35 @@ def load(self, system_code: str) -> CompiledMap: ... class ExecutionContext: - """State carried through one stage execution.""" + """State carried through one stage execution. + + `language` is the ISC-WF-1 W3 caller-selected language (§5.1); it + flows across `run` boundaries and is validated against each map's + own `languages` declaration by `effective_language`. + """ def __init__( self, map_: CompiledMap, current: str, loader: MapResolver | None = None, + language: str | None = None, ) -> None: self.map = map_ self.current = current self.loader = loader + self.language = language self._alias_cache: dict[str, Item] = {} + @property + def effective_language(self) -> str | None: + """The selected language when it is declared by the active map, + else None. Per ISC-WF-1 §5.1, selecting a language the map does + not declare makes every `when` guard INERT (as if unset).""" + if self.language is None or not self.map.languages: + return None + return self.language if self.language in self.map.languages else None + def resolve_alias(self, name: str) -> Item | None: """Resolve an alias against the current map, then (one level) against the map's declared dependencies.""" @@ -56,8 +72,8 @@ def resolve_alias(self, name: str) -> Item | None: def with_map(self, map_: CompiledMap) -> ExecutionContext: """Spawn a context for a different map (run rule with docName); - same loader, fresh alias cache.""" - return ExecutionContext(map_, self.current, self.loader) + same loader and language selection, fresh alias cache.""" + return ExecutionContext(map_, self.current, self.loader, self.language) def load_dependency(self, system_code: str) -> CompiledMap: if self.loader is None: diff --git a/interscript-py/src/interscript/errors.py b/interscript-py/src/interscript/errors.py index 50559262..1ff3c1e4 100644 --- a/interscript-py/src/interscript/errors.py +++ b/interscript-py/src/interscript/errors.py @@ -18,6 +18,21 @@ class MapFormatError(ValueError): """A compiled JSON-IR map file is not a valid `interscript.maps.ir.v1` document.""" +class IscParseError(MapFormatError): + """An .isc source could not be parsed (bad token, unclosed block, + malformed rule, `use` cycle...). + + Carries the byte `position` plus 1-based `line`/`col` of the failure, + mirroring the TypeScript `IscParseError`. + """ + + def __init__(self, message: str, position: int, line: int, col: int) -> None: + self.position = position + self.line = line + self.col = col + super().__init__(f"ISC parse error at line {line}:{col}: {message}") + + class ArtifactIntegrityError(ValueError): """The compiled artifact failed its SHA256SUMS verification.""" diff --git a/interscript-py/src/interscript/executor.py b/interscript-py/src/interscript/executor.py index 05593612..2f566136 100644 --- a/interscript-py/src/interscript/executor.py +++ b/interscript-py/src/interscript/executor.py @@ -72,16 +72,17 @@ def execute_stage( stage_name: str, text: str, loader: MapResolver | None = None, + language: str | None = None, ) -> str: """Run one stage by name; returns the transformed string. A missing stage is a no-op (returns the input unchanged), matching - the TS runtime. + the TS runtime. `language` is the ISC-WF-1 W3 caller selection (§5.1). """ stage = map_.stage(stage_name) if stage is None: return text - ctx = ExecutionContext(map_, text, loader) + ctx = ExecutionContext(map_, text, loader, language) execute_stage_rules(stage, ctx) return ctx.current @@ -100,6 +101,8 @@ def execute_stage_rules(stage: Stage, ctx: ExecutionContext) -> None: def execute_rule(rule: Rule, ctx: ExecutionContext) -> None: kind = rule.kind if kind == "sub": + if not _guard_active(rule, ctx): # type: ignore[arg-type] + return _execute_sub_rule(rule, ctx) # type: ignore[arg-type] elif kind == "run": _execute_run_rule(rule, ctx) # type: ignore[arg-type] @@ -114,6 +117,40 @@ def execute_rule(rule: Rule, ctx: ExecutionContext) -> None: raise MapLogicError(f"unknown rule kind: {kind!r}") +# --------------------------------------------------------------------------- +# ISC-WF-1 W3 language guards +# --------------------------------------------------------------------------- + + +def _guard_active(rule: SubRule, ctx: ExecutionContext) -> bool: + """A `when`-guarded sub rule applies only when the caller selected a + language the map declares AND the rule's `when` list contains it + (§5.1); unguarded rules always apply. Inert guards skip the rule + entirely, in every context (bare, sequence, parallel).""" + if rule.when is None: + return True + selected = ctx.effective_language + return selected is not None and selected in rule.when + + +def _drop_shadowed_unguarded(rules: list[SubRule]) -> list[SubRule]: + """§5.2 precedence: when a guard-applicable rule and an unguarded + rule share the same from-string, the guard-applicable rule wins — + the unguarded one is dropped so no first-alternative/insertion-order + tie can resurrect the generic value. Structural equality of the + `from` items is the same-from test (byte-exact for string rows).""" + guarded_froms = { + r.from_ for r in rules if r.when is not None and r.from_ is not None + } + if not guarded_froms: + return rules + return [ + r + for r in rules + if r.when is not None or r.from_ is None or r.from_ not in guarded_froms + ] + + # --------------------------------------------------------------------------- # sub # --------------------------------------------------------------------------- @@ -227,16 +264,20 @@ def _execute_parallel(rule: ParallelRule, ctx: ExecutionContext) -> None: trailing: list[Rule] = [] for inner in rule.rules: if inner.kind == "sub" and inner.from_ is not None: # type: ignore[union-attr] - sub_rules.append(inner) # type: ignore[arg-type] + # Inert W3 guards (§5.1) leave the rule out of the pass; a + # guard-applicable rule beats an unguarded same-from rule (§5.2). + if _guard_active(inner, ctx): # type: ignore[arg-type] + sub_rules.append(inner) # type: ignore[arg-type] else: trailing.append(inner) + sub_rules = _drop_shadowed_unguarded(sub_rules) any_constrained = any(not _tree_compatible(r, ctx) for r in sub_rules) if not any_constrained: # Tree mode: every `from` expands to literals; rules whose `to` # is an inline funcall or non-literal are skipped (mirrors TS). - trie_pairs: list[tuple[str, str]] = [] + flagged: list[tuple[str, str, bool]] = [] # (from_lit, to_lit, guarded) for r in sub_rules: if r.to is not None and r.to.kind == "funcall_inline": continue @@ -247,7 +288,13 @@ def _execute_parallel(rule: ParallelRule, ctx: ExecutionContext) -> None: if from_alts is None: continue for from_lit in from_alts: - trie_pairs.append((from_lit, to_lit)) + flagged.append((from_lit, to_lit, r.when is not None)) + # §5.2 at string level: an expanded literal shared by a + # guard-applicable rule and an unguarded rule keeps the guard. + guarded_literals = {f for f, _t, g in flagged if g} + trie_pairs = [ + (f, t) for f, t, g in flagged if g or f not in guarded_literals + ] if trie_pairs: ctx.current = parallel_replace_tree( ctx.current, compile_parallel_tree(trie_pairs) @@ -268,12 +315,20 @@ def sort_key(pair: tuple[int, SubRule]): return (-length, idx) mega_rules: list[MegaregexpRule] = [] + compiled: list[tuple[SubRule, str, str, str, str, str]] = [] for _idx, r in sorted(enumerate(sub_rules), key=sort_key): src = compile_item(r.from_, ctx).re # type: ignore[arg-type] before = compile_item(r.before, ctx).re if r.before else "" after = compile_item(r.after, ctx).re if r.after else "" not_before = compile_item(r.not_before, ctx).re if r.not_before else "" not_after = compile_item(r.not_after, ctx).re if r.not_after else "" + compiled.append((r, src, before, after, not_before, not_after)) + # §5.2 at pattern level: identical compiled `from` source under a + # guard-applicable rule shadows the unguarded one. + guarded_srcs = {c[1] for c in compiled if c[0].when is not None} + for r, src, before, after, not_before, not_after in compiled: + if r.when is None and src in guarded_srcs: + continue pattern = ( (f"(?<={before})" if before else "") + (f"(? { ... }` at map top level. +- **W2 use with override** — `use [override { }]`, + expanded at parse time: a macro splice of the referenced ruleset's + items at the use position (recursive; cycles are a parse error), + with child-wins row overrides on exact from-equality (spec §4.2). +- **W3 language conditioning** — `languages ...` inside the + system block and the `when ...` guard suffix on `sub` rules. + Guards are carried into the IR (`CompiledMap.languages`, + `SubRule.when`) and resolved at RUN time by the executor (§5). + +W1/W2 therefore never reach the executor: expanded maps compile to the +same IR shapes as before, plus the two optional W3 fields. The parser +emits `_ir_types` structures directly, so the parse path and the JSON-IR +loader share one runtime representation. + + doc = parse_isc(source, "map.isc") + doc = parse_isc(source, resolve_dependency=lambda code: other_doc) + cm = isc_to_compiled_map(doc) # -> executor.execute_stage(cm, ...) + +Parser fidelity notes (mirroring the TypeScript reference): + +- Unrecognized tokens in stage/block bodies are silently dropped to end + of line (Ruby Parslet parity), except the new keywords, which parse. +- `when` joins the keyword table (§2): it is never consumed as a bare + alias reference in item position. +- `languages` is declared once per map; a second declaration is a parse + error. `use` may appear in stage bodies, ruleset bodies and inside + `parallel`/`sequence` blocks; a ruleset whose expansion contains a + pass block may not itself be used inside a pass block (a nested pass + has no meaning in the one-simultaneous-pass model). +""" + +from __future__ import annotations + +import re +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +from ._ir_types import ( + AliasItem, + AnyCharClassItem, + AnyItem, + CaptureGroupItem, + CaptureRefItem, + CompiledMap, + FuncallInline, + FuncallRule, + GroupItem, + ParallelRule, + RepeatItem, + RunRule, + SequentialRule, + Stage, + StringItem, + SubRule, +) +from .errors import IscParseError, MapFormatError + +PRIMITIVES = frozenset( + ["boundary", "line_start", "line_end", "word_boundary", "space", "non_boundary"] +) + +FUNCTIONS = frozenset( + ["upcase", "downcase", "title_case", "reverse", "strip", "swapcase"] +) + +CONSTRAINTS = frozenset(["before", "after", "not_before", "not_after"]) + +# ISC-WF-1 §2: the four new keywords join the keyword table. `use` / +# `override` / `languages` / `ruleset` are handled at their grammar +# positions; `when` can otherwise be reached from item position (right +# after a compact rule's `to`), so it is excluded from alias references. +ITEM_KEYWORDS = frozenset(["when"]) + +# Words that terminate a comma-less `ident (","? ident)*` list (the +# `languages` declaration and `when` guards). Without this, a one-line +# `languages kur stage main { }` would swallow `stage` as a language. +RESERVED_WORDS = frozenset( + [ + "metadata", "tests", "aliases", "stage", "dependency", "languages", + "ruleset", "sub", "use", "run", "parallel", "sequence", "separate", + "compose", "decompose", "when", "override", "note", "description", + "notes", "from", "to", "before", "after", "not_before", "not_after", + "as", "map", + ] +) + +DependencyResolver = Callable[[str], "IscDocument | None"] + +# JavaScript \s (the TypeScript parser's whitespace class), kept explicit +# so Python's Unicode \s cannot diverge from the reference. +_WS = frozenset("\t\n\v\f\r \u00a0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000\ufeff") +_WORD = "A-Za-z0-9_" + +_RE_WORD = re.compile(f"[{_WORD}]") +_RE_IDENT = re.compile(f"[{_WORD}-]") +_RE_NUMBER = re.compile("[0-9]") + + +# --------------------------------------------------------------------------- +# Parse-level document types (mirror interscript-ts src/isc/types.ts) +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class IscTest: + input: str + expected: str + note: str | None = None + + +@dataclass(frozen=True) +class IscDependency: + target: str + alias_name: str | None = None + + +@dataclass(frozen=True) +class IscStage: + """One stage; `items` are expanded parse-level items (no `use`).""" + + name: str + items: tuple[Any, ...] + + +@dataclass(frozen=True) +class IscDocument: + system_code: str + metadata: dict[str, Any] + tests: tuple[IscTest, ...] + aliases: tuple[tuple[str, Any], ...] + stages: tuple[IscStage, ...] + dependencies: tuple[IscDependency, ...] + languages: tuple[str, ...] | None + rulesets: dict[str, tuple[Any, ...]] + + +# --------------------------------------------------------------------------- +# Parse-level stage items +# +# Rules carry their IR form plus the raw source text of the `from` atom, +# which is the W2 override key (byte-exact string equality, §4.2). +# Blocks hold rules until isc_to_compiled_map lowers them to IR. +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class ParsedRule: + rule: SubRule + from_source: str | None = None + + +@dataclass(frozen=True) +class ParallelBlock: + rules: tuple[Any, ...] + + +@dataclass(frozen=True) +class SequenceBlock: + rules: tuple[Any, ...] + + +@dataclass(frozen=True) +class RunItem: + dependency: str | None + stage: str + + +@dataclass(frozen=True) +class RawUse: + """Unexpanded `use` directive (resolved in the post-parse pass).""" + + name: str + dependency_alias: str | None = None + override: tuple[ParsedRule, ...] | None = None + + +# --------------------------------------------------------------------------- +# Parser +# --------------------------------------------------------------------------- + + +def parse_isc( + source: str, + filename: str | None = None, + resolve_dependency: DependencyResolver | None = None, +) -> IscDocument: + """Parse one .isc source into an expanded IscDocument. + + `resolve_dependency` maps a dependency's target system code to its + parsed IscDocument; it is called lazily when a `use alias.` + expansion needs the dependency's rulesets. W1 `use` splicing and W2 + override merging happen here, at parse time. + """ + return _Parser(source, filename, resolve_dependency).parse() + + +class _Parser: + def __init__( + self, + src: str, + filename: str | None, + resolve_dependency: DependencyResolver | None, + ) -> None: + self.src = src + self.pos = 0 + self.filename = filename + self.resolve_dependency = resolve_dependency + # Raw (unexpanded) map body, filled by parse(). + self._raw_rulesets: dict[str, list[Any]] = {} + self._dependencies: list[IscDependency] = [] + self._system_code = "" + # Memo of fully expanded rulesets (this map only; dependency + # rulesets arrive already expanded in their own IscDocument). + self._expanded: dict[str, list[Any]] = {} + + # -- Document --------------------------------------------------------- + + def parse(self) -> IscDocument: + self.skip_ws() + self.expect("system") + self.skip_ws() + self._system_code = self.parse_quoted_string() + self.skip_ws() + self.expect("{") + body = self.parse_block_items() + self.skip_ws() + self.expect("}") + self.skip_ws() + + metadata: dict[str, Any] = {} + tests: list[IscTest] = [] + aliases: list[tuple[str, Any]] = [] + raw_stages: list[tuple[str, list[Any]]] = [] + languages: list[str] | None = None + + for item in body: + tag = item[0] + if tag == "metadata": + metadata.update(item[1]) + elif tag == "tests": + tests = item[1] + elif tag == "aliases": + aliases = item[1] + elif tag == "stage": + raw_stages.append((item[1], item[2])) + elif tag == "dependency": + self._dependencies.append(item[1]) + elif tag == "languages": + if languages is not None: + self.error("languages declared more than once") + languages = item[1] + + # W1/W2 expansion pass (post-parse: forward references allowed). + for name in self._raw_rulesets: + self._expand_ruleset(name) + stages = [ + IscStage(name=name, items=tuple(self._expand_items(items, (), False))) + for name, items in raw_stages + ] + + return IscDocument( + system_code=self._system_code, + metadata=metadata, + tests=tuple(tests), + aliases=tuple(aliases), + stages=tuple(stages), + dependencies=tuple(self._dependencies), + languages=tuple(languages) if languages is not None else None, + rulesets={name: tuple(self._expanded[name]) for name in self._raw_rulesets}, + ) + + # -- Block items ------------------------------------------------------ + + def parse_block_items(self) -> list[Any]: + items: list[Any] = [] + while True: + self.skip_ws() + if self.peek() == "}" or self.pos >= len(self.src): + break + + keyword = self.peek_word() + if keyword == "metadata": + items.append(self.parse_metadata()) + elif keyword == "tests": + items.append(self.parse_tests()) + elif keyword == "aliases": + items.append(self.parse_aliases()) + elif keyword == "stage": + items.append(self.parse_stage()) + elif keyword == "dependency": + items.append(self.parse_dependency()) + elif keyword == "languages": + items.append(self.parse_languages()) + elif keyword == "ruleset": + items.append(self.parse_ruleset()) + else: + self.error(f"Unexpected keyword: {keyword}") + return items + + # -- Metadata --------------------------------------------------------- + + def parse_metadata(self) -> Any: + self.expect("metadata") + self.skip_ws() + self.expect("{") + metadata: dict[str, Any] = {} + + while True: + self.skip_ws() + if self.peek() == "}": + break + + if self.peek_word() == "description": + self.consume("description") + self.skip_inline_ws() + if self.peek() == "{": + self.expect("{") + metadata["description"] = self.parse_raw_text() + self.expect("}") + elif self.peek() == '"': + metadata["description"] = self.parse_quoted_string() + else: + metadata["description"] = self.parse_to_newline().strip() + continue + + if self.peek_word() == "notes": + self.consume("notes") + self.skip_inline_ws() + if self.peek() == "{": + self.expect("{") + # Note-list form (`note "..."`) vs raw free text. + saved = self.pos + self.skip_ws() + is_note_list = self.peek_word() == "note" + self.pos = saved + + if is_note_list: + notes: list[str] = [] + while True: + self.skip_ws() + if self.peek() == "}": + break + self.expect("note") + self.skip_ws() + notes.append(self.parse_quoted_string()) + self.expect("}") + metadata["notes"] = notes + else: + metadata["notes"] = self.parse_raw_text() + self.expect("}") + continue + + key = self.parse_identifier() + self.skip_inline_ws() + if self.peek() in ("\n", "}"): + metadata[key] = "" + continue + if self.peek() == "{": + self.expect("{") + metadata[key] = self.parse_raw_text() + self.expect("}") + elif self.peek() == '"': + metadata[key] = self.parse_quoted_string() + else: + metadata[key] = self.parse_to_newline().strip() + self.expect("}") + return ("metadata", metadata) + + # -- Tests ------------------------------------------------------------ + + def parse_tests(self) -> Any: + self.expect("tests") + self.skip_ws() + self.expect("{") + tests: list[IscTest] = [] + + while True: + self.skip_ws() + if self.peek() == "}": + break + if self.peek() == "#": + self.skip_line() + continue + + src = self.parse_string_literal() + self.skip_ws() + self.expect("->") + self.skip_ws() + expected = self.parse_string_literal() + self.skip_ws() + + if self.peek_word() == "note": + self.consume("note") + self.skip_ws() + tests.append(IscTest(input=src, expected=expected, note=self.parse_string_literal())) + else: + tests.append(IscTest(input=src, expected=expected)) + self.expect("}") + return ("tests", tests) + + # -- Aliases ---------------------------------------------------------- + + def parse_aliases(self) -> Any: + self.expect("aliases") + self.skip_ws() + self.expect("{") + aliases: list[tuple[str, Any]] = [] + + while True: + self.skip_ws() + if self.peek() == "}": + break + if self.peek() == "#": + self.skip_line() + continue + + name = self.parse_identifier() + self.skip_ws() + self.expect("=") + self.skip_ws() + aliases.append((name, self.parse_item(False))) + self.expect("}") + return ("aliases", aliases) + + # -- Languages (ISC-WF-1 W3) ------------------------------------------ + + def parse_languages(self) -> Any: + self.expect("languages") + self.skip_ws() + names = self.parse_ident_list() + return ("languages", names) + + def parse_ident_list(self) -> list[str]: + """`ident (","? ident)*` — the shared list form of `languages` and + `when`. A comma always continues the list; a comma-less word + continues it only when it is not a reserved keyword, so a + one-line declaration cannot swallow the next construct.""" + names = [self.parse_identifier()] + while True: + self.skip_inline_ws() + if self.peek() == ",": + self.pos += 1 + self.skip_ws() + names.append(self.parse_identifier()) + continue + if _RE_IDENT.match(self.peek()) and self.peek_word() not in RESERVED_WORDS: + names.append(self.parse_identifier()) + continue + break + return names + + # -- Stage and ruleset bodies ----------------------------------------- + + def parse_stage(self) -> Any: + self.expect("stage") + self.skip_ws() + name = self.parse_identifier() + self.skip_ws() + self.expect("{") + body = self.parse_stage_items() + self.expect("}") + return ("stage", name, body) + + def parse_ruleset(self) -> Any: + """ISC-WF-1 W1: `ruleset { }`.""" + self.expect("ruleset") + self.skip_ws() + name = self.parse_identifier() + self.skip_ws() + self.expect("{") + body = self.parse_stage_items() + self.expect("}") + if name in self._raw_rulesets: + self.error(f"ruleset declared more than once: {name}") + self._raw_rulesets[name] = body + return ("ruleset", name) + + def parse_stage_items(self) -> list[Any]: + """Shared body grammar of `stage` and `ruleset` blocks.""" + body: list[Any] = [] + while True: + self.skip_ws() + if self.peek() == "}": + break + if self.peek() == "#": + self.skip_line() + continue + + keyword = self.peek_word() + if keyword in ("parallel", "sequence"): + self.consume(keyword) + self.skip_ws() + self.expect("{") + rules: list[Any] = [] + while True: + self.skip_ws() + if self.peek() == "}": + break + if self.peek() == "#": + self.skip_line() + continue + if self.peek_word() == "sub": + rules.append(self.parse_rule()) + elif self.peek_word() == "use": + rules.append(self.parse_use()) + else: + # Ruby Parslet silently drops unrecognized tokens + # in rule blocks. + self.skip_line() + self.expect("}") + body.append(ParallelBlock(tuple(rules)) if keyword == "parallel" else SequenceBlock(tuple(rules))) + elif keyword == "run": + body.append(self.parse_run()) + elif keyword == "separate": + self.consume("separate") + self.skip_ws() + if self.peek_word() == "separator": + self.consume("separator") + self.skip_ws() + atom = self.parse_item_atom(False) + if isinstance(atom, StringItem): + body.append(FuncallRule(name="separate", kwargs={"separator": atom.value})) + else: + body.append(FuncallRule(name="separate")) + else: + body.append(FuncallRule(name="separate")) + elif keyword in ("compose", "decompose"): + self.consume(keyword) + body.append(FuncallRule(name=keyword)) + elif keyword == "rababa": + self.consume("rababa") + self.skip_inline_ws() + kwargs: dict[str, str] = {} + while self.peek() not in ("}", "", "\n"): + arg_name = self.parse_identifier() + self.skip_inline_ws() + self.expect(":") + self.skip_inline_ws() + kwargs[arg_name] = self.parse_string_literal() + self.skip_inline_ws() + if self.peek() == ",": + self.pos += 1 + self.skip_inline_ws() + body.append(FuncallRule(name="rababa", kwargs=kwargs)) + elif keyword in FUNCTIONS: + self.consume(keyword) + body.append(FuncallRule(name=keyword)) + elif keyword == "sub": + body.append(self.parse_rule()) + elif keyword == "use": + body.append(self.parse_use()) + else: + # Ruby Parslet silently drops unrecognized tokens in stage + # body. Skip to end of line for parity. + self.skip_line() + return body + + def parse_run(self) -> RunItem: + self.consume("run") + self.skip_ws() + dep: str | None = None + if self.peek_word() == "map": + self.consume("map") + self.expect(".") + dep = self.parse_identifier() + self.expect(".") + self.expect("stage") + self.expect(".") + stage_name = self.parse_identifier() + elif self.peek_word() == "stage": + self.consume("stage") + self.expect(".") + stage_name = self.parse_identifier() + else: + self.error("Expected map. or stage. after run") + return RunItem(dependency=dep, stage=stage_name) + + # -- Use (ISC-WF-1 W1/W2) ---------------------------------------------- + + def parse_use(self) -> RawUse: + self.expect("use") + self.skip_ws() + first = self.parse_identifier() + dependency_alias: str | None = None + self.skip_ws() + if self.peek() == ".": + self.pos += 1 + self.skip_ws() + dependency_alias = first + name = self.parse_identifier() + else: + name = first + + override: list[ParsedRule] | None = None + self.skip_ws() + if self.peek_word() == "override": + self.consume("override") + self.skip_ws() + self.expect("{") + override = [] + while True: + self.skip_ws() + if self.peek() == "}": + break + if self.peek() == "#": + self.skip_line() + continue + if self.peek_word() == "sub": + override.append(self.parse_rule()) + else: + self.skip_line() + self.expect("}") + return RawUse( + name=name, + dependency_alias=dependency_alias, + override=tuple(override) if override is not None else None, + ) + + # -- Rules -------------------------------------------------------------- + + def parse_rule(self) -> ParsedRule: + self.expect("sub") + self.skip_ws() + from_source: str | None = None + when: tuple[str, ...] | None = None + + # Block form: sub { from ... to ... } + if self.peek() == "{": + self.expect("{") + from_: Any = StringItem(value="") + to: Any = StringItem(value="") + constraints: list[tuple[str, Any]] = [] + + while True: + self.skip_ws() + if self.peek() == "}": + break + + word = self.peek_word() + if word == "from": + self.consume("from") + self.skip_ws() + from_ = self.parse_item(False) + elif word == "to": + self.consume("to") + self.skip_ws() + to = self.parse_item(True) + elif word in CONSTRAINTS: + self.consume(word) + self.skip_ws() + constraints.append((word, self.parse_item(False))) + else: + self.error(f"Unexpected keyword in sub block: {word}") + self.expect("}") + self.skip_ws() + if self.peek_word() == "when": + self.consume("when") + self.skip_ws() + when = tuple(self.parse_ident_list()) + return ParsedRule( + rule=_make_sub_rule(from_, to, constraints, when), + from_source=None, + ) + + # Compact form: sub [when ...] + start = self.pos + from_ = self.parse_item_atom(False) + from_source = self.src[start : self.pos].strip() + self.skip_ws() + to = self.parse_item_atom(True) + constraints: list[tuple[str, Any]] = [] + + while True: + self.skip_ws() + word = self.peek_word() + if word not in CONSTRAINTS: + break + self.consume(word) + self.skip_ws() + constraints.append((word, self.parse_item_atom(False))) + + self.skip_ws() + if self.peek_word() == "when": + self.consume("when") + self.skip_ws() + when = tuple(self.parse_ident_list()) + + return ParsedRule( + rule=_make_sub_rule(from_, to, constraints, when), + from_source=from_source, + ) + + # -- Items --------------------------------------------------------------- + + def parse_item(self, to_position: bool) -> Any: + first, first_kind = self.parse_item_atom_raw() + parts = [first] + kinds = [first_kind] + + while True: + self.skip_inline_ws() + if self.peek() == "+": + self.pos += 1 + self.skip_inline_ws() + atom, kind = self.parse_item_atom_raw() + parts.append(atom) + kinds.append(kind) + elif self._is_item_atom_start() and not self._is_keyword(): + atom, kind = self.parse_item_atom_raw() + parts.append(atom) + kinds.append(kind) + else: + break + + if len(parts) == 1: + return _function_to_inline(parts[0], kinds[0]) if to_position else parts[0] + return GroupItem(items=tuple(parts)) + + def parse_item_atom(self, to_position: bool) -> Any: + atom, kind = self.parse_item_atom_raw() + return _function_to_inline(atom, kind) if to_position else atom + + def parse_item_atom_raw(self) -> tuple[Any, str]: + """One item atom -> (IR item, parse kind). Function atoms keep + their kind so the `to` position can lower them to FuncallInline + (mirrors converter.ts); elsewhere they fall back to alias lookup + so a clear error surfaces at runtime.""" + c = self.peek() + + if c in ('"', "'"): + return StringItem(value=self.parse_string_literal()), "string" + + word = self.peek_word() + if not word: + self.error("Expected item atom") + + if word in ITEM_KEYWORDS: + self.error(f"Unexpected keyword in item position: {word}") + + if word == "none": + self.consume("none") + # converter.ts lowers `none` to the empty string. + return StringItem(value=""), "string" + if word == "any_character": + self.consume("any_character") + return AliasItem(name="any_character"), "function" + + if word == "any": + self.consume("any") + self.expect("(") + self.skip_ws() + inner = self.peek() + # any("...".."...") or any('...'..'...') — range from strings + if inner in ('"', "'"): + lo = self.parse_string_literal() + self.skip_ws() + if self.src.startswith("..", self.pos): + self.pos += 2 + self.skip_ws() + hi = self.parse_string_literal() + self.skip_ws() + self.expect(")") + return AnyCharClassItem(range=(lo, hi)), "item" + self.expect(")") + # any("abc") — split into per-char string items + return ( + AnyItem(of=tuple(StringItem(value=ch) for ch in lo)), + "item", + ) + if inner == "[": + self.expect("[") + items: list[Any] = [] + while True: + self.skip_ws() + if self.peek() == "]": + break + items.append(self.parse_item(False)) + self.skip_ws() + if self.peek() == ",": + self.pos += 1 + self.expect("]") + self.skip_ws() + self.expect(")") + return AnyItem(of=tuple(items)), "item" + # any(item), any(item + item), any(alias_name), any(range..) + item = self.parse_item(False) + self.skip_ws() + self.expect(")") + return AnyItem(of=(item,)), "item" + + if word == "capture": + self.consume("capture") + self.expect("(") + self.skip_ws() + inner = self.parse_item(False) + self.skip_ws() + self.expect(")") + return CaptureGroupItem(data=inner), "item" + + if word == "maybe": + self.consume("maybe") + self.expect("(") + self.skip_ws() + inner = self.parse_item(False) + self.skip_ws() + self.expect(")") + return RepeatItem(item=inner, min=0, max=1), "item" + + if word == "some": + self.consume("some") + self.expect("(") + self.skip_ws() + inner = self.parse_item(False) + self.skip_ws() + self.expect(")") + return RepeatItem(item=inner, min=1, max=None), "item" + + if word == "ref": + self.consume("ref") + self.expect("(") + self.skip_ws() + num = self.parse_number() + self.skip_ws() + self.expect(")") + return CaptureRefItem(id=num), "item" + + if word in PRIMITIVES: + self.consume(word) + return AliasItem(name=word), "item" + + if word in FUNCTIONS: + self.consume(word) + return AliasItem(name=word), "function" + + # Alias reference + self.consume(word) + return AliasItem(name=word), "item" + + # -- Dependency -------------------------------------------------------- + + def parse_dependency(self) -> Any: + self.expect("dependency") + self.skip_ws() + target = self.parse_quoted_string() + self.skip_inline_ws() + alias_name: str | None = None + if self.peek_word() == "as": + self.consume("as") + self.skip_ws() + alias_name = self.parse_identifier() + return ("dependency", IscDependency(target=target, alias_name=alias_name)) + + # -- String primitives --------------------------------------------------- + + def parse_string_literal(self) -> str: + c = self.peek() + if c == '"': + return self.parse_quoted_string() + if c == "'": + return self.parse_single_quoted_string() + self.error(f'Expected string but got "{self.src[self.pos : self.pos + 20]}"') + + def parse_quoted_string(self) -> str: + self.expect('"') + result: list[str] = [] + while self.pos < len(self.src): + c = self.src[self.pos] + self.pos += 1 + if c == '"': + return "".join(result) + if c == "\\": + if self.pos >= len(self.src): + break + nxt = self.src[self.pos] + self.pos += 1 + if nxt == "n": + result.append("\n") + elif nxt == "t": + result.append("\t") + elif nxt == "r": + result.append("\r") + elif nxt == '"': + result.append('"') + elif nxt == "\\": + result.append("\\") + elif nxt == "u": + hex_ = self.src[self.pos : self.pos + 4] + self.pos += 4 + try: + result.append(chr(int(hex_, 16))) + except ValueError: + result.append("\0") + elif nxt == "{": + result.append("{") + elif nxt == "}": + result.append("}") + else: + result.append(nxt) + else: + result.append(c) + self.error("Unterminated string") + + def parse_single_quoted_string(self) -> str: + self.expect("'") + start = self.pos + while self.pos < len(self.src) and self.src[self.pos] != "'": + self.pos += 1 + result = self.src[start : self.pos] + self.expect("'") + return result + + def parse_raw_text(self) -> str: + result: list[str] = [] + depth = 0 + while self.pos < len(self.src): + c = self.src[self.pos] + if c == "\\" and self.pos + 1 < len(self.src): + nxt = self.src[self.pos + 1] + if nxt in "{}": + result.append(nxt) + self.pos += 2 + continue + if c == "}" and depth == 0: + break + if c == "{": + depth += 1 + elif c == "}": + depth -= 1 + result.append(c) + self.pos += 1 + return "".join(result).strip() + + def parse_identifier(self) -> str: + self.skip_ws() + start = self.pos + while self.pos < len(self.src) and _RE_IDENT.match(self.src[self.pos]): + self.pos += 1 + if self.pos == start: + self.error("Expected identifier") + return self.src[start : self.pos] + + def parse_number(self) -> int: + start = self.pos + while self.pos < len(self.src) and _RE_NUMBER.match(self.src[self.pos]): + self.pos += 1 + return int(self.src[start : self.pos] or "0", 10) + + def parse_to_newline(self) -> str: + start = self.pos + while self.pos < len(self.src) and self.src[self.pos] != "\n": + self.pos += 1 + return self.src[start : self.pos] + + # -- W1/W2 expansion ----------------------------------------------------- + + def _expand_ruleset(self, name: str, chain: tuple[tuple[str, str], ...] = ()) -> list[Any]: + entry = (self._system_code, name) + if entry in chain: + self.error(f"use cycle: {' -> '.join(f'{m}/{r}' for m, r in (*chain, entry))}") + cached = self._expanded.get(name) + if cached is not None: + return cached + raw = self._raw_rulesets.get(name) + if raw is None: + self.error(f"unknown ruleset: {name}") + expanded = self._expand_items(raw, chain + (entry,), False) + self._expanded[name] = expanded + return expanded + + def _expand_items( + self, items: list[Any], chain: tuple[tuple[str, str], ...], in_block: bool + ) -> list[Any]: + out: list[Any] = [] + for it in items: + if isinstance(it, RawUse): + out.extend(self._expand_use(it, chain, in_block)) + continue + if in_block and isinstance(it, (ParallelBlock, SequenceBlock)): + self.error( + "a ruleset containing a parallel/sequence block cannot be " + "used inside a parallel/sequence block" + ) + if isinstance(it, (ParallelBlock, SequenceBlock)): + it = type(it)(rules=tuple(self._expand_items(list(it.rules), chain, True))) + out.append(it) + return out + + def _expand_use( + self, use: RawUse, chain: tuple[tuple[str, str], ...], in_block: bool + ) -> list[Any]: + if use.dependency_alias is not None: + expanded = self._expand_via_dependency(use, chain) + else: + expanded = self._expand_ruleset(use.name, chain) + + if in_block and any(isinstance(x, (ParallelBlock, SequenceBlock)) for x in expanded): + self.error( + "a ruleset containing a parallel/sequence block cannot be used " + "inside a parallel/sequence block" + ) + + if use.override is not None: + expanded = self._apply_override(expanded, list(use.override)) + return list(expanded) + + def _expand_via_dependency( + self, use: RawUse, chain: tuple[tuple[str, str], ...] + ) -> list[Any]: + target = next( + ( + d.target + for d in self._dependencies + if d.alias_name == use.dependency_alias + ), + None, + ) + if target is None: + self.error(f"unknown dependency alias: {use.dependency_alias}") + if self.resolve_dependency is None: + self.error( + f"cannot resolve dependency {target!r} for use " + f"{use.dependency_alias}.{use.name} (no resolver provided)" + ) + dep_doc = self.resolve_dependency(target) + if dep_doc is None: + self.error(f"cannot resolve dependency: {target}") + entry = (dep_doc.system_code, use.name) + if entry in chain: + self.error(f"use cycle: {' -> '.join(f'{m}/{r}' for m, r in (*chain, entry))}") + expanded = dep_doc.rulesets.get(use.name) + if expanded is None: + self.error(f"unknown ruleset in dependency {target!r}: {use.name}") + return list(expanded) + + @staticmethod + def _from_key(rule: ParsedRule) -> Any: + """The W2 override key: the byte-exact source text of the `from` + atom (spec §4.2); falls back to the IR item's structural equality + for rules built programmatically (no source span).""" + return rule.from_source if rule.from_source is not None else rule.rule.from_ + + def _apply_override( + self, items: list[Any], overrides: list[ParsedRule] + ) -> list[Any]: + """ISC-WF-1 §4.2 merge: R_kept (in R's order) followed by O (in + O's order); ALL rules sharing an override's from-string are + replaced. For rows inside a parallel/sequence block the matched + overrides are appended at that block's end so they stay in the + pass where the replaced row lived (§4.3); overrides matching + nothing are simply added at the end (§4.2 rule 4).""" + okeys = [self._from_key(o) for o in overrides] + placed: set[int] = set() + out: list[Any] = [] + for it in items: + if isinstance(it, ParsedRule): + if self._from_key(it) in okeys: + continue + out.append(it) + elif isinstance(it, (ParallelBlock, SequenceBlock)): + kept: list[Any] = [] + dropped_keys: set[Any] = set() + for r in it.rules: + if isinstance(r, ParsedRule) and self._from_key(r) in okeys: + dropped_keys.add(self._from_key(r)) + else: + kept.append(r) + block_overrides: list[ParsedRule] = [] + for i, o in enumerate(overrides): + if i not in placed and self._from_key(o) in dropped_keys: + block_overrides.append(o) + placed.add(i) + out.append(type(it)(rules=tuple(kept + block_overrides))) + else: + out.append(it) + for i, o in enumerate(overrides): + if i not in placed: + out.append(o) + return out + + # -- Scanner helpers (mirror the TypeScript parser) ---------------------- + + def char_at(self, i: int) -> str: + return self.src[i] if i < len(self.src) else "" + + def peek(self) -> str: + return self.char_at(self.pos) + + def peek_word(self) -> str: + start = self.pos + while self.pos < len(self.src) and _RE_WORD.match(self.src[self.pos]): + self.pos += 1 + word = self.src[start : self.pos] + self.pos = start + return word + + def expect(self, s: str) -> None: + self.skip_ws() + if not self.src.startswith(s, self.pos): + self.error(f'Expected "{s}" but got "{self.src[self.pos : self.pos + 20]}"') + self.pos += len(s) + + def consume(self, word: str) -> None: + if not self.src.startswith(word, self.pos): + self.error(f'Expected "{word}"') + self.pos += len(word) + + def skip_ws(self) -> None: + while self.pos < len(self.src): + c = self.src[self.pos] + if c in _WS: + self.pos += 1 + continue + if c == "#": + self.skip_line() + continue + break + + def skip_inline_ws(self) -> None: + while self.pos < len(self.src) and self.src[self.pos] in " \t": + self.pos += 1 + + def skip_line(self) -> None: + while self.pos < len(self.src) and self.src[self.pos] != "\n": + self.pos += 1 + if self.pos < len(self.src): + self.pos += 1 + + def _is_item_atom_start(self) -> bool: + c = self.peek() + if not c: + return False + return c in ('"', "'") or bool(_RE_WORD.match(c)) + + def _is_keyword(self) -> bool: + word = self.peek_word() + return word in ("to", "from", "before", "after", "not_before", "not_after", "note") + + def error(self, msg: str) -> None: + line = self.src[: self.pos].count("\n") + 1 + col = self.pos - self.src.rfind("\n", 0, self.pos) + full_msg = f"{self.filename}: {msg}" if self.filename else msg + raise IscParseError(full_msg, self.pos, line, col) + + +# --------------------------------------------------------------------------- +# Rule assembly + converter (mirror interscript-ts src/isc/converter.ts) +# --------------------------------------------------------------------------- + + +def _function_to_inline(item: Any, kind: str) -> Any: + """Lower a function atom in `to` position to a FuncallInline.""" + if kind == "function" and isinstance(item, AliasItem): + return FuncallInline(name=item.name) + return item + + +def _make_sub_rule( + from_: Any, + to: Any, + constraints: list[tuple[str, Any]], + when: tuple[str, ...] | None, +) -> SubRule: + picked = {kind: item for kind, item in constraints} + return SubRule( + from_=from_, + to=to, + before=picked.get("before"), + after=picked.get("after"), + not_before=picked.get("not_before"), + not_after=picked.get("not_after"), + when=when, + ) + + +def isc_to_compiled_map(doc: IscDocument) -> CompiledMap: + """Lower an expanded IscDocument to the compiled JSON-IR `CompiledMap` + consumed by the executor (port of `iscToCompiledMap`).""" + aliases = {name: item for name, item in doc.aliases} + dep_aliases = {d.alias_name: d.target for d in doc.dependencies if d.alias_name} + stages = tuple( + Stage( + name=s.name, + rules=tuple(_lower_item(it, dep_aliases) for it in s.items), + ) + for s in doc.stages + ) + return CompiledMap( + schema_version=1, + system_code=doc.system_code, + dependencies=tuple(d.target for d in doc.dependencies), + stages=stages, + aliases=aliases, + functions={}, + metadata=dict(doc.metadata) if doc.metadata else None, + languages=doc.languages, + ) + + +def _lower_item(item: Any, dep_aliases: dict[str, str]) -> Any: + if isinstance(item, ParallelBlock): + return ParallelRule(rules=tuple(_lower_rule(r) for r in item.rules)) + if isinstance(item, SequenceBlock): + return SequentialRule(rules=tuple(_lower_rule(r) for r in item.rules)) + if isinstance(item, ParsedRule): + return _lower_rule(item) + if isinstance(item, RunItem): + doc_name = dep_aliases.get(item.dependency, item.dependency) if item.dependency else None + return RunRule(stage=item.stage, doc_name=doc_name) + if isinstance(item, FuncallRule): + return item + raise MapFormatError(f"unrecognized parse-level stage item: {type(item).__name__}") + + +def _lower_rule(rule: ParsedRule) -> SubRule: + return rule.rule + + +# Re-export for callers that want the error without reaching into errors. +__all__ = [ + "IscDependency", + "IscDocument", + "IscParseError", + "IscStage", + "IscTest", + "ParallelBlock", + "ParsedRule", + "RawUse", + "RunItem", + "SequenceBlock", + "isc_to_compiled_map", + "parse_isc", +] diff --git a/interscript-py/tests/test_isc_parser.py b/interscript-py/tests/test_isc_parser.py new file mode 100644 index 00000000..087c4f2f --- /dev/null +++ b/interscript-py/tests/test_isc_parser.py @@ -0,0 +1,423 @@ +# Copyright 2026 Interscript Project. BSD-3-Clause (see LICENSE). +"""ISC parser unit tests — one class per construct (ISC-WF-1 §6.1). + +Covers the base grammar surface the py runtime needs (metadata, tests, +aliases, dependency, stage items, rules, items) plus the four WF-1 +constructs: ruleset (W1), use with override (W2), languages/when (W3 +parse side — executor semantics live in test_isc_wf1.py). +""" + +from __future__ import annotations + +import pytest + +from interscript._ir_types import ( + FuncallInline, + FuncallRule, + StringItem, + SubRule, +) +from interscript.errors import IscParseError +from interscript.executor import execute_stage +from interscript.isc import ( + IscDocument, + ParallelBlock, + ParsedRule, + SequenceBlock, + isc_to_compiled_map, + parse_isc, +) + + +def parse(src: str, **kw) -> IscDocument: + return parse_isc(src, "test.isc", **kw) + + +def bare_rules(doc: IscDocument, stage: str = "main") -> list[SubRule]: + """Expanded stage items flattened to (kind, from, to, when) tuples.""" + out = [] + st = next(s for s in doc.stages if s.name == stage) + for item in st.items: + if isinstance(item, ParallelBlock): + out.extend(item.rules) + elif isinstance(item, ParsedRule): + out.append(item) + return out + + +def row(r: ParsedRule) -> tuple[str | None, str | None, tuple[str, ...] | None]: + f = r.rule.from_ + t = r.rule.to + return ( + f.value if isinstance(f, StringItem) else None, + t.value if isinstance(t, StringItem) else None, + r.rule.when, + ) + + +class TestBaseGrammar: + def test_minimal_document(self): + doc = parse('system "ISO:rus-Cyrl:Latn:9-1995" { stage main { } }') + assert doc.system_code == "ISO:rus-Cyrl:Latn:9-1995" + assert doc.languages is None + assert [s.name for s in doc.stages] == ["main"] + + def test_metadata_description_notes_generic(self): + src = ''' +system "s" { + metadata { + authority_id iso + description { + A long description + over two lines + } + notes { + note "first note" + note "second note" + } + name Romanization Table + } +} +''' + doc = parse(src) + md = doc.metadata + assert md["authority_id"] == "iso" + assert "over two lines" in md["description"] + assert md["notes"] == ["first note", "second note"] + assert md["name"] == "Romanization Table" + + def test_notes_raw_text_form(self): + doc = parse('system "s" { metadata { notes { free text here } } }') + assert doc.metadata["notes"] == "free text here" + + def test_tests_block(self): + src = ''' +system "s" { + tests { + # a comment + "а" -> "a" + "б" -> "b" note "why" + } +} +''' + doc = parse(src) + assert [(t.input, t.expected) for t in doc.tests] == [("а", "a"), ("б", "b")] + assert doc.tests[1].note == "why" + + def test_aliases_block(self): + doc = parse('system "s" { aliases { vowels = any("ae") } stage main { } }') + assert "vowels" in dict(doc.aliases) + + def test_dependency_with_alias(self): + doc = parse( + 'system "s" { dependency "iso-9-base" as base9 stage main { } }' + ) + assert doc.dependencies[0].target == "iso-9-base" + assert doc.dependencies[0].alias_name == "base9" + + def test_dependency_without_alias(self): + doc = parse('system "s" { dependency "posix" stage main { } }') + assert doc.dependencies[0].target == "posix" + assert doc.dependencies[0].alias_name is None + + def test_compact_rule_constraints(self): + doc = parse('system "s" { stage main { sub "a" "b" before "c" not_after "d" } }') + rule = bare_rules(doc)[0].rule + assert rule.before == StringItem(value="c") + assert rule.not_after == StringItem(value="d") + + def test_block_rule_form(self): + doc = parse( + 'system "s" { stage main { sub { from "a" to upcase before "x" } } }' + ) + rule = bare_rules(doc)[0].rule + assert rule.from_ == StringItem(value="a") + assert rule.to == FuncallInline(name="upcase") + assert rule.before == StringItem(value="x") + + def test_parallel_and_sequence_blocks(self): + doc = parse( + 'system "s" { stage main { parallel { sub "a" "b" } sequence { sub "c" "d" } } }' + ) + st = doc.stages[0] + assert isinstance(st.items[0], ParallelBlock) + assert isinstance(st.items[1], SequenceBlock) + + def test_run_and_funcall_items(self): + src = ''' +system "s" { + dependency "dep-map" as d + stage main { + run stage.other + run map.d.stage.main + upcase + separate separator "-" + compose + } +} +''' + doc = parse(src) + items = doc.stages[0].items + assert items[0].stage == "other" and items[0].dependency is None + assert items[1].dependency == "d" and items[1].stage == "main" + assert isinstance(items[2], FuncallRule) and items[2].name == "upcase" + assert items[3].kwargs == {"separator": "-"} + assert items[4].name == "compose" + + def test_unterminated_string_error(self): + with pytest.raises(IscParseError, match="Unterminated string"): + parse('system "s { stage main { } }') + + def test_parse_error_has_position(self): + with pytest.raises(IscParseError) as exc: + parse('system "s" {\n bogus {\n}\n') + assert exc.value.line == 2 + + +class TestW3ParseSide: + def test_languages_declaration(self): + doc = parse( + 'system "s" {\n languages rus bel ukr\n stage main { }\n}' + ) + assert doc.languages == ("rus", "bel", "ukr") + + def test_languages_comma_form(self): + doc = parse('system "s" { languages kur, ckv stage main { } }') + assert doc.languages == ("kur", "ckv") + + def test_languages_duplicate_declaration_error(self): + with pytest.raises(IscParseError, match="more than once"): + parse('system "s" { languages a languages b stage main { } }') + + def test_languages_declaration_on_one_line(self): + doc = parse('system "s" { languages kur stage main { } }') + assert doc.languages == ("kur",) + assert [s.name for s in doc.stages] == ["main"] + + def test_when_suffix_compact_form(self): + doc = parse( + 'system "s" { languages kur ckv stage main { sub "қ" "k" when kur, ckv } }' + ) + assert row(bare_rules(doc)[0]) == ("қ", "k", ("kur", "ckv")) + + def test_when_suffix_space_separated(self): + doc = parse('system "s" { stage main { sub "қ" "k" when kur ckv } }') + assert row(bare_rules(doc)[0])[2] == ("kur", "ckv") + + def test_when_list_does_not_swallow_next_rule(self): + doc = parse( + 'system "s" { stage main { sub "x" "y" when kur sub "z" "w" } }' + ) + rows = [row(r) for r in bare_rules(doc)] + assert rows == [("x", "y", ("kur",)), ("z", "w", None)] + + def test_when_suffix_after_block_form(self): + doc = parse( + 'system "s" { stage main { sub { from "x" to "y" } when kur } }' + ) + assert row(bare_rules(doc)[0]) == ("x", "y", ("kur",)) + + def test_when_is_not_parsed_as_to(self): + # §2: `when` joins the keyword table — `sub "x" when kur` must not + # read `when` as the replacement string. + doc = parse('system "s" { stage main { sub "x" "y" when kur } }') + assert row(bare_rules(doc)[0]) == ("x", "y", ("kur",)) + + def test_when_never_an_alias_reference(self): + with pytest.raises(IscParseError, match="keyword"): + parse('system "s" { stage main { sub when "y" } }') + + def test_keywords_inside_strings_untouched(self): + doc = parse( + 'system "s" { stage main { sub "use when override languages ruleset" "y" } }' + ) + assert row(bare_rules(doc)[0])[0] == "use when override languages ruleset" + + def test_prose_with_keywords_in_metadata(self): + doc = parse( + 'system "s" { metadata { note "ruleset when languages use override appear in prose" } }' + ) + assert "ruleset" in doc.metadata["note"] + + +class TestW1Rulesets: + def test_ruleset_is_not_a_stage(self): + doc = parse( + 'system "s" { ruleset tbl { sub "a" "b" } stage main { use tbl } }' + ) + assert [s.name for s in doc.stages] == ["main"] + assert "tbl" in doc.rulesets + + def test_library_map_rulesets_only(self): + doc = parse('system "shared" { ruleset tbl { sub "a" "b" } }') + assert doc.stages == () + assert list(doc.rulesets) == ["tbl"] + + def test_use_expands_in_place_in_document_order(self): + doc = parse( + 'system "s" { ruleset tbl { sub "a" "A" } ' + 'stage main { sub "1" "!" use tbl sub "2" "@" } }' + ) + rows = [row(r)[:2] for r in bare_rules(doc)] + assert rows == [("1", "!"), ("a", "A"), ("2", "@")] + + def test_use_inside_parallel_participates_in_pass(self): + doc = parse( + 'system "s" { ruleset tbl { sub "bc" "Z" } ' + 'stage main { parallel { use tbl sub "abc" "Q" } } }' + ) + cm = isc_to_compiled_map(doc) + # one simultaneous pass: abc wins over bc + assert execute_stage(cm, "main", "xabc") == "xQ" + + def test_ruleset_with_parallel_block_is_one_pass_at_position(self): + doc = parse( + 'system "s" { ruleset tbl { parallel { sub "ab" "X" } } ' + 'stage main { sub "k" "ab" use tbl } }' + ) + cm = isc_to_compiled_map(doc) + # the bare rule runs first (document order); the expanded block + # is one pass at its position + assert execute_stage(cm, "main", "k") == "X" + + def test_recursive_expansion(self): + doc = parse( + 'system "s" { ruleset outer { use inner sub "z" "Z" } ' + 'ruleset inner { sub "a" "A" } ' + 'stage main { use outer } }' + ) + rows = [row(r)[:2] for r in bare_rules(doc)] + assert rows == [("a", "A"), ("z", "Z")] + + def test_forward_reference(self): + doc = parse( + 'system "s" { stage main { use later } ruleset later { sub "a" "A" } }' + ) + assert row(bare_rules(doc)[0])[:2] == ("a", "A") + + def test_cycle_error(self): + with pytest.raises(IscParseError, match="use cycle"): + parse( + 'system "s" { ruleset a { use b } ruleset b { use a } ' + 'stage main { use a } }' + ) + + def test_self_cycle_error(self): + with pytest.raises(IscParseError, match="use cycle"): + parse('system "s" { ruleset a { use a } stage main { use a } }') + + def test_unknown_ruleset_error(self): + with pytest.raises(IscParseError, match="unknown ruleset"): + parse('system "s" { stage main { use nope } }') + + def test_block_in_block_error(self): + with pytest.raises(IscParseError, match="parallel/sequence block"): + parse( + 'system "s" { ruleset tbl { parallel { sub "a" "b" } } ' + 'stage main { parallel { use tbl } } }' + ) + + def test_cross_dependency_use(self): + base = parse('system "base" { ruleset tbl { sub "a" "A" sub "b" "B" } }') + doc = parse( + 'system "kid" { dependency "base" as b9 ' + 'stage main { use b9.tbl } }', + resolve_dependency=lambda code: base if code == "base" else None, + ) + rows = [row(r)[:2] for r in bare_rules(doc)] + assert rows == [("a", "A"), ("b", "B")] + + def test_unknown_dependency_alias(self): + with pytest.raises(IscParseError, match="unknown dependency alias"): + parse('system "s" { stage main { use ghost.tbl } }') + + def test_unresolvable_dependency(self): + with pytest.raises(IscParseError, match="cannot resolve dependency"): + parse( + 'system "s" { dependency "missing" as m stage main { use m.tbl } }', + resolve_dependency=lambda code: None, + ) + + def test_no_resolver(self): + with pytest.raises(IscParseError, match="no resolver"): + parse('system "s" { dependency "d" as m stage main { use m.tbl } }') + + def test_missing_ruleset_in_dependency(self): + base = parse('system "base" { ruleset other { sub "a" "A" } }') + with pytest.raises(IscParseError, match="unknown ruleset in dependency"): + parse( + 'system "s" { dependency "base" as b stage main { use b.tbl } }', + resolve_dependency=lambda code: base, + ) + + +class TestW2Override: + def test_child_wins_on_exact_from_equality(self): + doc = parse( + 'system "s" { ruleset tbl { sub "ะ" "e" sub "ั" "a" } ' + 'stage main { use tbl override { sub "ะ" "a2011" } } }' + ) + rows = [row(r)[:2] for r in bare_rules(doc)] + # R_kept in R's order, then O in O's order (§4.2 rule 2) + assert rows == [("ั", "a"), ("ะ", "a2011")] + + def test_all_same_from_rows_replaced(self): + doc = parse( + 'system "s" { ruleset tbl { sub "k" "k1" before "a" sub "k" "k2" before "b" sub "m" "M" } ' + 'stage main { use tbl override { sub "k" "K" } } }' + ) + rows = [row(r)[:2] for r in bare_rules(doc)] + assert rows == [("m", "M"), ("k", "K")] + + def test_unmatched_overrides_added_at_end(self): + doc = parse( + 'system "s" { ruleset tbl { sub "a" "A" } ' + 'stage main { use tbl override { sub "x" "X" sub "a" "A2" } } }' + ) + rows = [row(r)[:2] for r in bare_rules(doc)] + # R_kept (empty — "a" was replaced) then O in O's order (§4.2 rules 2+4) + assert rows == [("x", "X"), ("a", "A2")] + + def test_override_is_a_complete_row(self): + # §4.2 rule 3: the override replaces the imported row's to value + # AND its constraints. + doc = parse( + 'system "s" { ruleset tbl { sub "k" "K1" before "v" } ' + 'stage main { use tbl override { sub "k" "K2" } } }' + ) + rule = bare_rules(doc)[0].rule + assert rule.before is None + assert rule.to == StringItem(value="K2") + + def test_override_row_carries_when(self): + doc = parse( + 'system "s" { languages kur ruleset tbl { sub "қ" "ķ" } ' + 'stage main { use tbl override { sub "қ" "k" when kur } } }' + ) + assert row(bare_rules(doc)[0]) == ("қ", "k", ("kur",)) + + def test_override_inside_block_appends_at_block_end(self): + doc = parse( + 'system "s" { ruleset tbl { parallel { sub "a" "A1" sub "b" "B" } } ' + 'stage main { use tbl override { sub "a" "A2" } } }' + ) + block = doc.stages[0].items[0] + rows = [row(r)[:2] for r in block.rules] + # the replaced row's override joins the same pass, at its end + assert rows == [("b", "B"), ("a", "A2")] + + def test_override_key_is_byte_exact(self): + doc = parse( + 'system "s" { ruleset tbl { sub "а" "a" sub "áb" "z" } ' + 'stage main { use tbl override { sub "ab" "AB" } } }' + ) + rows = [row(r)[:2] for r in bare_rules(doc)] + # "ab" matches nothing ("а" is Cyrillic; "áb" differs) — added at end + assert rows == [("а", "a"), ("áb", "z"), ("ab", "AB")] + + def test_duplicate_ruleset_declaration_error(self): + with pytest.raises(IscParseError, match="more than once"): + parse( + 'system "s" { ruleset t { sub "a" "b" } ruleset t { sub "c" "d" } ' + 'stage main { use t } }' + ) diff --git a/interscript-py/tests/test_isc_wf1.py b/interscript-py/tests/test_isc_wf1.py new file mode 100644 index 00000000..a41344a1 --- /dev/null +++ b/interscript-py/tests/test_isc_wf1.py @@ -0,0 +1,472 @@ +# Copyright 2026 Interscript Project. BSD-3-Clause (see LICENSE). +"""ISC-WF-1 end-to-end tests. + +Three layers: + +- **Executor guard semantics** (W3, spec §5) on synthetic CompiledMaps: + inert guards, guard-applicable beats unguarded same-from (parallel + tree + megaregexp), sequence skips inert guards, run carries the + selection into the dependency map's own declaration. +- **Additive JSON-IR decode**: `languages` on the document and `when` + arrays on rules (the only two IR extensions; the TypeScript converter + expands W1/W2 at build time). +- **Synthetic .isc fixtures through the full path**: base ruleset + + dependent override across a declared dependency, and a + language-conditioned table exercised with and without a selection, + including the public `transliterate(..., language=...)` API. +""" + +from __future__ import annotations + +import json + +import pytest + +from interscript._ir_types import ( + CompiledMap, + ParallelRule, + RunRule, + SequentialRule, + Stage, + StringItem, + SubRule, +) +from interscript.executor import execute_stage +from interscript.isc import isc_to_compiled_map, parse_isc + + +def make_map( + stages, + aliases=None, + dependencies=(), + system_code="test-map", + languages=None, +): + return CompiledMap( + schema_version=1, + system_code=system_code, + dependencies=tuple(dependencies), + stages=tuple(stages), + aliases=dict(aliases or {}), + functions={}, + languages=tuple(languages) if languages else None, + ) + + +def sub(src, dst, when=None, **kw): + return SubRule( + from_=StringItem(value=src) if isinstance(src, str) else src, + to=StringItem(value=dst) if isinstance(dst, str) else dst, + when=tuple(when) if when else None, + **kw, + ) + + +def stage(name, rules): + return Stage(name=name, rules=tuple(rules)) + + +class TestGuardFiltering: + def test_no_selection_all_guards_inert(self): + rules = [ + sub("қ", "ķ"), + sub("қ", "K", when=("kur",)), + sub("ҫ", "ş"), + sub("ҫ", "S", when=("chv",)), + ] + m = make_map([stage("main", [ParallelRule(rules=tuple(rules))])], languages=("kur", "chv")) + assert execute_stage(m, "main", "қ ҫ") == "ķ ş" + + def test_selected_guard_applies(self): + rules = [ + sub("қ", "ķ"), + sub("қ", "K", when=("kur",)), + sub("ҫ", "ş"), + sub("ҫ", "S", when=("chv",)), + ] + m = make_map([stage("main", [ParallelRule(rules=tuple(rules))])], languages=("kur", "chv")) + assert execute_stage(m, "main", "қ ҫ", language="kur") == "K ş" + assert execute_stage(m, "main", "қ ҫ", language="chv") == "ķ S" + + def test_undeclared_selection_inert(self): + # §5.1: a selection outside the map's `languages` declaration + # makes every guard inert (identical to no selection). + rules = [sub("қ", "ķ"), sub("қ", "K", when=("kur",))] + m = make_map([stage("main", [ParallelRule(rules=tuple(rules))])], languages=("kur",)) + assert execute_stage(m, "main", "қ", language="chv") == "ķ" + + def test_no_declaration_selection_inert(self): + rules = [sub("қ", "ķ"), sub("қ", "K", when=("kur",))] + m = make_map([stage("main", [ParallelRule(rules=tuple(rules))])]) + assert execute_stage(m, "main", "қ", language="kur") == "ķ" + + def test_guard_beats_unguarded_same_from_tree_mode(self): + # §5.2 precedence: guard-applicable wins over the unguarded + # same-from row without restating it, and regardless of order. + for order in ("generic_first", "generic_last"): + generic_first = order == "generic_first" + rules = [ + sub("қ", "ķ"), + sub("қ", "K", when=("kur",)), + ] if generic_first else [ + sub("қ", "K", when=("kur",)), + sub("қ", "ķ"), + ] + m = make_map([stage("main", [ParallelRule(rules=tuple(rules))])], languages=("kur",)) + assert execute_stage(m, "main", "ққ", language="kur") == "KK" + + def test_guard_beats_unguarded_same_from_megaregexp(self): + rules = [ + sub("қ", "ķ", before=StringItem(value="#")), + sub("қ", "K", when=("kur",)), + ] + m = make_map([stage("main", [ParallelRule(rules=tuple(rules))])], languages=("kur",)) + # the constrained generic row loses to the guard even on "#" context + assert execute_stage(m, "main", "#қ", language="kur") == "#K" + + def test_guard_literal_expansion_shadows_shared_literals(self): + # any("а","б") expands to literals а,б: the guard shadows BOTH + # shared literals but leaves unshared strings untouched. + from interscript._ir_types import AnyItem + + rules = [ + sub(AnyItem(of=(StringItem(value="а"), StringItem(value="б"))), "X"), + sub("а", "A", when=("kur",)), + ] + m = make_map([stage("main", [ParallelRule(rules=tuple(rules))])], languages=("kur",)) + assert execute_stage(m, "main", "абв", language="kur") == "AXв" + + def test_sequence_inert_guard_skipped(self): + rules = [ + sub("a", "A"), + sub("x", "X", when=("kur",)), + sub("b", "B", when=("kur",)), + ] + m = make_map([stage("main", [SequentialRule(rules=tuple(rules))])], languages=("kur",)) + assert execute_stage(m, "main", "axb") == "Axb" + + def test_sequence_guard_applies_in_order(self): + rules = [ + sub("1", "a"), + sub("a", "B", when=("kur",)), + ] + m = make_map([stage("main", [SequentialRule(rules=tuple(rules))])], languages=("kur",)) + # sequences apply their rules in order; the guard row runs on the + # output of the previous one + assert execute_stage(m, "main", "1", language="kur") == "B" + assert execute_stage(m, "main", "1") == "a" + + def test_bare_guarded_rule(self): + # bare stage rules apply in sequence order; an inert guard is skipped + m = make_map( + [ + stage("main", [sub("a", "A"), sub("b", "B", when=("kur",))]), + ], + languages=("kur",), + ) + assert execute_stage(m, "main", "ab") == "Ab" + assert execute_stage(m, "main", "ab", language="kur") == "AB" + + def test_selection_carries_into_run_dependency(self): + dep = make_map( + [ + stage( + "main", + [ParallelRule(rules=(sub("x", "X"), sub("x", "Y", when=("kur",))))], + ) + ], + system_code="dep-map", + languages=("kur",), + ) + loader = type("L", (), {"load": staticmethod(lambda code: dep)})() + m = make_map( + [ + stage( + "main", + [RunRule(stage="main", doc_name="dep-map")], + ) + ], + languages=("kur",), + ) + # the selection flows through the run rule; the dependency's own + # declared languages validate it, and the guard beats the + # unguarded same-from row inside the dependency's pass + assert execute_stage(m, "main", "x", loader, "kur") == "Y" + assert execute_stage(m, "main", "x", loader) == "X" + + def test_inert_vs_selected_identical_row_shapes(self): + # §6.5: inertness is all-or-nothing — a non-matching selection + # never partially applies. + rules = [ + sub("s", "1", when=("ckv",)), + sub("t", "2", when=("kur",)), + ] + m = make_map([stage("main", [ParallelRule(rules=tuple(rules))])], languages=("kur", "ckv")) + assert execute_stage(m, "main", "st") == "st" + assert execute_stage(m, "main", "st", language="kur") == "s2" + assert execute_stage(m, "main", "st", language="ckv") == "1t" + + +class TestIrDecode: + def test_decode_languages_and_when(self): + doc = { + "schemaVersion": 1, + "systemCode": "ISO:rus-Cyrl:Latn:9-1995", + "dependencies": [], + "languages": ["kur", "ckv"], + "stages": [ + { + "name": "main", + "rules": [ + { + "kind": "parallel", + "rules": [ + { + "kind": "sub", + "from": {"kind": "string", "value": "қ"}, + "to": {"kind": "string", "value": "k"}, + "when": ["kur"], + }, + { + "kind": "sub", + "from": {"kind": "string", "value": "ҫ"}, + "to": {"kind": "string", "value": "s"}, + }, + ], + } + ], + } + ], + "aliases": {}, + "functions": {}, + } + m = CompiledMap.__new__(CompiledMap) + from interscript._ir_types import map_from_json + + m = map_from_json(doc) + assert m.languages == ("kur", "ckv") + par = m.stages[0].rules[0] + assert par.rules[0].when == ("kur",) + assert par.rules[1].when is None + + def test_decode_without_new_fields_unchanged(self): + doc = { + "schemaVersion": 1, + "systemCode": "x", + "dependencies": [], + "stages": [ + { + "name": "main", + "rules": [ + { + "kind": "sub", + "from": {"kind": "string", "value": "a"}, + "to": {"kind": "string", "value": "b"}, + } + ], + } + ], + "aliases": {}, + "functions": {}, + } + from interscript._ir_types import map_from_json + + m = map_from_json(doc) + assert m.languages is None + assert m.stages[0].rules[0].when is None + + def test_artifact_corpus_has_no_new_fields(self): + # Additivity today: no bundled corpus map carries the new fields. + import os + + maps_dir = os.path.join(os.path.dirname(__file__), "data", "maps-ir", "maps") + for name in sorted(os.listdir(maps_dir))[:50]: + with open(os.path.join(maps_dir, name), encoding="utf-8") as fh: + doc = json.load(fh) + assert "languages" not in doc, name + for st in doc["stages"]: + for r in st["rules"]: + assert "when" not in r, name + + +# --------------------------------------------------------------------------- +# Synthetic .isc fixtures through the full path +# --------------------------------------------------------------------------- + +BASE_ISC = """ +system "ALALC:xxx-shared" { + ruleset leading_vowels { + sub "ะ" "e" + sub "ั" "a" + sub "ก" "k" + } +} +""" + +DEPENDENT_ISC = """ +system "ALALC:xxx-Thai:Latn:2011" { + dependency "ALALC:xxx-shared" as shared + + tests { + "ะกั" -> "a2011ka" + } + + stage main { + use shared.leading_vowels override { + sub "ะ" "a2011" + } + parallel { + sub "ง" "ng" + } + } +} +""" + + +class TestFixtureOverrideAcrossDependency: + @pytest.fixture() + def maps(self): + base = parse_isc(BASE_ISC, "shared.isc") + docs = {"ALALC:xxx-shared": base} + dependent = parse_isc( + DEPENDENT_ISC, + "dependent.isc", + resolve_dependency=docs.get, + ) + return base, dependent + + def test_base_ruleset_expands_alone(self, maps): + base, _ = maps + # a ruleset-only map is a library: no stages, no self-execution + assert base.stages == () + lib = parse_isc( + 'system "L" { ruleset l { sub "ะ" "e" sub "ั" "a" sub "ก" "k" } ' + 'stage main { use l } }' + ) + cm = isc_to_compiled_map(lib) + from interscript.executor import execute_stage + + assert execute_stage(cm, "main", "ะกั") == "eka" + + def test_dependent_override_via_dependency(self, maps): + _, dependent = maps + assert dependent.dependencies[0].target == "ALALC:xxx-shared" + cm = isc_to_compiled_map(dependent) + from interscript.executor import execute_stage + + # child-wins: the 2011 override replaces the shared row's value + assert execute_stage(cm, "main", "ะกั") == "a2011ka" + + def test_embedded_tests_replay(self, maps): + _, dependent = maps + cm = isc_to_compiled_map(dependent) + from interscript.executor import execute_stage + + for t in dependent.tests: + assert execute_stage(cm, "main", t.input) == t.expected + + def test_dependency_ruleset_used_without_override(self, maps): + base, _ = maps + src = """ +system "ALALC:xxx-plain" { + dependency "ALALC:xxx-shared" as shared + stage main { + use shared.leading_vowels + } +} +""" + doc = parse_isc(src, resolve_dependency={"ALALC:xxx-shared": base}.get) + cm = isc_to_compiled_map(doc) + from interscript.executor import execute_stage + + assert execute_stage(cm, "main", "ะกั") == "eka" + + +LANGUAGE_ISC = """ +system "ISO:xxx-9-1995" { + languages kur ckv + + tests { + "қҫ" -> "ķş" + } + + stage main { + parallel { + sub "қ" "ķ" + sub "қ" "k" when kur + sub "ҫ" "ş" + sub "ҫ" "s" when ckv + } + } +} +""" + + +class TestFixtureLanguageConditioned: + @pytest.fixture() + def compiled(self): + doc = parse_isc(LANGUAGE_ISC, "iso9.isc") + return doc, isc_to_compiled_map(doc) + + def test_map_declares_languages(self, compiled): + doc, cm = compiled + assert doc.languages == ("kur", "ckv") + assert cm.languages == ("kur", "ckv") + + def test_no_selection_generic_rows(self, compiled): + _, cm = compiled + assert execute_stage(cm, "main", "қҫ") == "ķş" + + def test_kur_selection(self, compiled): + _, cm = compiled + assert execute_stage(cm, "main", "қҫ", language="kur") == "kş" + + def test_ckv_selection(self, compiled): + _, cm = compiled + assert execute_stage(cm, "main", "қҫ", language="ckv") == "ķs" + + def test_undeclared_selection_generic_rows(self, compiled): + _, cm = compiled + assert execute_stage(cm, "main", "қҫ", language="bel") == "ķş" + + def test_public_api_language_parameter(self, tmp_path, compiled): + _, cm = compiled + import interscript + + maps_dir = tmp_path / "maps" + maps_dir.mkdir() + payload = { + "schemaVersion": 1, + "systemCode": cm.system_code, + "dependencies": [], + "languages": list(cm.languages), + "stages": [ + { + "name": "main", + "rules": [ + { + "kind": "parallel", + "rules": [ + { + "kind": "sub", + "from": {"kind": "string", "value": "қ"}, + "to": {"kind": "string", "value": "ķ"}, + }, + { + "kind": "sub", + "from": {"kind": "string", "value": "қ"}, + "to": {"kind": "string", "value": "k"}, + "when": ["kur"], + }, + ], + } + ], + } + ], + "aliases": {}, + "functions": {}, + } + (maps_dir / f"{cm.system_code}.json").write_text(json.dumps(payload), encoding="utf-8") + rt = interscript.Runtime([tmp_path]) + assert rt.transliterate(cm.system_code, "қ") == "ķ" + assert rt.transliterate(cm.system_code, "қ", language="kur") == "k"