test: keep webui CI hermetic offline
This commit is contained in:
+24
-5
@@ -228,10 +228,29 @@ def load_lease_snapshot(
|
||||
)
|
||||
|
||||
host = _host_from_url(project.remote_host)
|
||||
pr_fetch = fetch_prs or _fetch_prs
|
||||
issue_fetch = fetch_issues or _fetch_issues
|
||||
comment_fetch = fetch_comments or _fetch_comments
|
||||
using_live = fetch_prs is None or fetch_issues is None
|
||||
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
}
|
||||
|
||||
def _empty_pr_fetch(*_args, **_kwargs):
|
||||
return [], None
|
||||
|
||||
def _empty_issue_fetch(*_args, **_kwargs):
|
||||
return [], None
|
||||
|
||||
def _empty_comment_fetch(*_args, **_kwargs):
|
||||
return []
|
||||
|
||||
pr_fetch = fetch_prs or (_empty_pr_fetch if offline_test else _fetch_prs)
|
||||
issue_fetch = fetch_issues or (
|
||||
_empty_issue_fetch if offline_test else _fetch_issues
|
||||
)
|
||||
comment_fetch = fetch_comments or (
|
||||
_empty_comment_fetch if offline_test else _fetch_comments
|
||||
)
|
||||
using_live = not offline_test and (fetch_prs is None or fetch_issues is None)
|
||||
auth = get_auth_header(host) if using_live else "test-auth"
|
||||
|
||||
if using_live and not auth:
|
||||
@@ -328,4 +347,4 @@ def snapshot_to_dict(snapshot: LeaseSnapshot) -> dict[str, Any]:
|
||||
],
|
||||
"collision_history": list(snapshot.collision_history),
|
||||
"fetch_error": snapshot.fetch_error,
|
||||
}
|
||||
}
|
||||
|
||||
+19
-4
@@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
@@ -268,9 +269,23 @@ def load_queue_snapshot(
|
||||
)
|
||||
|
||||
host = _host_from_url(project.remote_host)
|
||||
pr_fetch = fetch_prs or _fetch_prs
|
||||
issue_fetch = fetch_issues or _fetch_issues
|
||||
using_live_fetch = fetch_prs is None or fetch_issues is None
|
||||
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
}
|
||||
|
||||
def _empty_fetch(*_args, **_kwargs):
|
||||
return [], _pagination_from_pages(
|
||||
per_page=50,
|
||||
pages_fetched=1,
|
||||
returned_count=0,
|
||||
is_final_page=True,
|
||||
)
|
||||
|
||||
pr_fetch = fetch_prs or (_empty_fetch if offline_test else _fetch_prs)
|
||||
issue_fetch = fetch_issues or (_empty_fetch if offline_test else _fetch_issues)
|
||||
using_live_fetch = not offline_test and (fetch_prs is None or fetch_issues is None)
|
||||
auth = get_auth_header(host) if using_live_fetch else "test-auth"
|
||||
if using_live_fetch and not auth:
|
||||
return QueueSnapshot(
|
||||
@@ -382,4 +397,4 @@ def snapshot_to_dict(snapshot: QueueSnapshot) -> dict[str, Any]:
|
||||
"prs": _page(snapshot.pr_pagination),
|
||||
"issues": _page(snapshot.issue_pagination),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
+18
-3
@@ -217,6 +217,11 @@ def load_runtime_snapshot(
|
||||
repo = _repo_root()
|
||||
host = _host_from_url(project.remote_host)
|
||||
remote = _remote_name_for_host(host)
|
||||
offline_test = (os.environ.get("WEBUI_TEST_OFFLINE") or "").strip() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
}
|
||||
profile = get_profile()
|
||||
allowed = profile.get("allowed_operations") or []
|
||||
forbidden = profile.get("forbidden_operations") or []
|
||||
@@ -247,10 +252,20 @@ def load_runtime_snapshot(
|
||||
fetch_error=str(exc),
|
||||
)
|
||||
|
||||
identity_fn = resolve_username or _resolve_username
|
||||
identity_fn = resolve_username
|
||||
if identity_fn is None:
|
||||
if offline_test:
|
||||
identity_fn = lambda _host: (None, "offline web UI test mode")
|
||||
else:
|
||||
identity_fn = _resolve_username
|
||||
username, identity_error = identity_fn(host)
|
||||
|
||||
git_fn = git_sync or _git_sync_status
|
||||
git_fn = git_sync
|
||||
if git_fn is None:
|
||||
if offline_test:
|
||||
git_fn = lambda _repo, _remote, _branch: (None, None, None)
|
||||
else:
|
||||
git_fn = _git_sync_status
|
||||
remote_ref = "prgs" if remote == "prgs" else "origin"
|
||||
local_sha, remote_sha, behind = git_fn(repo, remote_ref, project.default_branch)
|
||||
|
||||
@@ -317,4 +332,4 @@ def snapshot_to_dict(snapshot: RuntimeSnapshot) -> dict[str, Any]:
|
||||
"schema_hashes": [_hash_row(item) for item in snapshot.schema_hashes],
|
||||
"restart_guidance": snapshot.restart_guidance,
|
||||
"fetch_error": snapshot.fetch_error,
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user