"""Exact-head native Gitea Actions discovery without synthetic PR proof.""" from __future__ import annotations import copy from unittest.mock import patch import pytest from packages.connector_kit import GiteaError from tests.test_gitea_connector import connector, context HEAD = "e" * 40 BRANCH = "codex/workflows" def native_run(**changes): return { "id": 42, "run_number": 17, "run_attempt": 1, "name": None, "workflow_id": None, "display_title": "ci: validate actual commit message", "path": "managed-validation.yml@refs/pull/7/head", "event": "pull_request", "head_branch": "", "head_sha": HEAD, "status": "completed", "conclusion": "success", **changes, } def setup_native(runs=None): instance, fixture = connector() fixture._branches.append({"name": BRANCH, "commit": {"id": HEAD}, "protected": False}) fixture._action_run = native_run() native = runs if runs is not None else [fixture._action_run] original = instance.client._request requests = [] def request(path, *args, **kwargs): if path.endswith("/actions/runs"): query = args[0] if args else kwargs.get("params", {}) requests.append(dict(query)) if query.get("branch"): return {"workflow_runs": [], "total_count": 0} assert query.get("head_sha") == HEAD return {"workflow_runs": copy.deepcopy(native), "total_count": len(native)} return original(path, *args, **kwargs) instance.client._request = request return instance, fixture, requests def test_change_status_discovers_original_native_pr_run_with_null_workflow_id(): instance, _fixture, requests = setup_native() try: status = instance.change_status(context(), "hub-repository", BRANCH)["data"] assert status["validationRuns"][0]["id"] == 42 assert status["validationRuns"][0]["workflowId"] == "" assert status["validationRuns"][0]["workflowPath"] == native_run()["path"] assert status["validationState"] == "succeeded" assert status["readyToMerge"] is True assert "run-validation" not in status["nextActions"] assert requests and all("branch" not in item for item in requests) finally: instance.client.close() def test_validation_ensure_reuses_bound_native_canonical_pr_run(): instance, fixture, requests = setup_native() try: result = instance.pull_request_validation_ensure(context(), "hub-repository", 7, HEAD)[ "data" ] assert result["state"] == "success" assert result["runId"] == 42 assert result["reused"] is True assert requests and all(item.get("head_sha") == HEAD for item in requests) assert "POST" not in fixture.recorded_methods assert "PATCH" not in fixture.recorded_methods finally: instance.client.close() @pytest.mark.parametrize( "changes", [ {"path": "managed-validation.yml", "pull_requests": []}, {"path": "security.yml@refs/pull/7/head"}, {"path": "managed-validation.yml@refs/pull/8/head"}, {"pull_requests": [{"number": 8}]}, {"workflow_id": "security.yml", "status": "queued", "conclusion": ""}, {"path": "security.yml@refs/pull/7/head", "conclusion": "failure"}, ], ) def test_validation_ensure_does_not_reuse_or_rerun_unbound_native_run(changes): instance, fixture, _requests = setup_native([native_run(**changes)]) fixture._action_run.update(changes) try: with ( patch.object(instance._writer(), "rerun_action_run") as rerun, patch.object( instance._writer(), "close_pull_request", side_effect=GiteaError("GITEA_ACCESS_DENIED", "PR event trigger unavailable"), ) as close, pytest.raises(GiteaError) as error, ): instance.pull_request_validation_ensure(context(), "hub-repository", 7, HEAD) assert error.value.code == "GITEA_ACCESS_DENIED" close.assert_called_once() rerun.assert_not_called() assert "POST" not in fixture.recorded_methods assert "PATCH" not in fixture.recorded_methods finally: instance.client.close() @pytest.mark.parametrize( "changes", [ {"head_sha": "d" * 40}, {"head_branch": "other"}, {"path": "managed-validation.yml@refs/pull/8/head"}, {"pull_requests": [{"number": 8}]}, {"pull_requests": [{"number": 7, "head": {"ref": "other", "sha": HEAD}}]}, {"pull_requests": [{"number": 7, "head": {"ref": BRANCH, "sha": "d" * 40}}]}, {"pull_requests": "malformed"}, {"pull_requests": {}}, {"pull_requests": [{"number": 7, "head": {"ref": [], "sha": HEAD}}]}, {"path": "managed-validation.yml", "pull_requests": []}, ], ) def test_native_exact_head_read_rejects_contradictory_or_unbound_pr_metadata(changes): instance, fixture, _requests = setup_native([native_run(**changes)]) fixture._action_run.update(changes) try: repository = instance._repo(context(), "hub-repository") assert ( instance.client.exact_head_action_runs( repository, branch=BRANCH, head_sha=HEAD, pull_request_number=7, ) == [] ) with pytest.raises(GiteaError) as error: instance.pull_request_merge(context(), "hub-repository", 7, HEAD, "squash") assert error.value.code == "GITEA_PR_VALIDATION_REQUIRED" assert not any(body.get("Do") == "squash" for body in fixture.recorded_bodies) finally: instance.client.close() @pytest.mark.parametrize( "path", [ "security.yml@refs/pull/7/head", "untrusted/managed-validation.yml@refs/pull/7/head", "managed-validation.yml@refs/pull/7/head@extra", ], ) def test_native_workflow_title_cannot_substitute_for_canonical_path(path): instance, fixture, _requests = setup_native([native_run(path=path)]) fixture._action_run["path"] = path try: status = instance.change_status(context(), "hub-repository", BRANCH)["data"] assert status["validationRuns"] == [] assert status["readyToMerge"] is False with pytest.raises(GiteaError) as error: instance.pull_request_merge(context(), "hub-repository", 7, HEAD, "squash") assert error.value.code == "GITEA_PR_VALIDATION_REQUIRED" finally: instance.client.close() def test_newer_native_failed_attempt_controls_validation_projection(): runs = [native_run(), native_run(id=44, run_attempt=2, conclusion="failure")] instance, fixture, _requests = setup_native(runs) fixture._statuses[0]["status"] = "failure" try: status = instance.change_status(context(), "hub-repository", BRANCH)["data"] assert status["validationRuns"][0]["id"] == 44 assert status["validationRuns"][0]["runAttempt"] == 2 assert status["readyToMerge"] is False finally: instance.client.close() @pytest.mark.parametrize("failure", ["duplicate", "empty", "changed", "overflow"]) def test_native_inventory_rejects_incomplete_or_nonadvancing_pages(failure): instance, _fixture, _requests = setup_native() original = instance.client._request def request(path, *args, **kwargs): if path.endswith("/actions/runs"): query = args[0] if query["page"] == 1: return {"workflow_runs": [native_run()], "total_count": 2} if failure == "duplicate": return {"workflow_runs": [native_run()], "total_count": 2} if failure == "empty": return {"workflow_runs": [], "total_count": 2} if failure == "changed": return {"workflow_runs": [native_run(id=43)], "total_count": 3} return {"workflow_runs": [native_run(id=43), native_run(id=44)], "total_count": 2} return original(path, *args, **kwargs) instance.client._request = request try: repository = instance._repo(context(), "hub-repository") with pytest.raises(GiteaError) as error: instance.client.exact_head_action_runs( repository, branch=BRANCH, head_sha=HEAD, pull_request_number=7, ) assert error.value.code in {"GITEA_ACTION_RUNS_INVALID", "GITEA_ACTION_RUNS_TRUNCATED"} finally: instance.client.close() def test_validation_get_preserves_skipped_job_metadata_without_missing_logs(): instance, _fixture = connector() jobs = [ {"id": 43, "name": "preflight", "status": "completed", "conclusion": "failure"}, {"id": 44, "name": "build", "status": "completed", "conclusion": "skipped"}, ] try: with ( patch.object(instance.client, "action_run_jobs", return_value=jobs), patch.object( instance.client, "action_job_logs", return_value=("preflight failed", False), ) as logs, ): result = instance.validation_get(context(), "hub-repository", 42, True)["data"]["run"] assert len(result["jobs"]) == 2 assert result["jobs"][1]["conclusion"] == "skipped" assert result["jobs"][1]["logsUnavailable"] is True assert logs.call_count == 1 finally: instance.client.close() def test_validation_get_marks_expected_missing_logs_without_suppressing_other_errors(): instance, _fixture = connector() try: with patch.object( instance.client, "action_job_logs", side_effect=GiteaError( "GITEA_NOT_FOUND", "logs unavailable", ), ): result = instance.validation_get(context(), "hub-repository", 42, True)["data"]["run"] assert result["jobs"][0]["logsUnavailable"] is True assert result["jobs"][0]["status"] == "completed" with ( patch.object( instance.client, "action_job_logs", side_effect=GiteaError( "GITEA_ACCESS_DENIED", "access denied", ), ), pytest.raises(GiteaError) as error, ): instance.validation_get(context(), "hub-repository", 42, True) assert error.value.code == "GITEA_ACCESS_DENIED" finally: instance.client.close() def test_legacy_branch_read_prefers_actual_native_run_before_status_fallback(): instance, _fixture, _requests = setup_native() try: repository = instance._repo(context(), "hub-repository") runs = instance.client.action_runs(repository, branch=BRANCH, page=1, limit=25) assert len(runs) == 1 and runs[0]["id"] == 42 assert "validation_evidence" not in runs[0] finally: instance.client.close() def test_native_read_walks_all_actual_pages_and_returns_newest_attempt_first(): instance, _fixture, _requests = setup_native() original = instance.client._request seen = [] def request(path, *args, **kwargs): if path.endswith("/actions/runs"): page = args[0]["page"] seen.append(page) assert kwargs["max_attempts"] == 1 and 0 < kwargs["timeout_seconds"] <= 10 return {"workflow_runs": [native_run(id=41 + page, run_attempt=page)], "total_count": 2} return original(path, *args, **kwargs) instance.client._request = request try: repository = instance._repo(context(), "hub-repository") runs = instance.client.exact_head_action_runs( repository, branch=BRANCH, head_sha=HEAD, pull_request_number=7 ) assert seen == [1, 2] and [run["id"] for run in runs] == [43, 42] finally: instance.client.close() @pytest.mark.parametrize("changes", [{"id": True}, {"run_attempt": 0}, {"run_number": None}]) def test_native_inventory_rejects_malformed_run_ordering(changes): instance, _fixture, _requests = setup_native([native_run(**changes)]) try: repository = instance._repo(context(), "hub-repository") with pytest.raises(GiteaError) as error: instance.client.exact_head_action_runs( repository, branch=BRANCH, head_sha=HEAD, pull_request_number=7 ) assert error.value.code == "GITEA_ACTION_RUNS_INVALID" finally: instance.client.close() @pytest.mark.parametrize("length", [39, 41, 63, 65]) def test_native_read_rejects_non_git_sha_lengths(length): instance, fixture, _requests = setup_native() try: repository = instance._repo(context(), "hub-repository") with pytest.raises(GiteaError) as error: instance.client.exact_head_action_runs(repository, branch=BRANCH, head_sha="e" * length) assert error.value.code == "GITEA_TARGET_INVALID" assert not fixture.recorded_bodies finally: instance.client.close() def test_dispatch_exact_branch_ref_remains_bound_to_original_run(): instance, fixture = connector() try: fixture._branches.append({"name": BRANCH, "commit": {"id": HEAD}, "protected": False}) dispatched = instance.validation_run(context(), "hub-repository", BRANCH, "full")["data"] assert fixture._action_run["head_branch"] == f"refs/heads/{BRANCH}" result = instance.change_status(context(), "hub-repository", BRANCH)["data"] assert result["validationRuns"][0]["id"] == dispatched["id"] == 42 bad = dict(fixture._action_run, head_branch="refs/heads/other") assert not instance.client.matches_exact_head_action_run(bad, branch=BRANCH, head_sha=HEAD) finally: instance.client.close() def test_blank_native_pr_number_without_workflow_identity_cannot_authorize_merge(): run = native_run(path=None, workflow_id=None, pull_requests=[{"number": 7}]) instance, fixture, _requests = setup_native([run]) fixture._action_run = dict(run) try: result = instance.change_status(context(), "hub-repository", BRANCH)["data"] assert result["validationRuns"] == [] assert result["readyToMerge"] is False with pytest.raises(GiteaError) as error: instance.pull_request_merge(context(), "hub-repository", 7, HEAD, "squash") assert error.value.code == "GITEA_PR_VALIDATION_REQUIRED" assert not any(body.get("Do") == "squash" for body in fixture.recorded_bodies) finally: instance.client.close() def test_legacy_missing_workflow_identity_requires_exact_expected_branch_for_merge(): instance, fixture = connector() fixture._action_run.pop("workflow_id", None) fixture._action_run["head_branch"] = "other" try: with pytest.raises(GiteaError) as error: instance.pull_request_merge(context(), "hub-repository", 7, HEAD, "squash") assert error.value.code == "GITEA_PR_VALIDATION_REQUIRED" assert not any(body.get("Do") == "squash" for body in fixture.recorded_bodies) finally: instance.client.close() def test_actual_run_22404_null_head_branch_retains_pr236_original_proof(): instance, fixture = connector() actual_head = "c19a8620de6a24241b63aae4dbe52c0e5505aa62" run = native_run( id=22404, head_sha=actual_head, head_branch=None, run_number=1177, path="managed-validation.yml@refs/pull/236/head", ) fixture._action_run = run fixture._pulls[0]["number"] = 236 fixture._pulls[0]["head"]["sha"] = actual_head fixture._branches.append({"name": BRANCH, "commit": {"id": actual_head}, "protected": False}) fixture._statuses[0]["target_url"] = "https://gitea.example/acme/hub/actions/runs/22404" original_request = instance.client._request # The general fake transport serves only PR7; preserve actual PR236 payload. instance.client._request = lambda path, *args, **kwargs: original_request( path.replace("/pulls/236", "/pulls/7"), *args, **kwargs ) try: result = instance.change_status(context(), "hub-repository", BRANCH)["data"] assert result["validationRuns"][0]["id"] == 22404 assert result["validationRuns"][0]["headBranch"] == "" assert result["validationRuns"][0]["workflowPath"] == run["path"] assert result["readyToMerge"] is True merged = instance.pull_request_merge( context(), "hub-repository", 236, actual_head, "squash" )["data"] assert merged["validationRunId"] == 22404 finally: instance.client.close() @pytest.mark.parametrize("branch_metadata", [None, False, 0, [], {}]) def test_native_nonstring_branch_requires_null_pr_shape_and_workflow_identity(branch_metadata): run = native_run(head_branch=branch_metadata, path=None, pull_requests=[{"number": 7}]) instance, fixture, _requests = setup_native([run]) fixture._action_run = dict(run) try: result = instance.change_status(context(), "hub-repository", BRANCH)["data"] assert result["validationRuns"] == [] assert result["readyToMerge"] is False with pytest.raises(GiteaError): instance.pull_request_merge(context(), "hub-repository", 7, HEAD, "squash") assert not any(body.get("Do") == "squash" for body in fixture.recorded_bodies) finally: instance.client.close() def test_null_native_branch_does_not_hide_failed_required_workflow_context(): instance, fixture, _requests = setup_native([native_run(head_branch=None)]) fixture._action_run["head_branch"] = None fixture._branch_protections["acme/hub"][0]["status_check_contexts"].append( "Security / scan (pull_request)" ) fixture._statuses.append({"context": "Security / scan (pull_request)", "status": "failure"}) try: result = instance.change_status(context(), "hub-repository", BRANCH)["data"] assert result["validationRuns"][0]["id"] == 42 assert result["readyToMerge"] is False with pytest.raises(GiteaError) as error: instance.pull_request_merge(context(), "hub-repository", 7, HEAD, "squash") assert error.value.code == "GITEA_REQUIRED_CHECK_BLOCKED" assert not any(body.get("Do") == "squash" for body in fixture.recorded_bodies) finally: instance.client.close() def test_null_branch_is_not_absent_branch_for_push_or_dispatch(): instance, fixture = connector() try: for event in ("push", "workflow_dispatch"): assert not instance.client.matches_exact_head_action_run( native_run(head_branch=None, event=event), branch=BRANCH, head_sha=HEAD ) finally: instance.client.close()