How to become an AI Architect QA from a manual tester and automation tester, even as a beginner.

import ast
import operator

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool          # create custom tools

load_dotenv()

# eval() on text a model produced is arbitrary code execution. This walks the
# parsed expression instead, so only arithmetic can ever run.
_OPS = {
    ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
    ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg,
}


def _safe_eval(node):
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return node.value
    if isinstance(node, ast.BinOp) and type(node.op) in _OPS:
        return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
    if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
        return _OPS[type(node.op)](_safe_eval(node.operand))
    raise ValueError("only arithmetic is allowed")


@tool
def qa_metric_calculator(expression: str) -> str:
    """Calculate a QA metric from an arithmetic expression.

    Use this for pass rate, defect density, automation coverage, defect leakage
    or execution time. Input must be plain arithmetic with no words, for example
    '(438/500)*100' for a pass rate or '18/12' for defects per KLOC.
    """
    try:
        result = _safe_eval(ast.parse(expression, mode="eval").body)
        return f"{round(result, 2)}"
    except Exception:
        return "Error: give me a plain arithmetic expression, e.g. '(438/500)*100'"


agent = create_agent(
    model="google_genai:gemini-flash-lite-latest",
    tools=[qa_metric_calculator],          # list of available tools
    system_prompt=(
        "You are a QA metrics assistant for a test automation team. "
        "Use the qa_metric_calculator tool whenever a number must be computed. "
        "State the formula you used, then the result with its unit."
    ),
)

queries = [
    # maths -> the agent must call the tool
    "We ran 500 regression tests and 438 passed. What is the pass rate?",
    "A module of 12 KLOC has 18 defects. What is the defect density per KLOC?",
    # no maths -> the agent answers directly, no tool call
    "What is the difference between smoke testing and sanity testing?",
]

for q in queries:
    print(f"\nQuestion: {q}")
    result = agent.invoke({"messages": [{"role": "user", "content": q}]})
    print("Agent:", result["messages"][-1].text)

import re
import requests                            # call external APIs and pages
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.tools import tool

load_dotenv()
HEADERS = {"User-Agent": "Mozilla/5.0"}


# Tool 1: search the web and return the title of the results page
@tool
def search_tool(query: str) -> str:
    """Search the web for a query and return the title of the results page."""
    try:
        res = requests.get("https://html.duckduckgo.com/html/",
                           params={"q": query}, headers=HEADERS, timeout=10)
        match = re.search(r"<title>(.*?)</title>", res.text, re.S)
        return match.group(1).strip() if match else "No title found"
    except Exception:
        return "Error performing search"


# Tool 2: fetch a URL and return its plain text (first 1000 characters)
@tool
def web_content_tool(url: str) -> str:
    """Fetch the content of a web page from a URL and return its plain text."""
    try:
        res = requests.get(url, headers=HEADERS, timeout=10)
        text = re.sub(r"<[^>]*>", "", res.text)   # strip HTML tags
        return text[:1000]
    except Exception:
        return "Error fetching web content"


agent = create_agent(
    model="google_genai:gemini-flash-lite-latest",
    tools=[search_tool, web_content_tool],  # pass both tools in the list
    system_prompt=(
        "You are a helpful automation testing assistant. "
        "Use the available tools to search the web and fetch content when needed."
    ),
)

queries = [
    "What is Playwright?",                                # general info (may use search)
    "Search for the Appium official website",             # needs search_tool
    "Get content from https://playwright.dev/docs/intro", # needs web_content_tool
]

for q in queries:
    print(f"\nQ: {q}")
    result = agent.invoke({"messages": [{"role": "user", "content": q}]})
    print(f"Agent: {result['messages'][-1].content}")

from typing import Literal
from dotenv import load_dotenv
from pydantic import BaseModel
from langchain.agents import create_agent

load_dotenv()

# 1) Define the response schema with Pydantic
class TestCase(BaseModel):
    title: str
    description: str
    steps: list[str]
    expected_result: str
    priority: Literal["High", "Medium", "Low"]
    tags: list[str]

class TestCaseList(BaseModel):
    # Tip: always wrap arrays inside an object (better compatibility, especially with Anthropic)
    test_cases: list[TestCase]

# 2) Create the agent with response_format
agent = create_agent(
    model="google_genai:gemini-flash-lite-latest",
    # model="anthropic:claude-haiku-4-5",     # optional
    system_prompt=(
        "You are a senior automation testing engineer. Always return test cases "
        "in the exact structured JSON format requested. Be detailed and professional."
    ),
    response_format=TestCaseList,             # enforce the schema on the final output
)

