"""Google Workspace, Firebase push and SOAP Task nodes. Google, Firebase and
WNS are stubbed by an httpx MockTransport (no network); SOAP calls go through
real zeep and requests to a small local HTTP server."""

import base64
import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import httpx
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from jose import jwt

from app.crud import firebase_push as fp, google_apps as ga, google_auth, soap_client
from app.crud.google_auth import GoogleError


@pytest.fixture(scope="module")
def key_file() -> str:
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    pem = key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()).decode()
    return json.dumps({"type": "service_account", "project_id": "demo-project", "client_email": "bot@demo.iam.gserviceaccount.com", "private_key": pem, "token_uri": "https://evil.example.com/steal"})


class Google:
    """Records every call; `routes` maps "METHOD url-fragment" to a response."""

    def __init__(self):
        self.requests: list[httpx.Request] = []
        self.routes: list[tuple[str, object]] = []
        self.token_hosts: list[str] = []

    def on(self, match: str, response: object) -> None:
        self.routes.insert(0, (match, response))

    def __call__(self, request: httpx.Request) -> httpx.Response:
        url = str(request.url)
        if "oauth2.googleapis.com/token" in url or "accesstoken.srf" in url or "login.microsoftonline.com" in url:
            self.token_hosts.append(request.url.host)
            return httpx.Response(200, json={"access_token": "tok", "expires_in": 3600})
        self.requests.append(request)
        for match, response in self.routes:
            method, _, fragment = match.partition(" ")
            if request.method == method and fragment in url:
                result = response(request) if callable(response) else response
                if isinstance(result, httpx.Response):
                    return result
                return httpx.Response(200, json=result)
        return httpx.Response(404, json={"error": {"status": "NOT_FOUND", "message": f"unexpected {request.method} {url}"}})

    def bodies(self, fragment: str) -> list[dict]:
        return [json.loads(r.content) for r in self.requests if fragment in str(r.url) and r.content]


@pytest.fixture
def google(monkeypatch) -> Google:
    stub = Google()
    google_auth._tokens.clear()
    fp._wns_tokens.clear()
    monkeypatch.setattr(google_auth, "_client", lambda timeout=30.0: httpx.AsyncClient(transport=httpx.MockTransport(stub)))

    async def no_sleep(_):
        return None

    monkeypatch.setattr(google_auth, "_sleep", no_sleep)
    return stub


def _call(key_file, **settings) -> ga.Call:
    inputs = settings.pop("inputs", {})
    return ga.Call(config={"provider": "service_account", "service_account_json": key_file}, settings=settings, inputs=inputs)


# ---- sign-in ---------------------------------------------------------------------------------


async def test_service_account_signs_in_only_at_googles_own_endpoint(google, key_file):
    google.on("GET sheets.googleapis.com", {"values": []})
    await ga.run("google_sheets", "read", _call(key_file, spreadsheet_id="abc123", range="Sheet1"))
    assert google.token_hosts == ["oauth2.googleapis.com"]  # not the key file's token_uri
    assert google.requests[0].headers["authorization"] == "Bearer tok"


async def test_token_is_reused_but_not_across_credentials(google, key_file):
    google.on("GET sheets.googleapis.com", {"values": []})
    call = _call(key_file, spreadsheet_id="abc123", range="Sheet1")
    await ga.run("google_sheets", "read", call)
    await ga.run("google_sheets", "read", call)
    assert len(google.token_hosts) == 1
    other = json.loads(key_file)
    other["private_key_id"] = "another"
    await ga.run("google_sheets", "read", ga.Call(config={"service_account_json": json.dumps(other)}, settings=call.settings))
    assert len(google.token_hosts) == 2


async def test_delegated_user_goes_in_the_assertion(monkeypatch, key_file):
    seen = {}

    def handler(request: httpx.Request) -> httpx.Response:
        form = dict(pair.split("=", 1) for pair in request.content.decode().split("&"))
        seen.update(jwt.get_unverified_claims(form["assertion"]))
        return httpx.Response(200, json={"access_token": "t", "expires_in": 3600})

    google_auth._tokens.clear()
    monkeypatch.setattr(google_auth, "_client", lambda timeout=30.0: httpx.AsyncClient(transport=httpx.MockTransport(handler)))
    await google_auth.access_token({"service_account_json": key_file, "delegated_user": "asha@example.com"}, ("scope-a",))
    assert seen["sub"] == "asha@example.com" and seen["scope"] == "scope-a" and seen["aud"] == google_auth.TOKEN_URL


async def test_bad_key_never_appears_in_the_error():
    secret = '{"private_key": "TOPSECRET"'
    with pytest.raises(GoogleError) as err:
        google_auth.service_account({"service_account_json": secret})
    assert "TOPSECRET" not in str(err.value)


async def test_rate_limit_is_retried_and_expired_token_replaced(google, key_file):
    calls = {"n": 0}

    def flaky(_):
        calls["n"] += 1
        if calls["n"] == 1:
            return httpx.Response(429, headers={"retry-after": "1"}, json={"error": {"status": "RESOURCE_EXHAUSTED", "message": "slow down"}})
        if calls["n"] == 2:
            return httpx.Response(401, json={"error": {"status": "UNAUTHENTICATED", "message": "expired"}})
        return httpx.Response(200, json={"values": [["a"]]})

    google.on("GET sheets.googleapis.com", flaky)
    _, result = await ga.run("google_sheets", "read", _call(key_file, spreadsheet_id="abc123", range="Sheet1", first_row_headers="no"))
    assert result["rows"] == [["a"]] and calls["n"] == 3 and len(google.token_hosts) == 2


