lywsvip/ironclaw
1
0
Fork 0
mirror of https://github.com/nearai/ironclaw.git synced 2026-09-03 08:06:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

test(e2e): cover all Emulate provider operations (#6563)

Browse Source 
* test(e2e): cover all Emulate provider operations

* test(e2e): consolidate provider operation helpers
This commit is contained in:
firat.sertgoz
2026-07-23 15:39:51 +03:00
committed by GitHub
parent 346db3cc6a
commit 1fca2bd52c
17 changed files with 1957 additions and 437 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/reborn-e2e.yml vendored
View File

@@ -150,7 +150,7 @@ jobs:
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6
with:
repository: serrrfirat/emulate
ref: 92bd9c2157a74613c47549eb5cfcccc7c0740adf
ref: 7f0175ad65a884ad86b3897c527cc40336e1391b
path: .emulate
persist-credentials: false

21
tests/e2e/CLAUDE.md
View File

@@ -39,7 +39,7 @@ pip install -e .
playwright install chromium
```
Dependencies: `pytest`, `pytest-asyncio`, `pytest-playwright`, `pytest-timeout`, `playwright`, `aiohttp`, `httpx`. Optional: `anthropic` (vision extras). Requires Python >= 3.11. Emulate-backed provider tests also require Node.js. CI installs Node 24, builds `serrrfirat/emulate` at commit `92bd9c2157a74613c47549eb5cfcccc7c0740adf`, and passes its CLI through `IRONCLAW_EMULATE_CLI`. Without that override, unrelated local Emulate tests retain the `emulate@0.7.0` fallback.
Dependencies: `pytest`, `pytest-asyncio`, `pytest-playwright`, `pytest-timeout`, `playwright`, `aiohttp`, `httpx`. Optional: `anthropic` (vision extras). Requires Python >= 3.11. Emulate-backed provider tests also require Node.js. CI installs Node 24, builds `serrrfirat/emulate` at commit `7f0175ad65a884ad86b3897c527cc40336e1391b`, and passes its CLI through `IRONCLAW_EMULATE_CLI`. Without that override, unrelated local Emulate tests retain the `emulate@0.7.0` fallback.
## Running Tests
@@ -208,15 +208,16 @@ gates, and read provider state back from Emulate. This is the contract tier to
use when the behavior being protected is extension install/auth, model-to-tool
routing, tool execution, and provider mutation together.
The pinned `serrrfirat/emulate` fork adds the Google Docs and Sheets operations
and Slack `search.messages` used by the harvested trace catalog. Google Slides
remains outside the provider fixture because no harvested trace uses it. Manual
QA rows that mention Telegram, Twitter/X, or HN/web search are likewise
model-replay-only unless paired with a separate fake/provider fixture.
GitHub `/contents` file create/update/delete and workflow dispatch also remain
outside direct Emulate 0.7.0 coverage: the emulator exposes Git data APIs but no
`/contents` routes, and it exposes Actions routes but does not let fixture seeds
define workflows to dispatch.
The pinned `serrrfirat/emulate` fork adds the Google Calendar, Docs, Drive, and
Sheets operations and Slack `search.messages` used by the provider contract
catalog. Google Slides remains outside the provider fixture because Emulate has
no Slides routes. Manual QA rows that mention Telegram, Twitter/X, or HN/web
search are likewise model-replay-only unless paired with a separate
fake/provider fixture. GitHub `/contents` file create/update/delete, GraphQL
review threads, and workflow dispatch remain outside direct Emulate coverage:
the emulator exposes Git data APIs but no `/contents` or GraphQL routes, and it
exposes Actions routes but does not let fixture seeds define workflows to
dispatch.
### Environment passed to ironclaw in tests

24
tests/e2e/emulate_provider.py
View File

@@ -23,6 +23,30 @@ def github_headers(token: str = EMULATE_GITHUB_BEARER) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
async def google_json(
base_url: str,
method: str,
path: str,
*,
payload: dict | None = None,
params: dict | list[tuple[str, str | int]] | None = None,
expected_status: int = 200,
) -> dict | list:
async with httpx.AsyncClient(headers=google_headers(), timeout=15) as client:
response = await client.request(
method,
f"{base_url}{path}",
json=payload,
params=params,
)
assert response.status_code == expected_status, (
f"Google {method} {path} returned {response.status_code}: {response.text}"
)
if not response.content:
return {}
return response.json()
def gmail_header(message: dict, name: str) -> str | None:
for header in message.get("payload", {}).get("headers", []):
if header.get("name", "").lower() == name.lower():

16
tests/e2e/fixtures/emulate/google_gmail.yaml
View File

@@ -146,6 +146,18 @@ google:
parent_google_ids:
- root
data: "This Drive item belongs only to the secondary identity."
drive_permissions:
- id: perm_reborn_reader
user_email: e2e.google@example.com
file_id: drv_reborn_qa_brief
role: reader
type: user
email_address: seeded-reader@example.com
display_name: Seeded Reader
shared_drives:
- id: shared_reborn_engineering
user_email: e2e.google@example.com
name: Reborn Engineering
documents:
- id: doc_reborn_strategy
user_email: e2e.google@example.com
@@ -161,6 +173,10 @@ google:
values:
- [Company, Contact, Source, Status, QA Marker]
- [NEAR AI, partnerships@near.ai, Gmail, New, REBORN_QA_SEEDED]
- id: 7
title: DeleteMe
values:
- [Temporary]
- id: sheet_reborn_bug_tracker
user_email: e2e.google@example.com
title: Reborn QA Bug Tracker

140
tests/e2e/fixtures/provider_capability_coverage.toml
View File

@@ -4,30 +4,85 @@ schema_version = 1
# case executes the real extension boundary and observes Emulate.
[classifications]
tested = [
"github.add_issue_assignees",
"github.add_issue_labels",
"github.comment_issue",
"github.create_branch",
"github.create_issue",
"github.create_issue_comment",
"github.create_pr_review",
"github.create_pull_request",
"github.create_release",
"github.create_repo",
"github.fork_repo",
"github.get_authenticated_user",
"github.get_issue",
"github.get_pull_request",
"github.get_pull_request_files",
"github.get_pull_request_reviews",
"github.get_repo",
"github.get_workflow_runs",
"github.list_branches",
"github.list_issue_comments",
"github.list_issues",
"github.list_pull_request_comments",
"github.list_pull_requests",
"github.list_releases",
"github.list_repos",
"github.merge_pull_request",
"github.remove_issue_assignees",
"github.remove_issue_label",
"github.search_code",
"github.search_issues",
"github.search_issues_pull_requests",
"github.search_repositories",
"github.update_issue",
"github.update_pull_request",
"gmail.create_draft",
"gmail.get_message",
"gmail.list_messages",
"gmail.reply_to_message",
"gmail.send_message",
"gmail.trash_message",
"google-calendar.add_attendees",
"google-calendar.create_event",
"google-calendar.delete_event",
"google-calendar.find_free_slots",
"google-calendar.get_event",
"google-calendar.list_calendars",
"google-calendar.list_events",
"google-calendar.set_reminder",
"google-calendar.update_event",
"google-docs.batch_update",
"google-docs.create_document",
"google-docs.create_list",
"google-docs.delete_content",
"google-docs.format_paragraph",
"google-docs.format_text",
"google-docs.get_document",
"google-docs.insert_text",
"google-docs.insert_table",
"google-docs.read_content",
"google-docs.replace_text",
"google-drive.create_folder",
"google-drive.delete_file",
"google-drive.download_file",
"google-drive.get_file",
"google-drive.list_files",
"google-drive.list_permissions",
"google-drive.list_shared_drives",
"google-drive.remove_permission",
"google-drive.share_file",
"google-drive.trash_file",
"google-drive.update_file",
"google-drive.upload_file",
"google-sheets.add_sheet",
"google-sheets.append_values",
"google-sheets.batch_read_values",
"google-sheets.clear_values",
"google-sheets.create_spreadsheet",
"google-sheets.delete_sheet",
"google-sheets.format_cells",
"google-sheets.get_spreadsheet",
"google-sheets.read_values",
"google-sheets.rename_sheet",
@@ -47,80 +102,41 @@ unsupported = []
[[waivers]]
owner = "@serrrfirat"
reason = "No full-path hermetic provider-operation case exists yet."
reason = "Emulate lacks the matching GitHub route or seedable successful Actions state."
issue = "#6524"
review_condition = "Replace with tested, live_only, or unsupported classification before closing #6524."
review_condition = "Revisit when Emulate adds the missing route or seed configuration."
capabilities = [
"github.add_issue_assignees",
"github.add_issue_labels",
"github.comment_issue",
"github.create_branch",
"github.create_issue",
"github.create_issue_comment",
"github.create_or_update_file",
"github.create_pr_review",
"github.create_pull_request",
"github.create_release",
"github.create_repo",
"github.delete_file",
"github.fork_repo",
"github.get_combined_status",
"github.get_file_content",
"github.get_issue",
"github.get_job_logs",
"github.get_pull_request",
"github.get_pull_request_files",
"github.get_pull_request_reviews",
"github.get_workflow_run_artifacts",
"github.get_workflow_run_jobs",
"github.get_workflow_runs",
"github.handle_webhook",
"github.list_branches",
"github.list_issue_comments",
"github.list_issues",
"github.list_pull_request_comments",
"github.list_pull_request_review_threads",
"github.list_pull_requests",
"github.list_repos",
"github.merge_pull_request",
"github.remove_issue_assignees",
"github.remove_issue_label",
"github.reply_pull_request_comment",
"github.rerun_failed_workflow_run_jobs",
"github.rerun_workflow_job",
"github.resolve_review_thread",
"github.search_code",
"github.search_issues",
"github.search_issues_pull_requests",
"github.search_repositories",
"github.trigger_workflow",
"github.unresolve_review_thread",
"github.update_issue",
"github.update_pull_request",
"google-calendar.add_attendees",
"google-calendar.find_free_slots",
"google-calendar.get_event",
"google-calendar.set_reminder",
"google-calendar.update_event",
"google-docs.batch_update",
"google-docs.create_list",
"google-docs.delete_content",
"google-docs.format_paragraph",
"google-docs.format_text",
"google-docs.get_document",
"google-docs.insert_table",
"google-docs.replace_text",
"google-drive.delete_file",
"google-drive.list_permissions",
"google-drive.list_shared_drives",
"google-drive.remove_permission",
"google-drive.share_file",
"google-drive.trash_file",
"google-sheets.add_sheet",
"google-sheets.batch_read_values",
"google-sheets.clear_values",
"google-sheets.delete_sheet",
"google-sheets.format_cells",
]
[[waivers]]
owner = "@serrrfirat"
reason = "Webhook normalization is local logic and does not cross a provider boundary."
issue = "#6524"
review_condition = "Add a caller-path webhook ingress contract outside the Emulate provider runner."
capabilities = [
"github.handle_webhook",
]
[[waivers]]
owner = "@serrrfirat"
reason = "Emulate has no Google Slides provider routes."
issue = "#6524"
review_condition = "Revisit when a hermetic Slides provider double is available."
capabilities = [
"google-slides.batch_update",
"google-slides.create_presentation",
"google-slides.create_shape",
@@ -135,6 +151,14 @@ capabilities = [
"google-slides.insert_text",
"google-slides.replace_all_text",
"google-slides.replace_shapes_with_image",
]
[[waivers]]
owner = "@serrrfirat"
reason = "These search tools need a separate hermetic web-search/content provider double."
issue = "#6524"
review_condition = "Replace with tested when the non-Emulate provider fixture exists."
capabilities = [
"nearai.web_search",
"web-access.get_content",
"web-access.search",

29
tests/e2e/provider_operation_cases.py
View File

@@ -1,6 +1,29 @@
"""Registry for typed full-path provider operation cases."""
from provider_operation_google_cases import GOOGLE_PROVIDER_OPERATION_CASES
from provider_operation_types import ProviderOperationCase
from provider_operation_github_issue_cases import (
GITHUB_ISSUE_PROVIDER_OPERATION_CASES,
)
from provider_operation_github_pull_cases import (
GITHUB_PULL_PROVIDER_OPERATION_CASES,
)
from provider_operation_github_repo_cases import (
GITHUB_REPO_PROVIDER_OPERATION_CASES,
)
from provider_operation_gmail_cases import GMAIL_PROVIDER_OPERATION_CASES
from provider_operation_google_calendar_cases import (
GOOGLE_CALENDAR_PROVIDER_OPERATION_CASES,
)
from provider_operation_google_docs_cases import GOOGLE_DOCS_PROVIDER_OPERATION_CASES
from provider_operation_google_drive_cases import GOOGLE_DRIVE_PROVIDER_OPERATION_CASES
from provider_operation_google_sheets_cases import GOOGLE_SHEETS_PROVIDER_OPERATION_CASES
PROVIDER_OPERATION_CASES = GOOGLE_PROVIDER_OPERATION_CASES
PROVIDER_OPERATION_CASES = (
*GMAIL_PROVIDER_OPERATION_CASES,
*GOOGLE_CALENDAR_PROVIDER_OPERATION_CASES,
*GOOGLE_DOCS_PROVIDER_OPERATION_CASES,
*GOOGLE_DRIVE_PROVIDER_OPERATION_CASES,
*GOOGLE_SHEETS_PROVIDER_OPERATION_CASES,
*GITHUB_ISSUE_PROVIDER_OPERATION_CASES,
*GITHUB_PULL_PROVIDER_OPERATION_CASES,
*GITHUB_REPO_PROVIDER_OPERATION_CASES,
)

138
tests/e2e/provider_operation_github_common.py Normal file
View File

@@ -0,0 +1,138 @@
"""Shared GitHub provider setup and readback helpers."""
import httpx
from emulate_provider import github_json
OWNER = "nearai"
REPO = "ironclaw"
REPO_PATH = f"/repos/{OWNER}/{REPO}"
ISSUE_TITLE = "REBORN_PROVIDER_CASE_SEEDED_ISSUE"
PR_TITLE = "REBORN_PROVIDER_CASE_SEEDED_PR"
BRANCH = "provider-case-branch"
CODE_PATH = "docs/provider-case.md"
CODE_MARKER = "REBORN_PROVIDER_CASE_CODE_MARKER"
async def github_request(
emulate_url: str,
method: str,
path: str,
*,
payload: dict | None = None,
params: dict | None = None,
expected_status: int = 200,
) -> dict | list:
async with httpx.AsyncClient(timeout=15) as client:
return await github_json(
client,
emulate_url,
method,
path,
payload=payload,
params=params,
expected_status=expected_status,
)
async def seed_issue(emulate_url: str, *, title: str = ISSUE_TITLE) -> dict:
result = await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/issues",
payload={"title": title, "body": "Seeded by the provider operation case."},
expected_status=201,
)
assert isinstance(result, dict)
assert result["number"] == 1, result
return result
async def issue(emulate_url: str) -> dict:
result = await github_request(emulate_url, "GET", f"{REPO_PATH}/issues/1")
assert isinstance(result, dict)
return result
async def seed_branch(emulate_url: str, *, branch: str = BRANCH) -> dict:
main_ref = await github_request(
emulate_url, "GET", f"{REPO_PATH}/git/ref/heads/main"
)
assert isinstance(main_ref, dict)
main_sha = main_ref["object"]["sha"]
main_commit = await github_request(
emulate_url, "GET", f"{REPO_PATH}/git/commits/{main_sha}"
)
assert isinstance(main_commit, dict)
blob = await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/git/blobs",
payload={"content": CODE_MARKER, "encoding": "utf-8"},
expected_status=201,
)
assert isinstance(blob, dict)
tree = await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/git/trees",
payload={
"base_tree": main_commit["commit"]["tree"]["sha"],
"tree": [
{
"path": CODE_PATH,
"mode": "100644",
"type": "blob",
"sha": blob["sha"],
}
],
},
expected_status=201,
)
assert isinstance(tree, dict)
commit = await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/git/commits",
payload={
"message": "test: seed provider operation branch",
"tree": tree["sha"],
"parents": [main_sha],
},
expected_status=201,
)
assert isinstance(commit, dict)
result = await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/git/refs",
payload={"ref": f"refs/heads/{branch}", "sha": commit["sha"]},
expected_status=201,
)
assert isinstance(result, dict)
return result
async def seed_pull_request(emulate_url: str) -> dict:
await seed_branch(emulate_url)
result = await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/pulls",
payload={
"title": PR_TITLE,
"head": BRANCH,
"base": "main",
"body": "Seeded by the provider operation case.",
},
expected_status=201,
)
assert isinstance(result, dict)
assert result["number"] == 1, result
return result
async def pull_request(emulate_url: str) -> dict:
result = await github_request(emulate_url, "GET", f"{REPO_PATH}/pulls/1")
assert isinstance(result, dict)
return result

269
tests/e2e/provider_operation_github_issue_cases.py Normal file
View File

@@ -0,0 +1,269 @@
"""GitHub issue full-path provider operation cases."""
import json
from provider_operation_github_common import (
ISSUE_TITLE,
OWNER,
REPO,
REPO_PATH,
issue,
github_request,
seed_issue,
)
from provider_operation_types import ProviderOperationCase
CREATED_ISSUE = "REBORN_PROVIDER_CASE_CREATED_ISSUE"
UPDATED_ISSUE = "REBORN_PROVIDER_CASE_UPDATED_ISSUE"
COMMENT = "REBORN_PROVIDER_CASE_ISSUE_COMMENT"
SEEDED_COMMENT = "REBORN_PROVIDER_CASE_SEEDED_COMMENT"
LABEL = "provider-case"
ASSIGNEE = "reborn-reviewer"
BASE_ARGS = {"owner": OWNER, "repo": REPO}
async def _empty_issues(emulate_url: str) -> None:
items = await github_request(emulate_url, "GET", f"{REPO_PATH}/issues")
assert items == [], items
async def _seeded_issue(emulate_url: str) -> None:
await seed_issue(emulate_url)
async def _seeded_label_issue(emulate_url: str, *, assigned: bool) -> None:
await seed_issue(emulate_url)
await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/labels",
payload={"name": LABEL, "color": "3367d6"},
expected_status=201,
)
if assigned:
await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/issues/1/labels",
payload={"labels": [LABEL]},
)
async def _label_add_baseline(emulate_url: str) -> None:
await _seeded_label_issue(emulate_url, assigned=False)
async def _label_remove_baseline(emulate_url: str) -> None:
await _seeded_label_issue(emulate_url, assigned=True)
async def _seeded_assignee_issue(emulate_url: str, *, assigned: bool) -> None:
await seed_issue(emulate_url)
if assigned:
await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/issues/1/assignees",
payload={"assignees": [ASSIGNEE]},
)
async def _assignee_add_baseline(emulate_url: str) -> None:
await _seeded_assignee_issue(emulate_url, assigned=False)
async def _assignee_remove_baseline(emulate_url: str) -> None:
await _seeded_assignee_issue(emulate_url, assigned=True)
async def _seeded_comment_baseline(emulate_url: str) -> None:
await seed_issue(emulate_url)
await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/issues/1/comments",
payload={"body": SEEDED_COMMENT},
expected_status=201,
)
async def _created_issue_outcome(emulate_url: str, preview: dict) -> None:
created = await issue(emulate_url)
assert created["title"] == CREATED_ISSUE, created
assert CREATED_ISSUE in json.dumps(preview), preview
async def _seeded_issue_read_outcome(emulate_url: str, preview: dict) -> None:
created = await issue(emulate_url)
assert created["title"] == ISSUE_TITLE, created
assert ISSUE_TITLE in json.dumps(preview), preview
async def _updated_issue_outcome(emulate_url: str, preview: dict) -> None:
updated = await issue(emulate_url)
assert updated["title"] == UPDATED_ISSUE, updated
assert updated["state"] == "closed", updated
assert UPDATED_ISSUE in json.dumps(preview), preview
def _comment_outcome(marker: str):
async def assert_outcome(emulate_url: str, preview: dict) -> None:
comments = await github_request(
emulate_url, "GET", f"{REPO_PATH}/issues/1/comments"
)
assert isinstance(comments, list)
assert [comment["body"] for comment in comments] == [marker], comments
assert marker in json.dumps(preview), preview
return assert_outcome
async def _label_add_outcome(emulate_url: str, preview: dict) -> None:
updated = await issue(emulate_url)
assert [label["name"] for label in updated["labels"]] == [LABEL], updated
assert LABEL in json.dumps(preview), preview
async def _label_remove_outcome(emulate_url: str, preview: dict) -> None:
updated = await issue(emulate_url)
assert updated["labels"] == [], updated
assert LABEL in json.dumps(preview), preview
async def _assignee_add_outcome(emulate_url: str, preview: dict) -> None:
updated = await issue(emulate_url)
assert [assignee["login"] for assignee in updated["assignees"]] == [
ASSIGNEE
], updated
assert ASSIGNEE in json.dumps(preview), preview
async def _assignee_remove_outcome(emulate_url: str, preview: dict) -> None:
updated = await issue(emulate_url)
assert updated["assignees"] == [], updated
assert ASSIGNEE in json.dumps(preview), preview
GITHUB_ISSUE_PROVIDER_OPERATION_CASES = (
ProviderOperationCase(
case_id="github_create_issue",
provider_service="github",
capability_id="github.create_issue",
arguments={
**BASE_ARGS,
"title": CREATED_ISSUE,
"body": "Created through the provider operation runner.",
},
assert_baseline=_empty_issues,
assert_outcome=_created_issue_outcome,
),
ProviderOperationCase(
case_id="github_get_issue",
provider_service="github",
capability_id="github.get_issue",
arguments={**BASE_ARGS, "issue_number": 1},
assert_baseline=_seeded_issue,
assert_outcome=_seeded_issue_read_outcome,
),
ProviderOperationCase(
case_id="github_list_issues",
provider_service="github",
capability_id="github.list_issues",
arguments={**BASE_ARGS, "state": "open"},
assert_baseline=_seeded_issue,
assert_outcome=_seeded_issue_read_outcome,
),
ProviderOperationCase(
case_id="github_search_issues",
provider_service="github",
capability_id="github.search_issues",
arguments={
"query": ISSUE_TITLE,
"repository": f"{OWNER}/{REPO}",
"type": "issue",
},
assert_baseline=_seeded_issue,
assert_outcome=_seeded_issue_read_outcome,
),
ProviderOperationCase(
case_id="github_search_issues_pull_requests",
provider_service="github",
capability_id="github.search_issues_pull_requests",
arguments={
"query": ISSUE_TITLE,
"repository": f"{OWNER}/{REPO}",
"type": "issue",
},
assert_baseline=_seeded_issue,
assert_outcome=_seeded_issue_read_outcome,
),
ProviderOperationCase(
case_id="github_update_issue",
provider_service="github",
capability_id="github.update_issue",
arguments={
**BASE_ARGS,
"issue_number": 1,
"title": UPDATED_ISSUE,
"state": "closed",
},
assert_baseline=_seeded_issue,
assert_outcome=_updated_issue_outcome,
),
ProviderOperationCase(
case_id="github_create_issue_comment",
provider_service="github",
capability_id="github.create_issue_comment",
arguments={**BASE_ARGS, "issue_number": 1, "body": COMMENT},
assert_baseline=_seeded_issue,
assert_outcome=_comment_outcome(COMMENT),
),
ProviderOperationCase(
case_id="github_comment_issue",
provider_service="github",
capability_id="github.comment_issue",
arguments={**BASE_ARGS, "issue_number": 1, "body": COMMENT},
assert_baseline=_seeded_issue,
assert_outcome=_comment_outcome(COMMENT),
),
ProviderOperationCase(
case_id="github_list_issue_comments",
provider_service="github",
capability_id="github.list_issue_comments",
arguments={**BASE_ARGS, "issue_number": 1},
assert_baseline=_seeded_comment_baseline,
assert_outcome=_comment_outcome(SEEDED_COMMENT),
),
ProviderOperationCase(
case_id="github_add_issue_labels",
provider_service="github",
capability_id="github.add_issue_labels",
arguments={**BASE_ARGS, "issue_number": 1, "labels": [LABEL]},
assert_baseline=_label_add_baseline,
assert_outcome=_label_add_outcome,
),
ProviderOperationCase(
case_id="github_remove_issue_label",
provider_service="github",
capability_id="github.remove_issue_label",
arguments={**BASE_ARGS, "issue_number": 1, "name": LABEL},
assert_baseline=_label_remove_baseline,
assert_outcome=_label_remove_outcome,
),
ProviderOperationCase(
case_id="github_add_issue_assignees",
provider_service="github",
capability_id="github.add_issue_assignees",
arguments={**BASE_ARGS, "issue_number": 1, "assignees": [ASSIGNEE]},
assert_baseline=_assignee_add_baseline,
assert_outcome=_assignee_add_outcome,
),
ProviderOperationCase(
case_id="github_remove_issue_assignees",
provider_service="github",
capability_id="github.remove_issue_assignees",
arguments={**BASE_ARGS, "issue_number": 1, "assignees": [ASSIGNEE]},
assert_baseline=_assignee_remove_baseline,
assert_outcome=_assignee_remove_outcome,
),
)

212
tests/e2e/provider_operation_github_pull_cases.py Normal file
View File

@@ -0,0 +1,212 @@
"""GitHub pull request full-path provider operation cases."""
import json
from provider_operation_github_common import (
BRANCH,
CODE_PATH,
OWNER,
PR_TITLE,
REPO,
REPO_PATH,
pull_request,
github_request,
seed_branch,
seed_pull_request,
)
from provider_operation_types import ProviderOperationCase
CREATED_PR = "REBORN_PROVIDER_CASE_CREATED_PR"
UPDATED_PR = "REBORN_PROVIDER_CASE_UPDATED_PR"
REVIEW_BODY = "REBORN_PROVIDER_CASE_REVIEW"
INLINE_COMMENT = "REBORN_PROVIDER_CASE_INLINE_COMMENT"
BASE_ARGS = {"owner": OWNER, "repo": REPO}
async def _seeded_branch(emulate_url: str) -> None:
await seed_branch(emulate_url)
pulls = await github_request(emulate_url, "GET", f"{REPO_PATH}/pulls")
assert pulls == [], pulls
async def _seeded_pull(emulate_url: str) -> None:
await seed_pull_request(emulate_url)
async def _seeded_review(emulate_url: str) -> None:
await seed_pull_request(emulate_url)
await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/pulls/1/reviews",
payload={"body": REVIEW_BODY, "event": "COMMENT"},
expected_status=201,
)
async def _seeded_inline_comment(emulate_url: str) -> None:
await seed_pull_request(emulate_url)
await github_request(
emulate_url,
"POST",
f"{REPO_PATH}/pulls/1/reviews",
payload={
"body": REVIEW_BODY,
"event": "COMMENT",
"comments": [
{
"path": CODE_PATH,
"position": 1,
"body": INLINE_COMMENT,
}
],
},
expected_status=201,
)
async def _created_pull_outcome(emulate_url: str, preview: dict) -> None:
created = await pull_request(emulate_url)
assert created["title"] == CREATED_PR, created
assert created["head"]["ref"] == BRANCH, created
assert CREATED_PR in json.dumps(preview), preview
async def _seeded_pull_outcome(emulate_url: str, preview: dict) -> None:
seeded = await pull_request(emulate_url)
assert seeded["title"] == PR_TITLE, seeded
assert PR_TITLE in json.dumps(preview), preview
async def _updated_pull_outcome(emulate_url: str, preview: dict) -> None:
updated = await pull_request(emulate_url)
assert updated["title"] == UPDATED_PR, updated
assert UPDATED_PR in json.dumps(preview), preview
async def _pull_files_outcome(emulate_url: str, preview: dict) -> None:
files = await github_request(emulate_url, "GET", f"{REPO_PATH}/pulls/1/files")
assert isinstance(files, list)
# Emulate currently models this successful empty state but not generated
# pull-request diffs. The seeded branch still proves the PR is real.
assert files == [], files
assert preview["output_preview"] == "[]", preview
async def _create_review_outcome(emulate_url: str, preview: dict) -> None:
reviews = await github_request(
emulate_url, "GET", f"{REPO_PATH}/pulls/1/reviews"
)
assert isinstance(reviews, list)
assert [review["body"] for review in reviews] == [REVIEW_BODY], reviews
assert REVIEW_BODY in json.dumps(preview), preview
async def _list_comments_outcome(emulate_url: str, preview: dict) -> None:
comments = await github_request(
emulate_url, "GET", f"{REPO_PATH}/pulls/1/comments"
)
assert isinstance(comments, list)
assert [comment["body"] for comment in comments] == [
INLINE_COMMENT
], comments
assert INLINE_COMMENT in json.dumps(preview), preview
async def _merge_outcome(emulate_url: str, preview: dict) -> None:
merged = await pull_request(emulate_url)
assert merged["merged"] is True, merged
assert merged["state"] == "closed", merged
assert json.loads(preview["output_preview"])["merged"] is True, preview
GITHUB_PULL_PROVIDER_OPERATION_CASES = (
ProviderOperationCase(
case_id="github_create_pull_request",
provider_service="github",
capability_id="github.create_pull_request",
arguments={
**BASE_ARGS,
"title": CREATED_PR,
"head": BRANCH,
"base": "main",
"body": "Created through the provider operation runner.",
},
assert_baseline=_seeded_branch,
assert_outcome=_created_pull_outcome,
),
ProviderOperationCase(
case_id="github_get_pull_request",
provider_service="github",
capability_id="github.get_pull_request",
arguments={**BASE_ARGS, "pr_number": 1},
assert_baseline=_seeded_pull,
assert_outcome=_seeded_pull_outcome,
),
ProviderOperationCase(
case_id="github_list_pull_requests",
provider_service="github",
capability_id="github.list_pull_requests",
arguments={**BASE_ARGS, "state": "open"},
assert_baseline=_seeded_pull,
assert_outcome=_seeded_pull_outcome,
),
ProviderOperationCase(
case_id="github_update_pull_request",
provider_service="github",
capability_id="github.update_pull_request",
arguments={**BASE_ARGS, "pr_number": 1, "title": UPDATED_PR},
assert_baseline=_seeded_pull,
assert_outcome=_updated_pull_outcome,
),
ProviderOperationCase(
case_id="github_get_pull_request_files",
provider_service="github",
capability_id="github.get_pull_request_files",
arguments={**BASE_ARGS, "pr_number": 1},
assert_baseline=_seeded_pull,
assert_outcome=_pull_files_outcome,
),
ProviderOperationCase(
case_id="github_create_pr_review",
provider_service="github",
capability_id="github.create_pr_review",
arguments={
**BASE_ARGS,
"pr_number": 1,
"body": REVIEW_BODY,
"event": "COMMENT",
},
assert_baseline=_seeded_pull,
assert_outcome=_create_review_outcome,
),
ProviderOperationCase(
case_id="github_get_pull_request_reviews",
provider_service="github",
capability_id="github.get_pull_request_reviews",
arguments={**BASE_ARGS, "pr_number": 1},
assert_baseline=_seeded_review,
assert_outcome=_create_review_outcome,
),
ProviderOperationCase(
case_id="github_list_pull_request_comments",
provider_service="github",
capability_id="github.list_pull_request_comments",
arguments={**BASE_ARGS, "pr_number": 1},
assert_baseline=_seeded_inline_comment,
assert_outcome=_list_comments_outcome,
),
ProviderOperationCase(
case_id="github_merge_pull_request",
provider_service="github",
capability_id="github.merge_pull_request",
arguments={
**BASE_ARGS,
"pr_number": 1,
"merge_method": "squash",
"commit_title": "Merge provider operation case",
},
assert_baseline=_seeded_pull,
assert_outcome=_merge_outcome,
),
)

217
tests/e2e/provider_operation_github_repo_cases.py Normal file
View File

@@ -0,0 +1,217 @@
"""GitHub repository, branch, release, search, and Actions cases."""
import json
from provider_operation_github_common import (
BRANCH,
CODE_MARKER,
CODE_PATH,
OWNER,
REPO,
REPO_PATH,
github_request,
seed_branch,
)
from provider_operation_types import ProviderOperationCase
CREATED_BRANCH = "provider-case-created-branch"
CREATED_REPO = "reborn-provider-case-repo"
SEEDED_REPO = "reborn-provider-case-seeded-repo"
FORK_NAME = "reborn-provider-case-fork"
RELEASE_TAG = "provider-case-v1"
BASE_ARGS = {"owner": OWNER, "repo": REPO}
async def _repo_baseline(emulate_url: str) -> None:
repo = await github_request(emulate_url, "GET", REPO_PATH)
assert isinstance(repo, dict)
assert repo["full_name"] == f"{OWNER}/{REPO}", repo
async def _empty_release_baseline(emulate_url: str) -> None:
await _repo_baseline(emulate_url)
releases = await github_request(emulate_url, "GET", f"{REPO_PATH}/releases")
assert releases == [], releases
async def _seeded_user_repo(emulate_url: str) -> None:
await github_request(
emulate_url,
"POST",
"/user/repos",
payload={"name": SEEDED_REPO, "auto_init": True},
expected_status=201,
)
async def _seed_code(emulate_url: str) -> None:
await seed_branch(emulate_url)
async def _create_branch_outcome(emulate_url: str, preview: dict) -> None:
branches = await github_request(emulate_url, "GET", f"{REPO_PATH}/branches")
assert isinstance(branches, list)
assert {branch["name"] for branch in branches} == {
"main",
CREATED_BRANCH,
}, branches
assert CREATED_BRANCH in json.dumps(preview), preview
async def _list_branches_outcome(emulate_url: str, preview: dict) -> None:
branches = await github_request(emulate_url, "GET", f"{REPO_PATH}/branches")
assert isinstance(branches, list)
assert {branch["name"] for branch in branches} == {"main", BRANCH}, branches
rendered = json.dumps(preview)
assert "main" in rendered, preview
assert BRANCH in rendered, preview
async def _create_release_outcome(emulate_url: str, preview: dict) -> None:
releases = await github_request(emulate_url, "GET", f"{REPO_PATH}/releases")
assert isinstance(releases, list)
assert [release["tag_name"] for release in releases] == [RELEASE_TAG], releases
assert RELEASE_TAG in json.dumps(preview), preview
async def _create_repo_outcome(emulate_url: str, preview: dict) -> None:
repo = await github_request(
emulate_url, "GET", f"/repos/reborn-dev/{CREATED_REPO}"
)
assert isinstance(repo, dict)
assert repo["private"] is True, repo
assert CREATED_REPO in json.dumps(preview), preview
async def _fork_outcome(emulate_url: str, preview: dict) -> None:
fork = await github_request(emulate_url, "GET", f"/repos/reborn-dev/{FORK_NAME}")
assert isinstance(fork, dict)
assert fork["fork"] is True, fork
assert FORK_NAME in json.dumps(preview), preview
async def _list_repos_outcome(emulate_url: str, preview: dict) -> None:
repos = await github_request(emulate_url, "GET", "/user/repos")
assert isinstance(repos, list)
assert any(repo["name"] == SEEDED_REPO for repo in repos), repos
assert SEEDED_REPO in json.dumps(preview), preview
async def _search_repo_outcome(emulate_url: str, preview: dict) -> None:
await _repo_baseline(emulate_url)
rendered = json.dumps(preview)
assert f"{OWNER}/{REPO}" in rendered, preview
assert "Emulated IronClaw repo" in rendered, preview
async def _search_code_outcome(emulate_url: str, preview: dict) -> None:
result = await github_request(
emulate_url,
"GET",
"/search/code",
params={"q": f"{CODE_MARKER} repo:{OWNER}/{REPO}"},
)
assert isinstance(result, dict)
assert [item["path"] for item in result["items"]] == [CODE_PATH], result
rendered = json.dumps(preview)
assert CODE_PATH in rendered, preview
async def _workflow_runs_outcome(emulate_url: str, preview: dict) -> None:
result = await github_request(
emulate_url, "GET", f"{REPO_PATH}/actions/runs"
)
assert result == {"total_count": 0, "workflow_runs": []}, result
rendered = json.dumps(preview)
assert "workflow_runs" in rendered, preview
assert "total_count" in rendered, preview
GITHUB_REPO_PROVIDER_OPERATION_CASES = (
ProviderOperationCase(
case_id="github_create_branch",
provider_service="github",
capability_id="github.create_branch",
arguments={
**BASE_ARGS,
"branch": CREATED_BRANCH,
"from_ref": "main",
},
assert_baseline=_repo_baseline,
assert_outcome=_create_branch_outcome,
),
ProviderOperationCase(
case_id="github_list_branches",
provider_service="github",
capability_id="github.list_branches",
arguments=BASE_ARGS,
assert_baseline=_seed_code,
assert_outcome=_list_branches_outcome,
),
ProviderOperationCase(
case_id="github_create_release",
provider_service="github",
capability_id="github.create_release",
arguments={
**BASE_ARGS,
"tag_name": RELEASE_TAG,
"name": "Provider Case v1",
"body": "Created through the provider operation runner.",
},
assert_baseline=_empty_release_baseline,
assert_outcome=_create_release_outcome,
),
ProviderOperationCase(
case_id="github_create_repo",
provider_service="github",
capability_id="github.create_repo",
arguments={
"name": CREATED_REPO,
"description": "Created through the provider operation runner.",
"private": True,
"auto_init": True,
},
assert_baseline=_repo_baseline,
assert_outcome=_create_repo_outcome,
),
ProviderOperationCase(
case_id="github_fork_repo",
provider_service="github",
capability_id="github.fork_repo",
arguments={**BASE_ARGS, "name": FORK_NAME},
assert_baseline=_repo_baseline,
assert_outcome=_fork_outcome,
),
ProviderOperationCase(
case_id="github_list_repos",
provider_service="github",
capability_id="github.list_repos",
arguments={"type": "owner"},
assert_baseline=_seeded_user_repo,
assert_outcome=_list_repos_outcome,
),
ProviderOperationCase(
case_id="github_search_repositories",
provider_service="github",
capability_id="github.search_repositories",
arguments={"query": f"org:{OWNER} {REPO}"},
assert_baseline=_repo_baseline,
assert_outcome=_search_repo_outcome,
),
ProviderOperationCase(
case_id="github_search_code",
provider_service="github",
capability_id="github.search_code",
arguments={"query": f"{CODE_MARKER} repo:{OWNER}/{REPO}"},
assert_baseline=_seed_code,
assert_outcome=_search_code_outcome,
),
ProviderOperationCase(
case_id="github_get_workflow_runs",
provider_service="github",
capability_id="github.get_workflow_runs",
arguments=BASE_ARGS,
assert_baseline=_repo_baseline,
assert_outcome=_workflow_runs_outcome,
),
)

174
tests/e2e/provider_operation_gmail_cases.py Normal file
View File

@@ -0,0 +1,174 @@
"""Gmail full-path provider operation cases."""
import json
import httpx
from emulate_provider import gmail_header, google_headers, raw_mime
from provider_operation_types import ProviderOperationCase
GMAIL_REPLY_MARKER = "REBORN_PROVIDER_CASE_REPLY"
SEEDED_GMAIL_MESSAGE_ID = "<msg_emulate_unread@ironclaw.test>"
SEEDED_GMAIL_SUBJECT = "Emulate seeded unread"
SEEDED_GMAIL_THREAD_ID = "thr_emulate_unread"
async def _get(
emulate_url: str,
path: str,
*,
params: dict[str, str | int] | None = None,
) -> httpx.Response:
async with httpx.AsyncClient(headers=google_headers(), timeout=15) as client:
response = await client.get(f"{emulate_url}{path}", params=params)
response.raise_for_status()
return response
async def _drafts(emulate_url: str) -> list[dict]:
response = await _get(
emulate_url,
"/gmail/v1/users/me/drafts",
params={"maxResults": 100},
)
return response.json().get("drafts", [])
async def _assert_gmail_draft_baseline(emulate_url: str) -> None:
assert not await _drafts(emulate_url), "seeded provider unexpectedly has drafts"
async def _assert_gmail_draft_outcome(emulate_url: str, preview: dict) -> None:
drafts = await _drafts(emulate_url)
assert len(drafts) == 1, drafts
response = await _get(
emulate_url,
f"/gmail/v1/users/me/drafts/{drafts[0]['id']}",
params={"format": "full"},
)
draft = response.json()
assert gmail_header(draft["message"], "Subject") == "REBORN_PROVIDER_CASE_DRAFT"
assert "REBORN_PROVIDER_CASE_DRAFT" in json.dumps(preview), preview
async def _gmail_message(emulate_url: str, message_id: str) -> dict:
response = await _get(
emulate_url,
f"/gmail/v1/users/me/messages/{message_id}",
params={"format": "full"},
)
return response.json()
async def _gmail_thread_messages(emulate_url: str, thread_id: str) -> list[dict]:
response = await _get(
emulate_url,
"/gmail/v1/users/me/messages",
params={"includeSpamTrash": "true", "maxResults": 100},
)
messages = [
message
for message in response.json().get("messages", [])
if message["threadId"] == thread_id
]
return [
await _gmail_message(emulate_url, message["id"]) for message in messages
]
async def _assert_gmail_reply_baseline(emulate_url: str) -> None:
seeded = await _gmail_message(emulate_url, "msg_emulate_unread")
assert seeded["threadId"] == SEEDED_GMAIL_THREAD_ID, seeded
assert gmail_header(seeded, "Message-ID") == SEEDED_GMAIL_MESSAGE_ID, seeded
assert gmail_header(seeded, "Subject") == SEEDED_GMAIL_SUBJECT, seeded
thread_messages = await _gmail_thread_messages(
emulate_url, SEEDED_GMAIL_THREAD_ID
)
assert [message["id"] for message in thread_messages] == [
"msg_emulate_unread"
], thread_messages
async def _assert_gmail_reply_outcome(emulate_url: str, preview: dict) -> None:
thread_messages = await _gmail_thread_messages(
emulate_url, SEEDED_GMAIL_THREAD_ID
)
assert len(thread_messages) == 2, thread_messages
assert any(
message["id"] == "msg_emulate_unread" for message in thread_messages
), thread_messages
replies = [
message for message in thread_messages if "SENT" in message["labelIds"]
]
assert len(replies) == 1, thread_messages
reply = replies[0]
assert reply["threadId"] == SEEDED_GMAIL_THREAD_ID, reply
assert gmail_header(reply, "To") == "qa-sender@example.com", reply
assert gmail_header(reply, "Subject") == SEEDED_GMAIL_SUBJECT, reply
assert gmail_header(reply, "In-Reply-To") == SEEDED_GMAIL_MESSAGE_ID, reply
assert gmail_header(reply, "References") == SEEDED_GMAIL_MESSAGE_ID, reply
assert GMAIL_REPLY_MARKER in reply["snippet"], reply
assert GMAIL_REPLY_MARKER in json.dumps(preview), preview
async def _assert_gmail_trash_baseline(emulate_url: str) -> None:
message = await _gmail_message(emulate_url, "msg_emulate_unread")
assert "TRASH" not in message["labelIds"], message
async def _assert_gmail_trash_outcome(emulate_url: str, preview: dict) -> None:
message = await _gmail_message(emulate_url, "msg_emulate_unread")
assert "TRASH" in message["labelIds"], message
assert "msg_emulate_unread" in json.dumps(preview), preview
GMAIL_PROVIDER_OPERATION_CASES = (
ProviderOperationCase(
case_id="gmail_create_draft",
provider_service="google",
capability_id="gmail.create_draft",
arguments={
"draft": {
"message": {
"raw": raw_mime(
to="draft-recipient@example.com",
subject="REBORN_PROVIDER_CASE_DRAFT",
body="Created through the reusable provider operation runner.",
)
}
}
},
assert_baseline=_assert_gmail_draft_baseline,
assert_outcome=_assert_gmail_draft_outcome,
),
ProviderOperationCase(
case_id="gmail_trash_message",
provider_service="google",
capability_id="gmail.trash_message",
arguments={"message_id": "msg_emulate_unread"},
assert_baseline=_assert_gmail_trash_baseline,
assert_outcome=_assert_gmail_trash_outcome,
),
ProviderOperationCase(
case_id="gmail_reply_to_message",
provider_service="google",
capability_id="gmail.reply_to_message",
arguments={
"message": {
"raw": raw_mime(
to="qa-sender@example.com",
subject=SEEDED_GMAIL_SUBJECT,
body=(
f"{GMAIL_REPLY_MARKER}: Reply sent through the reusable "
"provider operation runner."
),
in_reply_to=SEEDED_GMAIL_MESSAGE_ID,
references=SEEDED_GMAIL_MESSAGE_ID,
),
"threadId": SEEDED_GMAIL_THREAD_ID,
}
},
assert_baseline=_assert_gmail_reply_baseline,
assert_outcome=_assert_gmail_reply_outcome,
),
)

204
tests/e2e/provider_operation_google_calendar_cases.py Normal file
View File

@@ -0,0 +1,204 @@
"""Google Calendar full-path provider operation cases."""
import json
from emulate_provider import google_json
from provider_operation_types import ProviderOperationCase
CALENDAR_ID = "primary"
EVENT_ID = "evt_reborn_planning_sync"
CREATED_EVENT = "REBORN_PROVIDER_CASE_CREATED_EVENT"
UPDATED_SUMMARY = "REBORN_PROVIDER_CASE_UPDATED_EVENT"
ADDED_ATTENDEE = "provider-case-attendee@example.com"
async def _events(emulate_url: str, query: str | None = None) -> list[dict]:
params: dict[str, str | int] = {"maxResults": 100}
if query is not None:
params["q"] = query
result = await google_json(
emulate_url,
"GET",
f"/calendar/v3/calendars/{CALENDAR_ID}/events",
params=params,
)
assert isinstance(result, dict)
return result.get("items", [])
async def _event(emulate_url: str) -> dict:
result = await google_json(
emulate_url,
"GET",
f"/calendar/v3/calendars/{CALENDAR_ID}/events/{EVENT_ID}",
)
assert isinstance(result, dict)
return result
async def _seeded_event_baseline(emulate_url: str) -> None:
event = await _event(emulate_url)
assert event["summary"] == "Reborn planning sync", event
async def _create_event_baseline(emulate_url: str) -> None:
assert not await _events(emulate_url, CREATED_EVENT)
async def _create_event_outcome(emulate_url: str, preview: dict) -> None:
matches = await _events(emulate_url, CREATED_EVENT)
assert len(matches) == 1, matches
event = matches[0]
assert event["summary"] == CREATED_EVENT, event
assert event["description"] == "Created by the provider operation runner.", event
assert event["start"]["dateTime"] == "2026-07-30T09:00:00.000Z", event
assert event["end"]["dateTime"] == "2026-07-30T09:30:00.000Z", event
assert [attendee["email"] for attendee in event["attendees"]] == [
"teammate@example.com"
], event
assert CREATED_EVENT in json.dumps(preview), preview
async def _delete_event_baseline(emulate_url: str) -> None:
matching_ids = {event["id"] for event in await _events(emulate_url)}
assert EVENT_ID in matching_ids, matching_ids
async def _delete_event_outcome(emulate_url: str, preview: dict) -> None:
matching_ids = {event["id"] for event in await _events(emulate_url)}
assert EVENT_ID not in matching_ids, matching_ids
assert EVENT_ID in json.dumps(preview), preview
async def _get_event_outcome(emulate_url: str, preview: dict) -> None:
await _seeded_event_baseline(emulate_url)
assert "Reborn planning sync" in json.dumps(preview), preview
async def _free_busy_baseline(emulate_url: str) -> None:
await _seeded_event_baseline(emulate_url)
async def _free_busy_outcome(emulate_url: str, preview: dict) -> None:
await _seeded_event_baseline(emulate_url)
rendered = json.dumps(preview)
assert EVENT_ID not in rendered
assert "2026-06-22T13:00:00" in rendered, preview
assert "2026-06-22T13:30:00" in rendered, preview
async def _update_event_outcome(emulate_url: str, preview: dict) -> None:
event = await _event(emulate_url)
assert event["summary"] == UPDATED_SUMMARY, event
assert UPDATED_SUMMARY in json.dumps(preview), preview
async def _add_attendees_outcome(emulate_url: str, preview: dict) -> None:
event = await _event(emulate_url)
emails = [attendee["email"] for attendee in event["attendees"]]
assert emails == ["teammate@example.com", ADDED_ATTENDEE], event
assert ADDED_ATTENDEE in json.dumps(preview), preview
async def _set_reminder_outcome(emulate_url: str, preview: dict) -> None:
event = await _event(emulate_url)
assert event["reminders"] == {
"useDefault": False,
"overrides": [{"method": "popup", "minutes": 15}],
}, event
assert EVENT_ID in json.dumps(preview), preview
GOOGLE_CALENDAR_PROVIDER_OPERATION_CASES = (
ProviderOperationCase(
case_id="google_calendar_create_event",
provider_service="google",
capability_id="google-calendar.create_event",
arguments={
"calendar_id": CALENDAR_ID,
"event": {
"summary": CREATED_EVENT,
"description": "Created by the provider operation runner.",
"start": {
"dateTime": "2026-07-30T09:00:00.000Z",
"timeZone": "UTC",
},
"end": {
"dateTime": "2026-07-30T09:30:00.000Z",
"timeZone": "UTC",
},
"attendees": [{"email": "teammate@example.com"}],
},
},
assert_baseline=_create_event_baseline,
assert_outcome=_create_event_outcome,
),
ProviderOperationCase(
case_id="google_calendar_delete_event",
provider_service="google",
capability_id="google-calendar.delete_event",
arguments={"calendar_id": CALENDAR_ID, "event_id": EVENT_ID},
assert_baseline=_delete_event_baseline,
assert_outcome=_delete_event_outcome,
),
ProviderOperationCase(
case_id="google_calendar_get_event",
provider_service="google",
capability_id="google-calendar.get_event",
arguments={"calendar_id": CALENDAR_ID, "event_id": EVENT_ID},
assert_baseline=_seeded_event_baseline,
assert_outcome=_get_event_outcome,
),
ProviderOperationCase(
case_id="google_calendar_find_free_slots",
provider_service="google",
capability_id="google-calendar.find_free_slots",
arguments={
"timeMin": "2026-06-22T12:45:00.000Z",
"timeMax": "2026-06-22T13:45:00.000Z",
"timeZone": "UTC",
"items": [{"id": CALENDAR_ID}],
},
assert_baseline=_free_busy_baseline,
assert_outcome=_free_busy_outcome,
),
ProviderOperationCase(
case_id="google_calendar_update_event",
provider_service="google",
capability_id="google-calendar.update_event",
arguments={
"calendar_id": CALENDAR_ID,
"event_id": EVENT_ID,
"event": {"summary": UPDATED_SUMMARY},
},
assert_baseline=_seeded_event_baseline,
assert_outcome=_update_event_outcome,
),
ProviderOperationCase(
case_id="google_calendar_add_attendees",
provider_service="google",
capability_id="google-calendar.add_attendees",
arguments={
"calendar_id": CALENDAR_ID,
"event_id": EVENT_ID,
"attendees": [{"email": ADDED_ATTENDEE}],
},
assert_baseline=_seeded_event_baseline,
assert_outcome=_add_attendees_outcome,
),
ProviderOperationCase(
case_id="google_calendar_set_reminder",
provider_service="google",
capability_id="google-calendar.set_reminder",
arguments={
"calendar_id": CALENDAR_ID,
"event_id": EVENT_ID,
"reminders": {
"useDefault": False,
"overrides": [{"method": "popup", "minutes": 15}],
},
},
assert_baseline=_seeded_event_baseline,
assert_outcome=_set_reminder_outcome,
),
)

364
tests/e2e/provider_operation_google_cases.py
View File

@@ -1,364 +0,0 @@
"""Google provider operation cases and provider-side oracles."""
import json
import httpx
from emulate_provider import gmail_header, google_headers, raw_mime
from provider_operation_types import ProviderOperationCase
CALENDAR_CREATE_MARKER = "REBORN_PROVIDER_CASE_CREATED_EVENT"
DRIVE_FOLDER_MARKER = "REBORN_PROVIDER_CASE_CREATED_FOLDER"
DRIVE_UPLOAD_CONTENT = "Uploaded through the reusable provider operation runner."
DRIVE_UPLOAD_MARKER = "REBORN_PROVIDER_CASE_UPLOADED_FILE.txt"
GMAIL_REPLY_MARKER = "REBORN_PROVIDER_CASE_REPLY"
SEEDED_GMAIL_MESSAGE_ID = "<msg_emulate_unread@ironclaw.test>"
SEEDED_GMAIL_SUBJECT = "Emulate seeded unread"
SEEDED_GMAIL_THREAD_ID = "thr_emulate_unread"
async def _get(
emulate_url: str,
path: str,
*,
params: dict[str, str | int] | None = None,
) -> httpx.Response:
async with httpx.AsyncClient(headers=google_headers(), timeout=15) as client:
response = await client.get(f"{emulate_url}{path}", params=params)
response.raise_for_status()
return response
async def _drive_file(emulate_url: str, file_id: str) -> dict:
response = await _get(emulate_url, f"/drive/v3/files/{file_id}")
return response.json()
async def _drive_files_named(emulate_url: str, name: str) -> list[dict]:
response = await _get(
emulate_url,
"/drive/v3/files",
params={"q": f"name = '{name}' and trashed = false", "pageSize": 100},
)
return response.json().get("files", [])
async def _assert_drive_get_baseline(emulate_url: str) -> None:
file = await _drive_file(emulate_url, "drv_reborn_qa_brief")
assert file["name"] == "Reborn QA Brief", file
async def _assert_drive_get_outcome(emulate_url: str, preview: dict) -> None:
await _assert_drive_get_baseline(emulate_url)
assert "Reborn QA Brief" in json.dumps(preview), preview
async def _assert_drive_update_baseline(emulate_url: str) -> None:
await _assert_drive_get_baseline(emulate_url)
async def _assert_drive_update_outcome(emulate_url: str, preview: dict) -> None:
file = await _drive_file(emulate_url, "drv_reborn_qa_brief")
assert file["name"] == "REBORN_PROVIDER_CASE_UPDATED_FILE", file
assert "REBORN_PROVIDER_CASE_UPDATED_FILE" in json.dumps(preview), preview
async def _assert_drive_folder_baseline(emulate_url: str) -> None:
assert not await _drive_files_named(emulate_url, DRIVE_FOLDER_MARKER)
async def _assert_drive_folder_outcome(emulate_url: str, preview: dict) -> None:
matches = await _drive_files_named(emulate_url, DRIVE_FOLDER_MARKER)
assert len(matches) == 1, matches
assert matches[0]["mimeType"] == "application/vnd.google-apps.folder", matches[0]
assert matches[0]["parents"] == ["root"], matches[0]
assert DRIVE_FOLDER_MARKER in json.dumps(preview), preview
async def _assert_drive_upload_baseline(emulate_url: str) -> None:
assert not await _drive_files_named(emulate_url, DRIVE_UPLOAD_MARKER)
async def _assert_drive_upload_outcome(emulate_url: str, preview: dict) -> None:
matches = await _drive_files_named(emulate_url, DRIVE_UPLOAD_MARKER)
assert len(matches) == 1, matches
uploaded = matches[0]
assert uploaded["mimeType"] == "text/plain", uploaded
assert uploaded["parents"] == ["root"], uploaded
assert uploaded["size"] == str(len(DRIVE_UPLOAD_CONTENT.encode())), uploaded
response = await _get(
emulate_url,
f"/drive/v3/files/{uploaded['id']}",
params={"alt": "media"},
)
assert response.text == DRIVE_UPLOAD_CONTENT
assert DRIVE_UPLOAD_MARKER in json.dumps(preview), preview
async def _drafts(emulate_url: str) -> list[dict]:
response = await _get(
emulate_url,
"/gmail/v1/users/me/drafts",
params={"maxResults": 100},
)
return response.json().get("drafts", [])
async def _assert_gmail_draft_baseline(emulate_url: str) -> None:
assert not await _drafts(emulate_url), "seeded provider unexpectedly has drafts"
async def _assert_gmail_draft_outcome(emulate_url: str, preview: dict) -> None:
drafts = await _drafts(emulate_url)
assert len(drafts) == 1, drafts
response = await _get(
emulate_url,
f"/gmail/v1/users/me/drafts/{drafts[0]['id']}",
params={"format": "full"},
)
draft = response.json()
assert gmail_header(draft["message"], "Subject") == "REBORN_PROVIDER_CASE_DRAFT"
assert "REBORN_PROVIDER_CASE_DRAFT" in json.dumps(preview), preview
async def _gmail_message(emulate_url: str, message_id: str) -> dict:
response = await _get(
emulate_url,
f"/gmail/v1/users/me/messages/{message_id}",
params={"format": "full"},
)
return response.json()
async def _gmail_thread_messages(emulate_url: str, thread_id: str) -> list[dict]:
response = await _get(
emulate_url,
"/gmail/v1/users/me/messages",
params={"includeSpamTrash": "true", "maxResults": 100},
)
messages = [
message
for message in response.json().get("messages", [])
if message["threadId"] == thread_id
]
return [
await _gmail_message(emulate_url, message["id"]) for message in messages
]
async def _assert_gmail_reply_baseline(emulate_url: str) -> None:
seeded = await _gmail_message(emulate_url, "msg_emulate_unread")
assert seeded["threadId"] == SEEDED_GMAIL_THREAD_ID, seeded
assert gmail_header(seeded, "Message-ID") == SEEDED_GMAIL_MESSAGE_ID, seeded
assert gmail_header(seeded, "Subject") == SEEDED_GMAIL_SUBJECT, seeded
thread_messages = await _gmail_thread_messages(
emulate_url, SEEDED_GMAIL_THREAD_ID
)
assert [message["id"] for message in thread_messages] == [
"msg_emulate_unread"
], thread_messages
async def _assert_gmail_reply_outcome(emulate_url: str, preview: dict) -> None:
thread_messages = await _gmail_thread_messages(
emulate_url, SEEDED_GMAIL_THREAD_ID
)
assert len(thread_messages) == 2, thread_messages
assert any(
message["id"] == "msg_emulate_unread" for message in thread_messages
), thread_messages
replies = [
message for message in thread_messages if "SENT" in message["labelIds"]
]
assert len(replies) == 1, thread_messages
reply = replies[0]
assert reply["threadId"] == SEEDED_GMAIL_THREAD_ID, reply
assert gmail_header(reply, "To") == "qa-sender@example.com", reply
assert gmail_header(reply, "Subject") == SEEDED_GMAIL_SUBJECT, reply
assert gmail_header(reply, "In-Reply-To") == SEEDED_GMAIL_MESSAGE_ID, reply
assert gmail_header(reply, "References") == SEEDED_GMAIL_MESSAGE_ID, reply
assert GMAIL_REPLY_MARKER in reply["snippet"], reply
assert GMAIL_REPLY_MARKER in json.dumps(preview), preview
async def _assert_gmail_trash_baseline(emulate_url: str) -> None:
message = await _gmail_message(emulate_url, "msg_emulate_unread")
assert "TRASH" not in message["labelIds"], message
async def _assert_gmail_trash_outcome(emulate_url: str, preview: dict) -> None:
message = await _gmail_message(emulate_url, "msg_emulate_unread")
assert "TRASH" in message["labelIds"], message
assert "msg_emulate_unread" in json.dumps(preview), preview
async def _calendar_events(emulate_url: str, query: str | None = None) -> list[dict]:
params: dict[str, str | int] = {"maxResults": 100}
if query is not None:
params["q"] = query
response = await _get(
emulate_url,
"/calendar/v3/calendars/primary/events",
params=params,
)
return response.json().get("items", [])
async def _assert_calendar_create_baseline(emulate_url: str) -> None:
assert not await _calendar_events(emulate_url, CALENDAR_CREATE_MARKER)
async def _assert_calendar_create_outcome(emulate_url: str, preview: dict) -> None:
matches = await _calendar_events(emulate_url, CALENDAR_CREATE_MARKER)
assert len(matches) == 1, matches
event = matches[0]
assert event["summary"] == CALENDAR_CREATE_MARKER, event
assert event["description"] == "Created by the provider operation runner.", event
assert event["start"]["dateTime"] == "2026-07-30T09:00:00.000Z", event
assert event["end"]["dateTime"] == "2026-07-30T09:30:00.000Z", event
assert [attendee["email"] for attendee in event["attendees"]] == [
"teammate@example.com"
], event
assert CALENDAR_CREATE_MARKER in json.dumps(preview), preview
async def _assert_calendar_delete_baseline(emulate_url: str) -> None:
matching_ids = {
event["id"] for event in await _calendar_events(emulate_url)
}
assert "evt_reborn_planning_sync" in matching_ids, matching_ids
async def _assert_calendar_delete_outcome(emulate_url: str, preview: dict) -> None:
matching_ids = {
event["id"] for event in await _calendar_events(emulate_url)
}
assert "evt_reborn_planning_sync" not in matching_ids, matching_ids
assert "evt_reborn_planning_sync" in json.dumps(preview), preview
GOOGLE_PROVIDER_OPERATION_CASES = (
ProviderOperationCase(
case_id="google_drive_get_file",
provider_service="google",
capability_id="google-drive.get_file",
arguments={"file_id": "drv_reborn_qa_brief"},
assert_baseline=_assert_drive_get_baseline,
assert_outcome=_assert_drive_get_outcome,
),
ProviderOperationCase(
case_id="gmail_create_draft",
provider_service="google",
capability_id="gmail.create_draft",
arguments={
"draft": {
"message": {
"raw": raw_mime(
to="draft-recipient@example.com",
subject="REBORN_PROVIDER_CASE_DRAFT",
body="Created through the reusable provider operation runner.",
)
}
}
},
assert_baseline=_assert_gmail_draft_baseline,
assert_outcome=_assert_gmail_draft_outcome,
),
ProviderOperationCase(
case_id="google_drive_update_file",
provider_service="google",
capability_id="google-drive.update_file",
arguments={
"file_id": "drv_reborn_qa_brief",
"name": "REBORN_PROVIDER_CASE_UPDATED_FILE",
},
assert_baseline=_assert_drive_update_baseline,
assert_outcome=_assert_drive_update_outcome,
),
ProviderOperationCase(
case_id="gmail_trash_message",
provider_service="google",
capability_id="gmail.trash_message",
arguments={"message_id": "msg_emulate_unread"},
assert_baseline=_assert_gmail_trash_baseline,
assert_outcome=_assert_gmail_trash_outcome,
),
ProviderOperationCase(
case_id="gmail_reply_to_message",
provider_service="google",
capability_id="gmail.reply_to_message",
arguments={
"message": {
"raw": raw_mime(
to="qa-sender@example.com",
subject=SEEDED_GMAIL_SUBJECT,
body=(
f"{GMAIL_REPLY_MARKER}: Reply sent through the reusable "
"provider operation runner."
),
in_reply_to=SEEDED_GMAIL_MESSAGE_ID,
references=SEEDED_GMAIL_MESSAGE_ID,
),
"threadId": SEEDED_GMAIL_THREAD_ID,
}
},
assert_baseline=_assert_gmail_reply_baseline,
assert_outcome=_assert_gmail_reply_outcome,
),
ProviderOperationCase(
case_id="google_calendar_create_event",
provider_service="google",
capability_id="google-calendar.create_event",
arguments={
"calendar_id": "primary",
"event": {
"summary": CALENDAR_CREATE_MARKER,
"description": "Created by the provider operation runner.",
"start": {
"dateTime": "2026-07-30T09:00:00.000Z",
"timeZone": "UTC",
},
"end": {
"dateTime": "2026-07-30T09:30:00.000Z",
"timeZone": "UTC",
},
"attendees": [{"email": "teammate@example.com"}],
},
},
assert_baseline=_assert_calendar_create_baseline,
assert_outcome=_assert_calendar_create_outcome,
),
ProviderOperationCase(
case_id="google_calendar_delete_event",
provider_service="google",
capability_id="google-calendar.delete_event",
arguments={
"calendar_id": "primary",
"event_id": "evt_reborn_planning_sync",
},
assert_baseline=_assert_calendar_delete_baseline,
assert_outcome=_assert_calendar_delete_outcome,
),
ProviderOperationCase(
case_id="google_drive_create_folder",
provider_service="google",
capability_id="google-drive.create_folder",
arguments={"name": DRIVE_FOLDER_MARKER, "parent_id": "root"},
assert_baseline=_assert_drive_folder_baseline,
assert_outcome=_assert_drive_folder_outcome,
),
ProviderOperationCase(
case_id="google_drive_upload_file",
provider_service="google",
capability_id="google-drive.upload_file",
arguments={
"name": DRIVE_UPLOAD_MARKER,
"content": DRIVE_UPLOAD_CONTENT,
"mime_type": "text/plain",
"parent_id": "root",
},
assert_baseline=_assert_drive_upload_baseline,
assert_outcome=_assert_drive_upload_outcome,
),
)

183
tests/e2e/provider_operation_google_docs_cases.py Normal file
View File

@@ -0,0 +1,183 @@
"""Google Docs full-path provider operation cases."""
import json
from emulate_provider import google_json
from provider_operation_types import ProviderOperationCase
DOCUMENT_ID = "doc_reborn_strategy"
SEEDED_TEXT = (
"NEAR AI Strategy: user-owned agents keep credentials and data under user control."
)
BATCH_MARKER = " REBORN_PROVIDER_CASE_BATCH"
REPLACEMENT = "customer-owned agents"
async def _document(emulate_url: str) -> dict:
result = await google_json(
emulate_url, "GET", f"/v1/documents/{DOCUMENT_ID}"
)
assert isinstance(result, dict)
return result
def _document_text(document: dict) -> str:
return "".join(
element.get("textRun", {}).get("content", "")
for item in document["body"]["content"]
for element in item.get("paragraph", {}).get("elements", [])
)
async def _baseline(emulate_url: str) -> None:
document = await _document(emulate_url)
assert document["revisionId"] == "1", document
assert _document_text(document) == SEEDED_TEXT, document
async def _get_outcome(emulate_url: str, preview: dict) -> None:
await _baseline(emulate_url)
rendered = json.dumps(preview)
assert "NEAR AI Strategy" in rendered, preview
assert DOCUMENT_ID in rendered, preview
def _revision_outcome(marker: str | None = None):
async def assert_outcome(emulate_url: str, preview: dict) -> None:
document = await _document(emulate_url)
assert document["revisionId"] == "2", document
if marker is not None:
assert marker in json.dumps(preview), preview
return assert_outcome
async def _batch_outcome(emulate_url: str, preview: dict) -> None:
document = await _document(emulate_url)
assert _document_text(document) == f"{SEEDED_TEXT}{BATCH_MARKER}", document
assert document["revisionId"] == "2", document
assert DOCUMENT_ID in json.dumps(preview), preview
async def _delete_outcome(emulate_url: str, preview: dict) -> None:
document = await _document(emulate_url)
assert _document_text(document) == SEEDED_TEXT[5:], document
assert document["revisionId"] == "2", document
assert DOCUMENT_ID in json.dumps(preview), preview
async def _replace_outcome(emulate_url: str, preview: dict) -> None:
document = await _document(emulate_url)
assert REPLACEMENT in _document_text(document), document
assert "user-owned agents" not in _document_text(document), document
assert document["revisionId"] == "2", document
assert DOCUMENT_ID in json.dumps(preview), preview
GOOGLE_DOCS_PROVIDER_OPERATION_CASES = (
ProviderOperationCase(
case_id="google_docs_get_document",
provider_service="google",
capability_id="google-docs.get_document",
arguments={"document_id": DOCUMENT_ID},
assert_baseline=_baseline,
assert_outcome=_get_outcome,
),
ProviderOperationCase(
case_id="google_docs_batch_update",
provider_service="google",
capability_id="google-docs.batch_update",
arguments={
"document_id": DOCUMENT_ID,
"requests": [
{
"insertText": {
"endOfSegmentLocation": {},
"text": BATCH_MARKER,
}
}
],
},
assert_baseline=_baseline,
assert_outcome=_batch_outcome,
),
ProviderOperationCase(
case_id="google_docs_delete_content",
provider_service="google",
capability_id="google-docs.delete_content",
arguments={
"document_id": DOCUMENT_ID,
"start_index": 1,
"end_index": 6,
},
assert_baseline=_baseline,
assert_outcome=_delete_outcome,
),
ProviderOperationCase(
case_id="google_docs_replace_text",
provider_service="google",
capability_id="google-docs.replace_text",
arguments={
"document_id": DOCUMENT_ID,
"find": "user-owned agents",
"replace": REPLACEMENT,
"match_case": True,
},
assert_baseline=_baseline,
assert_outcome=_replace_outcome,
),
ProviderOperationCase(
case_id="google_docs_format_text",
provider_service="google",
capability_id="google-docs.format_text",
arguments={
"document_id": DOCUMENT_ID,
"start_index": 1,
"end_index": 9,
"bold": True,
"foreground_color": "#3367D6",
},
assert_baseline=_baseline,
assert_outcome=_revision_outcome(DOCUMENT_ID),
),
ProviderOperationCase(
case_id="google_docs_format_paragraph",
provider_service="google",
capability_id="google-docs.format_paragraph",
arguments={
"document_id": DOCUMENT_ID,
"start_index": 1,
"end_index": 9,
"named_style": "HEADING_2",
"alignment": "CENTER",
},
assert_baseline=_baseline,
assert_outcome=_revision_outcome(DOCUMENT_ID),
),
ProviderOperationCase(
case_id="google_docs_insert_table",
provider_service="google",
capability_id="google-docs.insert_table",
arguments={
"document_id": DOCUMENT_ID,
"rows": 2,
"columns": 3,
"index": 1,
},
assert_baseline=_baseline,
assert_outcome=_revision_outcome(DOCUMENT_ID),
),
ProviderOperationCase(
case_id="google_docs_create_list",
provider_service="google",
capability_id="google-docs.create_list",
arguments={
"document_id": DOCUMENT_ID,
"start_index": 1,
"end_index": 9,
"bullet_preset": "BULLET_DISC_CIRCLE_SQUARE",
},
assert_baseline=_baseline,
assert_outcome=_revision_outcome(DOCUMENT_ID),
),
)

260
tests/e2e/provider_operation_google_drive_cases.py Normal file
View File

@@ -0,0 +1,260 @@
"""Google Drive full-path provider operation cases."""
import json
import httpx
from emulate_provider import google_headers, google_json
from provider_operation_types import ProviderOperationCase
FILE_ID = "drv_reborn_qa_brief"
CREATED_FOLDER = "REBORN_PROVIDER_CASE_CREATED_FOLDER"
UPLOADED_CONTENT = "Uploaded through the reusable provider operation runner."
UPLOADED_FILE = "REBORN_PROVIDER_CASE_UPLOADED_FILE.txt"
SEEDED_PERMISSION_ID = "perm_reborn_reader"
SHARED_DRIVE_ID = "shared_reborn_engineering"
SHARED_EMAIL = "provider-case-reader@example.com"
async def _file(emulate_url: str, *, expected_status: int = 200) -> dict:
result = await google_json(
emulate_url,
"GET",
f"/drive/v3/files/{FILE_ID}",
expected_status=expected_status,
)
assert isinstance(result, dict)
return result
async def _permissions(emulate_url: str) -> list[dict]:
result = await google_json(
emulate_url,
"GET",
f"/drive/v3/files/{FILE_ID}/permissions",
)
assert isinstance(result, dict)
return result["permissions"]
async def _files_named(emulate_url: str, name: str) -> list[dict]:
result = await google_json(
emulate_url,
"GET",
"/drive/v3/files",
params={"q": f"name = '{name}' and trashed = false", "pageSize": 100},
)
assert isinstance(result, dict)
return result.get("files", [])
async def _media(emulate_url: str, file_id: str) -> str:
async with httpx.AsyncClient(headers=google_headers(), timeout=15) as client:
response = await client.get(
f"{emulate_url}/drive/v3/files/{file_id}",
params={"alt": "media"},
)
response.raise_for_status()
return response.text
async def _seeded_file_baseline(emulate_url: str) -> None:
file = await _file(emulate_url)
assert file["name"] == "Reborn QA Brief", file
async def _baseline(emulate_url: str) -> None:
await _seeded_file_baseline(emulate_url)
permissions = await _permissions(emulate_url)
assert [permission["id"] for permission in permissions] == [
SEEDED_PERMISSION_ID
], permissions
async def _get_file_outcome(emulate_url: str, preview: dict) -> None:
await _seeded_file_baseline(emulate_url)
assert "Reborn QA Brief" in json.dumps(preview), preview
async def _update_file_outcome(emulate_url: str, preview: dict) -> None:
file = await _file(emulate_url)
assert file["name"] == "REBORN_PROVIDER_CASE_UPDATED_FILE", file
assert "REBORN_PROVIDER_CASE_UPDATED_FILE" in json.dumps(preview), preview
async def _create_folder_baseline(emulate_url: str) -> None:
assert not await _files_named(emulate_url, CREATED_FOLDER)
async def _create_folder_outcome(emulate_url: str, preview: dict) -> None:
matches = await _files_named(emulate_url, CREATED_FOLDER)
assert len(matches) == 1, matches
assert matches[0]["mimeType"] == "application/vnd.google-apps.folder", matches[0]
assert matches[0]["parents"] == ["root"], matches[0]
assert CREATED_FOLDER in json.dumps(preview), preview
async def _upload_file_baseline(emulate_url: str) -> None:
assert not await _files_named(emulate_url, UPLOADED_FILE)
async def _upload_file_outcome(emulate_url: str, preview: dict) -> None:
matches = await _files_named(emulate_url, UPLOADED_FILE)
assert len(matches) == 1, matches
uploaded = matches[0]
assert uploaded["mimeType"] == "text/plain", uploaded
assert uploaded["parents"] == ["root"], uploaded
assert uploaded["size"] == str(len(UPLOADED_CONTENT.encode())), uploaded
assert await _media(emulate_url, uploaded["id"]) == UPLOADED_CONTENT
assert UPLOADED_FILE in json.dumps(preview), preview
async def _delete_outcome(emulate_url: str, preview: dict) -> None:
missing = await _file(emulate_url, expected_status=404)
assert missing["error"]["code"] == 404, missing
assert FILE_ID in json.dumps(preview), preview
async def _trash_outcome(emulate_url: str, preview: dict) -> None:
file = await _file(emulate_url)
assert file["trashed"] is True, file
assert FILE_ID in json.dumps(preview), preview
async def _list_permissions_outcome(emulate_url: str, preview: dict) -> None:
await _baseline(emulate_url)
rendered = json.dumps(preview)
assert SEEDED_PERMISSION_ID in rendered, preview
assert "seeded-reader@example.com" in rendered, preview
async def _share_outcome(emulate_url: str, preview: dict) -> None:
permissions = await _permissions(emulate_url)
matches = [
permission
for permission in permissions
if permission.get("emailAddress") == SHARED_EMAIL
]
assert len(matches) == 1, permissions
assert matches[0]["role"] == "reader", matches[0]
assert SHARED_EMAIL in json.dumps(preview), preview
async def _remove_permission_outcome(emulate_url: str, preview: dict) -> None:
assert not await _permissions(emulate_url)
assert SEEDED_PERMISSION_ID in json.dumps(preview), preview
async def _shared_drives_outcome(emulate_url: str, preview: dict) -> None:
result = await google_json(emulate_url, "GET", "/drive/v3/drives")
assert isinstance(result, dict)
assert result["drives"] == [
{
"kind": "drive#drive",
"id": SHARED_DRIVE_ID,
"name": "Reborn Engineering",
}
], result
rendered = json.dumps(preview)
assert SHARED_DRIVE_ID in rendered, preview
assert "Reborn Engineering" in rendered, preview
GOOGLE_DRIVE_PROVIDER_OPERATION_CASES = (
ProviderOperationCase(
case_id="google_drive_get_file",
provider_service="google",
capability_id="google-drive.get_file",
arguments={"file_id": FILE_ID},
assert_baseline=_seeded_file_baseline,
assert_outcome=_get_file_outcome,
),
ProviderOperationCase(
case_id="google_drive_update_file",
provider_service="google",
capability_id="google-drive.update_file",
arguments={
"file_id": FILE_ID,
"name": "REBORN_PROVIDER_CASE_UPDATED_FILE",
},
assert_baseline=_seeded_file_baseline,
assert_outcome=_update_file_outcome,
),
ProviderOperationCase(
case_id="google_drive_create_folder",
provider_service="google",
capability_id="google-drive.create_folder",
arguments={"name": CREATED_FOLDER, "parent_id": "root"},
assert_baseline=_create_folder_baseline,
assert_outcome=_create_folder_outcome,
),
ProviderOperationCase(
case_id="google_drive_upload_file",
provider_service="google",
capability_id="google-drive.upload_file",
arguments={
"name": UPLOADED_FILE,
"content": UPLOADED_CONTENT,
"mime_type": "text/plain",
"parent_id": "root",
},
assert_baseline=_upload_file_baseline,
assert_outcome=_upload_file_outcome,
),
ProviderOperationCase(
case_id="google_drive_delete_file",
provider_service="google",
capability_id="google-drive.delete_file",
arguments={"file_id": FILE_ID},
assert_baseline=_baseline,
assert_outcome=_delete_outcome,
),
ProviderOperationCase(
case_id="google_drive_trash_file",
provider_service="google",
capability_id="google-drive.trash_file",
arguments={"file_id": FILE_ID},
assert_baseline=_baseline,
assert_outcome=_trash_outcome,
),
ProviderOperationCase(
case_id="google_drive_list_permissions",
provider_service="google",
capability_id="google-drive.list_permissions",
arguments={"file_id": FILE_ID},
assert_baseline=_baseline,
assert_outcome=_list_permissions_outcome,
),
ProviderOperationCase(
case_id="google_drive_share_file",
provider_service="google",
capability_id="google-drive.share_file",
arguments={
"file_id": FILE_ID,
"email": SHARED_EMAIL,
"role": "reader",
"message": "Provider contract test",
},
assert_baseline=_baseline,
assert_outcome=_share_outcome,
),
ProviderOperationCase(
case_id="google_drive_remove_permission",
provider_service="google",
capability_id="google-drive.remove_permission",
arguments={
"file_id": FILE_ID,
"permission_id": SEEDED_PERMISSION_ID,
},
assert_baseline=_baseline,
assert_outcome=_remove_permission_outcome,
),
ProviderOperationCase(
case_id="google_drive_list_shared_drives",
provider_service="google",
capability_id="google-drive.list_shared_drives",
arguments={"page_size": 10},
assert_baseline=_baseline,
assert_outcome=_shared_drives_outcome,
),
)

138
tests/e2e/provider_operation_google_sheets_cases.py Normal file
View File

@@ -0,0 +1,138 @@
"""Google Sheets full-path provider operation cases."""
import json
from emulate_provider import google_json
from provider_operation_types import ProviderOperationCase
SPREADSHEET_ID = "sheet_reborn_abc"
ADDED_SHEET = "ProviderCase"
async def _spreadsheet(emulate_url: str) -> dict:
result = await google_json(
emulate_url, "GET", f"/v4/spreadsheets/{SPREADSHEET_ID}"
)
assert isinstance(result, dict)
return result
async def _values(emulate_url: str, range_name: str) -> list[list]:
result = await google_json(
emulate_url,
"GET",
f"/v4/spreadsheets/{SPREADSHEET_ID}/values/{range_name}",
)
assert isinstance(result, dict)
return result.get("values", [])
async def _baseline(emulate_url: str) -> None:
spreadsheet = await _spreadsheet(emulate_url)
sheets = spreadsheet["sheets"]
assert [
(sheet["properties"]["sheetId"], sheet["properties"]["title"])
for sheet in sheets
] == [(0, "Sheet1"), (7, "DeleteMe")], sheets
values = await _values(emulate_url, "Sheet1!A1:E2")
assert values[1][-1] == "REBORN_QA_SEEDED", values
async def _batch_read_outcome(emulate_url: str, preview: dict) -> None:
await _baseline(emulate_url)
rendered = json.dumps(preview)
assert "REBORN_QA_SEEDED" in rendered, preview
assert "NEAR AI" in rendered, preview
async def _clear_outcome(emulate_url: str, preview: dict) -> None:
values = await _values(emulate_url, "Sheet1!A1:E2")
assert values == [
["Company", "Contact", "Source", "Status", "QA Marker"]
], values
assert "Sheet1!A2:E2" in json.dumps(preview), preview
async def _add_sheet_outcome(emulate_url: str, preview: dict) -> None:
spreadsheet = await _spreadsheet(emulate_url)
matches = [
sheet
for sheet in spreadsheet["sheets"]
if sheet["properties"]["title"] == ADDED_SHEET
]
assert len(matches) == 1, spreadsheet
assert matches[0]["properties"]["sheetId"] == 8, matches[0]
assert ADDED_SHEET in json.dumps(preview), preview
async def _delete_sheet_outcome(emulate_url: str, preview: dict) -> None:
spreadsheet = await _spreadsheet(emulate_url)
assert [
sheet["properties"]["sheetId"] for sheet in spreadsheet["sheets"]
] == [0], spreadsheet
assert SPREADSHEET_ID in json.dumps(preview), preview
async def _format_cells_outcome(emulate_url: str, preview: dict) -> None:
await _baseline(emulate_url)
assert SPREADSHEET_ID in json.dumps(preview), preview
GOOGLE_SHEETS_PROVIDER_OPERATION_CASES = (
ProviderOperationCase(
case_id="google_sheets_batch_read_values",
provider_service="google",
capability_id="google-sheets.batch_read_values",
arguments={
"spreadsheet_id": SPREADSHEET_ID,
"ranges": ["Sheet1!A1:B2", "Sheet1!E1:E2"],
},
assert_baseline=_baseline,
assert_outcome=_batch_read_outcome,
),
ProviderOperationCase(
case_id="google_sheets_clear_values",
provider_service="google",
capability_id="google-sheets.clear_values",
arguments={
"spreadsheet_id": SPREADSHEET_ID,
"range": "Sheet1!A2:E2",
},
assert_baseline=_baseline,
assert_outcome=_clear_outcome,
),
ProviderOperationCase(
case_id="google_sheets_add_sheet",
provider_service="google",
capability_id="google-sheets.add_sheet",
arguments={"spreadsheet_id": SPREADSHEET_ID, "title": ADDED_SHEET},
assert_baseline=_baseline,
assert_outcome=_add_sheet_outcome,
),
ProviderOperationCase(
case_id="google_sheets_delete_sheet",
provider_service="google",
capability_id="google-sheets.delete_sheet",
arguments={"spreadsheet_id": SPREADSHEET_ID, "sheet_id": 7},
assert_baseline=_baseline,
assert_outcome=_delete_sheet_outcome,
),
ProviderOperationCase(
case_id="google_sheets_format_cells",
provider_service="google",
capability_id="google-sheets.format_cells",
arguments={
"spreadsheet_id": SPREADSHEET_ID,
"sheet_id": 0,
"start_row": 0,
"end_row": 1,
"start_column": 0,
"end_column": 5,
"bold": True,
"background_color": "#D9EAD3",
"horizontal_alignment": "CENTER",
},
assert_baseline=_baseline,
assert_outcome=_format_cells_outcome,
),
)

3
tests/e2e/scenarios/test_reborn_qa_trace_full_path.py
View File

@@ -23,7 +23,8 @@ from provider_capability_inventory import (
EMULATE_SUPPORTED_TOOLS,
capability_id_to_wire_name,
)
from provider_operation_cases import PROVIDER_OPERATION_CASES, ProviderOperationCase
from provider_operation_cases import PROVIDER_OPERATION_CASES
from provider_operation_types import ProviderOperationCase
from reborn_webui_harness import (
YOLO_PROFILE,
capability_preview_payload,