LangFlow: every flow is an API

import { test, expect } from '@playwright/test';

const URL = 'https://app.thetestingacademy.com/ai/blueprint/learn/langflow-agents.html';

test('the bundled runs give FLAKY TEST COUNT: 1', async ({ page }) => {
  await page.goto(URL);
  await expect(page.getByTestId('lf-flaky-count')).toHaveText('1');
  await expect(page.getByTestId('lf-consistent-count')).toHaveText('2');
  await expect(page.getByTestId('lf-report')).toContainText('FLAKY TEST COUNT: 1');
  await expect(page.getByTestId('lf-report')).toContainText(
    '- [failed -> passed] loginTests/auth.spec.ts > loginTests/auth.spec.ts > @P0 Login > redirects to dashboard after successful login');
});

test('one changed result turns a stable suite flaky', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('lf-preset').selectOption('stable');
  await expect(page.getByTestId('lf-flaky-count')).toHaveText('0');

  await page.getByRole('button', { name: 'Run B: t3' }).click();   // passed -> failed
  await expect(page.getByTestId('lf-flaky-count')).toHaveText('1');
  await expect(page.getByTestId('lf-report')).toContainText('- [passed -> failed] suite.spec.ts > suite.spec.ts > t3');

  await page.getByRole('button', { name: 'Run B: t3' }).click();   // failed, then passed on retry
  await expect(page.getByTestId('lf-report')).toContainText('[passed -> flaky-in-run]');
});

test('a test in only one run is not comparable, and the REST body carries the folder', async ({ page }) => {
  await page.goto(URL);
  await page.getByTestId('lf-preset').selectOption('uneven');
  await expect(page.getByTestId('lf-nc-count')).toHaveText('1');
  await expect(page.getByTestId('lf-report')).toContainText('- only in result2.json: suite.spec.ts > suite.spec.ts > t_new');
  await expect(page.getByTestId('lf-request')).toContainText('"input_value": "/test-results/flaky_test_analyzer_ai_Agent/flow/tests/uneven"');

  await page.getByTestId('lf-api-llm').click();
  await expect(page.getByTestId('lf-llm-run')).toBeVisible();
  await expect(page.getByTestId('lf-llm-run')).toContainText('"File-daKW7": { "path": ["<file_path A>"] }');
});

PASSED, FAILED, SKIPPED, FLAKY_IN_RUN = "passed", "failed", "skipped", "flaky-in-run"


