"""RED->GREEN guard for the js/node-usage-analyzer.js innerHTML sinks.

The analyzer writes pack data and workflow data into innerHTML through turbogrid
cells, a flyover, and the status/message elements. Before the fix: the title
column built an UNQUOTED href (a bare space breaks out of the attribute); the
install-progress showStatus interpolated a raw pack key; the usage-details
flyover interpolated a raw workflow filename; and the install error path
interpolated both a raw pack key and raw server text.

One value needed fixing at its SOURCE rather than at the sink. The title column
is built as `pack.title || packKey`: /customnode/getlist runs populate_markdown
over each pack, so `pack.title` is server-escaped, but `packKey` is a dict key
that no server transform touches. Escaping at the formatter would have
double-escaped the common path into visible entity text ("Nodes for <Flux>
workflows"); leaving it bare would have shipped the raw fallback. The fix
normalises the fallback so the column carries ONE provenance, which is what
makes the bare formatter correct by construction.

This grid uses cellResizeObserver only — no rowFilter, no
highlightKeywordsFilter — so the keyword-highlight re-render does not reach it.
The two searchable managers use js/manager-grid.js to secure that path without
modifying the vendored TurboGrid bundle.

The status/message sinks take DISPLAY-READY text and do NOT escape, because
their callers carry mixed provenance; every caller is classified in
tests/test_message_sink_provenance.py. This analyzer is the only file that
reaches BOTH sink implementations — the shared one from common.js via `this.ui`
and its own class methods.

REPRODUCE THE RED HALF against the pre-fix revision — one command:
  mkdir -p /some/dir
  git show <base>:js/node-usage-analyzer.js > /some/dir/node-usage-analyzer.js
  git show <base>:js/common.js             > /some/dir/common.js
  MANAGER_JS_DIR=/some/dir pytest tests/test_node_usage_escaping.py
"""
import html
import json
import os
import unittest
from pathlib import Path

from js_lift import (
    NODE,
    JsSource,
    event_handlers,
    parse,
    run_node,
    slice_braced,
)

REPO_ROOT = Path(__file__).resolve().parent.parent
JS = JsSource(os.environ.get("MANAGER_JS_DIR") or (REPO_ROOT / "js"))

ANALYZER = "node-usage-analyzer.js"
COMMON = "common.js"
TITLE_FORMATTER_ANCHOR = "formatter: function (name, rowItem, columnItem, cellNode)"

BREAKOUT_HREF = "https://evil.example/a onmouseover=alert(1) x"
QUOTE_BREAKOUT = 'https://evil.example/a"><img src=x onerror=alert(1)>'
# An ordinary pack reference, for the legs that ask what a CORRECT link
# looks like rather than what a hostile one cannot do.
BENIGN_REFERENCE = "https://github.com/owner/repo"
TEXT_PAYLOAD = "<img src=x onerror=alert(1)>"


def _url_helper_home() -> str:
    """Which file currently owns sanitizeUrl/safeHref, or '' on a pre-fix tree.

    They were local to the analyzer when this guard was written and now live in
    common.js, imported by every client file. Resolving the home rather than
    hard-coding it keeps the MANAGER_JS_DIR reproduce working against both
    revisions, and makes a THIRD home fail loudly instead of being lifted twice.
    """
    for filename in (COMMON, ANALYZER):
        if "SAFE_URL_SCHEMES" in JS.text(filename):
            return filename
    return ""


def _preamble() -> str:
    parts = [JS.lift_declaration(COMMON, "export function sanitizeHTML(")]
    home = _url_helper_home()
    if home:
        span = JS.lift_span(
            home, "SAFE_URL_SCHEMES", "safeHref = (url) => sanitizeHTML(sanitizeUrl(url));")
        parts.append("const " + span.replace("export ", ""))
    return "\n".join(parts)


