deerflow-code/offline-backend-20260512/backend/tests/test_weknora_embed_script.py

"""Execute the generated iframe guard with a deterministic DOM/event-loop model."""

from __future__ import annotations

import ast
import json
import re
import shutil
import subprocess
from pathlib import Path
from types import SimpleNamespace

import pytest


def _embed_script(source: str | None = None) -> str:
    path = Path(__file__).parents[1] / "app/gateway/routers/llmwiki.py"
    tree = ast.parse(source if source is not None else path.read_text(encoding="utf-8"))
    function = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "_weknora_embed_injection")
    # Compile the actual function without importing the Gateway and its runtime
    # settings; only its surrounding serialization helpers need tiny stand-ins.
    module = ast.Module(body=[ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0), function], type_ignores=[])
    namespace = {
        "_tenant_id_from_session": lambda _: "tenant-test",
        "_embed_allowed_weknora_ids": lambda ctx: [ctx.weknora_id],
        "_with_public_prefix": lambda prefix, path: prefix + path,
        "_WEKNORA_EMBED_API_PREFIX": "/api/llmwiki/weknora",
        "_escape_script_json": json.dumps,
    }
    exec(compile(ast.fix_missing_locations(module), str(path), "exec"), namespace)
    html = namespace["_weknora_embed_injection"](
        allowed_path="/platform/knowledge-bases/base-test",
        tab="documents",
        session={},
        ctx=SimpleNamespace(can_write=True, weknora_id="base-test"),
    )
    return re.search(r'<script id="deerflow-weknora-embed-guard">(.*?)</script>', html, re.DOTALL).group(1)


