diff --git a/engine/overlay.py b/engine/overlay.py new file mode 100644 index 0000000..2065f70 --- /dev/null +++ b/engine/overlay.py @@ -0,0 +1,110 @@ +# engine/overlay.py +"""Resolve action names across commons + proprietaries. + +proprietaries first (override), commons second. Listing uses static ast scanning +(no import side effects). Dispatch loads exactly one module by path. +""" +from __future__ import annotations + +import ast +import importlib.util +import os +from dataclasses import dataclass +from typing import List, Optional + +ACTION_EXT = ".py" + + +@dataclass +class Resolved: + name: str + path: str + shadowed: Optional[str] # commons path when a proprietaries action overrides it + + +@dataclass +class ActionInfo: + name: str + source: str # "commons" | "local" | "override" + summary: str + disabled: object # None | True | "reason" + + +def _rel_names(root: Optional[str]): + if not root or not os.path.isdir(root): + return {} + out = {} + for dirpath, _dirs, files in os.walk(root): + for fn in files: + if not fn.endswith(ACTION_EXT): + continue + full = os.path.join(dirpath, fn) + rel = os.path.relpath(full, root)[: -len(ACTION_EXT)] + out[rel.replace(os.sep, "/")] = full + return out + + +def resolve(name: str, commons_root: str, prop_root: Optional[str]) -> Optional[Resolved]: + rel = name + ACTION_EXT + commons_path = os.path.join(commons_root, rel) + c = os.path.isfile(commons_path) + prop_path = os.path.join(prop_root, rel) if prop_root else None + p = bool(prop_path) and os.path.isfile(prop_path) + if p and c: + return Resolved(name, prop_path, commons_path) + if p: + return Resolved(name, prop_path, None) + if c: + return Resolved(name, commons_path, None) + return None + + +def discover(commons_root: str, prop_root: Optional[str]) -> List[ActionInfo]: + commons = _rel_names(commons_root) + prop = _rel_names(prop_root) + infos = [] + for name in sorted(set(commons) | set(prop)): + if name in prop and name in commons: + source, path = "override", prop[name] + elif name in prop: + source, path = "local", prop[name] + else: + source, path = "commons", commons[name] + summary, disabled = _scan(path) + infos.append(ActionInfo(name, source, summary, disabled)) + return infos + + +def _scan(path: str): + """Statically extract (summary, DISABLED) without importing the module.""" + try: + with open(path) as f: + tree = ast.parse(f.read(), path) + except (SyntaxError, OSError): + return "(unparseable)", None + summary, disabled = "", None + for node in tree.body: + if not isinstance(node, ast.Assign): + continue + targets = [t.id for t in node.targets if isinstance(t, ast.Name)] + if "METADATA" in targets and isinstance(node.value, ast.Call): + summary = _summary_of(node.value) + if "DISABLED" in targets and isinstance(node.value, ast.Constant): + disabled = node.value.value if node.value.value else None + return summary, disabled + + +def _summary_of(call: ast.Call) -> str: + for kw in call.keywords: + if kw.arg == "summary" and isinstance(kw.value, ast.Constant): + return kw.value.value + if call.args and isinstance(call.args[0], ast.Constant): + return call.args[0].value + return "" + + +def load_module(path: str, modname: str): + spec = importlib.util.spec_from_file_location(modname, path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod diff --git a/tests/test_overlay.py b/tests/test_overlay.py new file mode 100644 index 0000000..1b9993b --- /dev/null +++ b/tests/test_overlay.py @@ -0,0 +1,54 @@ +# tests/test_overlay.py +import os, tempfile, unittest +from engine import overlay + +def _write(root, rel, body): + p = os.path.join(root, rel) + os.makedirs(os.path.dirname(p), exist_ok=True) + with open(p, "w") as f: + f.write(body) + return p + +ACT = 'from engine.descriptor import Meta\nMETADATA = Meta(summary="{s}")\ndef run(ctx, args):\n return "{r}"\n' + +class TestOverlay(unittest.TestCase): + def setUp(self): + self.commons = tempfile.mkdtemp() + self.prop = tempfile.mkdtemp() + + def test_commons_only(self): + _write(self.commons, "configure/ufw.py", ACT.format(s="commons ufw", r="C")) + r = overlay.resolve("configure/ufw", self.commons, self.prop) + self.assertIsNotNone(r); self.assertIsNone(r.shadowed) + self.assertTrue(r.path.startswith(self.commons)) + + def test_override_exposes_shadowed(self): + _write(self.commons, "configure/ufw.py", ACT.format(s="commons", r="C")) + _write(self.prop, "configure/ufw.py", ACT.format(s="local", r="L")) + r = overlay.resolve("configure/ufw", self.commons, self.prop) + self.assertTrue(r.path.startswith(self.prop)) # local wins + self.assertTrue(r.shadowed.startswith(self.commons)) + + def test_unknown_returns_none(self): + self.assertIsNone(overlay.resolve("no/such", self.commons, self.prop)) + + def test_discover_marks_sources_and_disabled(self): + _write(self.commons, "configure/ufw.py", ACT.format(s="commons ufw", r="C")) + _write(self.commons, "sync/repository.py", ACT.format(s="sync repo", r="C")) + _write(self.prop, "configure/ufw.py", ACT.format(s="local ufw", r="L")) + _write(self.prop, "sync/minecraft.py", + 'from engine.descriptor import Meta\nMETADATA = Meta(summary="mc")\nDISABLED = "wip"\ndef run(ctx, args): pass\n') + infos = {i.name: i for i in overlay.discover(self.commons, self.prop)} + self.assertEqual(infos["configure/ufw"].source, "override") + self.assertEqual(infos["sync/repository"].source, "commons") + self.assertEqual(infos["sync/minecraft"].source, "local") + self.assertEqual(infos["sync/minecraft"].disabled, "wip") + self.assertEqual(infos["configure/ufw"].summary, "local ufw") + + def test_load_module_runs(self): + p = _write(self.commons, "x/y.py", ACT.format(s="s", r="Z")) + mod = overlay.load_module(p, "apex_action_x_y") + self.assertEqual(mod.run(None, None), "Z") + +if __name__ == "__main__": + unittest.main()