# 3) Provide a user story and invoke the agent
user_story = ("As a logged-in user, I want to add items to my shopping cart "
              "so that I can purchase them later.")

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": f"Generate 2 detailed test cases for this user story: {user_story}",
    }]
})

# 4) Work with the structured result: no manual parsing
print("=== Structured Output ===")
test_cases = result["structured_response"].test_cases
for tc in test_cases:
    print(tc.model_dump_json(indent=2))

print(f"\nSuccessfully generated {len(test_cases)} test cases!")

llm = ChatDeepSeek(
    model=os.getenv("DEEPSEEK_MODEL", "deepseek-flash"),
    api_key=os.environ["DEEPSEEK_API"],
    temperature=0,
)

SYSTEM_PROMPT="""
You are an end-to-end automation testing agent that controls a real browser
using Playwright.

Rules:
1. Always call launch_browser FIRST, before any other action.
2. Call snapshot_page whenever you land on a new page. It returns every role and
   name on that page, so you never have to guess a selector.
3. Prefer click_by_role and type_by_label over raw CSS. This app also exposes
   stable [data-test="..."] attributes if you need a CSS selector.
4. Verify as you go with assert_visible and assert_text_contains. Do not assume
   a step worked because the previous tool call returned no error.
5. Never invent a value you were not given, and never report a price you did not
   actually read off the page.
6. Before closing, call get_console_errors and take_screenshot.
7. Report every step, every assertion with its PASS/FAIL, and one final verdict:
   TEST PASSED or TEST FAILED."""

def _needs_page(fn):
    """Guard + error funnel, so 20 tools do not repeat the same 6 lines.

    Catching broad Exception is deliberate here: a tool that raises kills the
    agent run, while a tool that returns an error string lets the agent adapt.
    """
    @wraps(fn)
    async def wrapper(*args, **kwargs):
        if _page is None:
            return "Error: Browser not launched. Call launch_browser first."
        try:
            return await fn(*args, **kwargs)
        except PlaywrightTimeoutError as exc:
            return f"Timeout: {_brief(exc)}"
        except Exception as exc:
            return f"Error: {_brief(exc)}"
    return wrapper

# --------------------------------------------------------------------------
# Bundles - give an agent the smallest set that can do its job
# --------------------------------------------------------------------------

CORE_TOOLS = [
    launch_browser, close_browser, snapshot_page, get_page_info,
    navigate_to, click_by_role, click_by_text, type_by_label,
    click_element, type_text, get_text, take_screenshot,
]

INTERACTION_TOOLS = [
    hover_element, press_key, select_dropdown_option, set_checkbox,
    upload_file, drag_and_drop, scroll_to_element, wait_for_element,
    go_back, reload_page,
]

ASSERTION_TOOLS = [
    assert_visible, assert_text_contains, count_elements,
    get_all_texts, get_attribute_value,
]

DIAGNOSTIC_TOOLS = [get_console_errors, get_failed_requests]

ADVANCED_TOOLS = [set_viewport, mock_api_response, save_login_state]

PLAYWRIGHT_TOOLS = (
    CORE_TOOLS + INTERACTION_TOOLS + ASSERTION_TOOLS
    + DIAGNOSTIC_TOOLS + ADVANCED_TOOLS
)

async def main():
   
    agent = create_agent(
        model=llm,
        tools=PLAYWRIGHT_TOOLS,
        system_prompt=SYSTEM_PROMPT,
    )

    # Playwright's async API needs ainvoke: every tool call runs on the same event loop
    result = await agent.ainvoke({"messages": [{"role": "user", "content": TASK}]})

    # Show each tool call and what the browser answered
    print("\n===== Steps taken =====")
    for msg in result["messages"]:
        if msg.type == "ai" and msg.tool_calls:
            for call in msg.tool_calls:
                print(f"-> {call['name']}({call['args']})")
        elif msg.type == "tool":
            print(f"   {msg.content}")

    print("\n===== Final Result =====")
    print(result["messages"][-1].text)   # .text joins the reply's text blocks

# ---------------------------------------------------------------- stage 2
class TestCase(BaseModel):
    id: str = Field(description="e.g. TC-01")
    title: str
    priority: Literal["P0", "P1", "P2"]
    steps: list[str] = Field(description="Concrete UI steps against the target app")
    expected_result: str
    automatable: bool = Field(
        description="False when the ticket describes something absent from the target app")
    reason_if_not: str = Field(default="", description="Why it cannot be automated here")


class TestPlan(BaseModel):
    ticket_key: str
    scope: str = Field(description="2-3 sentences: what is tested and what is not")
    risks: list[str]
    test_cases: list[TestCase]