# ---- Sheets ----------------------------------------------------------------------------------


async def test_sheets_read_uses_first_row_as_field_names(google, key_file):
    google.on("GET values/Sheet1", {"values": [["Name", "City"], ["Asha", "Kochi"], ["Ravi"]]})
    detail, result = await ga.run("google_sheets", "read", _call(key_file, spreadsheet_id="https://docs.google.com/spreadsheets/d/1AbCdEfGhIjK/edit#gid=0", range="Sheet1"))
    assert "/spreadsheets/1AbCdEfGhIjK/" in str(google.requests[0].url)
    assert result["rows"] == [{"Name": "Asha", "City": "Kochi"}, {"Name": "Ravi", "City": ""}]
    assert result["headers"] == ["Name", "City"] and result["row_count"] == 2 and detail == "2 row(s)"


async def test_sheets_find_reports_real_row_numbers(google, key_file):
    google.on("GET values/Sheet1", {"values": [["Id", "State"], ["1", "new"], ["2", "done"], ["3", "new"]]})
    _, result = await ga.run("google_sheets", "find", _call(key_file, spreadsheet_id="abc123", range="Sheet1", column="State", value="new"))
    assert result["row_numbers"] == [2, 4] and result["first_row"]["Id"] == "1"
    with pytest.raises(GoogleError, match="no column"):
        await ga.run("google_sheets", "find", _call(key_file, spreadsheet_id="abc123", range="Sheet1", column="Nope", value="x"))


async def test_sheets_append_matches_objects_to_the_header_row_and_stays_raw(google, key_file):
    google.on("GET values/", {"values": [["Name", "Amount"]]})
    google.on("POST :append", {"updates": {"updatedRows": 1, "updatedRange": "Sheet1!A2:B2"}})
    detail, result = await ga.run("google_sheets", "append", _call(key_file, spreadsheet_id="abc123", range="My Sheet", rows=[{"Amount": 5, "Name": "=IMPORTXML(1)", "Extra": "x"}]))
    append = next(r for r in google.requests if ":append" in str(r.url))
    assert append.url.params["valueInputOption"] == "RAW"  # a form value can't become a live formula
    assert json.loads(append.content)["values"] == [["=IMPORTXML(1)", 5]]
    assert result["skipped_columns"] == ["Extra"] and "Extra" in detail
    header_read = next(r for r in google.requests if r.method == "GET")
    assert "%27My%20Sheet%27%211%3A1" in str(header_read.url)


async def test_sheets_append_to_an_empty_sheet_writes_a_header_first(google, key_file):
    google.on("GET values/", {"values": []})
    google.on("POST :append", {"updates": {"updatedRows": 2}})
    await ga.run("google_sheets", "append", _call(key_file, spreadsheet_id="abc123", range="Sheet1", rows='[{"a": 1}]'))
    assert google.bodies(":append")[0]["values"] == [["a"], [1]]


async def test_sheets_write_as_typed_only_when_asked(google, key_file):
    google.on("PUT values/", {"updatedCells": 2})
    await ga.run("google_sheets", "update", _call(key_file, spreadsheet_id="abc123", range="Sheet1!A1", rows=[["x", 1]], value_input="as_typed"))
    assert google.requests[-1].url.params["valueInputOption"] == "USER_ENTERED"


async def test_rows_are_required_and_must_parse(google, key_file):
    for rows in ("", "not json", None):
        with pytest.raises(GoogleError):
            await ga.run("google_sheets", "append", _call(key_file, spreadsheet_id="abc123", range="Sheet1", rows=rows))


@pytest.mark.parametrize("bad", ["../../drive/v3/files", "a/b", "x y", "abc?alt=media"])
async def test_an_id_cant_reach_another_api_path(google, key_file, bad):
    with pytest.raises(GoogleError, match="doesn't look right"):
        await ga.run("google_sheets", "read", _call(key_file, spreadsheet_id=bad, range="Sheet1"))
    assert google.requests == []


async def test_not_found_says_who_to_share_with(google, key_file):
    with pytest.raises(GoogleError, match="bot@demo.iam.gserviceaccount.com"):
        await ga.run("google_sheets", "read", _call(key_file, spreadsheet_id="abc123", range="Sheet1"))


# ---- Docs / Slides ---------------------------------------------------------------------------


async def test_docs_from_template_copies_then_fills_inputs(google, key_file):
    google.on("POST /tmpl123/copy", {"id": "newdoc", "name": "Offer"})
    google.on("POST newdoc:batchUpdate", {"replies": [{"replaceAllText": {"occurrencesChanged": 2}}, {"replaceAllText": {"occurrencesChanged": 1}}]})
    detail, result = await ga.run("google_docs", "from_template", _call(key_file, template_id="tmpl123", title="Offer", inputs={"name": "Asha", "amount": 500}))
    assert result["document_id"] == "newdoc" and result["replaced"] == 3 and "3" in detail
    requests_ = google.bodies("newdoc:batchUpdate")[0]["requests"]
    assert requests_[0]["replaceAllText"] == {"containsText": {"text": "{{name}}", "matchCase": True}, "replaceText": "Asha"}
    assert requests_[1]["replaceAllText"]["replaceText"] == "500"