class FlakyTestAnalyzer(Component):
    display_name = "Flaky Test Analyzer"
    description = "Read two Playwright results.json files from a folder and report flaky tests."
    icon = "activity"
    name = "FlakyTestAnalyzer"

    inputs = [
        MessageTextInput(
            name="folder",
            display_name="Results Folder",
            info="Folder holding result1.json and result2.json (host or container path).",
            value="/test-results/flaky_test_analyzer_ai_Agent",
        ),
    ]

    outputs = [Output(display_name="Report", name="report", method="analyze")]

    # ---------- path handling ----------

    def _resolve(self, raw: str) -> Path:
        text = (raw or "").strip().strip('"').strip("'")
        if not text:
            text = "/test-results/flaky_test_analyzer_ai_Agent"
        for host, container in HOST_TO_CONTAINER.items():
            if text == host:
                text = container
            elif text.startswith(host + "/"):
                text = container + text[len(host):]
        return Path(text)

    def _find_pair(self, folder: Path):
        if not folder.exists():
            raise ValueError(f"Folder not found inside the Langflow container: {folder}")
        if not folder.is_dir():
            raise ValueError(f"Not a folder: {folder}")

        first = folder / "result1.json"
        second = folder / "result2.json"
        if first.exists() and second.exists():
            return first, second

        candidates = sorted(p for p in folder.glob("*.json") if p.is_file())
        if len(candidates) < 2:
            raise ValueError(
                f"Need two Playwright JSON reports in {folder}, found {len(candidates)}."
            )
        return candidates[0], candidates[1]

    # ---------- playwright parsing ----------

    def _walk(self, node, trail, out):
        for suite in node.get("suites") or []:
            self._walk(suite, trail + [suite.get("title", "")], out)
        for spec in node.get("specs") or []:
            parts = [p for p in trail if p]
            key = f"{spec.get('file', '')} > {' > '.join(parts)} > {spec.get('title', '')}"
            statuses = []
            for test in spec.get("tests") or []:
                for result in test.get("results") or []:
                    statuses.append(result.get("status"))
            out[key] = statuses

    def _verdict(self, statuses):
        """Collapse Playwright retries into one verdict per test."""
        if not statuses:
            return SKIPPED
        bad = {"failed", "timedOut", "interrupted"}
        had_pass = any(s == "passed" for s in statuses)
        had_fail = any(s in bad for s in statuses)
        if had_pass and had_fail:
            return FLAKY_IN_RUN  # Playwright's own retry already proved instability
        if had_pass:
            return PASSED
        if had_fail:
            return FAILED
        return SKIPPED

    def _parse(self, path: Path):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as exc:
            raise ValueError(f"{path.name} is not valid JSON: {exc}") from exc
        if "suites" not in data:
            raise ValueError(f"{path.name} does not look like a Playwright JSON report.")
        raw = {}
        for suite in data.get("suites") or []:
            self._walk(suite, [suite.get("title", "")], raw)
        return {k: self._verdict(v) for k, v in raw.items()}, data.get("stats", {})

    # ---------- main ----------

    def analyze(self) -> Message:
        folder = self._resolve(self.folder)
        file_a, file_b = self._find_pair(folder)

        run_a, stats_a = self._parse(file_a)
        run_b, stats_b = self._parse(file_b)

        shared = set(run_a) & set(run_b)
        only_a = sorted(set(run_a) - set(run_b))
        only_b = sorted(set(run_b) - set(run_a))

        # Flaky = same test, different verdict across the two runs,
        # plus anything Playwright already flagged flaky via retries.
        flipped = sorted(t for t in shared if run_a[t] != run_b[t])
        retry_flaky = sorted(
            t for t in shared
            if FLAKY_IN_RUN in (run_a[t], run_b[t]) and t not in flipped
        )
        flaky = flipped + retry_flaky

        consistent_fail = sorted(
            t for t in shared if run_a[t] == FAILED and run_b[t] == FAILED
        )

        lines = []
        lines.append(f"FLAKY TEST COUNT: {len(flaky)}")
        lines.append("")
        lines.append(f"Run A: {file_a.name}  ({len(run_a)} tests)")
        lines.append(f"Run B: {file_b.name}  ({len(run_b)} tests)")
        lines.append(f"Compared: {len(shared)} tests present in both runs")
        lines.append("")

        lines.append(f"## FLAKY_TESTS ({len(flaky)})")
        if flaky:
            for t in flipped:
                lines.append(f"- [{run_a[t]} -> {run_b[t]}] {t}")
            for t in retry_flaky:
                lines.append(f"- [flaky on retry within a run] {t}")
        else:
            lines.append("- none")
        lines.append("")

        lines.append(f"## CONSISTENT_FAILURES ({len(consistent_fail)})")
        if consistent_fail:
            for t in consistent_fail:
                lines.append(f"- {t}")
        else:
            lines.append("- none")
        lines.append("")

        if only_a or only_b:
            lines.append(f"## NOT_COMPARABLE ({len(only_a) + len(only_b)})")
            for t in only_a:
                lines.append(f"- only in {file_a.name}: {t}")
            for t in only_b:
                lines.append(f"- only in {file_b.name}: {t}")
            lines.append("")

        lines.append("## RERUN_RECOMMENDATION")
        if flaky:
            lines.append(
                f"Quarantine and rerun the {len(flaky)} flaky test(s) above; "
                "they changed verdict between identical runs, so they are unreliable signals."
            )
        else:
            lines.append("No verdict changed between runs. Nothing to quarantine.")
        if consistent_fail:
            lines.append(
                f"The {len(consistent_fail)} consistent failure(s) are real bugs, not flakiness. Fix them."
            )

        lines.append("")
        lines.append("## RAW_STATS")
        lines.append(f"- {file_a.name}: {json.dumps(stats_a)}")
        lines.append(f"- {file_b.name}: {json.dumps(stats_b)}")

        report = "\n".join(lines)
        self.status = f"{len(flaky)} flaky, {len(consistent_fail)} consistent failures"
        return Message(text=report)

#!/usr/bin/env python3
"""
Test the Flaky Test Analyzer Langflow flow end to end.

Imports Flaky_Test_Analyzer.json into a running Langflow, runs it against
fixture folders with known answers, then deletes the imported copy.

    python3 test_flow.py                 # needs Langflow on :7860
    LANGFLOW_URL=... python3 test_flow.py
"""
import gzip
import json
import os
import re
import sys
import urllib.error
import urllib.request

BASE = os.environ.get("LANGFLOW_URL", "http://localhost:7860")
HERE = os.path.dirname(os.path.abspath(__file__))
FLOW_FILE = os.path.join(HERE, "Flaky_Test_Analyzer.json")
# Folder as seen from INSIDE the container (mounted by langflow-up.sh).
ROOT = "/test-results/flaky_test_analyzer_ai_Agent"