PLANNER_PROMPT = f"""You are a senior QA engineer writing a test plan from a Jira ticket.

The test cases will be executed by a browser agent against THIS app, which may be
a different product from the one the ticket describes:

    Target app: {TARGET_URL}
    Login:      {TARGET_USER} / {TARGET_PASS}

Rules:
1. Write 5-8 test cases covering the ticket's acceptance criteria.
2. Set automatable=true ONLY when every step can be performed on the target app
   above. If the ticket describes a feature the target app does not have (for
   example SSO or passkey buttons that are not on this login page), set
   automatable=false and say so in reason_if_not. Do NOT invent UI.
3. Steps must be concrete and clickable: name the field, button or text to check.
4. Never invent credentials beyond the ones given."""

# The .env calls the key DEEPSEEK_API, but the SDK looks for DEEPSEEK_API_KEY,
# so it has to be passed in explicitly. deepseek-flash is pinned on purpose:
# "deepseek-chat" is an alias and deepseek-v4-pro costs 4x more.
MODEL = os.getenv("DEEPSEEK_MODEL", "deepseek-flash")

# Stage 3 (tool calling) is happy with the default thinking mode.
llm = ChatDeepSeek(model=MODEL, api_key=os.environ["DEEPSEEK_API"], temperature=0)

# Stage 2 is NOT. response_format pins tool_choice to one function, and
# deepseek-flash rejects that while thinking is on:
#     400 "Thinking mode does not support this tool_choice"
# extra_body reaches the raw request body; model_kwargs does not.
planner_llm = ChatDeepSeek(
    model=MODEL,
    api_key=os.environ["DEEPSEEK_API"],
    temperature=0,
    extra_body={"thinking": {"type": "disabled"}},
)

    # ---- stage 2: plan --------------------------------------------------
    print(f"\n{'='*70}\nSTAGE 2  Writing the test plan\n{'='*70}")
    planner = create_agent(model=planner_llm, system_prompt=PLANNER_PROMPT,
                           response_format=TestPlan)
    planned = await planner.ainvoke({"messages": [{"role": "user", "content":
        f"Ticket {issue['key']} ({issue['issuetype']}, priority {issue['priority']})\n"
        f"Summary: {issue['summary']}\n\nDescription:\n{desc}"}]})
    plan: TestPlan = planned["structured_response"]

    print(f"\nScope: {plan.scope}\n")
    for r in plan.risks:
        print(f"  risk: {r}")
    runnable = [tc for tc in plan.test_cases if tc.automatable]
    skipped = [tc for tc in plan.test_cases if not tc.automatable]
    print(f"\n{len(plan.test_cases)} case(s): {len(runnable)} automatable, {len(skipped)} not\n")
    for tc in plan.test_cases:
        mark = "RUN " if tc.automatable else "SKIP"
        print(f"[{mark}] {tc.id} ({tc.priority}) {tc.title}")
        for s in tc.steps:
            print(f"         - {s}")
        print(f"         expected: {tc.expected_result}")
        if not tc.automatable:
            print(f"         not automatable: {tc.reason_if_not}")

    out = Path(f"test_plan_{key}.json")
    out.write_text(plan.model_dump_json(indent=2))
    print(f"\nPlan written to {out}")

    if args.dry_run:
        print("--dry-run: stopping before the browser.")
        return
    if not runnable:
        print("Nothing automatable against the target app. Stopping.")
        return
    if not confirm(f"\nRun {len(runnable)} case(s) against {TARGET_URL} in a REAL browser?", args.yes):
        sys.exit("Stopped before execution.")

# ------------------------------------------------------------------ schema
class TestCase(BaseModel):
    id: str = Field(description="e.g. TC-01")
    title: str
    priority: Literal["P0", "P1", "P2"]
    steps: list[str] = Field(description="Concrete UI steps against the target app")
    expected_result: str
    automatable: bool = Field(description="False if the target app lacks this UI")
    reason_if_not: str = ""
    rag_relation: Literal["duplicate", "extends", "new"] = Field(
        description="duplicate = an existing case already covers this exactly; "
                    "extends = it builds on one; new = nothing similar exists")
    related_existing_ids: list[str] = Field(
        default_factory=list, description="Ids from the retrieved library, e.g. TTA-002")