async def test_docs_read_flattens_paragraphs_and_tables(google, key_file):
    para = lambda t: {"paragraph": {"elements": [{"textRun": {"content": t}}]}}  # noqa: E731
    google.on("GET documents/doc123", {"title": "T", "body": {"content": [para("Hello\n"), {"table": {"tableRows": [{"tableCells": [{"content": [para("cell")]}]}]}}]}})
    _, result = await ga.run("google_docs", "read", _call(key_file, document_id="doc123"))
    assert result["text"] == "Hello\ncell" and result["title"] == "T"


async def test_create_in_a_folder_goes_through_drive(google, key_file):
    google.on("POST drive/v3/files", {"id": "made"})
    google.on("POST made:batchUpdate", {})
    _, result = await ga.run("google_docs", "create", _call(key_file, title="T", folder_id="https://drive.google.com/drive/folders/1FolderIdXyz", text="hi"))
    assert result["document_id"] == "made"
    assert google.bodies("drive/v3/files")[0] == {"name": "T", "mimeType": ga.NATIVE_DOC, "parents": ["1FolderIdXyz"]}


async def test_slides_add_slide_fills_the_layouts_placeholders(google, key_file):
    google.on("POST deck123:batchUpdate", {})
    await ga.run("google_slides", "add_slide", _call(key_file, presentation_id="deck123", layout="TITLE_AND_BODY", title="Q3", text="Numbers"))
    requests_ = google.bodies(":batchUpdate")[0]["requests"]
    mappings = requests_[0]["createSlide"]["placeholderIdMappings"]
    assert [m["layoutPlaceholder"]["type"] for m in mappings] == ["TITLE", "BODY"]
    assert [r["insertText"]["text"] for r in requests_[1:]] == ["Q3", "Numbers"]
    assert requests_[1]["insertText"]["objectId"] == mappings[0]["objectId"]


async def test_slides_export_saves_a_pptx(google, key_file):
    google.on("GET files/deck123", {"name": "Report"})
    google.on("GET /deck123/export", httpx.Response(200, content=b"PPTX"))
    saved = {}

    async def save(name, content_type, data):
        saved.update(name=name, type=content_type, data=data)
        return "file-1"

    call = _call(key_file, presentation_id="deck123", format="pptx")
    call.save_file = save
    _, result = await ga.run("google_slides", "export", call)
    assert result == {"file_id": "file-1", "file_name": "Report.pptx", "size": 4}
    assert saved["data"] == b"PPTX" and "presentationml" in saved["type"]


# ---- Drive -----------------------------------------------------------------------------------


async def test_drive_list_escapes_the_search_text(google, key_file):
    google.on("GET drive/v3/files", {"files": [{"id": "1", "name": "a", "mimeType": "x", "size": "5"}]})
    _, result = await ga.run("google_drive", "list", _call(key_file, name_contains="it's", folder_id="folder123"))
    query = google.requests[0].url.params["q"]
    assert "name contains 'it\\'s'" in query and "'folder123' in parents" in query and "trashed = false" in query
    assert result["files"][0]["size"] == 5 and result["file_ids"] == ["1"]


async def test_drive_upload_sends_the_files_bytes(google, key_file):
    google.on("POST upload/drive/v3/files", {"id": "up1", "name": "a.pdf"})

    async def load(file_id):
        return ("a.pdf", "application/pdf", b"%PDF-bytes") if file_id == "local1" else None

    call = _call(key_file, upload_file_id="local1", folder_id="folder123")
    call.load_file = load
    _, result = await ga.run("google_drive", "upload", call)
    body = google.requests[0].content
    assert result["drive_file_id"] == "up1" and b"%PDF-bytes" in body and b'"parents": ["folder123"]' in body
    missing = _call(key_file, upload_file_id="other")
    missing.load_file = load
    with pytest.raises(GoogleError, match="wasn't found"):
        await ga.run("google_drive", "upload", missing)


async def test_drive_trash_never_deletes_for_good(google, key_file):
    google.on("PATCH files/file123", {"id": "file123"})
    await ga.run("google_drive", "trash", _call(key_file, file_id="file123"))
    assert google.requests[0].method == "PATCH" and json.loads(google.requests[0].content) == {"trashed": True}


async def test_drive_share_with_anyone_needs_no_email(google, key_file):
    google.on("POST /permissions", {"id": "p1"})
    google.on("GET files/file123", {"webViewLink": "https://drive/x"})
    _, result = await ga.run("google_drive", "share", _call(key_file, file_id="file123", share_with="anyone", role="reader"))
    assert google.bodies("/permissions")[0] == {"type": "anyone", "role": "reader"} and result["url"] == "https://drive/x"
    with pytest.raises(GoogleError, match="email"):
        await ga.run("google_drive", "share", _call(key_file, file_id="file123", share_with="user"))


# ---- Calendar / Gmail / Forms ----------------------------------------------------------------


async def test_calendar_create_all_day_and_timed_with_meet(google, key_file):
    google.on("POST calendars/primary/events", {"id": "e1", "summary": "S", "hangoutLink": "https://meet/x", "start": {"date": "2026-12-01"}})
    _, result = await ga.run("google_calendar", "create_event", _call(key_file, summary="S", start="2026-12-01", end="2026-12-02T10:00:00", time_zone="Asia/Kolkata", attendees="a@x.com, b@x.com", add_meet="yes"))
    body = google.bodies("/events")[0]
    assert body["start"] == {"date": "2026-12-01"}
    assert body["end"] == {"dateTime": "2026-12-02T10:00:00", "timeZone": "Asia/Kolkata"}
    assert [a["email"] for a in body["attendees"]] == ["a@x.com", "b@x.com"]
    assert "createRequest" in body["conferenceData"]
    assert google.requests[0].url.params["conferenceDataVersion"] == "1"
    assert google.requests[0].url.params["sendUpdates"] == "none"
    assert result["event_id"] == "e1" and result["meet_link"] == "https://meet/x"