HARNESS = r"""
const vm = require('node:vm');
const assert = require('node:assert/strict');
const input = JSON.parse(require('node:fs').readFileSync(0, 'utf8'));
const observers = new Set();
const timers = new Map();
const events = new Map();
const metrics = { globalQueries: 0, subtreeQueries: 0, layoutReads: 0, styleWrites: 0, sourceAttempts: 0, clicks: 0 };
let now = 0, nextTimer = 0;
function changed(target, type, attributeName, addedNodes = []) {
  for (const observer of observers) {
    const { root, options } = observer;
    if (!root || (root !== target && (!options.subtree || !root.contains(target)))) continue;
    if (type === 'attributes' && (!options.attributes || (options.attributeFilter && !options.attributeFilter.includes(attributeName)))) continue;
    if (type === 'childList' && !options.childList) continue;
    observer.records.push({ target, type, attributeName, addedNodes });
  }
}
class Observer {
  constructor(callback) { this.callback = callback; this.records = []; observers.add(this); }
  observe(root, options) { this.root = root; this.options = options; }
  disconnect() { this.root = null; this.records = []; }
  takeRecords() { const records = this.records; this.records = []; return records; }
}
function flush() {
  for (let turn = 0; turn < 100; turn++) {
    const pending = [...observers].filter(observer => observer.records.length);
    if (!pending.length) return;
    for (const observer of pending) {
      const records = observer.takeRecords();
      if (records.length) observer.callback(records);
    }
  }
  throw new Error('MutationObserver failed to settle');
}
function timeout(callback, delay = 0, repeat = false) {
  const id = ++nextTimer;
  timers.set(id, { callback, due: now + delay, delay, repeat });
  return id;
}
function advance(duration) {
  const end = now + duration;
  flush();
  for (let steps = 0; steps < 10000; steps++) {
    const next = [...timers.entries()].filter(([, timer]) => timer.due <= end).sort((a, b) => a[1].due - b[1].due)[0];
    if (!next) { now = end; return; }
    const [id, timer] = next;
    now = timer.due;
    if (timer.repeat) timer.due += timer.delay;
    else timers.delete(id);
    timer.callback();
    flush();
  }
  throw new Error('Timer loop did not settle');
}
function attrName(key) { return 'data-' + key.replace(/[A-Z]/g, c => '-' + c.toLowerCase()); }
class Element {
  constructor(tag = 'div', attributes = {}) {
    this.nodeType = 1; this.tagName = tag; this.attrs = new Map(Object.entries(attributes)); this.children = []; this.parentElement = null; this.textContent = '';
    this.width = 1000; this.height = 800;
    this.dataset = new Proxy({}, {
      get: (_, key) => this.getAttribute(attrName(key)) ?? undefined,
      set: (_, key, value) => { this.setAttribute(attrName(key), value); return true; },
      deleteProperty: (_, key) => { this.removeAttribute(attrName(key)); return true; }
    });
    const values = new Map();
    this.style = new Proxy({
      getPropertyValue: key => values.get(key) || '',
      setProperty: (key, value) => { values.set(key, value); metrics.styleWrites++; changed(this, 'attributes', 'style'); }
    }, { get: (target, key) => key in target ? target[key] : values.get(key) || '', set: (_, key, value) => { values.set(key, value); metrics.styleWrites++; changed(this, 'attributes', 'style'); return true; } });
    this.classList = {
      contains: name => this.className.split(/\s+/).includes(name),
      add: name => { if (!this.classList.contains(name)) this.setAttribute('class', (this.className + ' ' + name).trim()); },
      remove: name => { if (this.classList.contains(name)) this.setAttribute('class', this.className.split(/\s+/).filter(value => value !== name).join(' ')); },
      toggle: (name, on) => on ? this.classList.add(name) : this.classList.remove(name)
    };
  }
  get className() { return this.getAttribute('class') || ''; }
  get src() { return this.getAttribute('src') || ''; }
  set src(value) { this.setAttribute('src', value); }
  getAttribute(name) { return this.attrs.get(name) ?? null; }
  setAttribute(name, value) { this.attrs.set(name, String(value)); changed(this, 'attributes', name); }
  removeAttribute(name) { if (this.attrs.delete(name)) changed(this, 'attributes', name); }
  contains(node) { return this === node || this.children.some(child => child.contains(node)); }
  appendChild(node) { node.parentElement = this; this.children.push(node); changed(this, 'childList', null, [node]); return node; }
  remove() { if (!this.parentElement) return; const parent = this.parentElement; parent.children = parent.children.filter(node => node !== this); this.parentElement = null; changed(parent, 'childList', null); }
  matches(selector) {
    return selector.split(',').some(part => {
      part = part.trim();
      if (part.includes('>')) return false;
      if (part.startsWith('.')) return this.classList.contains(part.slice(1));
      if (part.startsWith('#')) return this.getAttribute('id') === part.slice(1);
      const attribute = part.match(/^(?:([\w-]+))?\[([^\]=*^]+)(?:(\*=|\^=|=)["']?([^"'\]]*)["']?)?\]$/);
      if (attribute) {
        if (attribute[1] && attribute[1] !== this.tagName) return false;
        const value = this.getAttribute(attribute[2]);
        if (!attribute[3]) return value !== null;
        return value !== null && (attribute[3] === '*=' ? value.includes(attribute[4]) : attribute[3] === '^=' ? value.startsWith(attribute[4]) : value === attribute[4]);
      }
      return this.tagName === part;
    });
  }
  closest(selector) { for (let node = this; node; node = node.parentElement) if (node.matches(selector)) return node; return null; }
  querySelectorAll(selector) { metrics.subtreeQueries++; const result = []; const walk = node => { for (const child of node.children) { if (child.matches(selector)) result.push(child); walk(child); } }; walk(this); return result; }
  querySelector(selector) { return this.querySelectorAll(selector)[0] || null; }
  getBoundingClientRect() { metrics.layoutReads++; return { left: 0, right: this.width, width: this.width, height: this.height }; }
  click() { metrics.clicks++; this.onClick?.(); }
}
const root = new Element('html');
const body = root.appendChild(new Element('body'));
const app = body.appendChild(new Element('div', { id: 'app' }));
const main = app.appendChild(new Element('div', { class: 'main' }));
const outer = main.appendChild(new Element('aside', { class: 'sidebar' })); outer.width = 200;
const content = main.appendChild(new Element('div', { class: 'knowledge-layout' }));
const inner = content.appendChild(new Element('aside', { class: 'sidebar wiki-directory' })); inner.width = 200;
const document = {
  documentElement: root, body, readyState: 'complete',
  querySelectorAll: selector => { metrics.globalQueries++; return root.querySelectorAll(selector); },
  querySelector: selector => { metrics.globalQueries++; if (selector === '[data-id="missing"]') metrics.sourceAttempts++; return root.querySelector(selector); },
  getElementById: id => root.querySelector('#' + id),
  addEventListener() {}, contains: node => root.contains(node)
};
const storage = () => { const values = new Map(); return { getItem: key => values.get(key) ?? null, setItem: (key, value) => values.set(key, String(value)) }; };
const url = new URL('http://iframe.test/platform/knowledge-bases/base-test?theme=light' + (input.case.startsWith('source') ? '&documentId=' + (input.case === 'source-success' ? 'found' : 'missing') : ''));
const context = {
  document, location: url, history: { pushState() {}, replaceState() {} },
  localStorage: storage(), sessionStorage: storage(), MutationObserver: Observer,
  Node: { ELEMENT_NODE: 1 }, URL, URLSearchParams, Request, XMLHttpRequest: function () {},
  innerWidth: 1000, innerHeight: 800,
  getComputedStyle: node => ({ position: node.style.position || 'fixed' }),
  setTimeout: (callback, delay) => timeout(callback, delay), clearTimeout: id => timers.delete(id),
  setInterval: (callback, delay) => timeout(callback, delay, true), clearInterval: id => timers.delete(id),
  addEventListener: (type, callback) => { if (!events.has(type)) events.set(type, []); events.get(type).push(callback); },
  fetch: async () => ({}), CSS: { escape: value => value }
};
context.XMLHttpRequest.prototype.open = () => {};
context.window = context;
if (input.case === 'source-success') {
  const row = content.appendChild(new Element('button', { 'data-document-id': 'found' }));
  row.onClick = () => { const drawer = body.appendChild(new Element('div', { class: 't-drawer' })); drawer.textContent = '基本信息 文档内容'; };
}
vm.runInNewContext(input.script, context, { timeout: 2000 });
advance(1000);
const settled = { ...metrics };
if (input.case === 'idle' || input.case === 'baseline') {
  advance(5000);
  const delta = Object.fromEntries(Object.keys(metrics).map(key => [key, metrics[key] - settled[key]]));
  if (input.case === 'idle') {
    assert.equal(delta.globalQueries, 0, 'idle guard must stop full-document scans');
    assert.equal(delta.layoutReads, 0, 'idle guard must stop forcing layout');
    assert.equal(delta.styleWrites, 0, 'idle theme must stop rewriting CSS');
    assert.equal(timers.size, 0, 'ordinary loaded pages need no recurring timers');
  }
  process.stdout.write(JSON.stringify(delta));
} else if (input.case === 'theme') {
  const message = { type: 'deerflow-weknora-theme', theme: 'dark', vars: { primary: '#123456', background: '#010203' } };
  for (const callback of events.get('message')) callback({ data: message });
  advance(500);
  assert.equal(root.getAttribute('theme-mode'), 'dark');
  assert.equal(root.style.getPropertyValue('--td-brand-color'), '#123456');
  root.setAttribute('data-theme', 'light');
  root.classList.remove('dark');
  root.style.setProperty('--td-brand-color', 'red');
  advance(500);
  assert.equal(root.getAttribute('data-theme'), 'dark');
  assert.equal(root.classList.contains('dark'), true);
  assert.equal(root.style.getPropertyValue('--td-brand-color'), '#123456');
  const count = metrics.styleWrites;
  advance(5000);
  assert.equal(metrics.styleWrites, count);
  assert.equal(metrics.globalQueries, settled.globalQueries, 'root theme changes must not scan application content');
} else if (input.case === 'batch') {
  const globalBefore = metrics.globalQueries;
  const scansBefore = metrics.subtreeQueries;
  const section = content.appendChild(new Element('section'));
  for (let index = 0; index < 50; index++) section.appendChild(new Element('p'));
  flush();
  advance(119);
  const before = metrics.subtreeQueries;
  advance(1);
  assert.equal(metrics.subtreeQueries - before, 3, 'one inserted branch receives one guide/sidebar/content pass');
  assert.equal(metrics.globalQueries, globalBefore, 'new branches must not trigger global scans');
  assert.ok(metrics.subtreeQueries > scansBefore);
  const complete = metrics.subtreeQueries;
  advance(2000);
  assert.equal(metrics.subtreeQueries, complete);
} else if (input.case === 'shell-images') {
  assert.equal(outer.dataset.deerflowWeknoraChrome, 'hidden');
  assert.equal(inner.dataset.deerflowWeknoraChrome, undefined, 'Wiki directory is not application chrome');
  const sidebar = main.appendChild(new Element('aside', { class: 'sidebar' })); sidebar.width = 200;
  const guide = content.appendChild(new Element('div', { class: 't-guide' }));
  const image = content.appendChild(new Element('img', { 'data-protected-src': 'minio://bucket/image.png' }));
  advance(500);
  assert.equal(sidebar.dataset.deerflowWeknoraChrome, 'hidden');
  assert.equal(guide.parentElement, null);
  assert.match(image.src, /\/api\/llmwiki\/weknora\/api\/v1\/knowledge-bases\/base-test\/files\?file_path=/);
  image.src = 'transparent.gif';
  advance(200);
  assert.match(image.src, /file_path=/, 'image replacement is still repaired');
} else if (input.case === 'source-missing') {
  advance(10000);
  assert.equal(root.classList.contains('deerflow-source-detail-open'), true);
  assert.equal(metrics.sourceAttempts, 18);
  assert.equal(timers.size, 0, 'source fallback stops retries');
  const count = metrics.globalQueries;
  advance(10000);
  assert.equal(metrics.globalQueries, count);
} else if (input.case === 'source-success') {
  assert.equal(root.classList.contains('deerflow-source-detail-open'), true);
  assert.equal(metrics.clicks, 1);
  const drawer = root.querySelector('.t-drawer');
  assert.equal(drawer.dataset.deerflowSourceDetailPanel, 'true');
  assert.equal(timers.size, 0);
  const count = metrics.globalQueries;
  advance(10000);
  assert.equal(metrics.globalQueries, count);
}
"""


def _run_script(script: str, case: str) -> subprocess.CompletedProcess[str]:
    node = shutil.which("node")
    if node is None:
        pytest.skip("Node.js is required to execute the generated iframe script")
    return subprocess.run(
        [node, "-e", HARNESS],
        input=json.dumps({"script": script, "case": case}),
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=30,
        check=False,
    )


@pytest.mark.parametrize("case", ["idle", "theme", "batch", "shell-images", "source-missing", "source-success"])
def test_iframe_guard_settles_and_preserves_interactions(case: str) -> None:
    result = _run_script(_embed_script(), case)
    assert result.returncode == 0, result.stdout + result.stderr