@unittest.skipIf(NODE is None, "node is required to execute the lifted production JS")
class NodeUsageSinkEscapingTest(unittest.TestCase):
    """Every sink is exercised through the REAL lifted code, never a copy."""

    def _assert_inert(self, markup: str, allowed_tags):
        """No element outside the allow-list, and no event-handler attribute."""
        parsed = parse(markup)
        self.assertEqual(
            [t for t in parsed.tags if t not in allowed_tags], [],
            "live element injected into: %r" % (markup,),
        )
        self.assertEqual(
            event_handlers(markup), [],
            "event-handler attribute injected into: %r" % (markup,),
        )

    # --- C2: title column, previously UNQUOTED href -------------------------
    def _render_title(self, reference: str, name: str = "My Pack") -> str:
        return run_node(
            "%s\nconst titleFormatter = %s;\n"
            "console.log(JSON.stringify({markup: titleFormatter(%s, {reference: %s})}));\n"
            % (_preamble(),
               JS.lift_formatter(ANALYZER, TITLE_FORMATTER_ANCHOR),
               json.dumps(name), json.dumps(reference))
        )["markup"]

    def test_c2_href_survives_breakout_attempts(self):
        for reference in (BREAKOUT_HREF, QUOTE_BREAKOUT):
            markup = self._render_title(reference)
            self._assert_inert(markup, allowed_tags={"a", "b"})
            self.assertEqual(
                sorted(name for name, _ in parse(markup).attrs),
                ["href", "rel", "target"],
                "unexpected attributes on the anchor: %r" % (markup,),
            )

    def test_c2_title_anchor_severs_the_opener(self):
        """The pack chooses this host, so the page it opens must not keep us.

        `reference` is channel data, so the pack author picks the page that
        opens; without rel that page holds a live window.opener back into the
        Manager UI. Membership on the SPLIT attribute, not a literal match:
        'noreferrer noopener' is the same thing, and pinning token order would
        report a correct fix as a defect.
        """
        markup = self._render_title(BENIGN_REFERENCE)
        attrs = dict(parse(markup).attrs)
        self.assertEqual(
            attrs.get("target"), "_blank",
            "this guard is about _blank anchors; the target changed, so "
            "re-derive it rather than deleting the assertion: %r" % (markup,))
        self.assertIn(
            "rel", attrs,
            "the title anchor opens a pack-chosen page in a new tab without "
            "rel, so that page keeps a reference to this window: %r" % (markup,))
        tokens = attrs["rel"].lower().split()
        self.assertIn("noopener", tokens, markup)
        self.assertIn("noreferrer", tokens, markup)

    def test_c2_href_rejects_dangerous_schemes(self):
        for reference in ("javascript:alert(1)", "java\tscript:alert(1)",
                          "data:text/html,<script>alert(1)</script>"):
            href = dict(parse(self._render_title(reference)).attrs)["href"]
            self.assertEqual(href, "#", "%r survived as %r" % (reference, href))

    def test_c2_benign_reference_and_title_are_untouched(self):
        markup = self._render_title(BENIGN_REFERENCE, name="My Pack")
        self.assertEqual(dict(parse(markup).attrs)["href"], BENIGN_REFERENCE)
        self.assertIn("<b>My Pack</b>", markup)

    def test_c2_title_is_not_double_escaped_at_the_formatter(self):
        # The column value is already escaped HTML text by the time it arrives,
        # so re-escaping here would put entity text on screen.
        block = JS.lift_formatter(ANALYZER, TITLE_FORMATTER_ANCHOR)
        self.assertNotIn(
            "sanitizeHTML(String(name))", block,
            "title is normalised at its source; escaping it again renders '&lt;' to the user",
        )
        server_value = "Nodes for &lt;Flux&gt; workflows"
        markup = self._render_title("https://example.com/x", name=server_value)
        self.assertIn(server_value, markup)
        self.assertEqual(html.unescape(server_value), "Nodes for <Flux> workflows")

    def _run_panel(self, action, packs=None, status=500):
        """Run the production data, operation and rendering methods with stubbed I/O."""
        methods = ",\n".join(slice_braced(JS.text(ANALYZER), "\n\t" + marker) for marker in [
            "async loadData()", "async installModels(list, btn)", "async uninstallModels(list, btn)",
            "showUsageDetails(rowItem)", "showError(err)", "showMessage(msg, color)", "showStatus(msg, color)",
        ])
        return run_node("""
            %s
            %s
            %s
            const elements = {};
            const messages = {};
            const posts = [];
            const manager_instance = {datasrc_combo: {value: 'default'}};
            const fetchData = async () => ({data: {channel: 'default', node_packs: %s}});
            const analyzeWorkflowUsage = async () => ({success: true});
            const api = {fetchApi: async (url, options) => {
                if (options?.body) posts.push(JSON.parse(options.body));
                return {status: %d, json: async () => ({is_processing: false}), text: async () => %s};
            }};
            const customConfirm = async () => true;
            const md5 = () => 'hash';
            const show_message = message => messages.dialog = message;
            const btn = {classList: {add() {}, remove() {}}};
            const ctx = {
                %s,
                element: {querySelector: selector => elements[selector] ||= {innerHTML: '', style: {}}},
                grid: {scrollRowIntoView() {}, updateCell() {}, updateRow() {}},
                flyover: {show: (title, body) => Object.assign(messages, {title, body})},
                showLoading() {}, hideLoading() {}, renderGrid() {}, getModelList(rows) {return rows;}
            };
            ctx.ui = createUIStateManager(ctx.element, {
                status: '.nu-manager-status', message: '.nu-manager-message'
            });
            %s
            console.log(JSON.stringify({elements, messages, posts, rows: ctx.modelList}));
        """ % (_preamble(),
               JS.lift_declaration(COMMON, "export function createUIStateManager(element, selectors)"),
               JS.lift_declaration(COMMON, "export async function uninstallNodes(nodeList, options = {})"),
               json.dumps(packs or {}), status, json.dumps(TEXT_PAYLOAD), methods, action))

    def test_title_fallback_is_escaped_at_its_source(self):
        out = self._run_panel("await ctx.loadData();", {TEXT_PAYLOAD: {"state": "enabled"}})
        row = out["rows"][0]
        self._assert_inert(row["title"], allowed_tags=set())
        self.assertEqual(html.unescape(row["title"]), TEXT_PAYLOAD)
        self.assertEqual(row["name"], TEXT_PAYLOAD)

    def test_install_progress_and_errors_escape_raw_values(self):
        for status in (403, 500):
            with self.subTest(status=status):
                out = self._run_panel("await ctx.loadData(); await ctx.installModels(ctx.modelList, btn);",
                                      {TEXT_PAYLOAD: {"state": "enabled"}}, status)
                progress = out["elements"][".nu-manager-status"]["innerHTML"]
                self._assert_inert(progress, allowed_tags=set())
                self.assertEqual(html.unescape(progress), f"Install {TEXT_PAYLOAD} ...")
                errors = [out["elements"][".nu-manager-message"]["innerHTML"], out["messages"]["dialog"]]
                for error in errors:
                    self._assert_inert(error, allowed_tags={"font"})
                    self.assertIn(f"'{TEXT_PAYLOAD}': ", html.unescape(error))
                    if status == 500:
                        self.assertIn(": " + TEXT_PAYLOAD, html.unescape(error))

    def test_flyover_escapes_workflow_filename(self):
        for filename in (TEXT_PAYLOAD, "workflow <Flux> & friends.json"):
            with self.subTest(filename=filename):
                out = self._run_panel("ctx.showUsageDetails(%s);" % json.dumps({
                    "title": "My Pack", "workflowDetails": [{"filename": filename, "nodeCount": 2}],
                }))
                body = out["messages"]["body"]
                self._assert_inert(body, allowed_tags={"div"})
                self.assertIn(filename, html.unescape(body))
                self.assertIn("2 nodes", body)

    def test_uninstall_failure_keeps_normalized_titles_readable(self):
        for name in ("Pack <Flux> & friends", TEXT_PAYLOAD):
            for has_title in (False, True):
                for status in (403, 500):
                    with self.subTest(name=name, has_title=has_title, status=status):
                        pack = {"state": "enabled"}
                        if has_title:
                            pack["title"] = html.escape(name)
                        out = self._run_panel("await ctx.loadData(); await ctx.uninstallModels(ctx.modelList, btn);",
                                              {name: pack}, status)
                        progress = out["elements"][".nu-manager-status"]["innerHTML"]
                        self.assertEqual(html.unescape(progress), f"Uninstall {name} ...")
                        errors = [out["elements"][".nu-manager-message"]["innerHTML"], out["messages"]["dialog"]]
                        for error in errors:
                            self._assert_inert(error, allowed_tags={"font"})
                            self.assertIn(f"'{name}': ", html.unescape(error))
                            if status == 500:
                                self.assertIn(": " + TEXT_PAYLOAD, html.unescape(error))

    def test_uninstall_helper_escapes_a_raw_name_fallback(self):
        out = self._run_panel("""
            await uninstallNodes([{name: %s}], {
                onProgress: msg => messages.progress = msg,
                onError: msg => messages.error = msg
            });
        """ % json.dumps(TEXT_PAYLOAD))
        for message in out["messages"].values():
            self._assert_inert(message, allowed_tags=set())
            self.assertIn(TEXT_PAYLOAD, html.unescape(message))
        self.assertEqual(out["posts"][0]["name"], TEXT_PAYLOAD)

    def test_grid_has_no_keyword_highlight_path(self):
        # The premise the row states: this grid cannot hit the highlight defect.
        source = JS.text(ANALYZER)
        self.assertNotIn("highlightKeywordsFilter", source)
        self.assertNotIn("rowFilter:", source)


if __name__ == "__main__":
    unittest.main()