async def test_gmail_send_builds_a_real_message_and_refuses_header_injection(google, key_file):
    google.on("POST messages/send", {"id": "m1", "threadId": "t1"})

    async def load(_):
        return ("r.pdf", "application/pdf", b"PDF")

    call = _call(key_file, to="a@example.com", subject="Hello", body="<b>Hi</b>", body_format="html", attachment_file_id="f1")
    call.load_file = load
    _, result = await ga.run("google_gmail", "send", call)
    raw = base64.urlsafe_b64decode(google.bodies("messages/send")[0]["raw"]).decode()
    assert result["message_id"] == "m1" and "To: a@example.com" in raw and "r.pdf" in raw and "text/html" in raw
    with pytest.raises(GoogleError, match="header"):
        await ga.run("google_gmail", "send", _call(key_file, to="a@example.com", subject="Hi\nBcc: x@evil.com", body="b"))


async def test_gmail_read_decodes_the_text_part(google, key_file):
    text = base64.urlsafe_b64encode("héllo".encode()).decode().rstrip("=")
    google.on("GET messages/msg123", {"id": "msg123", "payload": {"headers": [{"name": "From", "value": "a@x"}], "parts": [{"mimeType": "text/plain", "body": {"data": text}}]}})
    _, result = await ga.run("google_gmail", "read", _call(key_file, message_id="msg123"))
    assert result["body"] == "héllo" and result["from"] == "a@x"


async def test_forms_responses_use_question_titles(google, key_file):
    google.on("GET forms/form1234", {"items": [{"title": "Your city", "questionItem": {"question": {"questionId": "q1", "textQuestion": {}}}}]})
    google.on("GET form1234/responses", {"responses": [{"responseId": "r1", "lastSubmittedTime": "2026-01-01", "answers": {"q1": {"textAnswers": {"answers": [{"value": "Kochi"}]}}}}]})
    _, result = await ga.run("google_forms", "responses", _call(key_file, form_id="form1234"))
    assert result["responses"][0]["Your city"] == "Kochi" and result["count"] == 1


async def test_unknown_operation_is_refused(key_file):
    with pytest.raises(GoogleError, match="unknown operation"):
        await ga.run("google_docs", "explode", _call(key_file))


# ---- Firebase --------------------------------------------------------------------------------


def _fb(key_file, **extra) -> dict:
    return {"provider": "fcm", "service_account_json": key_file, **extra}


def test_message_only_carries_what_was_set():
    message = fp.build_message({"title": "T", "body": "B", "web_link": "https://x.test", "android_channel": "alerts", "ios_badge": "3", "ttl_seconds": "60", "data": '{"id": 7, "s": "x"}'})
    assert message["notification"] == {"title": "T", "body": "B"}
    assert message["data"] == {"id": "7", "s": "x"}  # FCM takes only strings
    assert message["android"]["notification"] == {"channel_id": "alerts"} and message["android"]["ttl"] == "60s"
    assert message["apns"]["payload"]["aps"] == {"badge": 3}
    assert message["webpush"]["fcm_options"]["link"] == "https://x.test"


def test_silent_message_shows_nothing_and_needs_data():
    message = fp.build_message({"silent": "yes", "data": {"a": "1"}, "title": "ignored"})
    assert "notification" not in message and "notification" not in message["android"]
    assert message["apns"]["payload"]["aps"]["content-available"] == 1
    assert message["apns"]["headers"]["apns-priority"] == "5"
    with pytest.raises(GoogleError):
        fp.build_message({"silent": "yes"})
    with pytest.raises(GoogleError):
        fp.build_message({})
    with pytest.raises(GoogleError, match="https"):
        fp.build_message({"title": "T", "web_link": "http://x.test"})


def test_targets_accept_every_shape_and_drop_duplicates():
    assert fp.targets("a, b\nb") == ["a", "b"]
    assert fp.targets(["a", {"token": "b"}, {"nope": 1}, "a"]) == ["a", "b"]
    assert fp.targets('["x"]') == ["x"]
    assert fp.targets(None) == [] and fp.targets("") == []


async def test_push_reports_stale_tokens_and_keeps_going(google, key_file):
    def send(request):
        token = json.loads(request.content)["message"]["token"]
        if token == "stale":
            return httpx.Response(404, json={"error": {"status": "NOT_FOUND", "message": "gone", "details": [{"errorCode": "UNREGISTERED"}]}})
        return {"name": f"projects/demo/messages/{token}"}

    google.on("POST /projects/demo-project/messages:send", send)
    ok, detail, result = await fp.send(_fb(key_file), {"tokens": ["good", "stale"], "title": "T"})
    assert ok and result["sent"] == 1 and result["failed"] == 1 and result["invalid_tokens"] == ["stale"]
    assert "no longer registered" in detail and "stale" not in detail