def _call(path, data=None, headers=None, method=None):
    body = json.dumps(data).encode() if data is not None else None
    req = urllib.request.Request(
        BASE + path, data=body, method=method or ("POST" if data is not None else "GET")
    )
    req.add_header("Content-Type", "application/json")
    for k, v in (headers or {}).items():
        req.add_header(k, v)
    try:
        with urllib.request.urlopen(req, timeout=300) as resp:
            raw = resp.read()
            if resp.headers.get("Content-Encoding") == "gzip":
                raw = gzip.decompress(raw)
            text = raw.decode()
            try:
                return resp.status, json.loads(text)
            except json.JSONDecodeError:
                return resp.status, text
    except urllib.error.HTTPError as e:
        raw = e.read()
        if e.headers.get("Content-Encoding") == "gzip":
            raw = gzip.decompress(raw)
        return e.code, raw.decode()[:2000]


def bootstrap():
    """Auto-login, then mint an API key (run endpoints require one since v1.5)."""
    st, out = _call("/api/v1/auto_login")
    if st != 200:
        sys.exit(f"Langflow not reachable at {BASE} (auto_login -> {st})")
    bearer = {"Authorization": "Bearer " + out["access_token"]}
    st, out = _call("/api/v1/api_key/", {"name": "flaky-flow-test"}, headers=bearer)
    if st not in (200, 201):
        sys.exit(f"could not create api key: {st} {out}")
    return bearer, (out.get("api_key") or out.get("key"))


def run_flow(flow_id, key, folder):
    st, out = _call(
        f"/api/v1/run/{flow_id}?stream=false",
        {"output_type": "chat", "input_type": "chat", "input_value": folder},
        headers={"x-api-key": key},
    )
    if st != 200:
        return st, str(out)
    try:
        return 200, out["outputs"][0]["outputs"][0]["results"]["message"]["text"]
    except (KeyError, IndexError, TypeError):
        return 200, json.dumps(out)


def num(pattern, text):
    m = re.search(pattern, text)
    return int(m.group(1)) if m else None

ERROR_CASES = [
    ("missing folder",         "/test-results/does_not_exist"),
    ("folder with no reports", "/test-results"),
]


def main():
    bearer, key = bootstrap()

    flow = json.load(open(FLOW_FILE))
    flow.pop("id", None)
    flow["name"] = "ZZ_flaky_flow_undertest"
    st, created = _call("/api/v1/flows/", flow, headers=bearer)
    if st not in (200, 201):
        sys.exit(f"import of {os.path.basename(FLOW_FILE)} failed: {st} {created}")
    flow_id = created["id"]
    print(f"imported {os.path.basename(FLOW_FILE)} -> {flow_id}\n")

    passed = failed = 0
    try:
        for name, folder, want_f, want_c in CASES:
            st, text = run_flow(flow_id, key, folder)
            got_f = num(r"FLAKY TEST COUNT:\s*(\d+)", text) if st == 200 else None
            got_c = num(r"## CONSISTENT_FAILURES \((\d+)\)", text) if st == 200 else None
            ok = st == 200 and got_f == want_f and got_c == want_c
            print(f"[{'PASS' if ok else 'FAIL'}] {name}")
            print(f"       flaky {got_f}/{want_f}  consistent {got_c}/{want_c}")
            if not ok:
                print("       " + text[:300].replace("\n", "\n       "))
            passed += ok
            failed += not ok

        print()
        for name, folder in ERROR_CASES:
            st, text = run_flow(flow_id, key, folder)
            low = text.lower()
            ok = st != 200 or "not found" in low or "need two" in low or "error" in low
            print(f"[{'PASS' if ok else 'FAIL'}] {name} raises instead of answering wrongly")
            passed += ok
            failed += not ok
    finally:
        _call(f"/api/v1/flows/{flow_id}", headers=bearer, method="DELETE")
        print(f"\ncleaned up {flow_id}")

    print(f"\n{passed} passed, {failed} failed")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())

You are a senior test reliability engineer. You are given a comparison of two Playwright runs (Build 1 and Build 2) of the same suite.



COMPARISON REPORT:
{file1} - Build 1 JSON
{file2} - Build 2JSON  

Definitions you MUST follow:
- FLAKY = non-deterministic result: passed in one build and failed in the other, OR passed only after a retry. Flaky tests need a rerun / quarantine, not a code fix.
- CONSISTENT FAILURE = failed in BOTH builds. A real, reproducible bug, NOT flaky. Needs a fix.

Produce:
1. FLAKY_TESTS - names + one-line hypothesis of flake cause (timing, data, parallelism, network...).
2. CONSISTENT_FAILURES - tests failing in both builds, each with a probable root cause.
3. RERUN_RECOMMENDATION - which to rerun (flaky) vs send to engineering (bugs).
4. SUMMARY - counts + one sentence on suite health.

Base everything only on the comparison data. Do not invent test names.