92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
"""Safe access diagnostics preserve the stable client error contract."""
|
|
|
|
import json
|
|
from uuid import UUID
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from packages.connector_kit.gitea import (
|
|
GiteaClient,
|
|
GiteaError,
|
|
GiteaSettings,
|
|
RegisteredRepository,
|
|
StaticTokenProvider,
|
|
)
|
|
|
|
SECRET_REFERENCE = "docker-secret://" + "test"
|
|
|
|
|
|
@pytest.mark.parametrize("status", [401, 403])
|
|
@pytest.mark.parametrize(
|
|
("method", "path", "scope"),
|
|
[
|
|
("POST", "user/repos", "write:user"),
|
|
("GET", "user/repos", "read:user"),
|
|
("GET", "repos/private-owner/private-name", None),
|
|
],
|
|
)
|
|
def test_access_diagnostics_are_specific_sanitized_and_not_retried(status, method, path, scope):
|
|
calls = []
|
|
|
|
def upstream(request):
|
|
calls.append(request)
|
|
return httpx.Response(status, json={"message": "upstream-secret-value"})
|
|
|
|
client = GiteaClient(
|
|
GiteaSettings(base_url="https://gitea.example", token_reference=SECRET_REFERENCE),
|
|
StaticTokenProvider("credential-secret-value"),
|
|
transport=httpx.MockTransport(upstream),
|
|
)
|
|
try:
|
|
with pytest.raises(GiteaError) as caught:
|
|
client._request(path, method=method)
|
|
error = caught.value
|
|
envelope = error.envelope(operation="test")
|
|
assert error.code == "GITEA_ACCESS_DENIED"
|
|
assert error.upstream_status == status
|
|
assert error.category == ("authentication" if status == 401 else "permission")
|
|
assert not error.retryable and not error.safe_to_retry
|
|
assert len(calls) == 1
|
|
rendered = json.dumps(envelope)
|
|
for forbidden in [
|
|
"upstream-secret-value",
|
|
"credential-secret-value",
|
|
"private-owner",
|
|
"private-name",
|
|
]:
|
|
assert forbidden not in rendered
|
|
if status == 403 and scope:
|
|
assert scope in error.recovery_hint
|
|
else:
|
|
assert "write:user" not in error.recovery_hint
|
|
assert "read:user" not in error.recovery_hint
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
@pytest.mark.parametrize("status", [401, 403])
|
|
def test_log_access_uses_the_same_safe_diagnostics(status):
|
|
client = GiteaClient(
|
|
GiteaSettings(base_url="https://gitea.example", token_reference=SECRET_REFERENCE),
|
|
StaticTokenProvider("fixture-token"),
|
|
transport=httpx.MockTransport(
|
|
lambda request: httpx.Response(status, text="upstream-secret")
|
|
),
|
|
)
|
|
repo = RegisteredRepository(
|
|
repository_ref="test",
|
|
project_id=UUID(int=1),
|
|
owner="owner",
|
|
name="repo",
|
|
source_uri="itworx-git://test/test",
|
|
)
|
|
try:
|
|
with pytest.raises(GiteaError) as caught:
|
|
client.action_job_logs(repo, 1)
|
|
assert caught.value.upstream_status == status
|
|
assert caught.value.category == ("authentication" if status == 401 else "permission")
|
|
assert "upstream-secret" not in json.dumps(caught.value.envelope(operation="logs"))
|
|
finally:
|
|
client.close()
|