async def test_push_fails_when_nothing_arrives_unless_told_otherwise(google, key_file):
    google.on("POST messages:send", httpx.Response(403, json={"error": {"status": "PERMISSION_DENIED", "message": "nope"}}))
    ok, detail, _ = await fp.send(_fb(key_file), {"tokens": "secrettokenabcdef", "title": "T"})
    assert not ok and "PERMISSION_DENIED" in detail and "secrettokenabcdef" not in detail  # never a whole token
    ok, _, _ = await fp.send(_fb(key_file), {"tokens": "t", "title": "T", "stop_on_failure": "no"})
    assert ok


async def test_push_with_no_devices_is_not_a_failure(google, key_file):
    ok, detail, result = await fp.send(_fb(key_file), {"tokens": [], "title": "T"})
    assert ok and "nothing sent" in detail and result["sent"] == 0 and google.requests == []


async def test_push_to_topic_and_dry_run(google, key_file):
    google.on("POST messages:send", {"name": "m"})
    ok, detail, _ = await fp.send(_fb(key_file), {"target": "topic", "topic": "/topics/news", "title": "T", "dry_run": "yes"})
    body = google.bodies("messages:send")[0]
    assert ok and body["validate_only"] is True and body["message"]["topic"] == "news" and detail.startswith("validated")
    with pytest.raises(GoogleError):
        await fp.send(_fb(key_file), {"target": "topic", "topic": "bad topic!", "title": "T"})


async def test_windows_channel_gets_a_toast_with_escaped_text(google, key_file):
    channel = "https://wns2-by3p.notify.windows.com/?token=abc"
    google.on("POST notify.windows.com", httpx.Response(200, headers={"X-WNS-Msg-ID": "w1"}))
    config = _fb(key_file, wns_auth="entra", wns_client_id="id", wns_client_secret="s", wns_tenant_id="tenant-1")
    ok, _, result = await fp.send(config, {"target": "none", "windows_channels": [channel], "title": "A & B", "body": "<x>"})
    sent = google.requests[-1]
    assert ok and result["windows_sent"] == 1 and result["message_id"] == "w1"
    assert sent.headers["x-wns-type"] == "wns/toast" and sent.headers["authorization"] == "Bearer tok"
    assert b"<text>A &amp; B</text><text>&lt;x&gt;</text>" in sent.content
    assert "login.microsoftonline.com" in google.token_hosts


async def test_uwp_credentials_use_the_legacy_token_host(google, key_file):
    google.on("POST notify.windows.com", httpx.Response(200))
    config = _fb(key_file, wns_auth="package", wns_client_id="ms-app://S-1", wns_client_secret="s")
    await fp.send(config, {"target": "none", "windows_channels": "https://x.notify.windows.com/?token=1", "title": "T"})
    assert "login.live.com" in google.token_hosts


@pytest.mark.parametrize("uri", ["https://evil.example.com/?token=1", "http://x.notify.windows.com/", "https://notify.windows.com.evil.com/", "https://evilnotify.windows.com/"])
async def test_token_never_goes_to_a_channel_off_microsofts_domain(google, key_file, uri):
    config = _fb(key_file, wns_auth="entra", wns_client_id="id", wns_client_secret="s", wns_tenant_id="t")
    ok, _, result = await fp.send(config, {"target": "none", "windows_channels": [uri], "title": "T"})
    assert not ok and result["failed"] == 1
    assert google.requests == [] and google.token_hosts == []


async def test_expired_windows_channel_is_reported(google, key_file):
    google.on("POST notify.windows.com", httpx.Response(410))
    config = _fb(key_file, wns_auth="entra", wns_client_id="id", wns_client_secret="s", wns_tenant_id="t")
    channel = "https://x.notify.windows.com/?token=1"
    _, _, result = await fp.send(config, {"target": "none", "windows_channels": [channel], "title": "T"})
    assert result["expired_channels"] == [channel]


async def test_windows_without_credentials_says_what_to_set(google, key_file):
    with pytest.raises(GoogleError, match="WNS"):
        await fp.send(_fb(key_file), {"target": "none", "windows_channels": ["https://x.notify.windows.com/?token=1"], "title": "T"})


async def test_topic_subscribe_treats_already_subscribed_as_done(google, key_file):
    def subscribe(request):
        return httpx.Response(409, json={"error": {"status": "ALREADY_EXISTS", "message": "exists"}}) if "dup" in str(request.url) else {"topic_name": "news"}

    google.on("POST /topicSubscriptions", subscribe)
    ok, detail, result = await fp.manage_topic(_fb(key_file), {"operation": "subscribe", "tokens": ["one", "dup"], "topic": "news"})
    assert ok and result["done"] == 2 and "2 of 2" in detail
    assert google.requests[0].url.params["topic_name"] == "news"


async def test_topic_list_and_unsubscribe(google, key_file):
    google.on("GET /topicSubscriptions", {"topic_subscriptions": [{"topic_name": "projects/p/topics/news"}]})
    google.on("DELETE /topicSubscriptions/news", {})
    _, _, listed = await fp.manage_topic(_fb(key_file), {"operation": "list", "tokens": "t1"})
    assert listed["topics"] == ["news"]
    ok, _, result = await fp.manage_topic(_fb(key_file), {"operation": "unsubscribe", "tokens": "t1", "topic": "news"})
    assert ok and result["done"] == 1


async def test_connector_tests_touch_no_device(google, key_file):
    google.on("POST messages:send", {"name": "m"})
    ok, detail = await fp.test_credentials(_fb(key_file))
    assert ok and "demo-project" in detail
    assert google.bodies("messages:send")[0]["validate_only"] is True
    google.on("GET drive/v3/about", {"user": {"emailAddress": "bot@x"}})
    ok, detail = await ga.test_credentials({"service_account_json": key_file})
    assert ok and detail == "Connected as bot@x"