class TestPlan(BaseModel):
    ticket_key: str
    scope: str
    risks: list[str]
    coverage_gap: str = Field(description="What the ticket needs that the existing library misses")
    test_cases: list[TestCase]

    # ---- 2. RAG ---------------------------------------------------------
    rag_block = "RAG retrieval was skipped for this run."
    if not args.no_rag:
        banner(2, "Retrieve similar existing test cases (local RAG)")
        rag = TestCaseRAG()
        query = f"{issue['summary']}. {desc[:600]}"
        hits = rag.search(query, k=args.k)
        print(f"corpus  : {len(rag.cases)} cases | backend: {rag.backend}\n")
        for score, c in hits:
            print(f"  {score:.3f}  {c['id']}  [{c['type']:7}] {c['title']}")
            print(f"           last run: {c['last_result']}, {c['status']}, tags: {', '.join(c['tags'][:5])}")
        rag_block = format_for_prompt(hits)

    if not confirm(f"\nWrite a test plan for {key} grounded in these results?", args.yes):
        sys.exit("Stopped before planning.")

    # ---- 3. Plan --------------------------------------------------------
    banner(3, "Write the test plan (Groq + RAG context)")
    planner = create_agent(model=planner_llm, system_prompt=PLANNER_PROMPT,
                           response_format=TestPlan)
    planned = await planner.ainvoke({"messages": [{"role": "user", "content":
        f"TICKET {issue['key']} ({issue['issuetype']}, priority {issue['priority']})\n"
        f"Summary: {issue['summary']}\n\nDescription:\n{desc}\n\n"
        f"--- EXISTING TEST CASES RETRIEVED FROM THE TEAM LIBRARY ---\n{rag_block}"}]})
    plan: TestPlan = planned["structured_response"]

    print(f"\nScope: {plan.scope}")
    print(f"\nCoverage gap: {plan.coverage_gap}\n")
    for r in plan.risks:
        print(f"  risk: {r}")
    runnable = [tc for tc in plan.test_cases if tc.automatable]
    print(f"\n{len(plan.test_cases)} case(s): {len(runnable)} automatable\n")
    for tc in plan.test_cases:
        rel = f"{tc.rag_relation}"
        if tc.related_existing_ids:
            rel += f" of {', '.join(tc.related_existing_ids)}"
        print(f"[{'RUN ' if tc.automatable else 'SKIP'}] {tc.id} ({tc.priority}) {tc.title}")
        print(f"         vs library: {rel}")
        if not tc.automatable:
            print(f"         not automatable: {tc.reason_if_not}")

    dupes = [tc for tc in plan.test_cases if tc.rag_relation == "duplicate"]
    if dupes:
        print(f"\n{len(dupes)} case(s) already covered by the library: "
              f"{', '.join(tc.id + ' -> ' + ','.join(tc.related_existing_ids) for tc in dupes)}")

    Path(f"test_plan_{key}.json").write_text(plan.model_dump_json(indent=2))
    print(f"\nPlan written to test_plan_{key}.json")

    def search(self, query: str, k: int = 5) -> list[tuple[float, dict]]:
        """Return the k most similar test cases as (score, case), best first."""
        if self._vectors is None:
            return self._keyword_search(query, k)
        import numpy as np
        from fastembed import TextEmbedding
        q = np.array(next(iter(TextEmbedding(model_name=MODEL_NAME).embed([query]))), dtype="float32")
        q /= np.linalg.norm(q)
        scores = self._vectors @ q
        order = np.argsort(-scores)[:k]
        return [(float(scores[i]), self.cases[i]) for i in order]

    def _keyword_search(self, query: str, k: int) -> list[tuple[float, dict]]:
        """Jaccard overlap fallback. Crude, but it never fails to install."""
        def toks(s: str) -> set[str]:
            return {w for w in re.findall(r"[a-z0-9]+", s.lower()) if len(w) > 2}
        q = toks(query)
        scored = [(len(q & toks(d)) / max(len(q | toks(d)), 1), c)
                  for d, c in zip(self.docs, self.cases)]
        return sorted(scored, key=lambda x: -x[0])[:k]

    @classmethod
    def call_tool(cls, name: str, arguments: dict) -> dict:
        """The seam. Swap this body for a real MCP call and everything else holds."""
        if not cls.connected:
            cls.connect()
        ts = f"{datetime.now(timezone.utc).timestamp():.6f}"
        print(f"\n{'-'*66}\n[DUMMY SLACK MCP] call_tool({name})\n{'-'*66}")
        print(json.dumps(arguments, indent=2)[:2000])
        print(f"{'-'*66}\n[DUMMY SLACK MCP] not sent - dummy connection\n")
        record = {"tool": name, "arguments": arguments, "ts": ts}
        _SENT.append(record)
        return {"ok": True, "ts": ts, "simulated": True,
                "permalink": f"https://example.slack.com/archives/C0DUMMY/p{ts.replace('.', '')}"}