feat: complete small label contract and printing quality gates #2

Merged
james.maes merged 9 commits from release/printing-cohort-20261101 into main 2026-11-02 02:31:08 +00:00
31 changed files with 2903 additions and 470 deletions

No files matched your search

+15
View File
@@ -0,0 +1,15 @@
{
"typeCheckingMode": "strict",
"pythonVersion": "3.13",
"include": [
"scripts/*.py"
],
"reportMissingTypeStubs": "error",
"reportUnnecessaryTypeIgnoreComment": "error",
"reportImplicitOverride": "error",
"exclude": [
"**/__pycache__"
],
"venvPath": "..",
"venv": ".venv"
}
+95 -43
View File
@@ -2,55 +2,107 @@
# SPDX-License-Identifier: Apache-2.0
"""Reject missing/malformed CodeQL evidence before enforcing the severity gate."""
import json
import math
import os
from pathlib import Path
from typing import TypedDict
RuleProperties = TypedDict("RuleProperties", {"security-severity": str | float | None}, total=False)
def check(directory, language, threshold):
if not math.isfinite(threshold) or not 0 <= threshold <= 10:
raise ValueError("Invalid security severity threshold")
canonical = {"java-kotlin": "java", "javascript-typescript": "javascript", "c-cpp": "cpp"}.get(language, language)
files = sorted(Path(directory).glob("*.sarif"))
if not canonical or not any(path.name == canonical + ".sarif" for path in files):
raise ValueError("Missing expected CodeQL SARIF report for " + language)
blocking = []
for path in files:
report = json.loads(path.read_text(encoding="utf-8"))
runs = report.get("runs")
if report.get("version") != "2.1.0" or not isinstance(runs, list) or not runs:
raise ValueError("Invalid or empty SARIF runs: " + path.name)
for run in runs:
driver = run.get("tool", {}).get("driver", {})
results = run.get("results")
if driver.get("name") != "CodeQL" or not isinstance(driver.get("rules"), list) or not isinstance(results, list):
raise ValueError("Missing CodeQL tool/rules/results evidence: " + path.name)
rules = {}
for component in [driver, *run.get("tool", {}).get("extensions", [])]:
for rule in component.get("rules", []):
rules[rule["id"]] = rule
if not rules:
raise ValueError("CodeQL report has no analyzed rule metadata: " + path.name)
for invocation in run.get("invocations", []):
if invocation.get("executionSuccessful") is False:
raise ValueError("CodeQL invocation failed: " + path.name)
for result in results:
rule_id = result.get("ruleId") or result.get("rule", {}).get("id")
if rule_id not in rules:
raise ValueError("Result references missing rule metadata: " + path.name)
if result.get("suppressions"):
continue
severity = float(rules[rule_id].get("properties", {}).get("security-severity", 0) or 0)
if not math.isfinite(severity) or not 0 <= severity <= 10:
raise ValueError("Invalid rule severity: " + rule_id)
if severity >= threshold:
blocking.append((severity, rule_id))
for severity, rule_id in sorted(blocking, reverse=True):
print(f"::error title=CodeQL security finding::{rule_id} ({severity})")
print(f"Validated {len(files)} SARIF report(s); {len(blocking)} blocking result(s)")
return bool(blocking)
class Rule(TypedDict):
"""Keep the rule identity and optional severity metadata used by the gate."""
id: str
properties: RuleProperties
class Component(TypedDict, total=False):
"""Describe CodeQL driver or extension rule metadata."""
name: str
rules: list[Rule]
class Tool(TypedDict, total=False):
"""Group the driver and optional extension rule catalogs."""
driver: Component
extensions: list[Component]
class Result(TypedDict, total=False):
"""Describe only the result fields consumed by the severity gate."""
ruleId: str
rule: dict[str, str]
suppressions: list[object]
class Run(TypedDict, total=False):
"""Retain invocation success and findings for one analyzed language."""
tool: Tool
results: list[Result]
invocations: list[dict[str, bool]]
class Report(TypedDict, total=False):
"""Represent SARIF fields while retaining the explicit runtime validation."""
version: str
runs: list[Run]
MAX_SEVERITY = 10
def check(directory: str | Path, language: str, threshold: float) -> bool: # noqa: PLR0912 -- Keep the ordered evidence checks together; none may be skipped.
"""Validate complete CodeQL evidence and report findings at the threshold."""
if not math.isfinite(threshold) or not 0 <= threshold <= MAX_SEVERITY:
raise ValueError("Invalid security severity threshold")
canonical = {"java-kotlin": "java", "javascript-typescript": "javascript", "c-cpp": "cpp"}.get(language, language)
files = sorted(Path(directory).glob("*.sarif"))
if not canonical or not any(path.name == canonical + ".sarif" for path in files):
raise ValueError("Missing expected CodeQL SARIF report for " + language)
blocking: list[tuple[float, str]] = []
for path in files:
report: Report = json.loads(path.read_text(encoding="utf-8"))
runs = report.get("runs")
if report.get("version") != "2.1.0" or not isinstance(runs, list) or not runs:
raise ValueError("Invalid or empty SARIF runs: " + path.name)
for run in runs:
driver = run.get("tool", {}).get("driver", {})
results = run.get("results")
if driver.get("name") != "CodeQL" or not isinstance(driver.get("rules"), list) or not isinstance(results, list):
raise ValueError("Missing CodeQL tool/rules/results evidence: " + path.name)
rules: dict[str, Rule] = {}
for component in [driver, *run.get("tool", {}).get("extensions", [])]:
for rule in component.get("rules", []):
rules[rule["id"]] = rule
if not rules:
raise ValueError("CodeQL report has no analyzed rule metadata: " + path.name)
for invocation in run.get("invocations", []):
if invocation.get("executionSuccessful") is False:
raise ValueError("CodeQL invocation failed: " + path.name)
for result in results:
rule_id = result.get("ruleId") or result.get("rule", {}).get("id")
if rule_id not in rules:
raise ValueError("Result references missing rule metadata: " + path.name)
if result.get("suppressions"):
continue
severity = float(rules[rule_id].get("properties", {}).get("security-severity", 0) or 0)
if not math.isfinite(severity) or not 0 <= severity <= MAX_SEVERITY:
raise ValueError("Invalid rule severity: " + rule_id)
if severity >= threshold:
blocking.append((severity, rule_id))
for severity, rule_id in sorted(blocking, reverse=True):
print(f"::error title=CodeQL security finding::{rule_id} ({severity})")
print(f"Validated {len(files)} SARIF report(s); {len(blocking)} blocking result(s)")
return bool(blocking)
if __name__ == "__main__":
raise SystemExit(1 if check(os.environ["SARIF_DIR"], os.environ["EXPECTED_LANGUAGE"], float(os.environ["THRESHOLD"])) else 0)
raise SystemExit(1 if check(os.environ["SARIF_DIR"], os.environ["EXPECTED_LANGUAGE"], float(os.environ["THRESHOLD"])) else 0)
+144
View File
@@ -0,0 +1,144 @@
# Copyright 2026 Kingsrook, LLC
# SPDX-License-Identifier: Apache-2.0
"""Fail closed unless every tracked owned script is covered by, and really analyzed by, the quality tools."""
import hashlib
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path, PurePosixPath
from typing import Any
PYTHON_SUFFIXES = (".py", ".pyi", ".pyw")
SCRIPT_PATTERN = ".github/scripts/*.py"
PYRIGHT_CONFIG = ".github"
JAVASCRIPT_SUFFIXES = (".js", ".cjs", ".mjs", ".jsx", ".ts", ".mts", ".cts", ".tsx")
# EX-0015: the only npm package whose license classification the PR gate may exempt (the matcher ignores versions, so this guard pins it).
URI_JS_PURL = "pkg:npm/uri-js"
URI_JS_VERSION = "4.4.1"
URI_JS_INTEGRITY = "sha512-7rKUyy33Q1yc98pQ1DAmLtwX109F7TIfWlW1Ydo8Wl1ii1SeHieeh0HHfPeL2fMXK6z0s8ecKs9frCuLJvndBg=="
URI_JS_LICENSE_SHA256 = "0af366eff4c01ec147c9c61ea9e8ffad64a4294754c9d79355f3fd1b97cb2fb9"
PR_GATES = ".github/workflows/pr-gates.yml"
# Microsoft's official npm Pyright, restored from package-lock.json by `npm ci --ignore-scripts`.
PYRIGHT_ENTRY = Path(__file__).resolve().parents[2] / "node_modules" / "pyright" / "index.js"
def tracked_files(root: Path) -> list[str]:
"""List tracked and not-yet-ignored new paths so no source file can silently bypass the gates."""
# Fixed Git inventory operation; the root is an argument, never shell input.
output = subprocess.run(["git", "-C", str(root), "ls-files", "-z", "--cached", "--others", "--exclude-standard"], check=True, capture_output=True).stdout.decode("utf-8") # noqa: S603, S607 -- Resolve Git from the developer/CI PATH; no shell or user-supplied command.
return [name for name in output.split("\u0000") if name]
def python_scope(tracked: list[str]) -> list[str]:
"""Return the tracked Python sources, rejecting any outside the one checked script root."""
candidates = sorted(name for name in tracked if name.endswith(PYTHON_SUFFIXES))
outside = [name for name in candidates if not PurePosixPath(name).full_match(SCRIPT_PATTERN)]
if outside:
raise ValueError("Tracked Python outside the checked script root: " + ", ".join(outside))
if not candidates:
raise ValueError("No owned Python scripts selected")
return candidates
def javascript_scope(root: Path, tracked: list[str]) -> list[str]:
"""Return tracked JavaScript/TypeScript sources, rejecting any outside the tsconfig include list."""
candidates = sorted(name for name in tracked if name.endswith(JAVASCRIPT_SUFFIXES))
tsconfig = root / "tsconfig.json"
if not tsconfig.is_file():
if candidates:
raise ValueError("Tracked JavaScript/TypeScript without a tsconfig.json: " + ", ".join(candidates))
return []
includes: list[str] = json.loads(tsconfig.read_text(encoding="utf-8"))["include"]
outside = [name for name in candidates if not any(PurePosixPath(name).full_match(pattern) for pattern in includes)]
if outside:
raise ValueError("Tracked JavaScript/TypeScript outside the tsconfig include list: " + ", ".join(outside))
if not candidates:
raise ValueError("No owned JavaScript sources selected")
return candidates
def run_pyright(root: Path, config: str) -> dict[str, Any]:
"""Run the locked npm Pyright through Node with JSON output and return its full report."""
node = shutil.which("node")
if node is None:
raise ValueError("Node.js is required to run the locked Pyright")
if not PYRIGHT_ENTRY.is_file():
raise ValueError(f"Locked Pyright is missing; run npm ci --ignore-scripts ({PYRIGHT_ENTRY})")
# Resolved Node and locked entry script; the config is an argument, never shell input.
completed = subprocess.run([node, str(PYRIGHT_ENTRY), "-p", config, "--outputjson"], cwd=root, check=False, capture_output=True, text=True, encoding="utf-8") # noqa: S603 -- Resolved Node and locked entry script; no shell.
try:
report: dict[str, Any] = json.loads(completed.stdout[completed.stdout.index("{") :])
int(report["summary"]["filesAnalyzed"]) # Fail here when the summary shape is missing.
report["exitCode"] = completed.returncode
except (ValueError, KeyError) as error:
raise ValueError(f"Pyright produced no JSON summary (exit {completed.returncode}): {completed.stdout}{completed.stderr}") from error
return report
def check_pyright_report(report: dict[str, Any], expected: int) -> None:
"""Require the analyzed-file count to match the selected sources, with no errors or warnings."""
summary = report["summary"]
analyzed = summary["filesAnalyzed"]
if analyzed == 0 or analyzed != expected:
raise ValueError(f"Pyright analyzed {analyzed} files, expected {expected}")
if summary["errorCount"] or summary["warningCount"]:
details = [f"{item['file']}: {item['message']}" for item in report["generalDiagnostics"]]
raise ValueError("Pyright reported findings:\n" + "\n".join(details))
if report["exitCode"] != 0:
raise ValueError(f"Pyright exited {report['exitCode']} despite a clean summary")
def run_ruff(root: Path, arguments: list[str], sources: list[str]) -> None:
"""Run Ruff on the explicit source list so no selected file can be skipped."""
# Fixed interpreter module invocation; paths are arguments, never shell input.
if subprocess.run([sys.executable, "-m", "ruff", *arguments, "--no-cache", *sources], cwd=root, check=False).returncode != 0: # noqa: S603 -- Fixed interpreter and module; no shell.
raise ValueError("Ruff " + " ".join(arguments) + " failed")
def check_uri_js_exception(root: Path) -> None:
"""Allow the EX-0015 license exception only for the reviewed, unmodified, development-only uri-js."""
packages: dict[str, dict[str, Any]] = json.loads((root / "package-lock.json").read_text(encoding="utf-8"))["packages"]
copies = [entry for path, entry in packages.items() if path.rsplit("node_modules/", 1)[-1] == "uri-js"]
if len(copies) != 1:
raise ValueError(f"EX-0015 covers exactly one locked uri-js copy, found {len(copies)}")
if copies[0].get("version") != URI_JS_VERSION or copies[0].get("integrity") != URI_JS_INTEGRITY or copies[0].get("dev") is not True:
raise ValueError(f"EX-0015 covers only development-only uri-js {URI_JS_VERSION} with the reviewed integrity")
root_entry = packages.get("", {})
if "uri-js" in {*root_entry.get("dependencies", {}), *root_entry.get("devDependencies", {})}:
raise ValueError("EX-0015 covers uri-js only as a transitive tool dependency, not a direct one")
consumers = [path for path, entry in packages.items() if "uri-js" in entry.get("dependencies", {}) and entry.get("dev") is not True]
if consumers:
raise ValueError("uri-js is required by a non-development package: " + ", ".join(consumers))
digest = hashlib.sha256((root / "node_modules" / "uri-js" / "LICENSE").read_bytes()).hexdigest()
if digest != URI_JS_LICENSE_SHA256:
raise ValueError(f"Installed uri-js LICENSE bytes differ from the reviewed EX-0015 copy ({digest})")
caller = [line for line in (root / PR_GATES).read_text(encoding="utf-8").splitlines() if not line.lstrip().startswith("#")]
listed = re.findall(r"pkg:npm/[^\s,]+", "\n".join(caller))
if listed != [URI_JS_PURL]:
raise ValueError(f"EX-0015 permits only {URI_JS_PURL} in allow-dependencies-licenses, found {listed}")
def main() -> None:
"""Verify scope, then Ruff format/lint and strict Pyright with a positive analyzed count."""
root = Path(__file__).resolve().parents[2]
tracked = tracked_files(root)
sources = python_scope(tracked)
javascript = javascript_scope(root, tracked)
run_ruff(root, ["format", "--check"], sources)
run_ruff(root, ["check"], sources)
report = run_pyright(root, PYRIGHT_CONFIG)
check_pyright_report(report, len(sources))
check_uri_js_exception(root)
print(f"Script quality verified: {len(sources)} Python scripts analyzed by Ruff and Pyright, {len(javascript)} JavaScript files in tsconfig scope")
if __name__ == "__main__":
try:
main()
except (ValueError, OSError, subprocess.CalledProcessError) as error:
print(str(error), file=sys.stderr)
sys.exit(1)
+19 -17
View File
@@ -19,14 +19,14 @@ def selected(name: str, policy: dict[str, Any]) -> bool:
return any(fnmatch.fnmatchcase(name, pattern) for pattern in policy["includes"])
def leading_comments(text: str, style: str) -> str:
def leading_comments(text: str, style: str) -> str: # noqa: PLR0912 -- Each recognized leading-comment syntax has a distinct fail-closed branch.
"""Read comments only, preserving required interpreter/XML directive ordering."""
text = text.removeprefix("\ufeff")
if text.startswith("<?xml"):
declaration = re.match(r"<\?xml[^?]*\?>", text)
if style != "xml" or declaration is None:
raise ValueError("Invalid XML declaration")
text = text[declaration.end():]
text = text[declaration.end() :]
elif text.startswith("#!"):
if style not in ("hash", "powershell", "slash"):
raise ValueError("Unexpected interpreter directive")
@@ -35,7 +35,7 @@ def leading_comments(text: str, style: str) -> str:
if style != "powershell":
raise ValueError("Unexpected PowerShell directive")
text = text.partition("\n")[2]
comments = []
comments: list[str] = []
while text.strip():
text = text.lstrip()
if text.startswith("<?xml") or re.match(r"#!|#Requires\b", text, re.IGNORECASE):
@@ -46,11 +46,9 @@ def leading_comments(text: str, style: str) -> str:
elif style == "slash" and text.startswith("//"):
line, _, text = text.partition("\n")
comments.append(line[2:])
elif ((style == "slash" and text.startswith("/*"))
or (style == "xml" and text.startswith("<!--"))
or (style == "powershell" and text.startswith("<#"))):
elif (style == "slash" and text.startswith("/*")) or (style == "xml" and text.startswith("<!--")) or (style == "powershell" and text.startswith("<#")):
opening, closing = {"slash": ("/*", "*/"), "xml": ("<!--", "-->"), "powershell": ("<#", "#>")}[style]
body, separator, rest = text[len(opening):].partition(closing)
body, separator, rest = text[len(opening) :].partition(closing)
if not separator:
raise ValueError("Unterminated leading comment")
comments.append(body)
@@ -72,19 +70,22 @@ def check_notice(name: str, text: str, policy: dict[str, Any]) -> None:
approved = [item for item in policy["notices"] if any(fnmatch.fnmatchcase(name, pattern) for pattern in item["paths"])]
for notice in approved:
holder = re.escape(notice["holder"])
copyright_pattern = rf"(?m)^Copyright\s+(?:\([cC]\)\s*)?\d{{4}}(?:[-–]\d{{4}})?\.?\s+{holder}(?:\s+\(KofTwentyTwo\))?\.?(?:\s+Licensed under the MIT License \(see LICENSE\)\.)?$"
copyright_pattern = rf"(?m)^Copyright\s+(?:\([cC]\)\s*)?\d{{4}}(?:[-\u2013]\d{{4}})?\.?\s+{holder}(?:\s+\(KofTwentyTwo\))?\.?(?:\s+Licensed under the MIT License \(see LICENSE\)\.)?$"
if re.search(copyright_pattern, comments) is None:
continue
spdx = re.findall(r"(?m)^SPDX-License-Identifier:\s*(\S+)\s*$", comments)
if spdx and any(value != notice["license"] for value in spdx):
continue
apache = all(phrase in comments for phrase in (
"Licensed under the Apache License, Version 2.0",
"you may not use this file except in compliance with the License.",
"https://www.apache.org/licenses/LICENSE-2.0",
"WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND",
"limitations under the License.",
))
apache = all(
phrase in comments
for phrase in (
"Licensed under the Apache License, Version 2.0",
"you may not use this file except in compliance with the License.",
"https://www.apache.org/licenses/LICENSE-2.0",
"WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND",
"limitations under the License.",
)
)
if suffix == ".java" and not apache:
continue
if notice["license"] in spdx or (notice["license"] == "Apache-2.0" and apache):
@@ -94,8 +95,9 @@ def check_notice(name: str, text: str, policy: dict[str, Any]) -> None:
def check_repository(root: Path, policy: dict[str, Any]) -> int:
"""Check tracked sources without scanning build outputs or executing source."""
tracked = subprocess.run(["git", "-C", str(root), "ls-files", "-z"], check=True, capture_output=True).stdout.decode("utf-8").split("\u0000")
failures = []
# Fixed Git inventory operation; paths are arguments, never shell input.
tracked = subprocess.run(["git", "-C", str(root), "ls-files", "-z"], check=True, capture_output=True).stdout.decode("utf-8").split("\u0000") # noqa: S603, S607 -- Resolve Git from the developer/CI PATH; no shell or user-supplied command.
failures: list[str] = []
count = 0
for name in tracked:
if not name or not selected(name, policy):
+76 -61
View File
@@ -2,91 +2,106 @@
# SPDX-License-Identifier: Apache-2.0
"""Exercise the exact CI guards using synthetic evidence only."""
import contextlib
import copy
import io
import importlib.util
import io
import json
from pathlib import Path
import sys
import tempfile
import unittest
from pathlib import Path
from types import ModuleType
from typing import Any, override
sys.dont_write_bytecode = True
def load(name):
spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(name + ".py"))
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def load(name: str) -> ModuleType:
"""Load the real sibling guard for synthetic evidence tests."""
spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(name + ".py"))
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
codeql = load("check-codeql-sarif")
class SarifTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.directory = Path(self.temporary.name)
self.report = {"version": "2.1.0", "runs": [{"tool": {"driver": {"name": "CodeQL", "rules": [{"id": "synthetic", "properties": {"security-severity": "8.0"}}]}}, "results": []}]}
"""Reject incomplete SARIF evidence and preserve severity enforcement."""
def save(self, report):
(self.directory / "java.sarif").write_text(json.dumps(report), encoding="utf-8")
@override
def setUp(self) -> None:
"""Create isolated synthetic evidence with cleanup after each test."""
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.directory = Path(self.temporary.name)
self.report: dict[str, Any] = {"version": "2.1.0", "runs": [{"tool": {"driver": {"name": "CodeQL", "rules": [{"id": "synthetic", "properties": {"security-severity": "8.0"}}]}}, "results": []}]}
def test_clean_report_with_zero_findings_passes(self):
self.save(self.report)
self.assertFalse(codeql.check(self.directory, "java-kotlin", 7))
def save(self, report: object) -> None:
"""Write the test body as SARIF without validating it first."""
(self.directory / "java.sarif").write_text(json.dumps(report), encoding="utf-8")
def test_missing_report_or_wrong_language_fails(self):
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", 7)
self.save(self.report)
with self.assertRaises(ValueError):
codeql.check(self.directory, "csharp", 7)
def test_clean_report_with_zero_findings_passes(self) -> None:
"""Verify clean report with zero findings passes."""
self.save(self.report)
self.assertFalse(codeql.check(self.directory, "java-kotlin", 7))
def test_empty_or_malformed_report_fails(self):
for text in ["", "{", "null", "[]"]:
with self.subTest(text=text):
(self.directory / "java.sarif").write_text(text)
with self.assertRaises((ValueError, AttributeError)):
codeql.check(self.directory, "java-kotlin", 7)
def test_missing_report_or_wrong_language_fails(self) -> None:
"""Verify missing report or wrong language fails."""
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", 7)
self.save(self.report)
with self.assertRaises(ValueError):
codeql.check(self.directory, "csharp", 7)
def test_missing_runs_tool_rules_or_results_fails(self):
variants = [{"version": "2.1.0"}, {"version": "2.1.0", "runs": []}]
for field in ["tool", "results"]:
bad = copy.deepcopy(self.report)
del bad["runs"][0][field]
variants.append(bad)
bad = copy.deepcopy(self.report)
bad["runs"][0]["tool"]["driver"]["rules"] = []
variants.append(bad)
for bad in variants:
with self.subTest(report=bad):
self.save(bad)
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", 7)
def test_empty_or_malformed_report_fails(self) -> None:
"""Verify empty or malformed report fails."""
for text in ["", "{", "null", "[]"]:
with self.subTest(text=text):
(self.directory / "java.sarif").write_text(text)
with self.assertRaises((ValueError, AttributeError)):
codeql.check(self.directory, "java-kotlin", 7)
def test_high_finding_blocks_and_unknown_rule_fails(self):
self.report["runs"][0]["results"] = [{"ruleId": "synthetic"}]
self.save(self.report)
with contextlib.redirect_stdout(io.StringIO()):
self.assertTrue(codeql.check(self.directory, "java-kotlin", 7))
self.report["runs"][0]["results"] = [{"ruleId": "missing"}]
self.save(self.report)
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", 7)
def test_missing_runs_tool_rules_or_results_fails(self) -> None:
"""Verify missing runs tool rules or results fails."""
variants: list[dict[str, Any]] = [{"version": "2.1.0"}, {"version": "2.1.0", "runs": []}]
for field in ["tool", "results"]:
bad = copy.deepcopy(self.report)
del bad["runs"][0][field]
variants.append(bad)
bad = copy.deepcopy(self.report)
bad["runs"][0]["tool"]["driver"]["rules"] = []
variants.append(bad)
for bad in variants:
with self.subTest(report=bad):
self.save(bad)
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", 7)
def test_failed_invocation_and_invalid_threshold_fail(self):
self.report["runs"][0]["invocations"] = [{"executionSuccessful": False}]
self.save(self.report)
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", 7)
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", float("nan"))
def test_high_finding_blocks_and_unknown_rule_fails(self) -> None:
"""Verify high finding blocks and unknown rule fails."""
self.report["runs"][0]["results"] = [{"ruleId": "synthetic"}]
self.save(self.report)
with contextlib.redirect_stdout(io.StringIO()):
self.assertTrue(codeql.check(self.directory, "java-kotlin", 7))
self.report["runs"][0]["results"] = [{"ruleId": "missing"}]
self.save(self.report)
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", 7)
def test_failed_invocation_and_invalid_threshold_fail(self) -> None:
"""Verify failed invocation and invalid threshold fail."""
self.report["runs"][0]["invocations"] = [{"executionSuccessful": False}]
self.save(self.report)
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", 7)
with self.assertRaises(ValueError):
codeql.check(self.directory, "java-kotlin", float("nan"))
if __name__ == "__main__":
unittest.main()
unittest.main()
+297
View File
@@ -0,0 +1,297 @@
# Copyright 2026 Kingsrook, LLC
# SPDX-License-Identifier: Apache-2.0
"""Prove the quality gates reject real violations, using private temporary fixtures and the locked tools."""
import importlib.util
import json
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from types import ModuleType
from typing import Any, override
from unittest import mock
sys.dont_write_bytecode = True
ROOT = Path(__file__).resolve().parents[2]
GOOD_PYTHON = '"""Probe module."""\n\n\ndef add(left: int, right: int) -> int:\n """Add two integers."""\n return left + right\n'
UNUSED_IMPORT_PYTHON = '"""Probe module."""\n\nimport os\n\n\ndef add(left: int, right: int) -> int:\n """Add two integers."""\n return left + right\n'
TWO_SPACE_PYTHON = '"""Probe module."""\n\n\ndef add(left: int, right: int) -> int:\n """Add two integers."""\n return left + right\n'
INCOMPATIBLE_PYTHON = '"""Probe module."""\n\nVALUE: int = "text"\n'
UNANNOTATED_PYTHON = '"""Probe module."""\n\n\ndef echo(value):\n """Return the value."""\n return value\n'
GOOD_JAVASCRIPT = "export function addNumbers(left, right)\n{\n return left + right;\n}\n"
BAD_LAYOUT_JAVASCRIPT = "export function addNumbers(left, right) {\n return left + right\n}\n"
GOOD_TYPED_JAVASCRIPT = "/** @type {number} */\nexport const value = 1;\n"
INCOMPATIBLE_JAVASCRIPT = '/** @type {number} */\nexport const value = "text";\n'
def load(name: str) -> ModuleType:
"""Load the real sibling guard for tests."""
spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(name + ".py"))
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
guard = load("check-script-quality")
def run(command: list[str], cwd: Path, stdin: str | None = None) -> subprocess.CompletedProcess[str]:
"""Run one real tool invocation without a shell and capture its diagnostics."""
return subprocess.run(command, cwd=cwd, input=stdin, check=False, capture_output=True, text=True, encoding="utf-8") # noqa: S603 -- Fixed tool paths and private temporary arguments; no shell.
def ruff(arguments: list[str], source: Path) -> subprocess.CompletedProcess[str]:
"""Run the locked Ruff with the repository configuration on one private fixture."""
return run([sys.executable, "-m", "ruff", *arguments, "--no-cache", "--config", str(ROOT / "ruff.toml"), str(source)], source.parent)
class ScopeTests(unittest.TestCase):
"""Reject tracked sources that fall outside the explicit tool roots."""
def test_python_outside_script_root_is_rejected(self) -> None:
"""Reject a tracked Python file outside the script root instead of letting it disappear."""
for stray in ("tools/helper.py", ".github/scripts/nested/helper.py", "helper.pyi"):
with self.subTest(stray=stray), self.assertRaisesRegex(ValueError, "outside the checked script root"):
guard.python_scope([".github/scripts/check-a.py", stray])
def test_python_scope_returns_sorted_script_root_and_rejects_empty(self) -> None:
"""Select only the direct script root, sorted, and reject selecting nothing."""
self.assertEqual([".github/scripts/a.py", ".github/scripts/b.py"], guard.python_scope([".github/scripts/b.py", "README.md", ".github/scripts/a.py"]))
with self.assertRaisesRegex(ValueError, "No owned Python"):
guard.python_scope(["README.md"])
def test_javascript_requires_tsconfig_include_scope(self) -> None:
"""Reject tracked JavaScript outside tsconfig include patterns, including .js/.cjs/.ts."""
with tempfile.TemporaryDirectory() as name:
root = Path(name)
(root / "tsconfig.json").write_text(json.dumps({"include": ["tools/**/*.mjs", "eslint.config.mjs"]}), encoding="utf-8")
self.assertEqual(["eslint.config.mjs", "tools/a/b.mjs", "tools/c.mjs"], guard.javascript_scope(root, ["tools/c.mjs", "tools/a/b.mjs", "eslint.config.mjs", "README.md"]))
for stray in ("src/z.js", "tools/w.cjs", "tools/v.ts", "tools/u.js"):
with self.subTest(stray=stray), self.assertRaisesRegex(ValueError, "outside the tsconfig include list"):
guard.javascript_scope(root, ["tools/c.mjs", stray])
with self.assertRaisesRegex(ValueError, "No owned JavaScript"):
guard.javascript_scope(root, ["README.md"])
def test_javascript_without_tsconfig_is_rejected(self) -> None:
"""Reject JavaScript in a repository that has no tsconfig."""
with tempfile.TemporaryDirectory() as name:
root = Path(name)
self.assertEqual([], guard.javascript_scope(root, ["README.md"]))
with self.assertRaisesRegex(ValueError, "without a tsconfig.json"):
guard.javascript_scope(root, ["tools/a.mjs"])
class PyrightCountTests(unittest.TestCase):
"""Reject Pyright reports that analyzed too few files or found anything."""
@staticmethod
def report(analyzed: int, errors: int = 0, warnings: int = 0) -> dict[str, Any]:
"""Build the exact Pyright JSON shape consumed by the guard."""
return {"summary": {"filesAnalyzed": analyzed, "errorCount": errors, "warningCount": warnings}, "generalDiagnostics": [{"file": "x.py", "message": "finding"}] * (errors + warnings), "exitCode": 1 if errors or warnings else 0}
def test_nonzero_exit_fails_even_with_clean_summary(self) -> None:
"""Fail a nonzero Pyright exit even when every count is clean."""
clean = self.report(4)
clean["exitCode"] = 1
with self.assertRaisesRegex(ValueError, "exited 1"):
guard.check_pyright_report(clean, 4)
def test_count_must_match_selected_sources(self) -> None:
"""Fail zero or mismatched analyzed counts even with no diagnostics."""
guard.check_pyright_report(self.report(4), 4)
for analyzed in (0, 3, 5):
with self.subTest(analyzed=analyzed), self.assertRaisesRegex(ValueError, "analyzed"):
guard.check_pyright_report(self.report(analyzed), 4)
def test_errors_and_warnings_fail(self) -> None:
"""Fail on errors and on warnings at the exact count."""
for report in (self.report(4, errors=1), self.report(4, warnings=1)):
with self.subTest(report=report), self.assertRaisesRegex(ValueError, "reported findings"):
guard.check_pyright_report(report, 4)
class RuffProbeTests(unittest.TestCase):
"""Run the real locked Ruff on bounded private fixtures."""
def test_compliant_fixture_passes_and_violations_fail(self) -> None:
"""Pass a compliant control and reject unused imports and wrong indentation."""
with tempfile.TemporaryDirectory() as name:
probe = Path(name) / "probe.py"
probe.write_text(GOOD_PYTHON, encoding="utf-8", newline="\n")
self.assertEqual(0, ruff(["check"], probe).returncode)
self.assertEqual(0, ruff(["format", "--check"], probe).returncode)
probe.write_text(UNUSED_IMPORT_PYTHON, encoding="utf-8", newline="\n")
unused = ruff(["check", "--output-format", "concise"], probe)
self.assertNotEqual(0, unused.returncode)
self.assertIn("F401", unused.stdout)
probe.write_text(TWO_SPACE_PYTHON, encoding="utf-8", newline="\n")
indented = ruff(["format", "--check"], probe)
self.assertNotEqual(0, indented.returncode)
self.assertIn("would be reformatted", (indented.stdout + indented.stderr).lower())
class PyrightProbeTests(unittest.TestCase):
"""Run the real locked Pyright with the repository strict settings on private projects."""
@staticmethod
def project(root: Path, include: list[str] | None = None) -> None:
"""Write the repository Pyright config, without its venv location, into a private project."""
config: dict[str, Any] = json.loads((ROOT / ".github/pyrightconfig.json").read_text(encoding="utf-8"))
del config["venvPath"], config["venv"]
if include is not None:
config["include"] = include
(root / "pyrightconfig.json").write_text(json.dumps(config), encoding="utf-8")
def test_compliant_control_passes_and_strict_violations_fail(self) -> None:
"""Analyze the control once with no findings and reject incompatible and unannotated code."""
with tempfile.TemporaryDirectory() as name:
root = Path(name)
self.project(root)
scripts = root / "scripts"
scripts.mkdir()
probe = scripts / "probe.py"
probe.write_text(GOOD_PYTHON, encoding="utf-8", newline="\n")
clean = guard.run_pyright(root, ".")
self.assertEqual(0, clean["exitCode"])
guard.check_pyright_report(clean, 1)
clean["exitCode"] = 1
with self.assertRaisesRegex(ValueError, "exited 1"):
guard.check_pyright_report(clean, 1)
probe.write_text(INCOMPATIBLE_PYTHON, encoding="utf-8", newline="\n")
report = guard.run_pyright(root, ".")
self.assertIn("reportAssignmentType", {item.get("rule") for item in report["generalDiagnostics"]})
with self.assertRaisesRegex(ValueError, "reported findings"):
guard.check_pyright_report(report, 1)
probe.write_text(UNANNOTATED_PYTHON, encoding="utf-8", newline="\n")
strict = guard.run_pyright(root, ".")
self.assertIn("reportMissingParameterType", {item.get("rule") for item in strict["generalDiagnostics"]})
def test_missing_locked_pyright_or_node_is_rejected(self) -> None:
"""Reject a missing locked Pyright entry or Node instead of silently skipping the type gate."""
with tempfile.TemporaryDirectory() as name:
root = Path(name)
with mock.patch.object(guard, "PYRIGHT_ENTRY", root / "missing.js"), self.assertRaisesRegex(ValueError, "Locked Pyright is missing"):
guard.run_pyright(root, ".")
with mock.patch.object(guard.shutil, "which", return_value=None), self.assertRaisesRegex(ValueError, "Node.js is required"):
guard.run_pyright(root, ".")
def test_zero_analyzed_files_cannot_pass(self) -> None:
"""Reject an empty include and a hidden-directory root config in the wrapper."""
with tempfile.TemporaryDirectory() as name:
root = Path(name)
scripts = root / ".github" / "scripts"
scripts.mkdir(parents=True)
(scripts / "probe.py").write_text(GOOD_PYTHON, encoding="utf-8", newline="\n")
for include in (["nothing/*.py"], [".github/scripts/*.py"]):
self.project(root, include)
with self.subTest(include=include), self.assertRaises(ValueError):
guard.check_pyright_report(guard.run_pyright(root, "."), 1)
class JavaScriptProbeTests(unittest.TestCase):
"""Run the real locked ESLint and TypeScript on bounded private input."""
@override
def setUp(self) -> None:
"""Skip only where tsconfig is absent, because the scope guard forbids JavaScript there."""
if not (ROOT / "tsconfig.json").is_file():
self.skipTest("No JavaScript tooling in this repository")
node = shutil.which("node")
self.assertIsNotNone(node)
self.node = str(node)
def eslint(self, text: str) -> subprocess.CompletedProcess[str]:
"""Lint text as an in-scope tool file using the repository configuration."""
return run([self.node, str(ROOT / "node_modules/eslint/bin/eslint.js"), "--stdin", "--stdin-filename", "tools/probe.mjs", "--max-warnings", "0", "--format", "json"], ROOT, text)
def test_eslint_rejects_wrong_layout(self) -> None:
"""Pass a compliant control and reject K&R braces, two-space indent and a missing semicolon."""
good = self.eslint(GOOD_JAVASCRIPT)
self.assertEqual(0, good.returncode, good.stdout + good.stderr)
bad = self.eslint(BAD_LAYOUT_JAVASCRIPT)
self.assertNotEqual(0, bad.returncode)
rules = {message["ruleId"] for message in json.loads(bad.stdout)[0]["messages"]}
self.assertLessEqual({"@stylistic/brace-style", "@stylistic/semi", "@stylistic/indent"}, rules)
def test_checked_javascript_rejects_incompatible_assignment(self) -> None:
"""Reject a real incompatible assignment under the repository strict checkJs settings."""
with tempfile.TemporaryDirectory() as name:
root = Path(name)
(root / "tsconfig.json").write_text(json.dumps({"extends": (ROOT / "tsconfig.json").as_posix(), "compilerOptions": {"typeRoots": [(ROOT / "node_modules/@types").as_posix()]}, "include": ["probe.mjs"]}), encoding="utf-8")
tsc = [self.node, str(ROOT / "node_modules/typescript/bin/tsc"), "-p", str(root / "tsconfig.json")]
(root / "probe.mjs").write_text(GOOD_TYPED_JAVASCRIPT, encoding="utf-8", newline="\n")
good = run(tsc, root)
self.assertEqual(0, good.returncode, good.stdout + good.stderr)
(root / "probe.mjs").write_text(INCOMPATIBLE_JAVASCRIPT, encoding="utf-8", newline="\n")
bad = run(tsc, root)
self.assertNotEqual(0, bad.returncode)
self.assertIn("TS2322", bad.stdout)
URI_GATES = "jobs:\n pr:\n with:\n allow-dependencies-licenses: pkg:npm/uri-js\n"
URI_MIXED_GATES = "jobs:\n pr:\n with:\n allow-dependencies-licenses: >-\n pkg:nuget/Microsoft.WindowsAppSDK,\n pkg:npm/uri-js\n"
def uri_lock(version: str = "4.4.1", integrity: str | None = None, dev: bool = True, direct: bool = False, consumer_dev: bool = True) -> dict[str, Any]:
"""Build a minimal package-lock shape with uri-js under an ESLint-like consumer."""
root_entry: dict[str, Any] = {"devDependencies": {"uri-js": version} if direct else {"eslint": "10.14.0"}}
return {"packages": {"": root_entry, "node_modules/uri-js": {"version": version, "integrity": integrity or guard.URI_JS_INTEGRITY, "dev": dev}, "node_modules/eslint": {"dev": consumer_dev, "dependencies": {"uri-js": "^4.2.2"}}}}
def uri_project(root: Path, lock: dict[str, Any], gates: str = URI_GATES, license_bytes: bytes | None = None) -> None:
"""Write a private project with the lock, installed LICENSE and PR caller that the guard inspects."""
(root / "package-lock.json").write_text(json.dumps(lock), encoding="utf-8")
installed = root / "node_modules" / "uri-js"
installed.mkdir(parents=True)
installed.joinpath("LICENSE").write_bytes((ROOT / "node_modules" / "uri-js" / "LICENSE").read_bytes() if license_bytes is None else license_bytes)
workflows = root / ".github" / "workflows"
workflows.mkdir(parents=True)
(workflows / "pr-gates.yml").write_text("# EX-0015 pkg:npm/uri-js is cited here\n" + gates, encoding="utf-8")
class UriJsExceptionTests(unittest.TestCase):
"""Allow the EX-0015 license exception only for the reviewed, unmodified, development-only uri-js."""
def test_reviewed_uri_js_with_exact_caller_is_accepted(self) -> None:
"""Accept the reviewed package, installed license bytes and an exact caller list, mixed with NuGet entries."""
for gates in (URI_GATES, URI_MIXED_GATES):
with tempfile.TemporaryDirectory() as name, self.subTest(gates=gates):
root = Path(name)
uri_project(root, uri_lock(), gates)
guard.check_uri_js_exception(root)
def test_other_version_integrity_scope_or_consumer_is_rejected(self) -> None:
"""Reject another version, integrity, non-dev or direct use, a non-dev consumer and a second copy."""
duplicate = uri_lock()
duplicate["packages"]["node_modules/eslint/node_modules/uri-js"] = dict(duplicate["packages"]["node_modules/uri-js"])
for lock in (uri_lock(version="4.4.2"), uri_lock(integrity="sha512-other"), uri_lock(dev=False), uri_lock(direct=True), uri_lock(consumer_dev=False), duplicate):
with tempfile.TemporaryDirectory() as name, self.subTest(lock=lock), self.assertRaisesRegex(ValueError, "uri-js"):
root = Path(name)
uri_project(root, lock)
guard.check_uri_js_exception(root)
def test_modified_installed_license_is_rejected(self) -> None:
"""Reject installed license bytes that differ from the reviewed copy."""
with tempfile.TemporaryDirectory() as name:
root = Path(name)
uri_project(root, uri_lock(), license_bytes=(ROOT / "node_modules" / "uri-js" / "LICENSE").read_bytes() + b"\n")
with self.assertRaisesRegex(ValueError, "LICENSE bytes differ"):
guard.check_uri_js_exception(root)
def test_caller_must_list_only_uri_js(self) -> None:
"""Reject a caller that lists no npm exception, another package, a versioned form or a duplicate."""
for gates in ("jobs:\n pr:\n uses: shared\n", URI_GATES.rstrip() + ", pkg:npm/other\n", "jobs:\n pr:\n with:\n allow-dependencies-licenses: pkg:npm/other\n", URI_GATES.rstrip() + "@4.4.1\n", URI_GATES.rstrip() + ", pkg:npm/uri-js\n"):
with tempfile.TemporaryDirectory() as name, self.subTest(gates=gates), self.assertRaisesRegex(ValueError, "permits only"):
root = Path(name)
uri_project(root, uri_lock(), gates)
guard.check_uri_js_exception(root)
if __name__ == "__main__":
unittest.main()
+21 -17
View File
@@ -11,7 +11,6 @@ import unittest
from pathlib import Path
from typing import Any
SPEC = importlib.util.spec_from_file_location("headers", Path(__file__).with_name("check-source-headers.py"))
assert SPEC is not None and SPEC.loader is not None
HEADERS = importlib.util.module_from_spec(SPEC)
@@ -32,8 +31,7 @@ limitations under the License."""
def fixture_policy(holder: str = "James Maes", license_id: str = "MIT") -> dict[str, Any]:
"""Use the production styles with a small explicit owned fixture scope."""
return {"includes": ["src/*"], "exclusions": [], "styles": POLICY["styles"],
"notices": [{"holder": holder, "license": license_id, "paths": ["src/*"]}]}
return {"includes": ["src/*"], "exclusions": [], "styles": POLICY["styles"], "notices": [{"holder": holder, "license": license_id, "paths": ["src/*"]}]}
class HeaderTests(unittest.TestCase):
@@ -51,12 +49,23 @@ class HeaderTests(unittest.TestCase):
with self.subTest(name=name):
HEADERS.check_notice(name, text, fixture_policy())
def test_year_range_separator_is_hyphen_or_en_dash_only(self) -> None:
"""Accept a hyphen or en dash between years and reject any other separator."""
for separator, accepted in (("-", True), (chr(0x2013), True), ("/", False), (chr(0x2014), False)):
text = f"# Copyright 2025{separator}2026 James Maes\n# SPDX-License-Identifier: MIT"
with self.subTest(separator=separator):
if accepted:
HEADERS.check_notice("src/a.py", text, fixture_policy())
else:
with self.assertRaises(ValueError):
HEADERS.check_notice("src/a.py", text, fixture_policy())
def test_required_first_lines_and_bom(self) -> None:
"""Accept shebangs, PowerShell requirements and XML declarations first."""
for name, prefix, body in (
("src/a.sh", "#!/bin/sh\n", "# " + MIT.replace("\n", "\n# ")),
("src/a.ps1", "#Requires -Version 7.4\n", "# " + MIT.replace("\n", "\n# ")),
("src/a.xml", "<?xml version=\"1.0\"?>\n", "<!--" + MIT + "--><root/>"),
("src/a.xml", '<?xml version="1.0"?>\n', "<!--" + MIT + "--><root/>"),
("src/a.mjs", "#!/usr/bin/env node\n", "/*" + MIT + "*/"),
):
with self.subTest(name=name):
@@ -67,21 +76,20 @@ class HeaderTests(unittest.TestCase):
for name, text in (
("src/a.sh", "# " + MIT.replace("\n", "\n# ") + "\n#!/bin/sh"),
("src/a.ps1", "# " + MIT.replace("\n", "\n# ") + "\n#Requires -Version 7.4"),
("src/a.xml", "<!--" + MIT + "--><?xml version=\"1.0\"?><root/>"),
("src/a.xml", "<!--" + MIT + '--><?xml version="1.0"?><root/>'),
):
with self.subTest(name=name), self.assertRaises(ValueError):
HEADERS.check_notice(name, text, fixture_policy())
def test_code_strings_and_late_comments_do_not_count(self) -> None:
"""Reject docstrings and comments after executable declarations."""
for text in ("notice = \"" + MIT.replace("\n", "\\n") + "\"",
"\"\"\"" + MIT + "\"\"\"", "import os\n# " + MIT.replace("\n", "\n# ")):
for text in ('notice = "' + MIT.replace("\n", "\\n") + '"', '"""' + MIT + '"""', "import os\n# " + MIT.replace("\n", "\n# ")):
with self.subTest(text=text), self.assertRaises(ValueError):
HEADERS.check_notice("src/a.py", text, fixture_policy())
def test_wrong_syntax_and_unterminated_comments_fail(self) -> None:
"""Notice text alone is insufficient without a complete language comment."""
for text in ("# " + MIT, "/*" + MIT, "string text = \"" + MIT + "\";"):
for text in ("# " + MIT, "/*" + MIT, 'string text = "' + MIT + '";'):
with self.subTest(text=text), self.assertRaises(ValueError):
HEADERS.check_notice("src/a.cs", text, fixture_policy())
with self.assertRaises(ValueError):
@@ -89,8 +97,7 @@ class HeaderTests(unittest.TestCase):
def test_wrong_holder_or_license_and_conflicting_spdx_fail(self) -> None:
"""Do not conflate repository licenses or accept conflicting identifiers."""
for text in (MIT.replace("James Maes", "Someone Else"), MIT.replace("MIT", "Apache-2.0"),
MIT + "\nSPDX-License-Identifier: Apache-2.0", MIT.splitlines()[0]):
for text in (MIT.replace("James Maes", "Someone Else"), MIT.replace("MIT", "Apache-2.0"), MIT + "\nSPDX-License-Identifier: Apache-2.0", MIT.splitlines()[0]):
with self.subTest(text=text), self.assertRaises(ValueError):
HEADERS.check_notice("src/a.cs", "/*" + text + "*/", fixture_policy())
@@ -103,8 +110,7 @@ class HeaderTests(unittest.TestCase):
def test_java_requires_full_apache_block(self) -> None:
"""An abbreviated SPDX-only Java notice cannot replace scoped convention."""
policy = fixture_policy("Kingsrook, LLC", "Apache-2.0")
for notice in ("Copyright 2026 Kingsrook, LLC\nSPDX-License-Identifier: Apache-2.0",
APACHE.replace("limitations under the License.", "")):
for notice in ("Copyright 2026 Kingsrook, LLC\nSPDX-License-Identifier: Apache-2.0", APACHE.replace("limitations under the License.", "")):
with self.subTest(notice=notice), self.assertRaises(ValueError):
HEADERS.check_notice("src/A.java", "/*" + notice + "*/", policy)
@@ -121,9 +127,7 @@ class HeaderTests(unittest.TestCase):
"""New scripts enter the guard; data, generated and upstream files do not."""
self.assertTrue(HEADERS.selected(".github/scripts/new-tool.py", POLICY))
self.assertTrue(HEADERS.selected(".github/workflows/new-check.yml", POLICY))
for name in (".mvn/wrapper/MavenWrapperDownloader.java", "mvnw", "schemas/new.json",
"src/main/resources/openapi/new.yaml", "src/test/resources/openapi/old.yaml",
"src/generated/Example.cs", "src/Example/obj/Generated.cs", "packages.lock.json"):
for name in (".mvn/wrapper/MavenWrapperDownloader.java", "mvnw", "schemas/new.json", "src/main/resources/openapi/new.yaml", "src/test/resources/openapi/old.yaml", "src/generated/Example.cs", "src/Example/obj/Generated.cs", "packages.lock.json"):
with self.subTest(name=name):
self.assertFalse(HEADERS.selected(name, POLICY))
self.assertTrue(all(item["reason"].strip() for item in POLICY["exclusions"]))
@@ -140,11 +144,11 @@ class HeaderTests(unittest.TestCase):
"""A tracked new source fails while untracked generated output is ignored."""
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
subprocess.run(["git", "init", "-q", str(root)], check=True)
subprocess.run(["git", "init", "-q", str(root)], check=True) # noqa: S603, S607 -- Fixed Git command in the owned temporary test directory.
root.joinpath("src").mkdir()
source = root / "src/a.py"
source.write_text("print(1)", encoding="utf-8")
subprocess.run(["git", "-C", str(root), "add", "--", "src/a.py"], check=True)
subprocess.run(["git", "-C", str(root), "add", "--", "src/a.py"], check=True) # noqa: S603, S607 -- Fixed synthetic path; never a shell command.
with self.assertRaisesRegex(ValueError, "src/a.py"):
HEADERS.check_repository(root, fixture_policy())
source.write_text("# " + MIT.replace("\n", "\n# ") + "\nprint(1)", encoding="utf-8")
+13
View File
@@ -26,6 +26,19 @@ jobs:
node-version: '24'
package-manager-cache: false
- run: npm ci --ignore-scripts
- name: Set up uv
uses: astral-sh/setup-uv@c18668ad3cf93ea998bef934396af7bb5c839dc7 # v10.2.0
with:
version: '0.13.6'
enable-cache: false
- name: Restore locked Python tools
run: uv sync --locked
- name: Verify script quality gates and rejection paths
run: |
uv run --locked --no-sync python .github/scripts/test-script-quality.py
uv run --locked --no-sync python .github/scripts/check-script-quality.py
npm run typecheck
npm run lint
- name: Verify owned source notices and rejection paths
run: |
python3 .github/scripts/test-source-headers.py
+6
View File
@@ -18,6 +18,12 @@ concurrency:
jobs:
pr:
uses: KofTwentyTwo/standards/.github/workflows/pr.yml@fdf5ed029603c938615a32b1beca3628b8787945 # v0.1.2
with:
# EX-0015: development-only uri-js 4.4.1 (ESLint transitive; BSD-2-Clause plus an authors' views sentence).
# https://git.kof22.com/KofTwentyTwo/standards/src/commit/a18179e90f8e6d35ba9f5e583249885c066e52b8/exceptions/register.md#ex-0015
# The matcher ignores versions; check-script-quality.py pins the version, lock integrity and installed
# license bytes and rejects any other pkg:npm entry here.
allow-dependencies-licenses: pkg:npm/uri-js
permissions:
contents: read # read the commits being merged
pull-requests: read # read the pull request title and dependency diff
+5
View File
@@ -4,3 +4,8 @@ dist/
.env
.DS_Store
Thumbs.db
# Local script-quality environments and tool caches.
.venv/
.ruff_cache/
__pycache__/
+1
View File
@@ -0,0 +1 @@
3.13
+3
View File
@@ -14,6 +14,9 @@
acceptance remain required before a coordinated release.
- Add the [extension version registry](docs/VERSIONS.md). No schema or fixture shape
changed for the 2.2 to 2.3 reservation.
- Add `LABEL_4X2_*` and `LABEL_4X1_*` test profiles (203/300/600 dpi) with a 4 × 1
fixture; the 4 × 2 and 4 × 1 PDFs carry explicit actual-size job paper and never a
`mediaProfile`. Both consumers must come from this cohort.
## 0.3.0-rc.1 — 2026-10-31
+16 -1
View File
@@ -1,6 +1,6 @@
# Contributing
Use Node.js 24 or newer. Install the locked development dependencies with `npm ci`, then run `npm run check`. The runtime protocol has no JavaScript dependency; Ajv and ajv-formats are development-only JSON Schema validators.
Use Node.js 24 or newer. Install the locked development dependencies with `npm ci`, then run `npm run check` (`npm test` is the standard entry point for the same check). The runtime protocol has no JavaScript dependency; Ajv and ajv-formats are development-only JSON Schema validators.
Make changes through a pull request. Describe the client/server behavior affected and update README, schema, fixtures and changelog together. Keep credentials, customer documents, printer addresses and machine-specific data out of fixtures. Samples must remain synthetic.
@@ -9,3 +9,18 @@ Run both the Java and .NET consumers' protocol tests for a contract change. Chec
Keep wire names stable within `/printing/v1`. Optional response fields may be added; clients ignore unknown response fields. Removing/renaming a field, changing its type or changing an existing field's meaning requires a new protocol version. A prerelease package version is not permission to silently break an already deployed connector.
`npm run bundle` produces the release archive and checksum in `dist/`. Do not hand-edit generated checksums. Release tags must match the version in package.json, prefixed with `v`. Commits and releases are made by maintainers after review.
## Script quality
Owned scripts follow the KofTwentyTwo standard at revision `fbfc8d2cf3b5e16ae9750b04cf9de1794f1e7152` (configs adapted from its `python` and `typescript` profiles): three-space indentation, Ruff 0.17.3 format/lint (locked in `uv.lock`) and strict Pyright 1.1.414 (Microsoft's npm package, locked in `package-lock.json` and run through Node) for `.github/scripts/*.py`, and ESLint 10 with Stylistic (Allman) plus strict `tsc` over JSDoc for `tools/**/*.mjs` and `eslint.config.mjs`. Restore with `uv sync --locked` and `npm ci --ignore-scripts`, then run:
```text
uv run --locked --no-sync python .github/scripts/test-script-quality.py
uv run --locked --no-sync python .github/scripts/check-script-quality.py
npm run typecheck
npm run lint
```
`npm run quality` runs the repository's Python sequence in one step (`uv sync --locked`, both script-quality commands and both source-header commands); the shared Node CI calls it as its `quality-script`. It needs `uv` on `PATH` and `npm ci` first.
`check-script-quality.py` fails for any tracked Python outside `.github/scripts/` or JavaScript outside the `tsconfig.json` include list, and requires Pyright's analyzed-file count to equal the selected scripts. Pyright must run as `pyright -p .github`: it silently skips dot directories from a root config. Documented exceptions: CLI `print`/`console.log` only in `check-*.py` and `tools/`; synthetic test assertions and boundary constants; narrow inline `noqa` comments for fixed no-shell Git/interpreter calls and ordered fail-closed validators. Do not loosen a rule or add a type waiver to pass. `package.json` overrides `minimatch` to 11.0.1, the patched release for [GHSA-g3hv-75wg-m249](https://github.com/advisories/GHSA-g3hv-75wg-m249) (affected before 11.0.1); keep it until ESLint's own dependency range includes a patched release. The only npm license exception, [EX-0015](https://github.com/KofTwentyTwo/standards/blob/main/exceptions/register.md#ex-0015) for development-only `uri-js` 4.4.1, is enforced by `check-script-quality.py` (version, lock integrity, installed license bytes and the exact `pkg:npm/uri-js` caller entry); changing that package or the entry needs renewed approval.
+2 -2
View File
@@ -77,7 +77,7 @@ polling, and explicit receipts. These additions are included in the
* `EnrollResponse`: `protocolVersion` (1), `deviceId`, `deviceToken`, `pollIntervalSeconds` (default 5).
* `HeartbeatRequest`: `agentVersion`. Response: `serverTime`.
* `PrinterInventory`: full snapshot of the queues selected for cloud exposure in the local Windows companion, in `printers`. Each printer contains stable `printerId`, exact Windows queue `name`, `driverName`, `portName`, `status`, `formats`. An empty list withdraws every queue. There is no new required exposure field: presence in the snapshot expresses local exposure.
* `PrintOptions`: `copies` defaults to 1; `scale` defaults to `ACTUAL_SIZE`. The schema reserves optional `paperWidthMm`, `paperHeightMm` (which must appear together), but the initial implementation rejects them until per-job paper configuration is implemented. Configure paper on the Windows queue for now. First implementation supports `ACTUAL_SIZE` and `FIT`. ZPL dimensions are encoded in its payload; copies must be applied exactly once.
* `PrintOptions`: `copies` defaults to 1; `scale` defaults to `ACTUAL_SIZE`. Optional `paperWidthMm`, `paperHeightMm` (which must appear together) request actual-size job paper for PDF jobs; ZPL encodes its dimensions in the payload, and a `mediaProfile` selection cannot be combined with them. Without them, the Windows queue default applies. First implementation supports `ACTUAL_SIZE` and `FIT`. ZPL dimensions are encoded in its payload; copies must be applied exactly once.
* `SubmitJobRequest`: `deviceId`, `printerId`, `title`, `format`, `contentBase64`, `idempotencyKey`, optional `options`.
* `PrintJob`: `jobId`, `printerId`, `title`, `format`, lowercase hex `contentSha256`, `contentLength` (decoded bytes), normalized `options`, `createdAt`, `status`. Management responses additionally include `deviceId` and may include latest `message` and `spoolerJobId`.
* `JobEvent`: stable `eventId`, `status`, optional `message`, optional string `spoolerJobId`.
@@ -161,7 +161,7 @@ Error body: `{ "code": "ERROR_CODE", "message": "Human-readable explanation" }`.
## Branded test-document profiles
`DeviceTestPrintRequest` optionally includes `testProfile`. PDF accepts `LETTER`, `A4`, or a `LABEL_4X6_*` profile (all label profiles produce the same 101.6 × 152.4 mm PDF). ZPL accepts `LABEL_4X6_203`, `LABEL_4X6_300`, or `LABEL_4X6_600`; the suffix selects the printer resolution in dots per inch. Use possible-value selectors in operator interfaces and match the loaded stock and device resolution. The profile defines document layout; it does not reconfigure Windows paper stock.
`DeviceTestPrintRequest` optionally includes `testProfile`. PDF accepts `LETTER`, `A4`, or a `LABEL_4X6_*`, `LABEL_4X2_*` or `LABEL_4X1_*` profile (all profiles of one height produce the same PDF: 101.6 × 152.4, 50.8 or 25.4 mm). ZPL accepts those label profiles with `_203`, `_300` or `_600`; the suffix selects the printer resolution in dots per inch. For the 4 × 2 and 4 × 1 PDFs the server also submits `options.paperWidthMm`/`paperHeightMm` (101.6 × 50.8 or 25.4) so the client sets actual-size job paper; they are never combined with `mediaProfile`. The 4 × 6 PDF and all ZPL labels carry no job paper. These profiles require an agent and server from the same unreleased 0.4 cohort; older agents reject job paper. Use possible-value selectors in operator interfaces and match the loaded stock and device resolution. The profile defines document layout; it does not reconfigure Windows paper stock.
Requests that omit the field retain compatible defaults: PDF uses Letter, ZPL uses 4 × 6 at 203 dpi. An explicit null, unknown profile, or Letter/A4 with ZPL is invalid. A stable requestId protects transport retries; changing document media is a new operator request. Each action queues one test on each explicitly selected destination, subject to the existing exposure and approval checks. No discovery or connection event automatically prints.
+20 -1
View File
@@ -1,6 +1,25 @@
# Session state — 2026-11-01
## Echo printer media contract in progress
## Release-cohort integration (branch `release/printing-cohort-20261101`)
Unreleased cohort: protocol package `0.4.0-rc.1`, wire v1, native printing extension
2.3.0. This branch integrates the script-quality tooling (locked Ruff/Pyright,
ESLint/Stylistic, strict `checkJs`, the `uri-js` license guard) with the short-label
test profiles. Short labels add `LABEL_4X2_*` and `LABEL_4X1_*` (203/300/600 dpi)
for ZPL and PDF, a `device-test-profile-4x1.request.json` fixture, and explicit
actual-size job paper (101.6 × 50.8 or 25.4 mm) for the 4 × 2 and 4 × 1 PDFs, which
is never combined with `mediaProfile`. The strictly typed tools keep every existing
invalid-input check and add the short-label positive, unsupported-height and
job-paper assertions. Expected totals are 76 schema definitions and 79 fixtures.
State: implemented and locally verified on the merged tree. `npm run check` passed
76 schema definitions and 79 fixtures, and `npm run typecheck`, `npm run lint` and
`npm run bundle` all exited 0. Hosted Gitea CI, consumer pin updates, installed-client
acceptance and physical output are pending; nothing here claims them. The consumer
repositories pin this protocol by immutable commit after the final protocol commit
exists.
## Echo printer media contract in progress (earlier checkpoint)
Isolated branch `feat/echo-media-contract-20261101` adds configured Windows
queue-default profiles and an exact ID/revision job selection. See
+12
View File
@@ -1,5 +1,17 @@
# TODO
## Script quality gates
- [x] Pin Ruff, strict Pyright, ESLint/Stylistic and strict checkJs with locked restores; reject untracked-scope Python/JavaScript and zero-file Pyright runs; negative tool probes pass locally.
- [ ] Observe the new `verify` job steps pass on hosted CI before relying on them.
## Release cohort integration (0.4.0-rc.1, unreleased)
- [x] Integrate short-label (4 × 2 / 4 × 1) test profiles, fixture and job-paper assertions with the strictly typed tools in source.
- [x] Pass the merged-tree `npm run check` (76 definitions / 79 fixtures), `npm run typecheck`, `npm run lint` and `npm run bundle`.
- [ ] Pin the final protocol commit in both consumers, then observe hosted Gitea CI.
- [ ] Complete installed client and physical short-label acceptance separately; schema checks do not establish them.
## Echo media integration
- [x] Define bounded queue-default profile and exact revision selection schemas.
+84
View File
@@ -0,0 +1,84 @@
// Copyright 2026 Kingsrook, LLC
// SPDX-License-Identifier: Apache-2.0
// Based on K22 fbfc8d2cf3b5e16ae9750b04cf9de1794f1e7152.
// KofTwentyTwo TypeScript / JavaScript lint configuration (K22-CODE-TS).
// The Kingsrook layout (3-space indent, Allman braces, if(...) with no space, no more than
// one blank line inside code) is enforced by ESLint Stylistic rather than Prettier, which
// cannot express Allman braces. Type-aware strict rules come from typescript-eslint.
// Install: npm i -D eslint @eslint/js typescript-eslint @stylistic/eslint-plugin typescript
import js from "@eslint/js";
import stylistic from "@stylistic/eslint-plugin";
import tseslint from "typescript-eslint";
export default tseslint.config(
{
ignores: ["**", "!tools/", "!tools/**", "!eslint.config.mjs"],
},
js.configs.recommended,
tseslint.configs.strictTypeChecked,
tseslint.configs.stylisticTypeChecked,
{
languageOptions: {
globals: { process: "readonly", console: "readonly", Buffer: "readonly", URL: "readonly", fetch: "readonly", WebSocket: "readonly", setTimeout: "readonly", clearTimeout: "readonly", structuredClone: "readonly", performance: "readonly", AbortController: "readonly", Headers: "readonly" },
parserOptions: {
projectService: true,
tsconfigRootDir: import.meta.dirname,
},
},
plugins: { "@stylistic": stylistic },
rules: {
// Kingsrook layout
"@stylistic/indent": ["error", 3, { SwitchCase: 1 }],
"@stylistic/brace-style": ["error", "allman", { allowSingleLine: false }],
"@stylistic/keyword-spacing": ["error", {
before: true,
after: true,
overrides: {
if: { after: false },
for: { after: false },
while: { after: false },
switch: { after: false },
catch: { after: false },
},
}],
// 1 blank line inside code; up to 3 so methods can be separated by 3.
"@stylistic/no-multiple-empty-lines": ["error", { max: 3, maxBOF: 0, maxEOF: 0 }],
"@stylistic/lines-between-class-members": ["error", "always", { exceptAfterSingleLine: true }],
"@stylistic/semi": ["error", "always"],
"@stylistic/quotes": ["error", "double", { avoidEscape: true }],
"@stylistic/comma-dangle": ["error", "always-multiline"],
"@stylistic/eol-last": ["error", "always"],
"@stylistic/no-tabs": "error",
"@stylistic/no-trailing-spaces": "error",
"@stylistic/operator-linebreak": ["error", "before"],
"@stylistic/dot-location": ["error", "property"],
"@stylistic/max-len": "off",
// Kingsrook non-layout rules
"curly": ["error", "all"],
"eqeqeq": ["error", "always"],
"no-console": "error",
"no-warning-comments": ["error", { terms: ["todo", "fixme"], location: "start" }],
"@typescript-eslint/explicit-function-return-type": "error",
"@typescript-eslint/naming-convention": ["error",
{ selector: "default", format: ["camelCase"] },
{ selector: "variable", modifiers: ["const", "global"], format: ["camelCase", "UPPER_CASE"] },
{ selector: "typeLike", format: ["PascalCase"] },
{ selector: "enumMember", format: ["PascalCase", "UPPER_CASE"] },
{ selector: "import", format: null },
{ selector: "objectLiteralProperty", format: null },
],
},
},
{
files: ["**/*.js", "**/*.mjs", "**/*.cjs"],
extends: [tseslint.configs.disableTypeChecked],
// Checked JavaScript uses JSDoc plus strict tsc; TS-only return syntax is unavailable.
rules: { "@typescript-eslint/explicit-function-return-type": "off" },
},
{
files: ["tools/**/*.mjs"],
// These command-line tools intentionally print their results and diagnostics.
rules: { "no-console": ["error", { allow: ["log", "error"] }] },
},
);
@@ -0,0 +1 @@
{"printerId":"printer-zebra","format":"ZPL","requestId":"operator-branded-test-4x1","testProfile":"LABEL_4X1_300"}
+1442 -1
View File
File diff suppressed because it is too large. Load diff
+18 -3
View File
@@ -10,10 +10,25 @@
},
"scripts": {
"check": "node tools/verify-schema.mjs && node tools/verify-fixtures.mjs",
"bundle": "node tools/bundle.mjs"
"test": "npm run check",
"bundle": "node tools/bundle.mjs",
"lint": "eslint . --max-warnings 0",
"typecheck": "tsc --noEmit",
"quality": "uv sync --locked && uv run --locked --no-sync python .github/scripts/test-script-quality.py && uv run --locked --no-sync python .github/scripts/check-script-quality.py && uv run --locked --no-sync python .github/scripts/test-source-headers.py && uv run --locked --no-sync python .github/scripts/check-source-headers.py"
},
"devDependencies": {
"ajv": "8.20.0",
"ajv-formats": "3.0.1"
}
"ajv-formats": "3.0.1",
"@eslint/js": "10.0.1",
"typescript": "6.0.3",
"@stylistic/eslint-plugin": "5.10.0",
"@types/node": "24.10.1",
"typescript-eslint": "8.72.1",
"eslint": "10.14.0",
"pyright": "1.1.414"
},
"overrides": {
"minimatch": "11.0.1"
},
"packageManager": "npm@11.19.1"
}
+13
View File
@@ -0,0 +1,13 @@
# Copyright 2026 Kingsrook, LLC
# SPDX-License-Identifier: Apache-2.0
[project]
name = "qqq-print-protocol-script-quality"
version = "0.0.0"
requires-python = ">=3.13"
[dependency-groups]
dev = ["ruff==0.17.3"]
[tool.uv]
package = false
+56
View File
@@ -0,0 +1,56 @@
# Copyright 2026 Kingsrook, LLC
# SPDX-License-Identifier: Apache-2.0
# Based on K22 fbfc8d2cf3b5e16ae9750b04cf9de1794f1e7152.
# KofTwentyTwo Python lint + format configuration (K22-CODE-PY). Copy to the repo root
# (or merge into [tool.ruff] in pyproject.toml). `ruff format --check` and `ruff check`
# run in CI and pre-commit with zero findings allowed.
target-version = "py313"
include = [".github/scripts/*.py"]
extend-exclude = [".venv", "node_modules", "dist", "target", "bin", "obj"]
# Kingsrook: no practical line-length limit (ruff's maximum is 320).
line-length = 320
# Kingsrook: 3-space indentation. Ruff's formatter supports it fully; pycodestyle's
# "multiple of four" rules (E111/E114) are therefore not selected.
indent-width = 3
[format]
quote-style = "double"
indent-style = "space"
line-ending = "lf"
docstring-code-format = true
[lint]
select = [
"F", # pyflakes
"E4", "E7", "E9", "W", # pycodestyle errors that are layout-independent
"I", # isort: no wildcard imports, ordered
"N", # pep8-naming
"UP", # pyupgrade
"B", # bugbear
"S", # bandit security
"BLE", # blind except
"T20", # print statements: use logging (Kingsrook logging rule)
"ERA", # commented-out code (Kingsrook: no zombie code)
"FIX", # FIXME/TODO must be resolved or tracked
"TD", # TODO format: author and issue link
"PTH", # pathlib over os.path
"RET", "SIM", "C4", "PIE", "PL", "RUF",
"ANN", # type annotations on every function
"D", # docstrings on modules, classes, functions
"LOG", "G", # logging hygiene
]
ignore = [
"D203", # conflicts with D211
"D213", # conflicts with D212
"TD003", # issue link is enforced by review, not regex
]
[lint.per-file-ignores]
".github/scripts/check-*.py" = ["T201"] # CLI diagnostics are the public stdout contract, not application logging.
".github/scripts/test-*.py" = ["S101", "PLR2004"] # Synthetic assertions and explicit boundary values.
[lint.pydocstyle]
convention = "pep257"
[lint.flake8-tidy-imports]
ban-relative-imports = "all"
+14 -2
View File
@@ -65,7 +65,13 @@
"enum": [
"LABEL_4X6_203",
"LABEL_4X6_300",
"LABEL_4X6_600"
"LABEL_4X6_600",
"LABEL_4X2_203",
"LABEL_4X2_300",
"LABEL_4X2_600",
"LABEL_4X1_203",
"LABEL_4X1_300",
"LABEL_4X1_600"
]
}
}
@@ -79,7 +85,13 @@
"A4",
"LABEL_4X6_203",
"LABEL_4X6_300",
"LABEL_4X6_600"
"LABEL_4X6_600",
"LABEL_4X2_203",
"LABEL_4X2_300",
"LABEL_4X2_600",
"LABEL_4X1_203",
"LABEL_4X1_300",
"LABEL_4X1_600"
]
},
"DeviceRetryRequest": {
+21 -17
View File
@@ -1,25 +1,29 @@
// Copyright 2026 Kingsrook, LLC
// SPDX-License-Identifier: Apache-2.0
import assert from 'node:assert/strict';
import { createHash } from 'node:crypto';
import { mkdirSync, readFileSync, writeFileSync } from 'node:fs';
import { spawnSync } from 'node:child_process';
import { fileURLToPath } from 'node:url';
import assert from "node:assert/strict";
import { createHash } from "node:crypto";
import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { spawnSync } from "node:child_process";
import { fileURLToPath } from "node:url";
const root = fileURLToPath(new URL('../', import.meta.url));
const manifest = JSON.parse(readFileSync(new URL('../package.json', import.meta.url), 'utf8'));
assert(/^\d+\.\d+\.\d+(?:-[A-Za-z0-9.-]+)?$/.test(manifest.version), 'Invalid release version.');
if(process.env.GITHUB_REF_TYPE === 'tag')
const root = fileURLToPath(new URL("../", import.meta.url));
/** @type {{version:string}} */
const manifest = JSON.parse(readFileSync(new URL("../package.json", import.meta.url), "utf8"));
assert(/^\d+\.\d+\.\d+(?:-[A-Za-z0-9.-]+)?$/.test(manifest.version), "Invalid release version.");
if(process.env["GITHUB_REF_TYPE"] === "tag")
{
assert.equal(process.env.GITHUB_REF_NAME, `v${manifest.version}`, 'Release tag must match package.json.');
assert.equal(process.env["GITHUB_REF_NAME"], `v${manifest.version}`, "Release tag must match package.json.");
}
mkdirSync(new URL('../dist/', import.meta.url), { recursive: true });
mkdirSync(new URL("../dist/", import.meta.url), { recursive: true });
const name = `qqq-print-protocol-${manifest.version}.tar.gz`;
const files = ['README.md', 'LICENSE', 'NOTICE', 'CHANGELOG.md', 'CONTRIBUTING.md', 'SECURITY.md', 'package.json', 'package-lock.json', 'schemas', 'fixtures', 'tools', 'docs/INTEGRATION.md', 'docs/DEVICE-CONTROLS.md', 'docs/QQQ-AUTHENTICATION.md', 'docs/REALTIME.md', 'docs/PRINTER-MEDIA.md'];
const result = spawnSync('tar', ['-czf', `dist/${name}`, ...files], { cwd: root, stdio: 'inherit', shell: false });
if(result.error) throw result.error;
assert.equal(result.status, 0, 'Archive creation failed.');
const digest = createHash('sha256').update(readFileSync(new URL(`../dist/${name}`, import.meta.url))).digest('hex');
writeFileSync(new URL('../dist/SHA256SUMS', import.meta.url), `${digest} ${name}\n`);
const files = ["README.md", "LICENSE", "NOTICE", "CHANGELOG.md", "CONTRIBUTING.md", "SECURITY.md", "package.json", "package-lock.json", "tsconfig.json", "eslint.config.mjs", "schemas", "fixtures", "tools", "docs/INTEGRATION.md", "docs/DEVICE-CONTROLS.md", "docs/QQQ-AUTHENTICATION.md", "docs/REALTIME.md", "docs/PRINTER-MEDIA.md"];
const result = spawnSync("tar", ["-czf", `dist/${name}`, ...files], { cwd: root, stdio: "inherit", shell: false });
if(result.error)
{
throw result.error;
}
assert.equal(result.status, 0, "Archive creation failed.");
const digest = createHash("sha256").update(readFileSync(new URL(`../dist/${name}`, import.meta.url))).digest("hex");
writeFileSync(new URL("../dist/SHA256SUMS", import.meta.url), `${digest} ${name}\n`);
console.log(`Created dist/${name} and dist/SHA256SUMS.`);
+20 -20
View File
@@ -1,35 +1,35 @@
// Copyright 2026 Kingsrook, LLC
// SPDX-License-Identifier: Apache-2.0
import { createHash } from 'node:crypto';
import { writeFileSync } from 'node:fs';
import { createHash } from "node:crypto";
import { writeFileSync } from "node:fs";
// Keep printable fixtures synthetic and independent of a printer's settings.
const content = Buffer.from('^XA\n^FO40,40^A0N,36,36^FDQQQ cloud print test^FS\n^FO40,90^A0N,24,24^FDProtocol v1 - one test label^FS\n^XZ\n', 'utf8');
const root = new URL('../fixtures/', import.meta.url);
writeFileSync(new URL('test-label.zpl', root), content);
const content = Buffer.from("^XA\n^FO40,40^A0N,36,36^FDQQQ cloud print test^FS\n^FO40,90^A0N,24,24^FDProtocol v1 - one test label^FS\n^XZ\n", "utf8");
const root = new URL("../fixtures/", import.meta.url);
writeFileSync(new URL("test-label.zpl", root), content);
const job = {
jobId: 'job-example',
deviceId: 'device-example',
printerId: 'zebra-example',
title: 'QQQ cloud print test',
format: 'ZPL',
contentSha256: createHash('sha256').update(content).digest('hex'),
jobId: "job-example",
deviceId: "device-example",
printerId: "zebra-example",
title: "QQQ cloud print test",
format: "ZPL",
contentSha256: createHash("sha256").update(content).digest("hex"),
contentLength: content.length,
options: { copies: 1, scale: 'ACTUAL_SIZE' },
createdAt: '2026-10-29T18:00:00Z',
status: 'CLAIMED'
options: { copies: 1, scale: "ACTUAL_SIZE" },
createdAt: "2026-10-29T18:00:00Z",
status: "CLAIMED",
};
const request = {
deviceId: job.deviceId,
printerId: job.printerId,
title: job.title,
format: job.format,
contentBase64: content.toString('base64'),
idempotencyKey: 'protocol-fixture-label-v1',
options: job.options
contentBase64: content.toString("base64"),
idempotencyKey: "protocol-fixture-label-v1",
options: job.options,
};
writeFileSync(new URL('submit.request.json', root), JSON.stringify(request, null, 2) + '\n');
writeFileSync(new URL('claim.response.json', root), JSON.stringify({ job }, null, 2) + '\n');
writeFileSync(new URL('job.response.json', root), JSON.stringify(job, null, 2) + '\n');
writeFileSync(new URL("submit.request.json", root), JSON.stringify(request, null, 2) + "\n");
writeFileSync(new URL("claim.response.json", root), JSON.stringify({ job }, null, 2) + "\n");
writeFileSync(new URL("job.response.json", root), JSON.stringify(job, null, 2) + "\n");
+79 -78
View File
@@ -3,82 +3,83 @@
// Shared correspondence between wire examples and schema definitions.
export const fixtureTypes = {
'printing-realtime-extension.discovery.json': 'PrintingRealtimeExtensionDiscovery',
'version-legacy.response.json': 'ProtocolVersionResponse',
'version-controls.response.json': 'ProtocolVersionResponse',
'control-poll.request.json': 'DeviceControlPollRequest',
'control-poll.response.json': 'DeviceControlPollResponse',
'control-disconnected.response.json': 'DeviceControlPollResponse',
'device-control.response.json': 'Device',
'action-pause-dispatch.request.json': 'AdminDeviceActionRequest',
'action-resume-dispatch.request.json': 'AdminDeviceActionRequest',
'action-disconnect.request.json': 'AdminDeviceActionRequest',
'action-reconnect.request.json': 'AdminDeviceActionRequest',
'action-revoke.request.json': 'AdminDeviceActionRequest',
'action-refresh-inventory.request.json': 'AdminDeviceActionRequest',
'action-clear-queue.request.json': 'AdminDeviceActionRequest',
'action-cancel-spool.request.json': 'AdminDeviceActionRequest',
'action-refresh.response.json': 'DeviceActionRecord',
'action-cancel.response.json': 'DeviceActionRecord',
'action-clear-queue.response.json': 'DeviceActionRecord',
'action-pause-dispatch.response.json': 'DeviceActionRecord',
'action-revoke.response.json': 'DeviceActionRecord',
'control-inventory-refreshed.receipt.json': 'DeviceCommandReceipt',
'control-cancel-requested.receipt.json': 'DeviceCommandReceipt',
'control-already-absent.receipt.json': 'DeviceCommandReceipt',
'control-ownership-mismatch.receipt.json': 'DeviceCommandReceipt',
'control-unsupported.receipt.json': 'DeviceCommandReceipt',
'control-failed.receipt.json': 'DeviceCommandReceipt',
'control-expired.receipt.json': 'DeviceCommandReceipt',
'control-outcome-unknown.receipt.json': 'DeviceCommandReceipt',
'control-commands-page.response.json': 'DeviceActionList',
'control-cancel-completed.response.json': 'DeviceActionRecord',
'heartbeat-profile.request.json': 'HeartbeatRequest',
'printers-profile.request.json': 'PrinterInventory',
'device-test.request.json': 'DeviceTestPrintRequest',
'device-test-profile.request.json': 'DeviceTestPrintRequest',
'device-retry.request.json': 'DeviceRetryRequest',
'create-enrollment.request.json': 'CreateEnrollmentRequest',
'enrollment.response.json': 'EnrollmentCode',
'enroll.request.json': 'EnrollRequest',
'enroll.response.json': 'EnrollResponse',
'heartbeat.request.json': 'HeartbeatRequest',
'heartbeat.response.json': 'HeartbeatResponse',
'printers.request.json': 'PrinterInventory',
'printers-page.response.json': 'PrinterList',
'approval.request.json': 'PrinterApproval',
'claim-empty.response.json': 'ClaimResponse',
'claim.response.json': 'ClaimResponse',
'job.response.json': 'PrintJob',
'submit.request.json': 'SubmitJobRequest',
'received.event.json': 'JobEvent',
'submitting.event.json': 'JobEvent',
'spooled.event.json': 'JobEvent',
'ack.response.json': 'Acknowledgment',
'error.response.json': 'Error',
'printing-session.response.json': 'PrintingSessionIdentity',
'printing-session-bound.response.json': 'PrintingSessionIdentity',
'realtime-settings.payload.json': 'RealtimeSettings',
'realtime-hello.message.json': 'RealtimeHello',
'realtime-welcome.message.json': 'RealtimeWelcome',
'realtime-state-applied.message.json': 'RealtimeStateApplied',
'realtime-desired-state.message.json': 'RealtimeDesiredState',
'realtime-job-available.message.json': 'RealtimeJobAvailable',
'realtime-command.message.json': 'RealtimeCommand',
'realtime-command-receipt.message.json': 'RealtimeCommandReceipt',
'realtime-job-event.message.json': 'RealtimeJobEvent',
'realtime-job-event-replay.message.json': 'RealtimeJobEvent',
'realtime-settings.message.json': 'RealtimeSettingsMessage',
'realtime-ack.message.json': 'RealtimeAck',
'realtime-error.message.json': 'RealtimeError',
'realtime-heartbeat.message.json': 'RealtimeHeartbeat',
'realtime-inventory.message.json': 'RealtimeInventory',
'realtime-ping.message.json': 'RealtimePing',
'realtime-pong.message.json': 'RealtimePong',
'realtime-large-pdf.response.json': 'PrintJob',
'printer-media-profile.response.json': 'PrinterMediaProfile',
'media-profile-options.request.json': 'PrintOptions',
'operator-printers.response.json': 'OperatorPrinterList',
'operator-submit.request.json': 'OperatorSubmitJobRequest',
'realtime-media-hello.message.json': 'RealtimeHello'
"printing-realtime-extension.discovery.json": "PrintingRealtimeExtensionDiscovery",
"version-legacy.response.json": "ProtocolVersionResponse",
"version-controls.response.json": "ProtocolVersionResponse",
"control-poll.request.json": "DeviceControlPollRequest",
"control-poll.response.json": "DeviceControlPollResponse",
"control-disconnected.response.json": "DeviceControlPollResponse",
"device-control.response.json": "Device",
"action-pause-dispatch.request.json": "AdminDeviceActionRequest",
"action-resume-dispatch.request.json": "AdminDeviceActionRequest",
"action-disconnect.request.json": "AdminDeviceActionRequest",
"action-reconnect.request.json": "AdminDeviceActionRequest",
"action-revoke.request.json": "AdminDeviceActionRequest",
"action-refresh-inventory.request.json": "AdminDeviceActionRequest",
"action-clear-queue.request.json": "AdminDeviceActionRequest",
"action-cancel-spool.request.json": "AdminDeviceActionRequest",
"action-refresh.response.json": "DeviceActionRecord",
"action-cancel.response.json": "DeviceActionRecord",
"action-clear-queue.response.json": "DeviceActionRecord",
"action-pause-dispatch.response.json": "DeviceActionRecord",
"action-revoke.response.json": "DeviceActionRecord",
"control-inventory-refreshed.receipt.json": "DeviceCommandReceipt",
"control-cancel-requested.receipt.json": "DeviceCommandReceipt",
"control-already-absent.receipt.json": "DeviceCommandReceipt",
"control-ownership-mismatch.receipt.json": "DeviceCommandReceipt",
"control-unsupported.receipt.json": "DeviceCommandReceipt",
"control-failed.receipt.json": "DeviceCommandReceipt",
"control-expired.receipt.json": "DeviceCommandReceipt",
"control-outcome-unknown.receipt.json": "DeviceCommandReceipt",
"control-commands-page.response.json": "DeviceActionList",
"control-cancel-completed.response.json": "DeviceActionRecord",
"heartbeat-profile.request.json": "HeartbeatRequest",
"printers-profile.request.json": "PrinterInventory",
"device-test.request.json": "DeviceTestPrintRequest",
"device-test-profile.request.json": "DeviceTestPrintRequest",
"device-test-profile-4x1.request.json": "DeviceTestPrintRequest",
"device-retry.request.json": "DeviceRetryRequest",
"create-enrollment.request.json": "CreateEnrollmentRequest",
"enrollment.response.json": "EnrollmentCode",
"enroll.request.json": "EnrollRequest",
"enroll.response.json": "EnrollResponse",
"heartbeat.request.json": "HeartbeatRequest",
"heartbeat.response.json": "HeartbeatResponse",
"printers.request.json": "PrinterInventory",
"printers-page.response.json": "PrinterList",
"approval.request.json": "PrinterApproval",
"claim-empty.response.json": "ClaimResponse",
"claim.response.json": "ClaimResponse",
"job.response.json": "PrintJob",
"submit.request.json": "SubmitJobRequest",
"received.event.json": "JobEvent",
"submitting.event.json": "JobEvent",
"spooled.event.json": "JobEvent",
"ack.response.json": "Acknowledgment",
"error.response.json": "Error",
"printing-session.response.json": "PrintingSessionIdentity",
"printing-session-bound.response.json": "PrintingSessionIdentity",
"realtime-settings.payload.json": "RealtimeSettings",
"realtime-hello.message.json": "RealtimeHello",
"realtime-welcome.message.json": "RealtimeWelcome",
"realtime-state-applied.message.json": "RealtimeStateApplied",
"realtime-desired-state.message.json": "RealtimeDesiredState",
"realtime-job-available.message.json": "RealtimeJobAvailable",
"realtime-command.message.json": "RealtimeCommand",
"realtime-command-receipt.message.json": "RealtimeCommandReceipt",
"realtime-job-event.message.json": "RealtimeJobEvent",
"realtime-job-event-replay.message.json": "RealtimeJobEvent",
"realtime-settings.message.json": "RealtimeSettingsMessage",
"realtime-ack.message.json": "RealtimeAck",
"realtime-error.message.json": "RealtimeError",
"realtime-heartbeat.message.json": "RealtimeHeartbeat",
"realtime-inventory.message.json": "RealtimeInventory",
"realtime-ping.message.json": "RealtimePing",
"realtime-pong.message.json": "RealtimePong",
"realtime-large-pdf.response.json": "PrintJob",
"printer-media-profile.response.json": "PrinterMediaProfile",
"media-profile-options.request.json": "PrintOptions",
"operator-printers.response.json": "OperatorPrinterList",
"operator-submit.request.json": "OperatorSubmitJobRequest",
"realtime-media-hello.message.json": "RealtimeHello",
};
+177 -80
View File
@@ -1,148 +1,245 @@
// Copyright 2026 Kingsrook, LLC
// SPDX-License-Identifier: Apache-2.0
import assert from 'node:assert/strict';
import { createHash } from 'node:crypto';
import { readFileSync } from 'node:fs';
import { fixtureTypes } from './fixture-types.mjs';
import assert from "node:assert/strict";
import { createHash } from "node:crypto";
import { readFileSync } from "node:fs";
import { fixtureTypes } from "./fixture-types.mjs";
const schema = JSON.parse(readFileSync(new URL('../schemas/protocol.schema.json', import.meta.url), 'utf8'));
const root = new URL('../fixtures/', import.meta.url);
/** @typedef {{ $ref?:string,not?:Rule|boolean,required?:string[],type?:string,anyOf?:(Rule|boolean)[],const?:unknown,enum?:unknown[],additionalProperties?:boolean,properties?:Record<string,Rule|boolean>,dependentRequired?:Record<string,string[]>,minItems?:number,maxItems?:number,uniqueItems?:boolean,contains?:Rule|boolean,items?:Rule|boolean,minLength?:number,maxLength?:number,pattern?:string,format?:string,contentEncoding?:string,minimum?:number,maximum?:number,exclusiveMinimum?:number }} Rule */
/** @type {{$defs:Record<string,Rule>}} */
const schema = JSON.parse(readFileSync(new URL("../schemas/protocol.schema.json", import.meta.url), "utf8"));
const root = new URL("../fixtures/", import.meta.url);
// Validate the schema keywords used by this contract without installing dependencies.
/** @param {unknown} value @param {Rule|boolean} rule @param {string} path @returns {void} */
function validate(value, rule, path)
{
if(typeof rule === 'boolean')
if(typeof rule === "boolean")
{
assert(rule, `${path}: forbidden property`);
return;
}
if(rule.$ref)
{
return validate(value, schema.$defs[rule.$ref.split('/').at(-1)], path);
return validate(value, /** @type {Rule} */ (schema.$defs[/** @type {string} */ (rule.$ref.split("/").at(-1))]), path);
}
if(rule.not)
{
let matches = true;
try { validate(value, rule.not, path); }
catch { matches = false; }
try
{
validate(value, rule.not, path);
}
catch
{
matches = false;
}
assert(!matches, `${path}: forbidden combination`);
}
if(rule.required && !rule.type)
{
for(const name of rule.required) assert(name in value, `${path}.${name}: required`);
for(const name of rule.required)
{
assert(name in /** @type {object} */ (value), `${path}.${name}: required`);
}
}
if(rule.anyOf)
{
assert(rule.anyOf.some(candidate =>
{
try { validate(value, candidate, path); return true; }
catch { return false; }
try
{
validate(value, candidate, path); return true;
}
catch
{
return false;
}
}), `${path}: no anyOf variant matched`);
}
if(rule.const !== undefined) assert.deepEqual(value, rule.const, path);
if(rule.enum) assert(rule.enum.includes(value), `${path}: invalid enum`);
if(rule.type === 'null') assert.equal(value, null, path);
if(rule.type === 'object')
if(rule.const !== undefined)
{
assert(value !== null && typeof value === 'object' && !Array.isArray(value), `${path}: expected object`);
for(const name of rule.required ?? []) assert(name in value, `${path}.${name}: required`);
assert.deepEqual(value, rule.const, path);
}
if(rule.enum)
{
assert(rule.enum.includes(value), `${path}: invalid enum`);
}
if(rule.type === "null")
{
assert.equal(value, null, path);
}
if(rule.type === "object")
{
assert(value !== null && typeof value === "object" && !Array.isArray(value), `${path}: expected object`);
for(const name of rule.required ?? [])
{
assert(name in /** @type {object} */ (value), `${path}.${name}: required`);
}
if(rule.additionalProperties === false)
{
for(const name of Object.keys(value)) assert(name in (rule.properties ?? {}), `${path}.${name}: unknown property`);
for(const name of Object.keys(value))
{
assert(name in (rule.properties ?? {}), `${path}.${name}: unknown property`);
}
}
for(const [name, child] of Object.entries(rule.properties ?? {}))
{
if(name in value) validate(value[name], child, `${path}.${name}`);
if(name in /** @type {object} */ (value))
{
validate(/** @type {Record<string,unknown>} */ (value)[name], child, `${path}.${name}`);
}
}
for(const [name, dependencies] of Object.entries(rule.dependentRequired ?? {}))
{
if(name in value) for(const dependency of dependencies) assert(dependency in value, `${path}.${dependency}: required with ${name}`);
if(name in /** @type {object} */ (value))
{
for(const dependency of dependencies)
{
assert(dependency in value, `${path}.${dependency}: required with ${name}`);
}
}
}
}
if(rule.type === 'array')
if(rule.type === "array")
{
assert(Array.isArray(value), `${path}: expected array`);
if(rule.minItems !== undefined) assert(value.length >= rule.minItems, path);
if(rule.maxItems !== undefined) assert(value.length <= rule.maxItems, path);
if(rule.uniqueItems) assert.equal(new Set(value.map(item => JSON.stringify(item))).size, value.length, path);
if(rule.minItems !== undefined)
{
assert(value.length >= rule.minItems, path);
}
if(rule.maxItems !== undefined)
{
assert(value.length <= rule.maxItems, path);
}
if(rule.uniqueItems)
{
assert.equal(new Set(value.map(item => JSON.stringify(item))).size, value.length, path);
}
if(rule.contains)
{
assert(value.some(item =>
{
try { validate(item, rule.contains, path); return true; }
catch { return false; }
try
{
validate(item, /** @type {Rule|boolean} */ (rule.contains), path); return true;
}
catch
{
return false;
}
}), `${path}: required array member missing`);
}
value.forEach((item, index) => validate(item, rule.items, `${path}[${index}]`));
value.forEach((item, index) => validate(item, /** @type {Rule|boolean} */ (rule.items), `${path}[${index}]`));
}
if(rule.type === 'string')
if(rule.type === "string")
{
assert.equal(typeof value, 'string', path);
if(rule.minLength !== undefined) assert(value.length >= rule.minLength, path);
if(rule.maxLength !== undefined) assert(value.length <= rule.maxLength, path);
if(rule.pattern) assert(new RegExp(rule.pattern).test(value), path);
if(rule.format === 'date-time') assert(!Number.isNaN(Date.parse(value)) && value.endsWith('Z'), `${path}: UTC date-time required`);
if(rule.contentEncoding === 'base64') assert.equal(Buffer.from(value, 'base64').toString('base64'), value, path);
assert.equal(typeof value, "string", path);
if(rule.minLength !== undefined)
{
assert((/** @type {string} */ (value)).length >= rule.minLength, path);
}
if(rule.maxLength !== undefined)
{
assert((/** @type {string} */ (value)).length <= rule.maxLength, path);
}
if(rule.pattern)
{
assert(new RegExp(rule.pattern).test(/** @type {string} */ (value)), path);
}
if(rule.format === "date-time")
{
assert(!Number.isNaN(Date.parse(/** @type {string} */ (value))) && (/** @type {string} */ (value)).endsWith("Z"), `${path}: UTC date-time required`);
}
if(rule.contentEncoding === "base64")
{
assert.equal(Buffer.from(/** @type {string} */ (value), "base64").toString("base64"), value, path);
}
}
if(rule.type === 'boolean') assert.equal(typeof value, 'boolean', path);
if(rule.type === 'integer' || rule.type === 'number')
if(rule.type === "boolean")
{
assert.equal(typeof value, 'number', path);
assert.equal(typeof value, "boolean", path);
}
if(rule.type === "integer" || rule.type === "number")
{
assert.equal(typeof value, "number", path);
assert(Number.isFinite(value), path);
if(rule.type === 'integer') assert(Number.isInteger(value), path);
if(rule.minimum !== undefined) assert(value >= rule.minimum, path);
if(rule.maximum !== undefined) assert(value <= rule.maximum, path);
if(rule.exclusiveMinimum !== undefined) assert(value > rule.exclusiveMinimum, path);
if(rule.type === "integer")
{
assert(Number.isInteger(value), path);
}
if(rule.minimum !== undefined)
{
assert(/** @type {number} */ (value) >= rule.minimum, path);
}
if(rule.maximum !== undefined)
{
assert(/** @type {number} */ (value) <= rule.maximum, path);
}
if(rule.exclusiveMinimum !== undefined)
{
assert(/** @type {number} */ (value) > rule.exclusiveMinimum, path);
}
}
}
/** @type {Record<string,object>} */
const fixtures = {};
for(const [name, type] of Object.entries(fixtureTypes))
{
fixtures[name] = JSON.parse(readFileSync(new URL(name, root), 'utf8'));
validate(fixtures[name], schema.$defs[type], name);
fixtures[name] = JSON.parse(readFileSync(new URL(name, root), "utf8"));
validate(fixtures[name], /** @type {Rule} */ (schema.$defs[type]), name);
}
const content = readFileSync(new URL('test-label.zpl', root));
const job = fixtures['claim.response.json'].job;
const content = readFileSync(new URL("test-label.zpl", root));
const job = (/** @type {typeof import("../fixtures/claim.response.json")} */ (fixtures["claim.response.json"])).job;
assert.equal(content.length, job.contentLength);
assert.equal(createHash('sha256').update(content).digest('hex'), job.contentSha256);
assert.deepEqual(Buffer.from(fixtures['submit.request.json'].contentBase64, 'base64'), content);
assert.deepEqual(fixtures['job.response.json'], job);
assert.throws(() => validate({ copies: 0 }, schema.$defs.PrintOptions, 'invalidCopies'));
assert.throws(() => validate({ paperWidthMm: 100 }, schema.$defs.PrintOptions, 'missingHeight'));
assert.throws(() => validate({ eventId: 'test', status: 'PRINTED' }, schema.$defs.JobEvent, 'invalidStatus'));
assert.throws(() => validate({ name: 'x'.repeat(513) }, schema.$defs.DeviceProfile, 'oversizedName'));
assert.throws(() => validate({ description: 'x'.repeat(4097) }, schema.$defs.DeviceProfile, 'oversizedDescription'));
assert.throws(() => validate({ location: 'x'.repeat(513) }, schema.$defs.DeviceProfile, 'oversizedLocation'));
validate({ name: '', description: null }, schema.$defs.DeviceProfile, 'clearAndPreserve');
const brandedTest = fixtures['device-test-profile.request.json'];
assert.throws(() => validate({ ...brandedTest, testProfile: 'LETTER' }, schema.$defs.DeviceTestPrintRequest, 'incompatibleLabelMedia'));
assert.throws(() => validate({ ...brandedTest, testProfile: 'UNKNOWN' }, schema.$defs.DeviceTestPrintRequest, 'unknownTestProfile'));
assert.throws(() => validate({ ...brandedTest, testProfile: null }, schema.$defs.DeviceTestPrintRequest, 'nullTestProfile'));
assert.throws(() => validate({ format: 'PDF' }, schema.$defs.DeviceTestPrintRequest, 'missingTestIdentity'));
validate({ ...brandedTest, format: 'PDF', testProfile: 'A4' }, schema.$defs.DeviceTestPrintRequest, 'a4TestPage');
const clear = fixtures['action-clear-queue.request.json'];
assert.throws(() => validate({ ...clear, issuedAt: undefined }, schema.$defs.AdminDeviceActionRequest, 'missingIssuedAt'));
assert.throws(() => validate({ ...clear, jobId: 'job-example-1' }, schema.$defs.AdminDeviceActionRequest, 'clearCannotTargetSpool'));
assert.throws(() => validate({ ...clear, action: 'RESTART_SERVICE' }, schema.$defs.AdminDeviceActionRequest, 'arbitraryServiceControl'));
assert.throws(() => validate({ ...clear, command: 'shell' }, schema.$defs.AdminDeviceActionRequest, 'unknownCommandProperty'));
assert.throws(() => validate({ ...clear, action: 'CANCEL_SPOOL' }, schema.$defs.AdminDeviceActionRequest, 'cancelRequiresJobIdentity'));
assert.throws(() => validate({ ...fixtures['control-poll.request.json'], capabilities: ['CANCEL_SPOOL', 'CANCEL_SPOOL'] }, schema.$defs.DeviceControlPollRequest, 'duplicateCapability'));
assert.throws(() => validate({ ...fixtures['control-cancel-requested.receipt.json'], outcome: 'PRINT_RECALLED' }, schema.$defs.DeviceCommandReceipt, 'unsupportedPaperClaim'));
assert.equal(createHash("sha256").update(content).digest("hex"), job.contentSha256);
assert.deepEqual(Buffer.from((/** @type {typeof import("../fixtures/submit.request.json")} */ (fixtures["submit.request.json"])).contentBase64, "base64"), content);
assert.deepEqual(fixtures["job.response.json"], job);
assert.throws(() => validate({ copies: 0 }, /** @type {Rule} */ (schema.$defs["PrintOptions"]), "invalidCopies"));
assert.throws(() => validate({ paperWidthMm: 100 }, /** @type {Rule} */ (schema.$defs["PrintOptions"]), "missingHeight"));
assert.throws(() => validate({ eventId: "test", status: "PRINTED" }, /** @type {Rule} */ (schema.$defs["JobEvent"]), "invalidStatus"));
assert.throws(() => validate({ name: "x".repeat(513) }, /** @type {Rule} */ (schema.$defs["DeviceProfile"]), "oversizedName"));
assert.throws(() => validate({ description: "x".repeat(4097) }, /** @type {Rule} */ (schema.$defs["DeviceProfile"]), "oversizedDescription"));
assert.throws(() => validate({ location: "x".repeat(513) }, /** @type {Rule} */ (schema.$defs["DeviceProfile"]), "oversizedLocation"));
validate({ name: "", description: null }, /** @type {Rule} */ (schema.$defs["DeviceProfile"]), "clearAndPreserve");
const brandedTest = fixtures["device-test-profile.request.json"];
assert.throws(() => validate({ ...brandedTest, testProfile: "LETTER" }, /** @type {Rule} */ (schema.$defs["DeviceTestPrintRequest"]), "incompatibleLabelMedia"));
assert.throws(() => validate({ ...brandedTest, testProfile: "UNKNOWN" }, /** @type {Rule} */ (schema.$defs["DeviceTestPrintRequest"]), "unknownTestProfile"));
assert.throws(() => validate({ ...brandedTest, testProfile: null }, /** @type {Rule} */ (schema.$defs["DeviceTestPrintRequest"]), "nullTestProfile"));
assert.throws(() => validate({ format: "PDF" }, /** @type {Rule} */ (schema.$defs["DeviceTestPrintRequest"]), "missingTestIdentity"));
validate({ ...brandedTest, format: "PDF", testProfile: "A4" }, /** @type {Rule} */ (schema.$defs["DeviceTestPrintRequest"]), "a4TestPage");
for(const testProfile of ["LABEL_4X2_203", "LABEL_4X2_300", "LABEL_4X2_600", "LABEL_4X1_203", "LABEL_4X1_300", "LABEL_4X1_600"])
{
validate({ ...brandedTest, testProfile }, /** @type {Rule} */ (schema.$defs["DeviceTestPrintRequest"]), `zplShortLabel${testProfile}`);
validate({ ...brandedTest, format: "PDF", testProfile }, /** @type {Rule} */ (schema.$defs["DeviceTestPrintRequest"]), `pdfShortLabel${testProfile}`);
}
assert.throws(() => validate({ ...brandedTest, testProfile: "LABEL_4X3_300" }, /** @type {Rule} */ (schema.$defs["DeviceTestPrintRequest"]), "unsupportedLabelHeight"));
// Short PDF labels use explicit actual-size job paper, never a conflicting media selection.
validate({ copies: 1, scale: "ACTUAL_SIZE", paperWidthMm: 101.6, paperHeightMm: 25.4 }, /** @type {Rule} */ (schema.$defs["PrintOptions"]), "fourByOneJobPaper");
validate({ copies: 1, scale: "ACTUAL_SIZE", paperWidthMm: 101.6, paperHeightMm: 50.8 }, /** @type {Rule} */ (schema.$defs["PrintOptions"]), "fourByTwoJobPaper");
const clear = fixtures["action-clear-queue.request.json"];
assert.throws(() => validate({ ...clear, issuedAt: undefined }, /** @type {Rule} */ (schema.$defs["AdminDeviceActionRequest"]), "missingIssuedAt"));
assert.throws(() => validate({ ...clear, jobId: "job-example-1" }, /** @type {Rule} */ (schema.$defs["AdminDeviceActionRequest"]), "clearCannotTargetSpool"));
assert.throws(() => validate({ ...clear, action: "RESTART_SERVICE" }, /** @type {Rule} */ (schema.$defs["AdminDeviceActionRequest"]), "arbitraryServiceControl"));
assert.throws(() => validate({ ...clear, command: "shell" }, /** @type {Rule} */ (schema.$defs["AdminDeviceActionRequest"]), "unknownCommandProperty"));
assert.throws(() => validate({ ...clear, action: "CANCEL_SPOOL" }, /** @type {Rule} */ (schema.$defs["AdminDeviceActionRequest"]), "cancelRequiresJobIdentity"));
assert.throws(() => validate({ ...fixtures["control-poll.request.json"], capabilities: ["CANCEL_SPOOL", "CANCEL_SPOOL"] }, /** @type {Rule} */ (schema.$defs["DeviceControlPollRequest"]), "duplicateCapability"));
assert.throws(() => validate({ ...fixtures["control-cancel-requested.receipt.json"], outcome: "PRINT_RECALLED" }, /** @type {Rule} */ (schema.$defs["DeviceCommandReceipt"]), "unsupportedPaperClaim"));
console.log(`Validated ${Object.keys(fixtures).length} protocol fixtures, content integrity, and invalid-input checks.`);
const media = fixtures['printer-media-profile.response.json'];
for(const invalid of [{ dpiX: 0 }, { widthMicrons: 0 }, { revision: 'not-a-revision' }, { applyMode: 'CHANGE_STOCK' }])
const media = fixtures["printer-media-profile.response.json"];
for(const invalid of [{ dpiX: 0 }, { widthMicrons: 0 }, { revision: "not-a-revision" }, { applyMode: "CHANGE_STOCK" }])
{
assert.throws(() => validate({ ...media, ...invalid }, schema.$defs.PrinterMediaProfile, 'invalidMedia'));
assert.throws(() => validate({ ...media, ...invalid }, /** @type {Rule} */ (schema.$defs["PrinterMediaProfile"]), "invalidMedia"));
}
const selection = fixtures['media-profile-options.request.json'];
assert.throws(() => validate({ ...selection, paperWidthMm: 100, paperHeightMm: 150 }, schema.$defs.PrintOptions, 'ambiguousMedia'));
const selection = fixtures["media-profile-options.request.json"];
assert.throws(() => validate({ ...selection, paperWidthMm: 100, paperHeightMm: 150 }, /** @type {Rule} */ (schema.$defs["PrintOptions"]), "ambiguousMedia"));
const operatorRequest = fixtures['operator-submit.request.json'];
for(const invalid of [{ requestId: 'x'.repeat(129) }, { requestId: 'echo/scan' }, { actorId: 'forged' }, { tenantId: 'forged' }])
const operatorRequest = fixtures["operator-submit.request.json"];
for(const invalid of [{ requestId: "x".repeat(129) }, { requestId: "echo/scan" }, { actorId: "forged" }, { tenantId: "forged" }])
{
assert.throws(() => validate({ ...operatorRequest, ...invalid }, schema.$defs.OperatorSubmitJobRequest, 'invalidOperatorRequest'));
assert.throws(() => validate({ ...operatorRequest, ...invalid }, /** @type {Rule} */ (schema.$defs["OperatorSubmitJobRequest"]), "invalidOperatorRequest"));
}
validate({ ...operatorRequest, requestId: 'x'.repeat(128) }, schema.$defs.OperatorSubmitJobRequest, 'maximumRequestKey');
validate({ ...operatorRequest, requestId: "x".repeat(128) }, /** @type {Rule} */ (schema.$defs["OperatorSubmitJobRequest"]), "maximumRequestKey");
+65 -55
View File
@@ -1,62 +1,72 @@
// Copyright 2026 Kingsrook, LLC
// SPDX-License-Identifier: Apache-2.0
import assert from 'node:assert/strict';
import { readFileSync } from 'node:fs';
import Ajv2020 from 'ajv/dist/2020.js';
import addFormats from 'ajv-formats';
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import { Ajv2020 } from "ajv/dist/2020.js";
import addFormats from "ajv-formats";
const schema = JSON.parse(readFileSync(new URL('../schemas/protocol.schema.json', import.meta.url), 'utf8'));
const schema = JSON.parse(readFileSync(new URL("../schemas/protocol.schema.json", import.meta.url), "utf8"));
const ajv = new Ajv2020({ strict: true, allErrors: true });
addFormats(ajv);
// ajv-formats is callable CommonJS; its NodeNext declaration exposes a namespace.
/** @type {import("ajv-formats").FormatsPlugin} */ (/** @type {unknown} */ (addFormats))(ajv);
ajv.addSchema(schema);
const fixture = name => JSON.parse(readFileSync(new URL(`../fixtures/${name}`, import.meta.url), 'utf8'));
/** @param {string} name @returns {Record<string,unknown>} */
const fixture = name => JSON.parse(readFileSync(new URL(`../fixtures/${name}`, import.meta.url), "utf8"));
/** @param {string} type @param {unknown} value @returns {boolean|Promise<unknown>} */
const valid = (type, value) => ajv.validate(`${schema.$id}#/$defs/${type}`, value);
const settings = fixture('realtime-settings.payload.json');
const hello = fixture('realtime-hello.message.json');
const extension = fixture('printing-realtime-extension.discovery.json');
assert(valid('PrintingRealtimeExtensionDiscovery', extension), ajv.errorsText());
assert.equal(extension.version, '2.0.0');
assert(valid('PrintingRealtimeExtensionDiscovery', { ...extension, version: '2.7.3' }));
for(const version of ['1.0.0', '1.1.0', '3.0.0', '2.0', '2.01.0', '2.0.0-rc.1', null])
const settings = /** @type {typeof import("../fixtures/realtime-settings.payload.json")} */ (/** @type {unknown} */ (fixture("realtime-settings.payload.json")));
const hello = /** @type {typeof import("../fixtures/realtime-hello.message.json")} */ (/** @type {unknown} */ (fixture("realtime-hello.message.json")));
const extension = fixture("printing-realtime-extension.discovery.json");
assert(valid("PrintingRealtimeExtensionDiscovery", extension), ajv.errorsText());
assert.equal(extension["version"], "2.0.0");
assert(valid("PrintingRealtimeExtensionDiscovery", { ...extension, version: "2.7.3" }));
for(const version of ["1.0.0", "1.1.0", "3.0.0", "2.0", "2.01.0", "2.0.0-rc.1", null])
{
assert(!valid('PrintingRealtimeExtensionDiscovery', { ...extension, version }), 'Only the compatible stable native major is accepted.');
assert(!valid("PrintingRealtimeExtensionDiscovery", { ...extension, version }), "Only the compatible stable native major is accepted.");
}
assert(!valid('PrintingRealtimeExtensionDiscovery', { ...extension, name: 'another-extension' }));
assert(!valid('PrintingRealtimeExtensionDiscovery', { ...extension, pathPrefix: '/printing/v1' }));
assert(!valid("PrintingRealtimeExtensionDiscovery", { ...extension, name: "another-extension" }));
assert(!valid("PrintingRealtimeExtensionDiscovery", { ...extension, pathPrefix: "/printing/v1" }));
/*******************************************************************************
** Cross-field policy relationships require semantic checks beyond JSON Schema.
*******************************************************************************/
/** @param {typeof settings} policy @param {typeof hello.payload.limits} limits @returns {void} */
function assertPolicy(policy, limits)
{
assert(valid('RealtimeSettings', policy), ajv.errorsText());
assert(valid('RealtimeClientLimits', limits), ajv.errorsText());
assert(policy.maxQueuedFrameBytes >= policy.maxFrameBytes, 'Frame queue must hold one permitted message.');
assert(policy.maxStagedBytes >= Math.max(policy.maxPdfBytes, policy.maxZplBytes), 'Staging must admit one permitted document.');
assert(valid("RealtimeSettings", policy), ajv.errorsText());
assert(valid("RealtimeClientLimits", limits), ajv.errorsText());
assert(policy.maxQueuedFrameBytes >= policy.maxFrameBytes, "Frame queue must hold one permitted message.");
assert(policy.maxStagedBytes >= Math.max(policy.maxPdfBytes, policy.maxZplBytes), "Staging must admit one permitted document.");
assert(policy.downloadTimeoutMillis >= policy.downloadConnectTimeoutMillis);
assert(policy.downloadTimeoutMillis >= policy.downloadIdleTimeoutMillis);
assert(policy.heartbeatTimeoutSeconds >= 2 * policy.heartbeatIntervalSeconds);
assert(policy.reconnectMaxDelayMillis >= policy.reconnectMinDelayMillis);
for(const name of schema.$defs.RealtimeClientLimits.required)
for(const name of /** @type {(keyof typeof hello.payload.limits)[]} */ (schema.$defs.RealtimeClientLimits.required))
{
assert(policy[name] <= limits[name], `${name}: policy exceeds declared client capability`);
}
}
assertPolicy(settings, hello.payload.limits);
assert.equal(settings.maxConcurrentDownloads, 1, 'Initial deployment defaults to one configurable download lane.');
assert.equal(settings.maxConcurrentDownloads, 1, "Initial deployment defaults to one configurable download lane.");
assert.equal(settings.maxCopies, 100);
assert.equal(settings.nativeSubmissionTimeoutMillis, 120000);
for(const [name, rule] of Object.entries(schema.$defs.RealtimeSettings.properties))
for(const [name, rule] of Object.entries(/** @type {Record<keyof typeof settings,{default?:number,minimum:number,maximum:number}>} */ (schema.$defs.RealtimeSettings.properties)))
{
if(rule.default !== undefined) assert.equal(settings[name], rule.default, `${name}: fixture/schema default mismatch`);
if(rule.default !== undefined)
{
assert.equal(settings[/** @type {keyof typeof settings} */ (name)], rule.default, `${name}: fixture/schema default mismatch`);
}
/** @type {Partial<typeof settings>} */
const missing = { ...settings };
delete missing[name];
assert(!valid('RealtimeSettings', missing), `${name}: missing value cannot inherit a default`);
// Negative evidence deliberately removes each required schema field.
// eslint-disable-next-line @typescript-eslint/no-dynamic-delete
delete missing[/** @type {keyof typeof settings} */ (name)];
assert(!valid("RealtimeSettings", missing), `${name}: missing value cannot inherit a default`);
for(const value of [rule.minimum - 1, rule.maximum + 1, 1.5, null])
{
assert(!valid('RealtimeSettings', { ...settings, [name]: value }), `${name}: invalid bound/type accepted`);
assert(!valid("RealtimeSettings", { ...settings, [name]: value }), `${name}: invalid bound/type accepted`);
}
}
for(const change of [
@@ -64,50 +74,50 @@ for(const change of [
{ maxStagedBytes: 1 },
{ downloadTimeoutMillis: 1000 },
{ heartbeatTimeoutSeconds: 10 },
{ reconnectMinDelayMillis: 60000, reconnectMaxDelayMillis: 1000 }
{ reconnectMinDelayMillis: 60000, reconnectMaxDelayMillis: 1000 },
])
{
assert.throws(() => assertPolicy({ ...settings, ...change }, hello.payload.limits));
}
for(const name of schema.$defs.RealtimeClientLimits.required)
for(const name of /** @type {(keyof typeof hello.payload.limits)[]} */ (schema.$defs.RealtimeClientLimits.required))
{
assert.throws(() => assertPolicy(settings, { ...hello.payload.limits, [name]: settings[name] - 1 }), name);
assert.throws(() => assertPolicy(settings, { ...hello.payload.limits, [name]: settings[/** @type {keyof typeof settings} */ (name)] - 1 }), name);
}
assertPolicy({ ...settings, maxConcurrentDownloads: 16, maxConcurrentPrints: 16, maxCopies: 32767 }, hello.payload.limits);
assert(!valid('RealtimeHello', { ...hello, payload: { ...hello.payload, capabilities: [] } }));
assert(!valid('RealtimeHello', { ...hello, payload: { ...hello.payload, capabilities: ['CANCEL_SPOOL'] } }));
assert(!valid('RealtimeHello', { ...hello, payload: { ...hello.payload, limits: {} } }));
assert(!valid('RealtimeHello', { ...hello, messageId: 'hello:unsafe-prefix' }));
assert(!valid('RealtimeHello', { ...hello, token: 'not-a-real-token' }));
assert(!valid('RealtimeClientMessage', { type: 'RUN_SHELL', messageId: 'unsupported-1', payload: {} }));
assert(!valid("RealtimeHello", { ...hello, payload: { ...hello.payload, capabilities: [] } }));
assert(!valid("RealtimeHello", { ...hello, payload: { ...hello.payload, capabilities: ["CANCEL_SPOOL"] } }));
assert(!valid("RealtimeHello", { ...hello, payload: { ...hello.payload, limits: {} } }));
assert(!valid("RealtimeHello", { ...hello, messageId: "hello:unsafe-prefix" }));
assert(!valid("RealtimeHello", { ...hello, token: "not-a-real-token" }));
assert(!valid("RealtimeClientMessage", { type: "RUN_SHELL", messageId: "unsupported-1", payload: {} }));
const clientFixtures = ['hello', 'state-applied', 'job-event', 'command-receipt', 'heartbeat', 'inventory', 'ping'];
const serverFixtures = ['welcome', 'desired-state', 'job-available', 'command', 'settings', 'ack', 'error', 'pong'];
for(const [files, accepted, rejected] of [
[clientFixtures, 'RealtimeClientMessage', 'RealtimeServerMessage'],
[serverFixtures, 'RealtimeServerMessage', 'RealtimeClientMessage']
])
const clientFixtures = ["hello", "state-applied", "job-event", "command-receipt", "heartbeat", "inventory", "ping"];
const serverFixtures = ["welcome", "desired-state", "job-available", "command", "settings", "ack", "error", "pong"];
for(const [files, accepted, rejected] of /** @type {[string[],string,string][]} */ ([
[clientFixtures, "RealtimeClientMessage", "RealtimeServerMessage"],
[serverFixtures, "RealtimeServerMessage", "RealtimeClientMessage"],
]))
{
for(const name of files)
{
const message = fixture(`realtime-${name}.message.json`);
assert(valid(accepted, message), `${name}: ${ajv.errorsText()}`);
assert(!valid(rejected, message), `${name}: wrong-direction frame accepted`);
assert(Buffer.byteLength(JSON.stringify(message), 'utf8') <= settings.maxFrameBytes);
assert(Buffer.byteLength(JSON.stringify(message), "utf8") <= settings.maxFrameBytes);
const missing = { ...message };
delete missing.payload;
delete missing["payload"];
assert(!valid(accepted, missing), `${name}: missing payload accepted`);
assert(!valid(accepted, { ...message, contentUrl: 'https://unrelated.invalid/document' }));
assert(!valid(accepted, { ...message, contentUrl: "https://unrelated.invalid/document" }));
}
}
const event = fixture('realtime-job-event.message.json');
assert.deepEqual(fixture('realtime-job-event-replay.message.json'), event, 'Lost ACK replay preserves the exact envelope.');
assert.equal(fixture('realtime-ack.message.json').payload.acknowledgedMessageId, event.messageId);
assert.equal(fixture('realtime-error.message.json').messageId, event.messageId, 'Frame errors retain request correlation.');
const largeJob = fixture('realtime-large-pdf.response.json');
assert(valid('PrintJob', largeJob), ajv.errorsText());
const event = fixture("realtime-job-event.message.json");
assert.deepEqual(fixture("realtime-job-event-replay.message.json"), event, "Lost ACK replay preserves the exact envelope.");
assert.equal((/** @type {typeof import("../fixtures/realtime-ack.message.json")} */ (/** @type {unknown} */ (fixture("realtime-ack.message.json")))).payload.acknowledgedMessageId, event["messageId"]);
assert.equal(fixture("realtime-error.message.json")["messageId"], event["messageId"], "Frame errors retain request correlation.");
const largeJob = /** @type {typeof import("../fixtures/realtime-large-pdf.response.json")} */ (/** @type {unknown} */ (fixture("realtime-large-pdf.response.json")));
assert(valid("PrintJob", largeJob), ajv.errorsText());
assert.equal(largeJob.contentLength, settings.maxPdfBytes);
assert(largeJob.contentLength > 20 * 1024 * 1024);
assert(valid('PrintOptions', { copies: 32767 }));
assert(!valid('PrintOptions', { copies: 32768 }));
console.log('Realtime directional frames, defaults/ranges, policy/ceiling refusal and exact replay fixtures passed; runtime concurrency acceptance remains consumer-owned.');
assert(valid("PrintOptions", { copies: 32767 }));
assert(!valid("PrintOptions", { copies: 32768 }));
console.log("Realtime directional frames, defaults/ranges, policy/ceiling refusal and exact replay fixtures passed; runtime concurrency acceptance remains consumer-owned.");
+80 -72
View File
@@ -1,16 +1,17 @@
// Copyright 2026 Kingsrook, LLC
// SPDX-License-Identifier: Apache-2.0
import assert from 'node:assert/strict';
import { readFileSync } from 'node:fs';
import Ajv2020 from 'ajv/dist/2020.js';
import addFormats from 'ajv-formats';
import { fixtureTypes } from './fixture-types.mjs';
import './verify-realtime.mjs';
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import { Ajv2020 } from "ajv/dist/2020.js";
import addFormats from "ajv-formats";
import { fixtureTypes } from "./fixture-types.mjs";
import "./verify-realtime.mjs";
const schema = JSON.parse(readFileSync(new URL('../schemas/protocol.schema.json', import.meta.url), 'utf8'));
const schema = JSON.parse(readFileSync(new URL("../schemas/protocol.schema.json", import.meta.url), "utf8"));
const ajv = new Ajv2020({ strict: true, allErrors: true });
addFormats(ajv);
// ajv-formats is callable CommonJS; its NodeNext declaration exposes a namespace.
/** @type {import("ajv-formats").FormatsPlugin} */ (/** @type {unknown} */ (addFormats))(ajv);
assert(ajv.validateSchema(schema), ajv.errorsText());
ajv.addSchema(schema);
@@ -21,91 +22,98 @@ for(const name of Object.keys(schema.$defs))
}
for(const [file, type] of Object.entries(fixtureTypes))
{
const value = JSON.parse(readFileSync(new URL(`../fixtures/${file}`, import.meta.url), 'utf8'));
const value = JSON.parse(readFileSync(new URL(`../fixtures/${file}`, import.meta.url), "utf8"));
const validate = ajv.compile({ $ref: `${schema.$id}#/$defs/${type}` });
assert(validate(value), `${file}: ${ajv.errorsText(validate.errors)}`);
}
const validateTest = ajv.compile({ $ref: `${schema.$id}#/$defs/DeviceTestPrintRequest` });
const testRequest = { printerId: 'printer-zebra', format: 'ZPL', requestId: 'test-1' };
assert(validateTest(testRequest), 'Existing requests without a profile remain valid.');
assert(validateTest({ ...testRequest, testProfile: 'LABEL_4X6_300' }));
assert(!validateTest({ ...testRequest, testProfile: 'LETTER' }));
const testRequest = { printerId: "printer-zebra", format: "ZPL", requestId: "test-1" };
assert(validateTest(testRequest), "Existing requests without a profile remain valid.");
assert(validateTest({ ...testRequest, testProfile: "LABEL_4X6_300" }));
assert(validateTest({ ...testRequest, testProfile: "LABEL_4X1_203" }));
assert(validateTest({ ...testRequest, testProfile: "LABEL_4X2_600" }));
assert(validateTest({ ...testRequest, format: "PDF", testProfile: "LABEL_4X2_300" }));
assert(!validateTest({ ...testRequest, testProfile: "LABEL_4X3_300" }));
assert(!validateTest({ ...testRequest, testProfile: "LETTER" }));
assert(!validateTest({ ...testRequest, testProfile: null }));
assert(!validateTest({ format: 'PDF' }));
assert(!validateTest({ format: "PDF" }));
const validatePrinters = ajv.compile({ $ref: `${schema.$id}#/$defs/PrinterList` });
const printerPage = JSON.parse(readFileSync(new URL('../fixtures/printers-page.response.json', import.meta.url), 'utf8'));
/** @type {typeof import("../fixtures/printers-page.response.json")} */
const printerPage = JSON.parse(readFileSync(new URL("../fixtures/printers-page.response.json", import.meta.url), "utf8"));
assert(validatePrinters(printerPage));
assert(validatePrinters({ printers: printerPage.printers }), 'Older responses without a cursor remain valid.');
assert(validatePrinters({ ...printerPage, nextCursor: 'opaque-server-cursor' }));
assert(validatePrinters({ printers: printerPage.printers }), "Older responses without a cursor remain valid.");
assert(validatePrinters({ ...printerPage, nextCursor: "opaque-server-cursor" }));
assert(!validatePrinters({ ...printerPage, nextCursor: 17 }));
assert(!validatePrinters({ ...printerPage, nextCursor: 'x'.repeat(12001) }));
assert(!validatePrinters({ ...printerPage, nextCursor: "x".repeat(12001) }));
assert(!validatePrinters({ printers: Array(201).fill(printerPage.printers[0]) }));
const control = name => JSON.parse(readFileSync(new URL(`../fixtures/${name}`, import.meta.url), 'utf8'));
/** @param {string} name @returns {Record<string,unknown>} */
const control = name => JSON.parse(readFileSync(new URL(`../fixtures/${name}`, import.meta.url), "utf8"));
/** @param {string} type @param {unknown} value @returns {boolean|Promise<unknown>} */
const validControl = (type, value) => ajv.validate(`${schema.$id}#/$defs/${type}`, value);
assert(!validControl('PrintingSessionIdentity', { tenantId: 'test-tenant' }), 'Display or device identity cannot replace the authoritative principal.');
assert(!validControl('PrintingSessionIdentity', { principalId: 'test-operator', tenantId: '' }));
assert(!validControl('PrintingSessionIdentity', { principalId: 'test-operator', tenantId: 'test-tenant', deviceId: null }));
assert(validControl('PrintingSessionIdentity', { principalId: 'test-operator', tenantId: 'test-tenant', futureResponseField: true }));
const action = control('action-clear-queue.request.json');
assert(validControl('AdminDeviceActionRequest', action));
assert(!validControl("PrintingSessionIdentity", { tenantId: "test-tenant" }), "Display or device identity cannot replace the authoritative principal.");
assert(!validControl("PrintingSessionIdentity", { principalId: "test-operator", tenantId: "" }));
assert(!validControl("PrintingSessionIdentity", { principalId: "test-operator", tenantId: "test-tenant", deviceId: null }));
assert(validControl("PrintingSessionIdentity", { principalId: "test-operator", tenantId: "test-tenant", futureResponseField: true }));
const action = control("action-clear-queue.request.json");
assert(validControl("AdminDeviceActionRequest", action));
const withoutIssuedAt = { ...action };
delete withoutIssuedAt.issuedAt;
assert(!validControl('AdminDeviceActionRequest', withoutIssuedAt), 'New controls require their durable freshness timestamp.');
delete withoutIssuedAt["issuedAt"];
assert(!validControl("AdminDeviceActionRequest", withoutIssuedAt), "New controls require their durable freshness timestamp.");
for(const invalid of [
{ issuedAt: '2026-10-31T16:00:00-05:00' },
{ issuedAt: "2026-10-31T16:00:00-05:00" },
{ issuedAt: null },
{ action: 'RESTART_SERVICE' },
{ action: 'CANCEL_SPOOL' },
{ jobId: 'job-example-1' },
{ spoolerJobId: '17' },
{ queueName: 'Some other queue' },
{ command: 'shell' },
{ requestId: '../device' }
{ action: "RESTART_SERVICE" },
{ action: "CANCEL_SPOOL" },
{ jobId: "job-example-1" },
{ spoolerJobId: "17" },
{ queueName: "Some other queue" },
{ command: "shell" },
{ requestId: "../device" },
])
{
assert(!validControl('AdminDeviceActionRequest', { ...action, ...invalid }), JSON.stringify(invalid));
assert(!validControl("AdminDeviceActionRequest", { ...action, ...invalid }), JSON.stringify(invalid));
}
const poll = control('control-poll.request.json');
assert(!validControl('DeviceControlPollRequest', { ...poll, controlVersion: 2 }));
assert(!validControl('DeviceControlPollRequest', { ...poll, capabilities: ['CANCEL_SPOOL', 'CANCEL_SPOOL'] }));
assert(!validControl('DeviceControlPollRequest', { ...poll, capabilities: ['RESTART_SERVICE'] }));
assert(!validControl('DeviceControlPollRequest', { ...poll, lastAppliedRevision: -1 }));
assert(!validControl('DeviceControlPollRequest', { ...poll, lastAppliedRevision: 9007199254740992 }));
const pollResult = control('control-poll.response.json');
assert(validControl('DeviceControlPollResponse', { ...pollResult, futureResponseField: true }), 'Additive response fields remain compatible.');
assert(!validControl('DeviceControlPollResponse', { ...pollResult, commands: Array(21).fill(pollResult.commands[0]) }));
assert(!validControl('DeviceControlPollResponse', { ...pollResult, desiredState: { ...pollResult.desiredState, connectionState: 'REVOKED' } }), 'Revocation is credential invalidation, not reversible desired state.');
assert(!validControl('DeviceCommand', { ...pollResult.commands[1], jobId: null }));
assert(!validControl('DeviceCommand', { ...pollResult.commands[0], jobId: 'job-example-1' }));
assert(!validControl('DeviceCommand', { ...pollResult.commands[0], action: 'CLEAR_QUEUE' }), 'Server queue clear is never an arbitrary client command.');
const receipt = control('control-cancel-requested.receipt.json');
assert(!validControl('DeviceCommandReceipt', { ...receipt, outcome: 'PRINT_RECALLED' }));
assert(!validControl('DeviceCommandReceipt', { ...receipt, message: 'x'.repeat(1025) }));
assert(!validControl('DeviceCommandReceipt', { ...receipt, completedAt: 'not-a-date' }));
assert(!validControl('DeviceCommandReceipt', { ...receipt, jobId: 'other-job' }), 'Receipt cannot override its route-bound command target.');
const history = control('control-commands-page.response.json');
assert(!validControl('DeviceActionList', { ...history, commands: Array(201).fill(history.commands[0]) }));
assert(!validControl('DeviceActionList', { ...history, nextCursor: 'x'.repeat(12001) }));
const poll = control("control-poll.request.json");
assert(!validControl("DeviceControlPollRequest", { ...poll, controlVersion: 2 }));
assert(!validControl("DeviceControlPollRequest", { ...poll, capabilities: ["CANCEL_SPOOL", "CANCEL_SPOOL"] }));
assert(!validControl("DeviceControlPollRequest", { ...poll, capabilities: ["RESTART_SERVICE"] }));
assert(!validControl("DeviceControlPollRequest", { ...poll, lastAppliedRevision: -1 }));
assert(!validControl("DeviceControlPollRequest", { ...poll, lastAppliedRevision: 9007199254740992 }));
const pollResult = /** @type {typeof import("../fixtures/control-poll.response.json")} */ (/** @type {unknown} */ (control("control-poll.response.json")));
assert(validControl("DeviceControlPollResponse", { ...pollResult, futureResponseField: true }), "Additive response fields remain compatible.");
assert(!validControl("DeviceControlPollResponse", { ...pollResult, commands: Array(21).fill(pollResult.commands[0]) }));
assert(!validControl("DeviceControlPollResponse", { ...pollResult, desiredState: { ...pollResult.desiredState, connectionState: "REVOKED" } }), "Revocation is credential invalidation, not reversible desired state.");
assert(!validControl("DeviceCommand", { ...pollResult.commands[1], jobId: null }));
assert(!validControl("DeviceCommand", { ...pollResult.commands[0], jobId: "job-example-1" }));
assert(!validControl("DeviceCommand", { ...pollResult.commands[0], action: "CLEAR_QUEUE" }), "Server queue clear is never an arbitrary client command.");
const receipt = control("control-cancel-requested.receipt.json");
assert(!validControl("DeviceCommandReceipt", { ...receipt, outcome: "PRINT_RECALLED" }));
assert(!validControl("DeviceCommandReceipt", { ...receipt, message: "x".repeat(1025) }));
assert(!validControl("DeviceCommandReceipt", { ...receipt, completedAt: "not-a-date" }));
assert(!validControl("DeviceCommandReceipt", { ...receipt, jobId: "other-job" }), "Receipt cannot override its route-bound command target.");
const history = /** @type {typeof import("../fixtures/control-commands-page.response.json")} */ (/** @type {unknown} */ (control("control-commands-page.response.json")));
assert(!validControl("DeviceActionList", { ...history, commands: Array(201).fill(history.commands[0]) }));
assert(!validControl("DeviceActionList", { ...history, nextCursor: "x".repeat(12001) }));
console.log(`Compiled ${Object.keys(schema.$defs).length} JSON Schema 2020-12 definitions; ${Object.keys(fixtureTypes).length} fixtures passed full schema validation.`);
const media = control('printer-media-profile.response.json');
for(const invalid of [{ dpiX: 0 }, { widthMicrons: 0 }, { revision: 'not-a-revision' }, { applyMode: 'CHANGE_STOCK' }])
const media = control("printer-media-profile.response.json");
for(const invalid of [{ dpiX: 0 }, { widthMicrons: 0 }, { revision: "not-a-revision" }, { applyMode: "CHANGE_STOCK" }])
{
assert(!validControl('PrinterMediaProfile', { ...media, ...invalid }));
assert(!validControl("PrinterMediaProfile", { ...media, ...invalid }));
}
const selection = control('media-profile-options.request.json');
assert(!validControl('PrintOptions', { ...selection, paperWidthMm: 100, paperHeightMm: 150 }));
assert(!validControl('PrinterMediaSelection', { ...selection.mediaProfile, futureOption: true }));
const selection = /** @type {typeof import("../fixtures/media-profile-options.request.json")} */ (/** @type {unknown} */ (control("media-profile-options.request.json")));
assert(!validControl("PrintOptions", { ...selection, paperWidthMm: 100, paperHeightMm: 150 }));
assert(!validControl("PrinterMediaSelection", { ...selection.mediaProfile, futureOption: true }));
const operatorRequest = control('operator-submit.request.json');
for(const invalid of [{ requestId: 'x'.repeat(129) }, { requestId: 'echo/scan' }, { actorId: 'forged' }, { tenantId: 'forged' }])
const operatorRequest = control("operator-submit.request.json");
for(const invalid of [{ requestId: "x".repeat(129) }, { requestId: "echo/scan" }, { actorId: "forged" }, { tenantId: "forged" }])
{
assert(!validControl('OperatorSubmitJobRequest', { ...operatorRequest, ...invalid }));
assert(!validControl("OperatorSubmitJobRequest", { ...operatorRequest, ...invalid }));
}
assert(validControl('OperatorSubmitJobRequest', { ...operatorRequest, requestId: 'x'.repeat(128) }));
assert(validControl("OperatorSubmitJobRequest", { ...operatorRequest, requestId: "x".repeat(128) }));
const catalogEntry = control('operator-printers.response.json').printers[0];
assert(validControl('OperatorPrinter', { ...catalogEntry, deviceName: 'x'.repeat(1024) }));
assert(!validControl('OperatorPrinter', { ...catalogEntry, deviceName: 'x'.repeat(1025) }));
assert(validControl('PrinterMediaProfile', { ...media, id: 'x'.repeat(128), futureMetadata: { supported: true } }));
assert(!validControl('PrinterMediaProfile', { ...media, id: 'x'.repeat(129) }));
const catalogEntry = (/** @type {typeof import("../fixtures/operator-printers.response.json")} */ (/** @type {unknown} */ (control("operator-printers.response.json")))).printers[0];
assert(validControl("OperatorPrinter", { ...catalogEntry, deviceName: "x".repeat(1024) }));
assert(!validControl("OperatorPrinter", { ...catalogEntry, deviceName: "x".repeat(1025) }));
assert(validControl("PrinterMediaProfile", { ...media, id: "x".repeat(128), futureMetadata: { supported: true } }));
assert(!validControl("PrinterMediaProfile", { ...media, id: "x".repeat(129) }));
+45
View File
@@ -0,0 +1,45 @@
{
"$schema": "https://json.schemastore.org/tsconfig",
"compilerOptions": {
"target": "ES2024",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"lib": [
"ES2024"
],
"strict": true,
"noUncheckedIndexedAccess": true,
"exactOptionalPropertyTypes": true,
"noImplicitOverride": true,
"noImplicitReturns": true,
"noFallthroughCasesInSwitch": true,
"noUnusedLocals": true,
"noUnusedParameters": true,
"noPropertyAccessFromIndexSignature": true,
"useUnknownInCatchVariables": true,
"verbatimModuleSyntax": true,
"isolatedModules": true,
"forceConsistentCasingInFileNames": true,
"skipLibCheck": true,
"declaration": true,
"sourceMap": true,
"newLine": "lf",
"resolveJsonModule": true,
"allowJs": true,
"checkJs": true,
"noEmit": true,
"types": [
"node"
]
},
"include": [
"tools/**/*.mjs",
"eslint.config.mjs"
],
"exclude": [
"node_modules",
"dist",
"build",
"coverage"
]
}
Generated
+43
View File
@@ -0,0 +1,43 @@
version = 1
revision = 5
requires-python = ">=3.13"
[[package]]
name = "qqq-print-protocol-script-quality"
version = "0.0.0"
source = { virtual = "." }
[package.dev-dependencies]
dev = [
{ name = "ruff" },
]
[package.metadata]
[package.metadata.requires-dev]
dev = [{ name = "ruff", specifier = "==0.17.3" }]
[[package]]
name = "ruff"
version = "0.17.3"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/59/73/81f140c4edd9cc323de5478e6bbd3bbcfae9a82692df1a258de8cec6a6d2/ruff-0.17.3.tar.gz", hash = "sha256:a57049c8addb3662073859bb55a7ad35aa54f8a7c9ed268911d1333c5aadaaa9", size = 5339444, upload-time = "2026-10-27T00:34:05.914Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/9e/8c/083365310c4e79c842f25bf5a842facbd9b0c0ec336f96dffa5074fe723d/ruff-0.17.3-py3-none-linux_armv6l.whl", hash = "sha256:ab86b84cd9d05e56d329742c0cabdcbe2d1d56c98f4a04aab9b5b66196febfb4", size = 10216522, upload-time = "2026-10-27T00:33:29.822Z" },
{ url = "https://files.pythonhosted.org/packages/72/26/4b2edf31cfebe9605d8f6ed8f1d6f376d3a53d4c9a123e62661c64c440f0/ruff-0.17.3-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:e407b4a6852853a9a2cdce96963e94ad9775578bb44e7561a0f2829f3cc04b11", size = 10217786, upload-time = "2026-10-27T00:33:33.949Z" },
{ url = "https://files.pythonhosted.org/packages/a2/7e/537aba4b7f54397a374d2f6a6fc93e9c8015e18553f31a4a7b7dbced48b5/ruff-0.17.3-py3-none-macosx_11_0_arm64.whl", hash = "sha256:b75cc113b3aca388933d8296b667b1ac401a80317ea3074c9b0a66541199da29", size = 9440362, upload-time = "2026-10-27T00:33:35.784Z" },
{ url = "https://files.pythonhosted.org/packages/3d/66/283e39734f66c8a5ebaa93c22eb7b7515779a0b1d5a5c0ad20e05c0edb41/ruff-0.17.3-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:ab5fa7ed5ce18554ea69d850ad162cbaa0c85fc47c08d83a569f806681c9a6ad", size = 10094281, upload-time = "2026-10-27T00:33:38.216Z" },
{ url = "https://files.pythonhosted.org/packages/21/4c/1ffc768497bd025a32c72152771e40c0b99bddc5cc6e4f50c7949a395346/ruff-0.17.3-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:c673c170e27834d1262e3a9c1e97bdc9b9c3b180c63d89173ce69747232474c1", size = 10022543, upload-time = "2026-10-27T00:33:40.181Z" },
{ url = "https://files.pythonhosted.org/packages/73/c3/f90db547301080fcb82c92abf914c8bd57b89911d9b65a6c4cc6b52d0635/ruff-0.17.3-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:6a754c0341bc9ca151b7b335988b3263921c08859f84ca219b70cee825571b6f", size = 10683921, upload-time = "2026-10-27T00:33:42.182Z" },
{ url = "https://files.pythonhosted.org/packages/e5/98/92d4a8a26c6c96cf1a5b34dd59036d8a71e52d650aab065c58867e964e63/ruff-0.17.3-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:4590b9d86c0b1543ebb09094254e191cd43fe8edaf330baa035aab03da251532", size = 11373408, upload-time = "2026-10-27T00:33:44.19Z" },
{ url = "https://files.pythonhosted.org/packages/43/02/1f24d2e3064c9753926deeb901afa57777b37ee46799b09ddd7aee6e5145/ruff-0.17.3-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:bbde18beccd98c29a227c63c99022e3443b99258e08326df6425dfe0443705ef", size = 10977469, upload-time = "2026-10-27T00:33:46.179Z" },
{ url = "https://files.pythonhosted.org/packages/f0/20/fecedea2209f574ab92033b344e01a7f0c2dc76c6d6d19a14c9f8141173f/ruff-0.17.3-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7fa4b0f931cdbcb8a52044f370a3eb429fb42eb88807b0e800fcaa98dd42b24e", size = 10577491, upload-time = "2026-10-27T00:33:48.233Z" },
{ url = "https://files.pythonhosted.org/packages/33/36/b93fdf0bc6f84f0cd487aadfb0354deb40013669ea35ac3d26855de1e959/ruff-0.17.3-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:39405f341deec934b4289634324ad695f72df2970f9121582872eaf5b722f0f5", size = 10608748, upload-time = "2026-10-27T00:33:50.284Z" },
{ url = "https://files.pythonhosted.org/packages/5d/b7/7d3f5f14145daead1efb0b9a8289ed1daa100f419d60774375951723ce00/ruff-0.17.3-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:68074094f55ca3d29ab19491f640282f7a84061c4c7977aa5094b968ca457e9a", size = 10075057, upload-time = "2026-10-27T00:33:52.074Z" },
{ url = "https://files.pythonhosted.org/packages/b0/ed/93ff7a5909ee56e2bd4b982deae9e0efb2724417874a90c9e41ad66c6dd7/ruff-0.17.3-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:27bb571ac78fef178fe5118d2827df95c2aae74a48fff2b15f6acd03af5ff5bb", size = 10022431, upload-time = "2026-10-27T00:33:54.061Z" },
{ url = "https://files.pythonhosted.org/packages/9b/07/a83b9eeec8c1c3c0e4eda15f06c3a91001bfd2687ea2f07b750719bf9712/ruff-0.17.3-py3-none-musllinux_1_2_i686.whl", hash = "sha256:029c37cbd3de48ca37aa12117bead2d065f15da17d66bf84e0d4274013705db2", size = 10394903, upload-time = "2026-10-27T00:33:56.158Z" },
{ url = "https://files.pythonhosted.org/packages/af/b2/05c39857c2526fcbebe46ad93a8dc97424656dfef14d3ef933abe1ffd331/ruff-0.17.3-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:692fdb46d0fd9c71e380947dc5fc72f2c3f881073bc1f739e4579376231affbe", size = 10799780, upload-time = "2026-10-27T00:33:58.204Z" },
{ url = "https://files.pythonhosted.org/packages/ec/d9/47171a16739eb7a353936b31c20cf54340531e68c9af7d72fd7a784e235d/ruff-0.17.3-py3-none-win32.whl", hash = "sha256:46539cb031b01fd29ae0a02142483c46640b233ee12bb1c1b690bffa1555d5ca", size = 9908481, upload-time = "2026-10-27T00:34:00.087Z" },
{ url = "https://files.pythonhosted.org/packages/d0/a0/f08c1631bfd81faa6a90eaf70c37c786c11bed2046302b56bb506c25e1a0/ruff-0.17.3-py3-none-win_amd64.whl", hash = "sha256:56b8ffb3f3b3cbccb8b76b71b38261b7ed982ef5d20f5b447b1b61a7f4949b24", size = 10341333, upload-time = "2026-10-27T00:34:02.099Z" },
{ url = "https://files.pythonhosted.org/packages/99/d5/4ad4ed31413c301430918e70d3d38ae6f097af0766cb9ab8c9d502b2a4d6/ruff-0.17.3-py3-none-win_arm64.whl", hash = "sha256:81fce74bcd6be0e9da7c67c7553fd0f1f12964a2052cd728ef917b937a5ca1d9", size = 9854332, upload-time = "2026-10-27T00:34:03.847Z" },
]