# ---- through a real task run -----------------------------------------------------------------


async def _connector(client, auth_headers, ctype, config):
    resp = await client.post("/connectors", headers=auth_headers, json={"name": f"Test {ctype}", "type": ctype, "configuration": config, "status": "enabled"})
    assert resp.status_code == 201, resp.text
    return resp.json()["id"]


async def _run(client, auth_headers, node, context=None):
    nodes = [{"key": "start", "type": "start", "label": "Start", "next_nodes": [{"to": "n"}]}, {"key": "n", "label": "Node", **node}]
    task = await client.post("/tasks", headers=auth_headers, json={"name": "Service node test", "nodes": nodes})
    assert task.status_code == 201, task.text
    resp = await client.post(f"/tasks/{task.json()['id']}/run", headers=auth_headers, json={"context": context or {}, "allow_draft": True})
    assert resp.status_code == 200, resp.text
    return resp.json()


async def test_sheets_node_passes_a_list_through_whole(client, auth_headers, google, key_file):
    google.on("POST :append", {"updates": {"updatedRows": 2}})
    connector = await _connector(client, auth_headers, "google", {"provider": "service_account", "service_account_json": key_file})
    result = await _run(
        client,
        auth_headers,
        {
            "type": "google_sheets",
            "connector_id": connector,
            "service_operation": "append",
            "service_config": {"spreadsheet_id": "{{sheet}}", "range": "Sheet1", "rows": "{{rows}}"},
            "output_params": [{"name": "updated_rows", "type": "int"}],
        },
        {"sheet": "abc123", "rows": [["a", 1], ["b", 2]]},
    )
    assert result["ok"], result["trace"]
    assert result["context"]["updated_rows"] == 2
    assert google.bodies(":append")[0]["values"] == [["a", 1], ["b", 2]]


async def test_docs_export_becomes_a_flowfabric_file(client, auth_headers, google, key_file):
    google.on("GET files/doc123", {"name": "Letter"})
    google.on("GET /doc123/export", httpx.Response(200, content=b"%PDF"))
    connector = await _connector(client, auth_headers, "google", {"provider": "service_account", "service_account_json": key_file})
    result = await _run(
        client,
        auth_headers,
        {"type": "google_docs", "connector_id": connector, "service_operation": "export", "service_config": {"document_id": "doc123", "format": "pdf"}, "output_params": [{"name": "file_id", "type": "string"}, {"name": "file_name", "type": "string"}]},
    )
    assert result["ok"], result["trace"]
    assert result["context"]["file_name"] == "Letter.pdf"


async def test_service_nodes_refuse_secrets_and_the_wrong_connector(client, auth_headers, google, key_file):
    connector = await _connector(client, auth_headers, "firebase", {"provider": "fcm", "service_account_json": key_file})
    wrong = await _run(client, auth_headers, {"type": "google_sheets", "connector_id": connector, "service_operation": "read", "service_config": {"spreadsheet_id": "abc123", "range": "A"}})
    assert not wrong["ok"] and "not a Google connector" in wrong["trace"][-1]["detail"]
    leak = await _run(client, auth_headers, {"type": "firebase_push", "connector_id": connector, "service_config": {"tokens": "t", "title": "{{secret.API_KEY}}"}})
    assert not leak["ok"] and "secrets can only be used" in leak["trace"][-1]["detail"]
    assert google.requests == []


async def test_push_node_end_to_end(client, auth_headers, google, key_file):
    google.on("POST messages:send", {"name": "projects/demo-project/messages/1"})
    connector = await _connector(client, auth_headers, "firebase", {"provider": "fcm", "service_account_json": key_file})
    test = await client.post(f"/connectors/{connector}/test", headers=auth_headers)
    assert test.json()["success"], test.text
    result = await _run(
        client,
        auth_headers,
        {"type": "firebase_push", "connector_id": connector, "service_config": {"tokens": "{{devices}}", "title": "Hi {{name}}"}, "output_params": [{"name": "sent", "type": "int"}, {"name": "invalid_tokens", "type": "list"}]},
        {"devices": ["d1", "d2"], "name": "Asha"},
    )
    assert result["ok"], result["trace"]
    assert result["context"]["sent"] == 2 and result["context"]["invalid_tokens"] == []
    assert google.bodies("messages:send")[-1]["message"]["notification"]["title"] == "Hi Asha"


async def test_connector_providers_are_listed(client, auth_headers):
    for ctype, ids in (("google", {"service_account", "oauth"}), ("firebase", {"fcm"})):
        resp = await client.get(f"/connector-providers/{ctype}", headers=auth_headers)
        assert resp.status_code == 200 and {p["id"] for p in resp.json()} == ids


# ---- SOAP ------------------------------------------------------------------------------------

