diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index 01668c1..79560aa 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "$schema": "https://anthropic.com/claude-code/plugin.schema.json", "name": "plc-programming", - "version": "0.2.1", + "version": "0.2.2", "description": "PLC programming across IEC 61131-3 (ST, LD, FBD, SFC, IL), PLCopen XML / IEC 61131-10, Siemens TIA (SCL, SimaticML, SIMATIC SD), Schneider (Machine Expert, Control Expert), CODESYS, TwinCAT, ABB Freelance, Rockwell L5X/L5K, IEC 61499, AutomationML, DEXPI, PackML/ISA-88 and OPC UA: design, write, convert, review and debug programs, with a stdlib-only structural validator.", "author": { "name": "SmailG", diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 517b174..7e1abd8 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -14,7 +14,7 @@ privately, as described in [SECURITY.md](SECURITY.md). Pull requests use the tem | `scripts/plc_validate.py` | Validator CLI and hook entry point | | `scripts/codesys_project.py` | Offline reader for CODESYS 3.5 / Machine Expert `.project` files (`list`, `st`, `dump-st`, `graph`, `info`, `diff`, `xref`); `scripts/codesysproj/` holds `binary.py` (varint, string table), `project.py` (objects, kinds, text), `graph.py` (LD/FBD records) and `graphview.py` (their text and JSON), `export.py` (`dump-st`), `buildinfo.py` (`info`), `diff.py` (`diff`), `xref.py` and `stscan.py` (`xref`) | | `scripts/plccheck/` | One module per format (`st.py`, `plcopen.py`, `simaticml.py`, `twincat.py`, `logix.py`, `schneider.py`, `iec61499.py`, `aml.py`, `dexpi.py`, `tabular.py`); `findings.py` holds the finding type, `xmlutil.py` the line-numbered XML parser | -| `tests/` | `test_st.py`, `test_formats.py` (validator rules), `test_codesys_project.py` and `test_codesys_tools.py` (the `.project` reader and its `info`, `diff` and `xref`, on synthetic projects only), `test_skills.py` (frontmatter, templates, every `iecst` block lints clean) | +| `tests/` | `test_st.py`, `test_formats.py` (validator rules), `test_codesys_project.py` and `test_codesys_tools.py` (the `.project` reader and its `info`, `diff` and `xref`, on synthetic projects only), `test_codesys_hardening.py` (hostile and damaged `.project` input and the reader's CLI contracts, synthetic projects only), `test_linear_xml.py`, `test_linear_aml.py` and `test_linear_text.py` (CPU-time scaling and ordering of the format checkers on crafted input; `linear_support.py` holds their ratio helper and XML builders), `test_skills.py` (frontmatter, templates, every `iecst` block lints clean) | | `evals/` | `claude plugin eval` cases comparing the plugin against a no-plugin baseline | ## Rules diff --git a/plugin.json b/plugin.json index b97a096..e4b35fd 100644 --- a/plugin.json +++ b/plugin.json @@ -1,7 +1,7 @@ { "name": "plc-programming", "displayName": "PLC Programming", - "version": "0.2.1", + "version": "0.2.2", "description": "PLC programming across IEC 61131-3 (ST, LD, FBD, SFC, IL), PLCopen XML / IEC 61131-10, Siemens TIA, Schneider, CODESYS, TwinCAT, ABB Freelance, Rockwell L5X/L5K, IEC 61499, AutomationML, DEXPI, PackML/ISA-88 and OPC UA with a stdlib-only structural validator.", "author": { "name": "SmailG", diff --git a/scripts/plccheck/aml.py b/scripts/plccheck/aml.py index c3f6ecc..4a53d2c 100644 --- a/scripts/plccheck/aml.py +++ b/scripts/plccheck/aml.py @@ -21,6 +21,23 @@ def _resolve(ref: str, base: Path) -> tuple[Path | None, str]: return (base / p) if p else None, u.fragment +def _global_ids(target: Path, cache: dict[Path, set[str] | None]) -> set[str] | None: + """The globalId and uuid values of a referenced PLCopen file, or None if it is not well-formed XML. + + Parsed once per file: every interface that links to the same file would otherwise parse and walk + it again, so a file of n links to itself costs n parses of an n-link document. + """ + key = target.resolve() # `x.xml`, `./x.xml` and `a/../x.xml` are one file + if key not in cache: + try: + doc = parse_file(target) + except XmlError: + cache[key] = None + else: + cache[key] = {v for n in doc.iter() for v in (n.get("globalId"), n.get("uuid")) if v is not None} + return cache[key] + + def check_aml(root: Node, path: Path | None = None) -> list[Finding]: out: list[Finding] = [] if root.tag != "CAEXFile": @@ -56,6 +73,7 @@ def check_aml(root: Node, path: Path | None = None) -> list[Finding]: out.append(warning("AM005", f"InternalLink {side} '{ref}' matches no interface ID in this file", link.line)) base = path.parent if path is not None else None + plcopen_ids: dict[Path, set[str] | None] = {} for iface in root.iter("ExternalInterface"): cls = iface.get("RefBaseClassPath") or "" if not any(cls.endswith(c) for c in DATA_CONNECTORS): @@ -74,11 +92,8 @@ def check_aml(root: Node, path: Path | None = None) -> list[Finding]: out.append(warning("AM011", f"refURI target {target.name} not found next to this file", value.line)) elif fragment and cls.endswith("PLCopenXMLInterface") and target.is_file(): # An absolute refURI can name a directory, FIFO or device; only a regular file is read. - try: - doc = parse_file(target) - except XmlError: - continue - if not any(n.get("globalId") == fragment or n.get("uuid") == fragment for n in doc.iter()): + ids = _global_ids(target, plcopen_ids) + if ids is not None and fragment not in ids: out.append(warning("AM012", f"refURI fragment #{fragment} matches no globalId in {target.name}", value.line)) for ext in root.findall("ExternalReference"): if base is not None and ext.get("Path") and not (base / ext.get("Path")).exists(): diff --git a/scripts/plccheck/logix.py b/scripts/plccheck/logix.py index 6e31422..dcb37eb 100644 --- a/scripts/plccheck/logix.py +++ b/scripts/plccheck/logix.py @@ -3,6 +3,9 @@ from __future__ import annotations import re +from bisect import bisect_left +from collections import Counter +from itertools import tee, zip_longest from .findings import Finding, error, warning, with_context from .st import lint_st @@ -23,23 +26,66 @@ "JMP": 1, "LBL": 1, "AFI": 0, "NOP": 0, "TND": 0, "MCR": 0, "UID": 0, "UIE": 0, "GSV": 4, "SSV": 4, "MSG": 1, "PSC": 0, "PCMD": 3, } -_INSTR = re.compile(r"([A-Za-z_][A-Za-z0-9_]*)\(") - - -def _split_operands(s: str) -> list[str]: - parts, depth, cur = [], 0, "" - for ch in s: - if ch in "([": +# `NAME(`, found at the start of an identifier run (leading digits skipped). Without the lookbehind every +# position of a long run would rescan it for a `(`. +_INSTR = re.compile(r"(? list[tuple[str, int, str | None]]: + """(mnemonic, operand count, first operand) for each `NAME(` in the neutral text, in order. + + The operands are the comma-separated parts, at bracket depth 0, between the instruction's + parentheses; a missing `)` runs to the end of the text less its last character. The count is 0 + when every part is blank. The first operand is "" when there is none, and None when it holds + another instruction (a `NAME(`): that is a call, not a tag, and keeping a copy of it for every + instruction around it would grow with the square of the nesting depth. A bare `(` does not + count, so `Arr[(i+1)]` stays an operand. + + One pass over the text: parentheses matched with a stack, commas kept by bracket depth, and the + operands of each instruction read from those, so no instruction rescans its own operand text. + """ + n = len(t) + depth_before = [0] * (n + 1) # bracket depth ( [ ) ] before each character + text_before = [0] * (n + 1) # non-whitespace characters before each character + close: dict[int, int] = {} # index of each `(` -> index of its `)` + commas: dict[int, list[int]] = {} # bracket depth -> indexes of the commas at that depth + opened: list[int] = [] + depth = chars = 0 + for k, ch in enumerate(t): + depth_before[k], text_before[k] = depth, chars + if ch == ",": + commas.setdefault(depth, []).append(k) + elif ch == "(": + opened.append(k) + depth += 1 + elif ch == ")": + if opened: + close[opened.pop()] = k + depth -= 1 + elif ch == "[": depth += 1 - elif ch in ")]": + elif ch == "]": depth -= 1 - if ch == "," and depth == 0: - parts.append(cur) - cur = "" - else: - cur += ch - parts.append(cur) - return [p.strip() for p in parts] if any(p.strip() for p in parts) else [] + if not ch.isspace(): + chars += 1 + depth_before[n], text_before[n] = depth, chars + found: list[tuple[str, int, str | None]] = [] + # Each match is paired with the next. Matches do not overlap, so only the next one can lie inside this one's operands. + current, following = tee(_INSTR.finditer(t)) + next(following, None) + for m, after in zip_longest(current, following): + start = m.end() + end = close.get(start - 1, n - 1) + top = commas.get(depth_before[start], []) # a comma at the depth the operands start at is a separator + first, last = bisect_left(top, start), bisect_left(top, end) + separators = max(last - first, 0) + has_operand = end > start and text_before[end] - text_before[start] > separators + first_operand: str | None = "" + if has_operand: + stop = top[first] if separators else end + first_operand = None if after is not None and after.end() <= stop else t[start:stop].strip() + found.append((m.group(1).upper(), separators + 1 if has_operand else 0, first_operand)) + return found def check_rung(text: str, line: int, ctx: str) -> tuple[list[Finding], list[str]]: @@ -69,20 +115,29 @@ def check_rung(text: str, line: int, ctx: str) -> tuple[list[Finding], list[str] if bracket: out.append(error("LX021", "unbalanced branch brackets", line, ctx)) otes: list[str] = [] - for m in _INSTR.finditer(t): - start, depth, i = m.end(), 1, m.end() - while i < len(t) and depth: - depth += {"(": 1, ")": -1}.get(t[i], 0) - i += 1 - ops = _split_operands(t[start:i - 1]) - mnem = m.group(1).upper() - if mnem in ARITY and len(ops) != ARITY[mnem]: - out.append(warning("LX022", f"{mnem} takes {ARITY[mnem]} operand(s), found {len(ops)}", line, ctx)) - if mnem == "OTE" and ops: - otes.append(ops[0]) + for mnem, count, first_operand in _instructions(t): + if mnem in ARITY and count != ARITY[mnem]: + out.append(warning("LX022", f"{mnem} takes {ARITY[mnem]} operand(s), found {count}", line, ctx)) + if mnem == "OTE" and count and first_operand is not None: + otes.append(first_operand) return out, otes +def _own_rungs(routine: Node): + """Rung elements of this routine in document order, not descending into a nested . + + A nested is visited by the caller as a scope of its own. Iterating the whole subtree + per routine would check a nested rung once for every routine around it. + """ + stack = list(reversed(routine.children)) + while stack: + node = stack.pop() + if node.tag == "Rung": + yield node + if node.tag != "Routines": + stack.extend(reversed(node.children)) + + def check_l5x(root: Node) -> list[Finding]: out: list[Finding] = [] if root.tag != "RSLogix5000Content": @@ -115,7 +170,7 @@ def check_l5x(root: Node) -> list[Finding]: if rtype not in ROUTINE_TYPES: out.append(warning("LX010", f"unknown routine Type '{rtype}'", r.line, ctx)) if rtype == "RLL": - for rung in r.iter("Rung"): + for rung in _own_rungs(r): text_node = rung.find("Text") if text_node is None: continue @@ -147,6 +202,7 @@ def check_l5k(text: str) -> list[Finding]: rungs are `N: neutral text;`; ST routine lines start with a single quote.""" out: list[Finding] = [] stack: list[tuple[str, int]] = [] + open_words: Counter[str] = Counter() # open frames per keyword, so "is X open" is not a stack scan ends = set(re.findall(r"\bEND_([A-Z][A-Z0-9_]*)\b", text)) st_lines: list[tuple[int, str]] = [] for lineno, raw in enumerate(text.splitlines(), 1): @@ -165,18 +221,20 @@ def check_l5k(text: str) -> list[Finding]: if not end: if word in L5K_BLOCKS or word in ends: stack.append((word, lineno)) + open_words[word] += 1 continue - pos = next((k for k in range(len(stack) - 1, -1, -1) if stack[k][0] == word), None) - if pos is None: + if not open_words[word]: # else the scan would walk the whole stack to find nothing out.append(error("LX050", f"END_{word} without an open {word}", lineno)) continue + pos = next(k for k in range(len(stack) - 1, -1, -1) if stack[k][0] == word) for kw, ln in stack[pos + 1:]: out.append(error("LX050", f"{kw} opened at line {ln} is not closed before END_{word}", ln)) closed = stack[pos] + for kw, _ in stack[pos:]: + open_words[kw] -= 1 del stack[pos:] if closed[0] == "ST_ROUTINE" and st_lines: - src = "\n" * (st_lines[0][0] - 1) + "\n".join(t for _, t in st_lines) - f, _ = lint_st(src, "body") + f, _ = lint_st("\n".join(t for _, t in st_lines), "body", st_lines[0][0]) out.extend(with_context(f, "L5K ST routine")) st_lines = [] for kw, ln in stack: diff --git a/scripts/plccheck/plcopen.py b/scripts/plccheck/plcopen.py index e1f9885..f35f741 100644 --- a/scripts/plccheck/plcopen.py +++ b/scripts/plccheck/plcopen.py @@ -242,15 +242,38 @@ def check_iec61131_10(root: Node) -> list[Finding]: return [error("PX101", f"root element is <{root.tag}>, expected ", root.line)] if root.get("schemaVersion") is None: out.append(warning("PX102", " has no schemaVersion", root.line)) - for bc in root.iter("BodyContent"): + for bc, outs, conns in _bodies_with_connections(root): kind = (bc.get("type") or "").rsplit(":", 1)[-1] st = bc.find("ST") if kind == "ST" and st is not None: f, _ = lint_st(st.text, "body") out.extend(with_context(f, "IEC 61131-10 ST body", (st.text_line or st.line) - 1)) - outs = {n.get("connectionPointOutId") for n in bc.iter() if n.get("connectionPointOutId")} - for conn in bc.iter("Connection"): + for conn in conns: ref = conn.get("refConnectionPointOutId") if ref and ref not in outs: out.append(warning("PX121", f"Connection refers to connectionPointOutId {ref}, not found in this body", conn.line)) return out + + +def _bodies_with_connections(root: Node) -> list[tuple[Node, set[str], list[Node]]]: + """Each BodyContent in document order, with its connectionPointOutId values and Connection elements. + + One walk over the document: a node counts for its nearest enclosing BodyContent. Scanning the + subtree per BodyContent is quadratic when BodyContent elements nest, which a valid file never + does. For such a file the scope of PX121 is the nearest body, not every body around it. + """ + bodies: list[tuple[Node, set[str], list[Node]]] = [] + stack: list[tuple[Node, tuple[Node, set[str], list[Node]] | None]] = [(root, None)] + while stack: + node, current = stack.pop() + if node.tag == "BodyContent": + current = (node, set(), []) + bodies.append(current) + if current is not None: + out_id = node.get("connectionPointOutId") + if out_id: + current[1].add(out_id) + if node.tag == "Connection": + current[2].append(node) + stack.extend((child, current) for child in reversed(node.children)) + return bodies diff --git a/scripts/plccheck/schneider.py b/scripts/plccheck/schneider.py index 2284da6..c01f6bd 100644 --- a/scripts/plccheck/schneider.py +++ b/scripts/plccheck/schneider.py @@ -10,6 +10,22 @@ ROOTS = {"FEFExchangeFile", "VariablesExchangeFile"} +def _own_st_sources(fb: Node): + """STSource elements of this DFB, in document order, not descending into a nested FBSource. + + A nested FBSource is visited by the caller's own loop. Iterating the whole subtree per + FBSource would lint an STSource once for every FBSource around it. + """ + stack = list(reversed(fb.children)) + while stack: + node = stack.pop() + if node.tag == "FBSource": + continue + if node.tag == "STSource": + yield node + stack.extend(reversed(node.children)) + + def check_control_expert(root: Node) -> list[Finding]: out: list[Finding] = [] if root.tag not in ROOTS: @@ -40,7 +56,7 @@ def check_control_expert(root: Node) -> list[Finding]: if name.upper() in seen: out.append(error("CX012", f"DFB parameter/variable '{name}' declared twice", var.line, ctx)) seen[name.upper()] = var.line - for st in fb.iter("STSource"): + for st in _own_st_sources(fb): f, _ = lint_st(st.text, "body") out.extend(with_context(f, f"{ctx} (ST)", (st.text_line or st.line) - 1)) diff --git a/scripts/plccheck/simaticml.py b/scripts/plccheck/simaticml.py index 8d95c97..db73294 100644 --- a/scripts/plccheck/simaticml.py +++ b/scripts/plccheck/simaticml.py @@ -8,6 +8,7 @@ from __future__ import annotations import re +from collections import Counter from .findings import Finding, error, warning from .xmlutil import Node @@ -100,7 +101,7 @@ def _check_block(block: Node) -> list[Finding]: iface = attrs.find("Interface") if iface is not None: out.extend(_check_interface(block.tag, iface, ctx)) - for cu in block.iter("SW.Blocks.CompileUnit"): + for cu in block.iter(COMPILE_UNIT): out.extend(_check_compile_unit(cu, ctx)) return out @@ -140,22 +141,37 @@ def _check_interface(kind: str, iface: Node, ctx: str) -> list[Finding]: PAYLOAD_ROOTS = ("FlgNet", "StructuredText", "StatementList", "Graph") +COMPILE_UNIT = "SW.Blocks.CompileUnit" + + +def _own_scopes(cu: Node) -> list[Node]: + """Payload roots in this CompileUnit, in document order, not inside a nested CompileUnit. + + A nested CompileUnit is visited by the caller as a unit of its own. Iterating the whole subtree + per unit would lint a payload once for every CompileUnit around it. + """ + scopes: list[Node] = [] + stack = list(reversed(cu.children)) + while stack: + node = stack.pop() + if node.tag == COMPILE_UNIT: + continue + if node.tag in PAYLOAD_ROOTS: + scopes.append(node) + stack.extend(reversed(node.children)) + return scopes def _check_compile_unit(cu: Node, ctx: str) -> list[Finding]: """UIds are unique per network payload; GRAPH nests one FlgNet per condition, each its own scope.""" out: list[Finding] = [] - scopes: dict[int, Node] = {} - for node in cu.iter(): - if node.tag in PAYLOAD_ROOTS: - scopes[id(node)] = node - for scope in scopes.values(): + for scope in _own_scopes(cu): nodes = [] stack = list(scope.children) - while stack: # stop at nested payload roots: they are separate scopes + while stack: # stop at nested payload roots and compile units: they are separate scopes n = stack.pop() nodes.append(n) - if n.tag not in PAYLOAD_ROOTS: + if n.tag not in PAYLOAD_ROOTS and n.tag != COMPILE_UNIT: stack.extend(n.children) uids: dict[str, int] = {} wires: set[str] = set() @@ -179,6 +195,7 @@ def check_simatic_sd(text: str) -> list[Finding]: """SIMATIC SD (.s7dcl): only NETWORK/RUNG nesting and brace balance are checked.""" out: list[Finding] = [] stack: list[tuple[str, int]] = [] + open_words: Counter[str] = Counter() # open frames per keyword, so "is X open" is not a stack scan depth = 0 for lineno, raw in enumerate(text.splitlines(), 1): line = re.sub(r"//.*$", "", raw) @@ -187,14 +204,17 @@ def check_simatic_sd(text: str) -> list[Finding]: for word in re.findall(r"\b(END_NETWORK|END_RUNG|NETWORK|RUNG)\b", line): if word in ("NETWORK", "RUNG"): stack.append((word, lineno)) + open_words[word] += 1 continue opener = word[4:] - pos = next((k for k in range(len(stack) - 1, -1, -1) if stack[k][0] == opener), None) - if pos is None: + if not open_words[opener]: # else the scan would walk the whole stack to find nothing out.append(error("SD001", f"{word} without matching {opener}", lineno)) continue + pos = next(k for k in range(len(stack) - 1, -1, -1) if stack[k][0] == opener) for inner, ln in stack[pos + 1:]: out.append(error("SD001", f"{inner} opened at line {ln} is not closed before {word}", ln)) + for kw, _ in stack[pos:]: + open_words[kw] -= 1 del stack[pos:] for word, lineno in stack: out.append(error("SD001", f"{word} opened here is never closed", lineno)) diff --git a/scripts/plccheck/st.py b/scripts/plccheck/st.py index 7910754..a73c09f 100644 --- a/scripts/plccheck/st.py +++ b/scripts/plccheck/st.py @@ -18,6 +18,7 @@ from __future__ import annotations import re +from collections import Counter from dataclasses import dataclass, field from .findings import Finding, error, warning @@ -109,10 +110,10 @@ def shifted(self, offset: int) -> "StAnalysis": ) -def tokenize(src: str) -> tuple[list[Tok], list[Finding]]: +def tokenize(src: str, first_line: int) -> tuple[list[Tok], list[Finding]]: toks: list[Tok] = [] findings: list[Finding] = [] - i, n, line = 0, len(src), 1 + i, n, line = 0, len(src), first_line while i < n: c = src[i] if c == "\n": @@ -242,6 +243,7 @@ class _Frame: line: int scope: dict[str, int] | None = None # declared names, for scope openers body: bool = False # statements allowed at this level + owner: "_Frame | None" = field(default=None, repr=False, compare=False) # nearest scope frame at or below this one def _is_zero(t: Tok) -> bool: @@ -313,20 +315,31 @@ def _designator_then_eq(stmt: list[Tok]) -> bool: return i < len(stmt) and _is_op(stmt[i], "=") -def lint_st(src: str, mode: str = "file") -> tuple[list[Finding], StAnalysis]: - toks, findings = tokenize(src) +def lint_st(src: str, mode: str = "file", first_line: int = 1) -> tuple[list[Finding], StAnalysis]: + """Lint ST text. `first_line` is the file line of its first line, for text taken from the + middle of a file: every line number in a finding, message text included, then names a file line.""" + toks, findings = tokenize(src, first_line) an = StAnalysis() root = _Frame("", 0, scope={}, body=(mode != "declaration")) + root.owner = root stack: list[_Frame] = [root] + open_kw: Counter[str] = Counter() # open frames per keyword, so "is any X open" is not a stack scan pending: tuple[str, str, int] | None = None # (awaited keyword, opener, line) stmt: list[Tok] = [] paren = 0 + def push(fr: _Frame) -> None: + fr.owner = fr if fr.scope is not None else stack[-1].owner + stack.append(fr) + open_kw[fr.kw] += 1 + + def is_open(kinds: set[str]) -> bool: + return any(open_kw[k] for k in kinds) + def scope_frame() -> _Frame: - for fr in reversed(stack): - if fr.scope is not None: - return fr - return root + owner = stack[-1].owner + assert owner is not None + return owner def in_decl() -> bool: return stack[-1].kw in DECL_OPENERS @@ -415,8 +428,8 @@ def _declaration(decl: list[Tok]) -> None: an.has_body = True name = t.text.lstrip("#").upper() nxt = toks[idx + 1] if idx + 1 < len(toks) else None - conditional = any(fr.kw in CONDITIONALS for fr in stack) - in_loop = any(fr.kw in LOOPS for fr in stack) + conditional = is_open(CONDITIONALS) + in_loop = is_open(LOOPS) if not t.member: if _is_op(nxt, "("): an.calls.append((name, t.line, conditional, in_loop)) @@ -459,7 +472,7 @@ def _declaration(decl: list[Tok]) -> None: k += 1 if k < len(toks) and toks[k].kind in ("word", "dstr"): an.header = (kw, toks[k].text.strip('"')) - stack.append(fr) + push(fr) stmt = [] continue @@ -480,7 +493,7 @@ def _declaration(decl: list[Tok]) -> None: stmt = [t] continue if kw in ("EXIT", "CONTINUE"): - if not any(fr.kw in LOOPS for fr in stack): + if not is_open(LOOPS): findings.append(error("ST005", f"{kw} outside FOR/WHILE/REPEAT", t.line)) stmt.append(t) continue @@ -491,13 +504,17 @@ def _declaration(decl: list[Tok]) -> None: if kw in CLOSERS: expected = CLOSERS[kw] - pos = next((k for k in range(len(stack) - 1, 0, -1) if stack[k].kw in expected), None) + pos = None + if is_open(expected): # else the scan would walk the whole stack to find nothing + pos = next((k for k in range(len(stack) - 1, 0, -1) if stack[k].kw in expected), None) if pos is None: findings.append(error("ST002", f"{kw} without a matching {'/'.join(sorted(expected))}", t.line)) else: for fr in stack[pos + 1:]: findings.append(error("ST002", f"{fr.kw} opened at line {fr.line} is not closed before {kw}", fr.line)) closed = stack[pos] + for fr in stack[pos:]: + open_kw[fr.kw] -= 1 del stack[pos:] if closed.kw in VAR_KINDS: scope_frame().body = True diff --git a/scripts/plccheck/twincat.py b/scripts/plccheck/twincat.py index cb09f60..28a4061 100644 --- a/scripts/plccheck/twincat.py +++ b/scripts/plccheck/twincat.py @@ -20,11 +20,53 @@ _FIRST_KW = re.compile(r"\b(PROGRAM|FUNCTION_BLOCK|FUNCTION|INTERFACE|METHOD|PROPERTY|TYPE|VAR_GLOBAL)\b", re.I) +def _blank_spans(text: str, opener: str, closer: str) -> str: + """Replace each non-nested `opener ... closer` span by one space. + + Single pass with str.find. The closer is searched after the opener's own characters, so + `(*)` is not a comment. An opener with no closer after it ends the scan: a later opener has + none either, and retrying from each one (as a lazy regex does) is quadratic. + """ + parts: list[str] = [] + pos = 0 + while True: + start = text.find(opener, pos) + if start < 0: + break + end = text.find(closer, start + len(opener)) + if end < 0: + break + parts.append(text[pos:start]) + parts.append(" ") + pos = end + len(closer) + parts.append(text[pos:]) + return "".join(parts) + + +def _blank_line_comments(text: str) -> str: + """Replace each `//` up to, not including, the end of its line by one space.""" + parts: list[str] = [] + pos = 0 + while True: + start = text.find("//", pos) + if start < 0: + break + end = text.find("\n", start + 2) + parts.append(text[pos:start]) + parts.append(" ") + if end < 0: + pos = len(text) + break + pos = end + parts.append(text[pos:]) + return "".join(parts) + + def _strip_comments(text: str) -> str: - text = re.sub(r"\(\*.*?\*\)", " ", text, flags=re.S) - text = re.sub(r"/\*.*?\*/", " ", text, flags=re.S) - text = re.sub(r"//[^\n]*", " ", text) - return re.sub(r"\{[^}]*\}", " ", text) + text = _blank_spans(text, "(*", "*)") + text = _blank_spans(text, "/*", "*/") + text = _blank_line_comments(text) + return _blank_spans(text, "{", "}") def check_twincat(root: Node, path: Path | None = None) -> list[Finding]: diff --git a/tests/linear_support.py b/tests/linear_support.py new file mode 100644 index 0000000..5f12c39 --- /dev/null +++ b/tests/linear_support.py @@ -0,0 +1,61 @@ +"""Shared pieces of the CPU-time scaling tests: the ratio helper and two XML builders. + +Not a test module (no `test_` prefix), like codesys_synth.py; test_linear_xml.py, test_linear_aml.py +and test_linear_text.py import it. +""" + +from __future__ import annotations + +import sys +import time +import unittest +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parent / "scripts")) + +from plccheck.xmlutil import Node, parse_bytes # noqa: E402 + + +class LinearTime(unittest.TestCase): + """Base class: the ratio helper.""" + + RUNS = 3 + MAX_RATIO = 8 + TIMER_FLOOR = 0.01 # seconds; keeps the ratio meaningful if the small run is below timer noise + + def best_of_runs(self, run) -> float: + best = float("inf") + for _ in range(self.RUNS): + start = time.process_time() + run() + best = min(best, time.process_time() - start) + return best + + def assert_linear(self, make, small, big): + """`make(n)` builds the input outside the timer and returns the thunk to time. + + A ratio, not an absolute bound, so a slow CI runner does not flake: both sizes slow down + together. Process (CPU) time drops scheduler noise, and the best of three drops a run a + hiccup inflated. Linear work lands near big/small, a rescan per element near its square. + """ + run_small, run_big = make(small), make(big) + t_small, t_big = self.best_of_runs(run_small), self.best_of_runs(run_big) + self.assertLess(t_big, self.MAX_RATIO * max(t_small, self.TIMER_FLOOR), + f"size {small} took {t_small:.3f}s, size {big} took {t_big:.3f}s") + + +def xml_root(text: str): + return parse_bytes(text.encode()) + + +def xef(body: str) -> str: + return f"{body}" + + +def xml_root(text: str) -> Node: + return parse_bytes(text.encode()) + + +def xef(body: str) -> str: + return f"{body}" diff --git a/tests/test_formats.py b/tests/test_formats.py index 7032694..cb7bab8 100644 --- a/tests/test_formats.py +++ b/tests/test_formats.py @@ -18,6 +18,7 @@ sys.path.insert(0, str(HERE.parent / "scripts")) import plc_validate # noqa: E402 +from plccheck.st import lint_st # noqa: E402 from plccheck.xmlutil import XmlError, parse_bytes, sniff_root # noqa: E402 TC6_NS = "http://www.plcopen.org/xml/tc6_0201" @@ -580,7 +581,7 @@ def test_text_line_is_the_first_chunk_line(self): class XmlTextJoinScaling(unittest.TestCase): - """Text split into many chunks (one per entity reference) must parse in linear time.""" + """Text split into many chunks (one per entity reference) must parse in linear time (CPU time).""" RUNS = 3 SMALL, BIG = 40_000, 160_000 # BIG is 4x SMALL: linear work is about 4x, quadratic about 16x @@ -591,14 +592,15 @@ def best_of_runs(self, n: int) -> float: data = (f'' + "x<" * n + "").encode() best = float("inf") for _ in range(self.RUNS): - start = time.perf_counter() + start = time.process_time() parse_bytes(data) - best = min(best, time.perf_counter() - start) + best = min(best, time.process_time() - start) return best def test_text_join_time_grows_linearly_with_the_chunk_count(self): # A ratio, not an absolute bound, so a slow CI runner does not flake: both sizes slow - # down together. Best of three drops a run that a scheduling hiccup inflated, which would + # down together. Process (CPU) time, not wall time: both are single-thread CPU-bound, and it + # drops scheduler noise. Best of three drops a run that a hiccup inflated, which would # otherwise push the ratio up. Appending each chunk to node.text copies the text so far # each time (quadratic) and lands near 16x; one join per element lands near 4x. t_small, t_big = self.best_of_runs(self.SMALL), self.best_of_runs(self.BIG) @@ -606,6 +608,45 @@ def test_text_join_time_grows_linearly_with_the_chunk_count(self): f"{self.SMALL} chunks took {t_small:.3f}s, {self.BIG} chunks took {t_big:.3f}s") +class StNestingScaling(unittest.TestCase): + """ST lint CPU time must grow linearly with the nesting depth: no rescan of the open blocks per token.""" + + RUNS = 3 + SMALL, BIG = 2_000, 8_000 # BIG is 4x SMALL: linear work is about 4x, quadratic about 16x + MAX_RATIO = 8 + TIMER_FLOOR = 0.01 # seconds; keeps the ratio meaningful if the small run is below timer noise + + # Each input leaves every block open, so the stack is n deep where a rescan would run. A scan for + # a keyword stops at the first match, so the IF input only exposes the loop scan and the FOR input the + # conditional scan. + INPUTS = { + "word tokens and scope lookup in n open IF": lambda n: "IF a THEN " * n, + "word tokens in n open FOR (no IF/CASE to stop the conditional scan early)": lambda n: "FOR a DO " * n, + "EXIT with n open IF and no loop": lambda n: "IF a THEN " * n + "EXIT;" * n, + "closer matching nothing over n open IF": lambda n: "IF a THEN " * n + "END_FOR;" * n, + } + + def best_of_runs(self, src: str) -> float: + best = float("inf") + for _ in range(self.RUNS): + start = time.process_time() + lint_st(src) + best = min(best, time.process_time() - start) + return best + + def test_lint_time_grows_linearly_with_the_nesting_depth(self): + # A ratio, not an absolute bound, so a slow CI runner does not flake: both sizes slow + # down together. Process (CPU) time, not wall time: both are single-thread CPU-bound, and it + # drops scheduler noise. Best of three drops a run that a hiccup inflated, which would + # otherwise push the ratio up. Rescanning the whole stack for every token is quadratic and + # lands near 16x; per-block counts and a scope pointer land near 4x. + for name, make in self.INPUTS.items(): + with self.subTest(name): + t_small, t_big = self.best_of_runs(make(self.SMALL)), self.best_of_runs(make(self.BIG)) + self.assertLess(t_big, self.MAX_RATIO * max(t_small, self.TIMER_FLOOR), + f"depth {self.SMALL} took {t_small:.3f}s, depth {self.BIG} took {t_big:.3f}s") + + class ShippedTemplates(unittest.TestCase): """Every template a skill hands to the model must itself pass the validator.""" diff --git a/tests/test_linear_aml.py b/tests/test_linear_aml.py new file mode 100644 index 0000000..202f750 --- /dev/null +++ b/tests/test_linear_aml.py @@ -0,0 +1,95 @@ +"""A crafted AutomationML file must not make the checker spin: a PLCopen file linked from many interfaces. + +Each scaling test feeds a checker an input whose size grows 4x and asserts the CPU time grows +about 4x, not 16x (a rescan per element). Each semantics test beside it pins what the linear +version must keep returning on legitimate input. +""" + +from __future__ import annotations + +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parent / "scripts")) +sys.path.insert(0, str(HERE)) + +from linear_support import LinearTime # noqa: E402 +from plccheck import aml # noqa: E402 +from plccheck.xmlutil import parse_bytes # noqa: E402 + +class AmlLinkedFileParsing(LinearTime): + """A PLCopen file linked from many interfaces is parsed and walked once, not once per link.""" + + PLCOPEN = '' + + def interface(self, i: int, ref: str, cls: str = "AMLBase/PLCopenXMLInterface") -> str: + return (f'' + f'{ref}') + + def aml_text(self, interfaces: list[str]) -> str: + return ('' + + "".join(interfaces) + "") + + def check(self, d: str, refs: list[str], files: dict[str, str] | None = None) -> list[tuple[str, str]]: + base = Path(d) + for name, text in {"plc.xml": self.PLCOPEN, **(files or {})}.items(): + (base / name).parent.mkdir(exist_ok=True) + (base / name).write_text(text, encoding="utf-8") + path = base / "plant.aml" + path.write_text(self.aml_text([self.interface(i, ref) for i, ref in enumerate(refs)]), encoding="utf-8") + return [(f.code, f.message) for f in aml.check_aml(parse_bytes(path.read_bytes()), path)] + + def test_check_time_grows_linearly_with_the_links_to_one_file(self): + # Every link to the same file parsed it again, so n links to the file itself (n links + # in it) cost n parses of an n-link document: quadratic (about 16x for 4x). + with tempfile.TemporaryDirectory() as d: + def make(n): # one file per size: the thunks run after both are built + path = Path(d) / f"plant{n}.aml" + path.write_text(self.aml_text([self.interface(i, f"{path.name}#zzz") for i in range(n)]), encoding="utf-8") + root = parse_bytes(path.read_bytes()) + return lambda: aml.check_aml(root, path) + self.assert_linear(make, 80, 320) + + def test_a_file_linked_many_times_is_parsed_once_whatever_the_spelling_of_its_path(self): + refs = ["plc.xml#zz", "./plc.xml#zz", "sub/../plc.xml#zz", "plc.xml#yy"] + with tempfile.TemporaryDirectory() as d, mock.patch.object(aml, "parse_file", wraps=aml.parse_file) as parse: + got = self.check(d, refs, {"sub/x.xml": ""}) + self.assertEqual(parse.call_count, 1) + self.assertEqual([code for code, _ in got], ["AM012"] * 4) + + def test_a_malformed_file_is_parsed_once_and_gives_no_finding(self): + with tempfile.TemporaryDirectory() as d, mock.patch.object(aml, "parse_file", wraps=aml.parse_file) as parse: + got = self.check(d, ["bad.xml#a", "bad.xml#b", "bad.xml#c"], {"bad.xml": ""}) + self.assertEqual(parse.call_count, 1) + self.assertEqual(got, []) + + def test_each_file_is_checked_for_its_own_fragments(self): + files = {"other.xml": ''} + refs = ["plc.xml#g1", "plc.xml#u1", "plc.xml#g2", "other.xml#g2", "other.xml#g1", "missing.xml#g1"] + with tempfile.TemporaryDirectory() as d: + got = self.check(d, refs, files) + # plc.xml has globalId g1 and uuid u1 but not g2; other.xml has g2 only; missing.xml is AM011 + self.assertEqual(got, [ + ("AM012", "refURI fragment #g2 matches no globalId in plc.xml"), + ("AM012", "refURI fragment #g1 matches no globalId in other.xml"), + ("AM011", "refURI target missing.xml not found next to this file"), + ]) + + def test_only_plcopen_links_with_a_fragment_are_followed(self): + with tempfile.TemporaryDirectory() as d, mock.patch.object(aml, "parse_file", wraps=aml.parse_file) as parse: + base = Path(d) + (base / "plc.xml").write_text(self.PLCOPEN, encoding="utf-8") + (base / "plant.aml").write_text(self.aml_text([ + self.interface(0, "plc.xml", "AMLBase/PLCopenXMLInterface"), + self.interface(1, "plc.xml#zz", "AMLBase/COLLADAInterface"), + ]), encoding="utf-8") + got = aml.check_aml(parse_bytes((base / "plant.aml").read_bytes()), base / "plant.aml") + self.assertEqual((parse.call_count, got), (0, [])) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_linear_text.py b/tests/test_linear_text.py new file mode 100644 index 0000000..1584db2 --- /dev/null +++ b/tests/test_linear_text.py @@ -0,0 +1,291 @@ +"""A crafted text file must not make a checker spin: TwinCAT comments, L5K and SIMATIC SD blocks, Logix +rungs and ST routines in L5K. + +Each scaling test feeds a checker an input whose size grows 4x and asserts the CPU time grows +about 4x, not 16x (a rescan per element). Each semantics test beside it pins what the linear +version must keep returning on legitimate input. +""" + +from __future__ import annotations + +import sys +import tracemalloc +import unittest +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parent / "scripts")) +sys.path.insert(0, str(HERE)) + +from linear_support import LinearTime # noqa: E402 +from plccheck import twincat # noqa: E402 +from plccheck.logix import check_l5k, check_rung # noqa: E402 +from plccheck.simaticml import check_simatic_sd # noqa: E402 +from plccheck.st import lint_st # noqa: E402 + + +class TwinCatCommentStripping(LinearTime): + """Comment stripping scans once; an opener that never closes must not rescan the rest of the text.""" + + # (opener repeated, small, big). Every opener is unclosed, so a retry from each one walks to the + # end of the text. The pragma case needs a bigger text: its scan is a single character class. + INPUTS = { + "(* never closed": ("(* ", 1_500, 6_000), + "/* never closed": ("/* ", 1_500, 6_000), + "{ never closed": ("{ ", 5_000, 20_000), + } + + def test_strip_time_grows_linearly_with_the_unclosed_openers(self): + for name, (unit, small, big) in self.INPUTS.items(): + with self.subTest(name): + self.assert_linear(lambda n, unit=unit: (lambda text=unit * n: twincat._strip_comments(text)), small, big) + + def test_an_opener_without_a_closer_ends_the_scan(self): + # The timing test cannot see a retry loop built on str.find (about 10x cheaper than a regex + # retry, so below timer noise at a test-sized input), so count the searches directly. + class CountingStr(str): + calls = 0 + + def find(self, *args): + CountingStr.calls += 1 + return super().find(*args) + + text = CountingStr("(* " * 1000) + self.assertEqual(twincat._blank_spans(text, "(*", "*)"), text) + self.assertLessEqual(CountingStr.calls, 3) # one opener found, one failed closer search; a retry per opener is 1000 + CountingStr.calls = 0 + twincat._blank_spans(CountingStr("(* a *) " * 1000), "(*", "*)") + self.assertLessEqual(CountingStr.calls, 2 * 1000 + 1) # an opener and a closer search per span, one last opener search + + def test_terminated_unterminated_and_adjacent_comments(self): + cases = { + "a (* c *) b": "a b", + "a (* c": "a (* c", + "(* x *)(* y *)": " ", + "a /* c */ b /* d": "a b /* d", + "/* x *//* y */z": " z", + "a (* x\ny *) b": "a b", + "{ pragma } a { open": " a { open", + "{a\n}b": " b", + "x // c\ny": "x \ny", + "x // c": "x ", + "//\n//\n": " \n \n", + } + for text, want in cases.items(): + with self.subTest(text=text): + self.assertEqual(twincat._strip_comments(text), want) + + def test_an_opener_is_not_closed_by_its_own_characters(self): + for text in ("(*)", "a (*) b", "/*/", "a /*/ b"): + with self.subTest(text=text): + self.assertEqual(twincat._strip_comments(text), text) + self.assertEqual(twincat._strip_comments("(*)*)"), " ") + self.assertEqual(twincat._strip_comments("/*/*/"), " ") + + def test_openers_do_not_nest_and_passes_run_in_order(self): + self.assertEqual(twincat._strip_comments("(* (* *) *)"), " *)") + self.assertEqual(twincat._strip_comments("(* // *)\nx"), " \nx") # the (* *) pass runs before // + self.assertEqual(twincat._strip_comments("(* /* *) */"), " */") # (* *) first: it takes the /* inside + self.assertEqual(twincat._strip_comments("/* (* */ *)"), "/* ") # then /* */ finds nothing left to pair with + self.assertEqual(twincat._strip_comments("// (* \n *)"), " ") # so the (* *) span crosses the line end + self.assertEqual(twincat._strip_comments("// x (* y\nz"), " \nz") # an unclosed (* is left for the // pass + self.assertEqual(twincat._strip_comments("{ // }\nz"), "{ \nz") # { } runs last: // already took the } + self.assertEqual(twincat._strip_comments("/* { */ }"), " }") + + +class LogixL5kBlockStack(LinearTime): + """A closer that matches no open block is found without scanning the stack.""" + + def errors(self, text: str) -> list[tuple[int | None, str]]: + found = check_l5k(text) + self.assertEqual({f.code for f in found}, {"LX050"} if found else set()) + return [(f.line, f.message) for f in found] + + def test_check_time_grows_linearly_with_the_open_blocks(self): + # Every END_ZZ matches nothing, so each one used to walk all the open PROGRAM frames: + # quadratic in the depth (about 16x for 4x), a count lookup is linear (about 4x). + def make(n): + text = "PROGRAM " * n + "END_ZZ " * n + return lambda: check_l5k(text) + self.assert_linear(make, 1_000, 4_000) + + def test_a_closer_matching_nothing_reports_its_error(self): + self.assertEqual(self.errors("END_ROUTINE"), [(1, "END_ROUTINE without an open ROUTINE")]) + self.assertEqual(self.errors("PROGRAM p\nEND_ROUTINE\nEND_PROGRAM"), [(2, "END_ROUTINE without an open ROUTINE")]) + + def test_a_matched_closer_closes_its_block(self): + self.assertEqual(self.errors("PROGRAM p\nROUTINE r\nEND_ROUTINE\nEND_PROGRAM"), []) + self.assertEqual(self.errors("PROGRAM p\nEND_PROGRAM\nEND_PROGRAM"), [(3, "END_PROGRAM without an open PROGRAM")]) + self.assertEqual(self.errors("PROGRAM a PROGRAM b END_PROGRAM END_PROGRAM END_PROGRAM"), + [(1, "END_PROGRAM without an open PROGRAM")]) + + def test_a_closer_closes_the_blocks_inside_its_own_and_none_more(self): + got = self.errors("PROGRAM p\nROUTINE r\nTAG t\nEND_PROGRAM\nEND_ROUTINE\nEND_TAG\nEND_PROGRAM") + self.assertEqual(got, [ + (2, "ROUTINE opened at line 2 is not closed before END_PROGRAM"), + (3, "TAG opened at line 3 is not closed before END_PROGRAM"), + (5, "END_ROUTINE without an open ROUTINE"), + (6, "END_TAG without an open TAG"), + (7, "END_PROGRAM without an open PROGRAM"), + ]) + + def test_blocks_left_open_are_reported_at_the_end(self): + self.assertEqual(self.errors("PROGRAM p\nROUTINE r\nEND_ROUTINE"), [(1, "PROGRAM opened here is never closed")]) + + +class LogixL5kStRoutineLines(LinearTime): + """An ST routine is linted from its own line, not behind a padding of one newline per file line before it.""" + + def test_lint_time_grows_linearly_with_the_number_of_routines(self): + # Routine k used to be padded with about 3k newlines, so n routines cost n^2 characters: + # quadratic in the file size (about 16x for 4x). + def make(n): + text = "ST_ROUTINE r\n'a := 1;\nEND_ST_ROUTINE\n" * n + return lambda: check_l5k(text) + self.assert_linear(make, 700, 2_800) + + def test_findings_name_file_lines_in_their_position_and_their_message(self): + text = "PROGRAM p\nROUTINE r\nEND_ROUTINE\nST_ROUTINE s\n'IF a THEN\n'x := 1;\nEND_ST_ROUTINE\n" \ + "ST_ROUTINE t\n'(* open\nEND_ST_ROUTINE\nEND_PROGRAM" + got = [(f.code, f.line, f.message, f.context) for f in check_l5k(text)] + self.assertEqual(got, [ + ("ST002", 5, "IF opened at line 5 is never closed", "L5K ST routine"), + ("ST001", 9, "unterminated comment opened with '(*'", "L5K ST routine"), + ]) + + def test_lint_st_numbers_lines_from_first_line(self): + findings, _ = lint_st("x := 1;\nIF a THEN", "body", 10) + self.assertEqual([(f.line, f.message) for f in findings], [(11, "IF opened at line 11 is never closed")]) + findings, _ = lint_st("x := 1;\nIF a THEN", "body") + self.assertEqual([(f.line, f.message) for f in findings], [(2, "IF opened at line 2 is never closed")]) + + +class SimaticSdBlockStack(LinearTime): + """SIMATIC SD: a closer that matches no open NETWORK/RUNG is found without scanning the stack.""" + + def errors(self, text: str) -> list[tuple[int | None, str]]: + found = check_simatic_sd(text) + self.assertEqual({f.code for f in found}, {"SD001"} if found else set()) + return [(f.line, f.message) for f in found] + + def test_check_time_grows_linearly_with_the_open_blocks(self): + # Every END_RUNG matches nothing, so each one used to walk all the open NETWORK frames: + # quadratic in the depth (about 16x for 4x), a count lookup is linear (about 4x). + def make(n): + text = "NETWORK " * n + "END_RUNG " * n + return lambda: check_simatic_sd(text) + self.assert_linear(make, 1_000, 4_000) + + def test_a_closer_matching_nothing_reports_its_error(self): + self.assertEqual(self.errors("END_RUNG"), [(1, "END_RUNG without matching RUNG")]) + self.assertEqual(self.errors("NETWORK\nEND_RUNG\nEND_NETWORK"), [(2, "END_RUNG without matching RUNG")]) + + def test_a_matched_closer_closes_its_block(self): + self.assertEqual(self.errors("NETWORK\nRUNG\nEND_RUNG\nEND_NETWORK"), []) + self.assertEqual(self.errors("NETWORK\nEND_NETWORK\nEND_NETWORK"), [(3, "END_NETWORK without matching NETWORK")]) + self.assertEqual(self.errors("RUNG RUNG END_RUNG END_RUNG END_RUNG"), [(1, "END_RUNG without matching RUNG")]) + + def test_a_closer_closes_the_blocks_inside_its_own_and_none_more(self): + got = self.errors("NETWORK\nRUNG\nEND_NETWORK\nEND_RUNG\nEND_NETWORK") + self.assertEqual(got, [ + (2, "RUNG opened at line 2 is not closed before END_NETWORK"), + (4, "END_RUNG without matching RUNG"), + (5, "END_NETWORK without matching NETWORK"), + ]) + + def test_blocks_left_open_are_reported_at_the_end(self): + self.assertEqual(self.errors("NETWORK\nRUNG\nEND_RUNG"), [(1, "NETWORK opened here is never closed")]) + + +class LogixRungOperands(LinearTime): + """Operands of every instruction are read in one pass, not by rescanning from each `NAME(`.""" + + # (text for n, small n, big n). Nested and unclosed instructions rescanned the rest of the rung + # per `(`; a long identifier rescanned its run from every character; sequential instructions with + # commas must not search all commas at their depth. + INPUTS = { + "n nested instructions": (lambda n: "XIC(" * n + ")" * n + ";", 300, 1_200), + "n unclosed instructions": (lambda n: "XIC(" * n + ";", 300, 1_200), + "one identifier of n characters": (lambda n: "a" * n + ";", 2_500, 10_000), + "n sequential instructions with commas": (lambda n: "XIC(a,b)" * n + ";", 3_000, 12_000), + } + + def warnings(self, text: str) -> list[str]: + return [f.message for f in check_rung(text, 1, "r")[0] if f.code == "LX022"] + + def test_check_time_grows_linearly_with_the_rung_length(self): + for name, (make, small, big) in self.INPUTS.items(): + with self.subTest(name): + self.assert_linear(lambda n, make=make: (lambda text=make(n): check_rung(text, 1, "r")), small, big) + + def test_operands_are_counted_at_the_instruction_s_own_bracket_depth(self): + self.assertEqual(self.warnings("XIC(a);XIO(b);"), []) + self.assertEqual(self.warnings("XIC(a,b);"), ["XIC takes 1 operand(s), found 2"]) + self.assertEqual(self.warnings("MOV(a,,b);"), ["MOV takes 2 operand(s), found 3"]) + self.assertEqual(self.warnings("COP([a,b],c,d);"), []) # a comma inside [ ] separates nothing + self.assertEqual(self.warnings("XIC(MOV(a,b));"), []) # nor does one inside a nested instruction + self.assertEqual(self.warnings("XIC(MOV(a,b,c));"), ["MOV takes 2 operand(s), found 3"]) + self.assertEqual(self.warnings("XIC(a));"), []) # a stray ) closes nothing + self.assertEqual(self.warnings("MOV(ABS(x),y);"), []) # the comma after a nested call is at the outer depth + self.assertEqual(self.warnings("MOV(ABS(x),y,z);"), ["MOV takes 2 operand(s), found 3"]) + self.assertEqual(self.warnings("MOV(ABS(ABS(x)),y);"), []) + self.assertEqual(self.warnings("MOV(x,ABS(y));XIC(a);"), []) # depth returns to 0 after each instruction + + def test_blank_operands_count_as_none(self): + self.assertEqual(self.warnings("XIC();"), ["XIC takes 1 operand(s), found 0"]) + self.assertEqual(self.warnings("XIC( , );"), ["XIC takes 1 operand(s), found 0"]) + self.assertEqual(self.warnings("MOV(a, );"), []) # one blank part among others still counts + self.assertEqual(self.warnings("XIC(,a);"), ["XIC takes 1 operand(s), found 2"]) # a comma right after the `(` + + def test_an_instruction_without_its_closing_parenthesis_drops_the_last_character(self): + # How the checker has always read `XIC(a,b` (operands "a,"): pinned, not endorsed. + self.assertEqual(self.warnings("XIC(a,b"), ["XIC takes 1 operand(s), found 2"]) + self.assertEqual(self.warnings("XIC(a"), ["XIC takes 1 operand(s), found 0"]) + self.assertEqual(self.warnings("XIC(a,"), []) # the dropped last character is the comma: one operand, `a` + self.assertEqual(self.warnings("XIC(a,,"), ["XIC takes 1 operand(s), found 2"]) + self.assertEqual(self.warnings("XIC("), ["XIC takes 1 operand(s), found 0"]) + + def test_the_mnemonic_skips_leading_digits(self): + self.assertEqual(self.warnings("12XIC(a,b);"), ["XIC takes 1 operand(s), found 2"]) + self.assertEqual(self.warnings("9(a,b);"), []) + + def test_output_coil_operands_are_the_first_operand_stripped(self): + self.assertEqual(check_rung("XIC(a)OTE(b);", 1, "r")[1], ["b"]) + self.assertEqual(check_rung("OTE( x [1] , y );", 1, "r")[1], ["x [1]"]) + self.assertEqual(check_rung("ote(a);", 1, "r")[1], ["a"]) + self.assertEqual(check_rung("OTE( , a);", 1, "r")[1], [""]) # as before: an empty first operand is kept + self.assertEqual(check_rung("OTE( );", 1, "r")[1], []) + + def test_an_operand_holding_an_instruction_is_not_a_coil(self): + # A copy of every enclosing operand grew with the square of the nesting depth (20 KB of + # `OTE(` held 50 MB of operands). An operand that holds another `NAME(` is a call, not a tag. + self.assertEqual(check_rung("OTE(OTE(a));", 1, "r")[1], ["a"]) + self.assertEqual(check_rung("OTE(a(b),c);", 1, "r")[1], []) + self.assertEqual(check_rung("OTE(12XIC(a));", 1, "r")[1], []) + self.assertEqual(check_rung("OTE(a, OTE(b));", 1, "r")[1], ["a", "b"]) # a first operand ends at its comma + self.assertEqual(check_rung("OTE(a(", 1, "r")[1], ["a"]) # the unclosed text drops its last character, the `(` + self.assertEqual(check_rung("OTE(a()", 1, "r")[1], []) # here the dropped last character is a `)`: `a(` is inside + self.assertEqual(check_rung("OTE(" * 400 + ")" * 400 + ";", 1, "r")[1], []) + + def test_a_parenthesis_that_starts_no_instruction_keeps_the_operand(self): + # Whether Logix accepts a parenthesised subscript is not verified; the checker must not + # lose the double-coil warning for it (v0.2.1 kept these operands). + self.assertEqual(check_rung("OTE((a));", 1, "r")[1], ["(a)"]) + self.assertEqual(check_rung("OTE(Arr[(i+1)]);", 1, "r")[1], ["Arr[(i+1)]"]) + self.assertEqual(check_rung("OTE(Arr[(i+1)],b);", 1, "r")[1], ["Arr[(i+1)]"]) + + def test_memory_grows_linearly_with_the_nesting_depth(self): + # About 2 MB for 5000 nested `OTE(` (the lists indexed by character); the operand copies + # the rescan kept were 50 MB. + for text in ("OTE(" * 5_000, "OTE(" * 5_000 + ")" * 5_000 + ";"): + tracemalloc.start() + try: + check_rung(text, 1, "r") + peak = tracemalloc.get_traced_memory()[1] + finally: + tracemalloc.stop() + self.assertLess(peak, 10_000_000) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_linear_xml.py b/tests/test_linear_xml.py new file mode 100644 index 0000000..e134cc7 --- /dev/null +++ b/tests/test_linear_xml.py @@ -0,0 +1,274 @@ +"""A crafted XML document must not make a checker spin or multiply its memory: Control Expert, +IEC 61131-10, L5X and SimaticML. + +Each scaling test feeds a checker an input whose size grows 4x and asserts the CPU time grows +about 4x, not 16x (a rescan per element). Each semantics test beside it pins what the linear +version must keep returning on legitimate input. +""" + +from __future__ import annotations + +import sys +import unittest +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE.parent / "scripts")) +sys.path.insert(0, str(HERE)) + +from linear_support import LinearTime, xef, xml_root # noqa: E402 +from plccheck.logix import check_l5x # noqa: E402 +from plccheck.plcopen import check_iec61131_10 # noqa: E402 +from plccheck.schneider import check_control_expert # noqa: E402 +from plccheck.simaticml import check_simaticml # noqa: E402 + + +class ControlExpertNestedDfb(LinearTime): + """An STSource is linted once, for the nearest FBSource around it, not once per FBSource around it. + + The shape is the one Control Expert writes: FBSource > FBProgram > STSource. + """ + + OPEN = '' + CLOSE = "" + + def dfb(self, name: str, *sources: str) -> str: + return self.OPEN.format(name) + "".join(f"{s}" for s in sources) + self.CLOSE + + def nested(self, depth: int, st: str) -> str: + return xef(self.OPEN.format("D") * depth + f"{st}" + self.CLOSE * depth) + + def check(self, body: str) -> list[tuple[str, int, str | None]]: + return [(f.code, f.line, f.context) for f in check_control_expert(xml_root(xef(body)))] + + def test_lint_time_grows_linearly_with_depth_and_source_size_together(self): + # Depth and ST length grow together: relinting the ST once per enclosing FBSource is + # quadratic in the file size (about 16x for 4x), one lint per STSource is linear (about 4x). + def make(n): + root = xml_root(self.nested(n, "IF a THEN " * (10 * n))) + return lambda: check_control_expert(root) + self.assert_linear(make, 15, 60) + + def test_a_dfb_body_reports_its_unclosed_block_once_with_the_dfb_context(self): + self.assertEqual(self.check(self.dfb("Pump", "IF a THEN ")), [("ST002", 1, "DFB Pump (ST)")]) + self.assertEqual(self.check(self.dfb("Pump", "IF a THEN x := 1; END_IF;")), []) + + def test_deeply_nested_dfb_reports_the_findings_of_a_single_lint(self): + st = "IF a THEN " + single = check_control_expert(xml_root(self.nested(1, st))) + self.assertEqual(len(single), 1, single) + self.assertEqual(check_control_expert(xml_root(self.nested(50, st))), single) + + def test_an_st_source_under_two_wrappers_is_found(self): + got = self.check('
IF a THEN
' + "") + self.assertEqual(got, [("ST002", 1, "DFB Deep (ST)")]) + + def test_two_st_sources_of_one_dfb_are_reported_in_document_order(self): + got = self.check(self.OPEN.format("Pump") + "\nIF a THEN \nFOR i := 1 TO 2 DO " + + self.CLOSE) + self.assertEqual(got, [("ST002", 2, "DFB Pump (ST)"), ("ST002", 3, "DFB Pump (ST)")]) + + def test_st_sources_of_sibling_wrappers_are_reported_in_document_order(self): + got = self.check('\nIF a THEN \n' + 'FOR i := 1 TO 2 DO ') + self.assertEqual(got, [("ST002", 2, "DFB Pump (ST)"), ("ST002", 3, "DFB Pump (ST)")]) + + def test_nested_st_source_belongs_to_the_nearest_dfb(self): + got = self.check(self.OPEN.format("Outer") + "IF a THEN " + + self.dfb("Inner", "FOR i := 1 TO 2 DO ") + self.CLOSE) + self.assertEqual(sorted((code, ctx) for code, _, ctx in got), [("ST002", "DFB Inner (ST)"), ("ST002", "DFB Outer (ST)")]) + + def test_sibling_dfbs_keep_their_own_context_and_line(self): + got = self.check("\n" + self.dfb("A", "IF a THEN ") + "\n" + self.dfb("B", "FOR i := 1 TO 2 DO ")) + self.assertEqual(got, [("ST002", 2, "DFB A (ST)"), ("ST002", 3, "DFB B (ST)")]) + + +class Iec61131_10NestedBodies(LinearTime): + """Connections and ids are collected in one pass, each for its nearest BodyContent.""" + + def project(self, body: str) -> str: + return f'{body}' + + def px121(self, body: str) -> list[tuple[str, int]]: + found = check_iec61131_10(xml_root(self.project(body))) + self.assertEqual({f.code for f in found}, {"PX121"} if found else set()) + return [(f.message.split()[4], f.line) for f in found] + + def test_check_time_grows_linearly_with_the_nesting_depth(self): + # Scanning the whole subtree of every BodyContent for ids and Connection elements is + # quadratic in the depth (about 16x for 4x); one walk is linear (about 4x). + def make(n): + root = xml_root(self.project('' * n + "" * n)) + return lambda: check_iec61131_10(root) + self.assert_linear(make, 500, 2000) + + def test_a_missing_ref_warns_and_a_resolving_one_does_not(self): + got = self.px121( + '\n\n' + '\n') + self.assertEqual(got, [("9,", 4)]) + + def test_ids_resolve_only_inside_their_own_body(self): + got = self.px121( + '\n' + '') + self.assertEqual(got, [("2,", 1), ("1,", 2)]) + + def test_the_body_element_itself_and_deeper_elements_supply_ids(self): + got = self.px121( + '' + '') + self.assertEqual(got, []) + + def test_elements_outside_any_body_are_ignored(self): + got = self.px121('' + '') + self.assertEqual(got, [("1,", 1)]) + + def test_nested_body_is_its_own_scope(self): + # Not a valid structure; pinned so the scope is a decision, not an accident of the walk. + got = self.px121( + '' + '' + '') + self.assertEqual(got, [("1,", 1), ("3,", 1), ("2,", 1)]) + + +class LogixL5xNestedRoutines(LinearTime): + """A rung is checked for the routine it is in, not once for every routine around it.""" + + def l5x(self, programs: str) -> str: + return ('' + f'{programs}') + + def routine(self, name: str, rungs: str, inner: str = "") -> str: + return f'{rungs}{inner}' + + def rung(self, number: int, text: str) -> str: + return f'{text}' + + def nested(self, depth: int, text: str) -> str: + """`depth` routines, each in the of the one before and each with one rung.""" + out = "" + for _ in range(depth): + out = self.routine("r", self.rung(0, text), f"{out}" if out else "") + return self.l5x(f'{out}') + + def check(self, programs: str) -> list[tuple[str, int, str | None]]: + return [(f.code, f.line, f.context) for f in check_l5x(xml_root(self.l5x(programs)))] + + def test_check_time_grows_linearly_with_the_nesting_depth(self): + # Rescanning the whole subtree of every routine for rungs is quadratic in the depth + # (about 16x for 4x); one visit per rung is linear (about 4x). + def make(n): + root = xml_root(self.nested(n, "XIC(a);")) + return lambda: check_l5x(root) + self.assert_linear(make, 150, 600) + + def test_deeply_nested_routines_report_each_rung_once(self): + found = check_l5x(xml_root(self.nested(50, "XIC(a)"))) # no ';': one LX020 per rung + self.assertEqual([f.code for f in found], ["LX020"] * 50) + + def test_sibling_routines_keep_their_own_rungs_and_context(self): + got = self.check('\n' + self.routine("a", self.rung(0, "XIC(a)")) + "\n" + + self.routine("b", self.rung(0, "XIC(b)") + self.rung(1, "OTE(c)")) + "") + self.assertEqual(got, [("LX020", 2, "p/a rung 0"), ("LX020", 3, "p/b rung 0"), ("LX020", 3, "p/b rung 1")]) + + def test_rungs_are_reported_in_document_order(self): + # Two children of one routine, each holding a rung, and two rungs in one child. + rungs = lambda *texts: "".join(f'{t}' for i, t in enumerate(texts)) + routine = ('\n' + rungs("XIC(a)", "XIC(b)") + "\n" + "" + rungs("XIC(c)", "XIC(d)") + "") + got = self.check('' + routine + "") + self.assertEqual([(line, ctx) for _, line, ctx in got], + [(2, "p/a rung 0"), (2, "p/a rung 1"), (3, "p/a rung 0"), (3, "p/a rung 1")]) + + def test_coils_are_counted_per_routines_scope(self): + got = self.check('' + self.routine("a", self.rung(0, "OTE(x);") + self.rung(1, "OTE(x);")) + + self.routine("b", self.rung(0, "OTE(x);")) + "") + self.assertEqual(got, [("LX040", 1, "p")]) + + def test_a_double_coil_on_a_parenthesised_subscript_is_found(self): + two = self.routine("a", self.rung(0, "OTE(Arr[(i+1)]);") + self.rung(1, "OTE(Arr[(i+1)]);")) + self.assertEqual(self.check('' + two + ""), [("LX040", 1, "p")]) + + def test_a_routine_nested_without_a_routines_wrapper_keeps_its_rungs_with_the_routine_around_it(self): + # Only a is visited as a scope of its own; a bare inside a routine is not. + got = self.check('' + self.routine("a", self.rung(0, "XIC(a);") + + self.routine("bare", self.rung(1, "XIC(b)"))) + "") + self.assertEqual(got, [("LX020", 1, "p/a rung 1")]) + + def test_rungs_of_a_nested_routines_scope_belong_to_it(self): + # Not a valid structure; pinned so the scope is a decision, not an accident of the walk. + got = self.check('' + self.routine("a", self.rung(0, "XIC(a)"), + "" + self.routine("in", self.rung(0, "XIC(b)")) + "") + "") + self.assertEqual(got, [("LX020", 1, "p/a rung 0"), ("LX020", 1, "a/in rung 0")]) + + +class SimaticMlNestedCompileUnits(LinearTime): + """A network payload is checked for the CompileUnit it is in, not once for every CompileUnit around it.""" + + UNIT = "SW.Blocks.CompileUnit" + PART = '' + DUP = '' # one SM020 + + def unit(self, *payloads: str) -> str: + """A CompileUnit with each payload under its NetworkSource, as TIA writes it.""" + return f"<{self.UNIT}>" + "".join(f"{p}" for p in payloads) + f"" + + def document(self, units: str) -> str: + return ('B' + f"{units}") + + def nested(self, depth: int, payload: str) -> str: + return self.document(f"<{self.UNIT}>{payload}" * depth + f"" * depth) + + def check(self, units: str) -> list[tuple[str, int, str | None]]: + return [(f.code, f.line, f.context) for f in check_simaticml(xml_root(self.document(units)))] + + def test_check_time_grows_linearly_with_the_nesting_depth(self): + # Collecting the payloads of the whole subtree for every CompileUnit is quadratic in the + # depth (about 16x for 4x); one visit per payload is linear (about 4x). + def make(n): + root = xml_root(self.nested(n, self.PART)) + return lambda: check_simaticml(root) + self.assert_linear(make, 150, 600) + + def test_deeply_nested_compile_units_report_each_network_once(self): + found = check_simaticml(xml_root(self.nested(50, self.DUP))) + self.assertEqual([f.code for f in found], ["SM020"] * 50) + + def test_sibling_compile_units_are_separate_scopes(self): + self.assertEqual(self.check(self.unit(self.PART) + self.unit(self.PART)), []) + self.assertEqual(self.check(f"\n{self.unit(self.DUP)}\n{self.unit(self.PART)}"), [("SM020", 2, "SW.Blocks.FB B")]) + + def test_networks_of_one_compile_unit_are_reported_in_document_order(self): + dup = '' + got = self.check(f"<{self.UNIT}>\n{dup}\n{dup}") + self.assertEqual([line for _, line, _ in got], [2, 3]) + + def test_payloads_of_one_wrapper_are_reported_in_document_order(self): + dup = '' + got = self.check(f"<{self.UNIT}>\n{dup}\n{dup}") + self.assertEqual([line for _, line, _ in got], [2, 3]) + + def test_unresolved_references_and_nested_graph_payloads_keep_their_scopes(self): + graph = ('' + '') + self.assertEqual(self.check(self.unit(graph)), [("SM021", 1, "SW.Blocks.FB B")]) + + def test_payloads_of_a_nested_compile_unit_belong_to_it(self): + # Not a valid structure; pinned so the scope is a decision, not an accident of the walk. The + # outer unit's payload reuses UId 1 with the inner one's and neither sees the other. + inner = self.unit(self.DUP) + self.assertEqual(self.check(f"<{self.UNIT}>{self.PART}{inner}"), + [("SM020", 1, "SW.Blocks.FB B")]) + + def test_a_compile_unit_inside_a_payload_does_not_lend_its_parts_to_it(self): + part_in_unit = f'<{self.UNIT}>' + self.assertEqual(self.check(self.unit(f"{part_in_unit}")), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_st.py b/tests/test_st.py index f01413b..656be17 100644 --- a/tests/test_st.py +++ b/tests/test_st.py @@ -192,6 +192,62 @@ def test_split_declaration_and_body(self): self.assertEqual(codes(scan_call_findings([an1, an2])), ["ST030"]) +class OpenBlockStateTests(unittest.TestCase): + """What the open-block bookkeeping answers: conditional, in a loop, in a loop at all, which scope.""" + + def prog(self, body: str) -> str: + return ("PROGRAM P\nVAR\n tonA : TON;\n rtB : R_TRIG;\n x : BOOL;\n i : INT;\nEND_VAR\nrtB(CLK := x);\n" + + body + "\nEND_PROGRAM\n") + + def test_timer_after_end_if_is_unconditional(self): + self.assertEqual(codes(check_st_file(self.prog("IF x THEN i := 1; END_IF;\ntonA(IN := x, PT := T#1S);"))), []) + + def test_timer_in_for_inside_if_is_conditional_and_in_loop(self): + f = check_st_file(self.prog("IF x THEN FOR i := 1 TO 3 DO tonA(IN := x, PT := T#1S); END_FOR; END_IF;")) + self.assertEqual(codes(f), ["ST030", "ST032"]) + + def test_timer_after_end_for_inside_if_is_conditional_but_not_in_loop(self): + f = check_st_file(self.prog("IF x THEN FOR i := 1 TO 3 DO i := i; END_FOR; tonA(IN := x, PT := T#1S); END_IF;")) + self.assertEqual(codes(f), ["ST030"]) + + def test_timer_inside_case_is_conditional(self): + self.assertEqual(codes(check_st_file(self.prog("CASE i OF 1: tonA(IN := x, PT := T#1S); END_CASE;"))), ["ST030"]) + + def test_exit_inside_repeat_is_inside_a_loop(self): + self.assertEqual(codes(check_st_file("REPEAT EXIT; UNTIL x END_REPEAT;\n")), []) + + def test_timer_after_closer_that_discards_an_open_if_is_unconditional(self): + f = check_st_file(self.prog("FOR i := 1 TO 3 DO IF x THEN END_FOR;\ntonA(IN := x, PT := T#1S);")) + self.assertEqual(codes(f), ["ST002"]) + self.assertIn("IF opened", f[0].message) + + def test_exit_after_the_loop_closed_is_outside_a_loop(self): + f = check_st_file("WHILE a DO x := 1; END_WHILE;\nEXIT;\n") + self.assertEqual(codes(f), ["ST005"]) + self.assertEqual(f[0].line, 2) + + def test_exit_after_closer_that_discards_an_open_loop_is_outside_a_loop(self): + f = check_st_file("IF a THEN WHILE b DO END_IF;\nEXIT;\n") + self.assertEqual(codes(f), ["ST002", "ST005"]) + + def test_closer_of_another_kind_matches_nothing_and_keeps_the_stack(self): + f = check_st_file("IF a THEN END_FOR; x := 1; END_IF;\n") + self.assertEqual(codes(f), ["ST002"]) + self.assertIn("END_FOR without a matching FOR", f[0].message) + + def test_closer_finds_its_opener_under_other_open_blocks(self): + self.assertEqual(codes(check_st_file("WHILE a DO IF b THEN CASE c OF 1: EXIT; END_CASE; END_IF; END_WHILE;\n")), []) + + def test_scope_follows_the_nearest_pou_through_nested_blocks_and_closers(self): + src = ("FUNCTION_BLOCK F\nVAR a : INT; END_VAR\n" + "METHOD M\nVAR a : INT; END_VAR\n" # own scope: not a duplicate of F's a + "IF a THEN IF a THEN IF a THEN x = 1; END_IF; END_IF; END_IF;\n" # body seen 3 blocks deep: ST010 + "END_METHOD\n" + "VAR a : INT; END_VAR\n" # back in F's scope: duplicate + "END_FUNCTION_BLOCK\n") + self.assertEqual(codes(check_st_file(src)), ["ST010", "ST040"]) + + class DialectTests(unittest.TestCase): def test_siemens_scl_source(self): src = '''FUNCTION_BLOCK "FB_Conveyor"