_WSDL = """<?xml version="1.0"?>
<definitions name="Orders" targetNamespace="urn:orders" xmlns="http://schemas.xmlsoap.org/wsdl/" xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/" xmlns:tns="urn:orders" xmlns:xsd="http://www.w3.org/2001/XMLSchema">
  <types><xsd:schema targetNamespace="urn:orders" elementFormDefault="qualified">
    <xsd:element name="GetStatus"><xsd:complexType><xsd:sequence>
      <xsd:element name="OrderId" type="xsd:string"/>
      <xsd:element name="Note" type="xsd:string" minOccurs="0"/>
    </xsd:sequence></xsd:complexType></xsd:element>
    <xsd:element name="GetStatusResponse"><xsd:complexType><xsd:sequence>
      <xsd:element name="Status" type="xsd:string"/>
      <xsd:element name="Amount" type="xsd:decimal"/>
    </xsd:sequence></xsd:complexType></xsd:element>
  </xsd:schema></types>
  <message name="In"><part name="parameters" element="tns:GetStatus"/></message>
  <message name="Out"><part name="parameters" element="tns:GetStatusResponse"/></message>
  <portType name="OrdersPort"><operation name="GetStatus"><input message="tns:In"/><output message="tns:Out"/></operation></portType>
  <binding name="OrdersBinding" type="tns:OrdersPort">
    <soap:binding style="document" transport="http://schemas.xmlsoap.org/soap/http"/>
    <operation name="GetStatus"><soap:operation soapAction="urn:GetStatus"/><input><soap:body use="literal"/></input><output><soap:body use="literal"/></output></operation>
  </binding>
  <service name="OrdersService"><port name="OrdersSoap" binding="tns:OrdersBinding"><soap:address location="http://127.0.0.1:{port}/soap"/></port></service>
</definitions>"""

_OK = b"""<?xml version="1.0"?><soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"><soap:Body>
<GetStatusResponse xmlns="urn:orders"><Status>SHIPPED</Status><Amount>12.50</Amount></GetStatusResponse></soap:Body></soap:Envelope>"""

_FAULT = b"""<?xml version="1.0"?><soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"><soap:Body>
<soap:Fault><faultcode>soap:Client</faultcode><faultstring>No such order</faultstring></soap:Fault></soap:Body></soap:Envelope>"""


@pytest.fixture
def soap_server():
    seen: list[dict] = []

    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def _send(self, status: int, body: bytes):
            self.send_response(status)
            self.send_header("Content-Type", "text/xml")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self):
            if self.server.require_auth and "Authorization" not in self.headers:
                return self._send(401, b"")
            self._send(200, _WSDL.format(port=self.server.server_address[1]).encode())

        def do_POST(self):
            body = self.rfile.read(int(self.headers.get("Content-Length") or 0))
            seen.append({"body": body.decode(), "headers": dict(self.headers)})
            self._send(500, _FAULT) if b"MISSING" in body else self._send(200, _OK)

    server = HTTPServer(("127.0.0.1", 0), Handler)
    server.require_auth = False
    server.seen = seen
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    yield server
    server.shutdown()


def _wsdl(server) -> str:
    return f"http://127.0.0.1:{server.server_address[1]}/?wsdl"


@pytest.mark.parametrize("url", ["C:/Windows/win.ini", "/etc/passwd", "file:///etc/passwd", "ftp://example.com/x", ""])
def test_soap_refuses_anything_but_http(url):
    with pytest.raises(soap_client.SoapError):
        soap_client.check_url(url)


def test_soap_refuses_the_metadata_address():
    with pytest.raises(soap_client.SoapError):
        soap_client.check_url("http://169.254.169.254/latest/meta-data")


def test_soap_local_file_wsdl_is_refused_before_loading():
    with pytest.raises(soap_client.SoapError):
        soap_client.inspect(soap_client.SoapSpec(wsdl_url="C:/Windows/win.ini"))


def test_entities_are_not_expanded():
    xml = b'<!DOCTYPE d [<!ENTITY x "EXPANDED">]><a><b>&x;</b></a>'
    assert b"EXPANDED" not in soap_client.etree.tostring(soap_client._parse(xml))


def test_xml_becomes_plain_data():
    root = soap_client._parse('<r xmlns="urn:x" id="1"><a>1</a><a>2</a><b><c>t</c></b><d/></r>')
    assert soap_client.element_to_data(root) == {"@id": "1", "a": ["1", "2"], "b": {"c": "t"}, "d": None}


async def test_soap_node_from_wsdl(client, auth_headers, soap_server):
    result = await _run(
        client,
        auth_headers,
        {
            "type": "api_soap",
            "soap_mode": "wsdl",
            "wsdl_url": _wsdl(soap_server),
            "operation": "GetStatus",
            "input_params": [{"name": "OrderId", "type": "string", "source_field": "order"}],
            "output_params": [{"name": "Status", "type": "string"}, {"name": "Amount", "type": "number"}, {"name": "result", "type": "object"}],
        },
        {"order": "A1"},
    )
    assert result["ok"], result["trace"]
    assert result["context"]["Status"] == "SHIPPED" and result["context"]["Amount"] == 12.5
    assert result["context"]["result"] == {"Status": "SHIPPED", "Amount": 12.5}
    assert "<ns0:OrderId>A1</ns0:OrderId>" in soap_server.seen[0]["body"] or "OrderId>A1<" in soap_server.seen[0]["body"]


async def test_soap_node_saved_before_modes_keeps_its_text_result(client, auth_headers, soap_server):
    result = await _run(
        client,
        auth_headers,
        {"type": "api_soap", "wsdl_url": _wsdl(soap_server), "operation": "GetStatus", "input_params": [{"name": "OrderId", "type": "string", "source_field": "order"}], "output_params": [{"name": "response", "type": "string"}]},
        {"order": "A1"},
    )
    assert result["ok"], result["trace"]
    assert "SHIPPED" in result["context"]["response"]


async def test_soap_fault_stops_or_branches(client, auth_headers, soap_server):
    node = {"type": "api_soap", "soap_mode": "wsdl", "wsdl_url": _wsdl(soap_server), "operation": "GetStatus", "input_params": [{"name": "OrderId", "type": "string", "source_field": "order"}]}
    stopped = await _run(client, auth_headers, node, {"order": "MISSING"})
    assert not stopped["ok"] and "No such order" in stopped["trace"][-1]["detail"]
    carried = await _run(client, auth_headers, {**node, "soap_fail_on_fault": False, "output_params": [{"name": "fault_message", "type": "string"}]}, {"order": "MISSING"})
    assert carried["ok"], carried["trace"]
    assert carried["context"]["fault_message"] == "No such order"


async def test_soap_unknown_operation_lists_what_exists(client, auth_headers, soap_server):
    result = await _run(client, auth_headers, {"type": "api_soap", "soap_mode": "wsdl", "wsdl_url": _wsdl(soap_server), "operation": "Nope"})
    assert not result["ok"] and "GetStatus" in result["trace"][-1]["detail"]


async def test_soap_raw_mode_escapes_values_and_reads_the_answer(client, auth_headers, soap_server):
    envelope = '<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"><soap:Body><Q><Id>{{order}}</Id></Q></soap:Body></soap:Envelope>'
    result = await _run(
        client,
        auth_headers,
        {
            "type": "api_soap",
            "soap_mode": "raw",
            "soap_endpoint": f"http://127.0.0.1:{soap_server.server_address[1]}/soap",
            "soap_action": "urn:Q",
            "soap_envelope": envelope,
            "input_params": [{"name": "order", "type": "string", "source_field": "order"}],
            "output_params": [{"name": "Status", "type": "string"}, {"name": "status_code", "type": "any", "source": "xml"}],
        },
        {"order": "a</Id><Evil>1</Evil><Id>&"},
    )
    assert result["ok"], result["trace"]
    assert result["context"]["Status"] == "SHIPPED"
    sent = soap_server.seen[0]
    assert "<Evil>" not in sent["body"] and "&lt;Evil&gt;" in sent["body"]
    assert sent["headers"]["SOAPAction"] == '"urn:Q"' and sent["headers"]["Content-Type"].startswith("text/xml")


async def test_soap_raw_12_uses_its_content_type(soap_server):
    spec = soap_client.SoapSpec(endpoint=f"http://127.0.0.1:{soap_server.server_address[1]}/soap", envelope="<a/>", soap_action="urn:X", version="1.2")
    # The stub answers with an Envelope either way; only the request matters here.
    soap_client.call_raw(spec)
    assert soap_server.seen[0]["headers"]["Content-Type"] == 'application/soap+xml; charset=utf-8; action="urn:X"'


async def test_soap_password_comes_from_a_secret_and_stays_out_of_the_trace(client, auth_headers, soap_server):
    made = await client.post("/secrets", headers=auth_headers, json={"name": "SOAP_PASS", "value": "s3cret-value-123"})
    assert made.status_code in (200, 201), made.text
    soap_server.require_auth = True
    result = await _run(
        client,
        auth_headers,
        {"type": "api_soap", "soap_mode": "wsdl", "wsdl_url": _wsdl(soap_server), "operation": "GetStatus", "soap_auth": "basic", "soap_username": "u", "soap_password": "{{secret.SOAP_PASS}}", "input_params": [{"name": "OrderId", "type": "string", "value": "A1"}]},
    )
    assert result["ok"], result["trace"]
    expected = "Basic " + base64.b64encode(b"u:s3cret-value-123").decode()
    assert soap_server.seen[0]["headers"]["Authorization"] == expected
    assert "s3cret-value-123" not in json.dumps(result)


async def test_soap_wsse_goes_in_the_envelope(soap_server):
    spec = soap_client.SoapSpec(wsdl_url=_wsdl(soap_server), operation="GetStatus", auth="wsse_text", username="u", password="p")
    answer = soap_client.call_wsdl(spec, {"OrderId": "A1"})
    assert answer["ok"] and "UsernameToken" in soap_server.seen[0]["body"]


async def test_soap_custom_header_and_endpoint_override(soap_server):
    port = soap_server.server_address[1]
    spec = soap_client.SoapSpec(wsdl_url=_wsdl(soap_server), operation="GetStatus", endpoint=f"http://localhost:{port}/other", header_xml='<K xmlns="urn:k">v</K>', headers={"X-Api": "1"})
    soap_client.call_wsdl(spec, {"OrderId": "A1"})
    sent = soap_server.seen[0]
    assert "urn:k" in sent["body"] and sent["headers"]["X-Api"] == "1" and sent["headers"]["Host"].startswith("localhost")


async def test_inspect_lists_operations_and_their_arguments(client, auth_headers, soap_server):
    resp = await client.post("/tasks/soap/inspect", headers=auth_headers, json={"wsdl_url": _wsdl(soap_server)})
    assert resp.status_code == 200, resp.text
    op = resp.json()["operations"][0]
    assert op["name"] == "GetStatus" and op["port"] == "OrdersSoap"
    assert {i["name"]: i["required"] for i in op["inputs"]} == {"OrderId": True, "Note": False}
    bad = await client.post("/tasks/soap/inspect", headers=auth_headers, json={"wsdl_url": "C:/Windows/win.ini"})
    assert bad.status_code == 400