243 lines
11 KiB
Python
243 lines
11 KiB
Python
# SPDX-License-Identifier: BSD-2-Clause
|
|
# SPDX-FileCopyrightText: 2023 KDE e.V.
|
|
# SPDX-FileContributor: Ingo Klöcker <dev@ingo-kloecker.de>
|
|
|
|
from unittest.mock import ANY, create_autospec
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
import sftpnotary.authentication
|
|
from sftpnotary import projects
|
|
from sftpnotary.authentication import AuthenticationFailed, GitLabJobTokenAuthentication
|
|
from sftpnotary.core import Request
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def no_requests(monkeypatch):
|
|
"""Remove requests.sessions.Session.request for all tests to prevent "requests"
|
|
from performing actual HTTP requests."""
|
|
monkeypatch.delattr("requests.sessions.Session.request")
|
|
|
|
|
|
def monkeyPatchWithAutospec(monkeyPatch, obj, name):
|
|
mock = create_autospec(getattr(obj, name))
|
|
monkeyPatch.setattr(obj, name, mock)
|
|
return mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_requests_get(monkeypatch):
|
|
return monkeyPatchWithAutospec(monkeypatch, requests, "get")
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_apiRequest(monkeypatch):
|
|
return monkeyPatchWithAutospec(monkeypatch, GitLabJobTokenAuthentication, "apiRequest")
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_getJob(monkeypatch):
|
|
return monkeyPatchWithAutospec(monkeypatch, GitLabJobTokenAuthentication, "getJob")
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_getProject(monkeypatch):
|
|
return monkeyPatchWithAutospec(monkeypatch, GitLabJobTokenAuthentication, "getProject")
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_projects_settings(monkeypatch):
|
|
mock = create_autospec(projects.ProjectSettings)
|
|
monkeypatch.setattr(sftpnotary.authentication.projects, "settings", mock)
|
|
return mock
|
|
|
|
|
|
class MockResponse:
|
|
def __init__(self, status_code=200, headers=None, json=None):
|
|
self.status_code = status_code
|
|
self.ok = status_code < 400
|
|
self.text = ""
|
|
self.headers = headers if headers is not None else {}
|
|
self._json = json
|
|
|
|
def json(self):
|
|
if self._json == requests.JSONDecodeError:
|
|
raise requests.JSONDecodeError("", "", 0)
|
|
return self._json
|
|
|
|
def raise_for_status(self):
|
|
if not self.ok:
|
|
raise requests.HTTPError(f"{self.status_code} HTTP Error", response=self)
|
|
|
|
|
|
class TestAuthentication:
|
|
def test_init(self):
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4", userAgent="SFTPNotary Test")
|
|
|
|
assert auth.apiV4Url == "https://gitlab.example.net/api/v4"
|
|
assert auth.userAgent == "SFTPNotary Test"
|
|
|
|
def test_apiRequest(self, mock_requests_get):
|
|
mock_requests_get.return_value = MockResponse(json={"foo": "bar"}, headers={})
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
r = auth.apiRequest("/path/")
|
|
|
|
assert r == {"foo": "bar"}
|
|
assert mock_requests_get.call_count == 1
|
|
assert mock_requests_get.call_args.args[0] == f"{auth.apiV4Url}/path/"
|
|
headers = mock_requests_get.call_args.kwargs.get("headers")
|
|
assert headers is not None, "headers not passed to request function"
|
|
assert headers == {}
|
|
timeout = mock_requests_get.call_args.kwargs.get("timeout")
|
|
assert timeout is not None, "timeout not passed to request function"
|
|
assert timeout == 5, "timeout doesn't default to 5"
|
|
|
|
def test_apiRequest_with_keyword_arguments(self, mock_requests_get):
|
|
mock_requests_get.return_value = MockResponse(json={"foo": "bar"}, headers={})
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4", userAgent="SFTPNotary Test")
|
|
auth.apiRequest("/path/", headers={"Authorization": "Bearer foo"}, timeout=42, keyword="value")
|
|
|
|
headers = mock_requests_get.call_args.kwargs.get("headers")
|
|
assert headers.get("Authorization") == "Bearer foo"
|
|
assert headers.get("User-Agent") == auth.userAgent
|
|
timeout = mock_requests_get.call_args.kwargs.get("timeout")
|
|
assert timeout == 42
|
|
keyword = mock_requests_get.call_args.kwargs.get("keyword")
|
|
assert keyword == "value", "keyword argument 'keyword' not passed as-is to request function"
|
|
|
|
def test_apiRequest_failed_request(self, mock_requests_get):
|
|
mock_requests_get.return_value = MockResponse(status_code=400, json={}, headers={})
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(requests.HTTPError, match="HTTP Error"):
|
|
auth.apiRequest("/path/")
|
|
|
|
def test_apiRequest_invalid_response(self, mock_requests_get):
|
|
mock_requests_get.return_value = MockResponse(status_code=200, json=requests.JSONDecodeError, headers={})
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(requests.JSONDecodeError):
|
|
auth.apiRequest("/path/")
|
|
|
|
def test_getJob(self, mock_apiRequest):
|
|
mock_apiRequest.return_value = {"foo": "bar"}
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
job = auth.getJob("job-token")
|
|
|
|
assert job == {"foo": "bar"}
|
|
mock_apiRequest.assert_called_once_with(auth, "/job", headers=ANY)
|
|
headers = mock_apiRequest.call_args.kwargs.get("headers")
|
|
assert headers.get("Authorization") == "Bearer job-token"
|
|
|
|
def test_getJob_failed_request(self, mock_apiRequest):
|
|
mock_apiRequest.side_effect = requests.RequestException
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
job = auth.getJob("job-token")
|
|
|
|
assert job is None
|
|
|
|
def test_getProject(self, mock_apiRequest):
|
|
mock_apiRequest.return_value = {"foo": "bar"}
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
project = auth.getProject(42)
|
|
|
|
assert project == {"foo": "bar"}
|
|
mock_apiRequest.assert_called_once_with(auth, "/projects/42")
|
|
|
|
def test_getProject_failed_request(self, mock_apiRequest):
|
|
mock_apiRequest.side_effect = requests.RequestException
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
project = auth.getProject(42)
|
|
|
|
assert project is None
|
|
|
|
def test_authenticate(self, mock_getJob, mock_getProject, mock_projects_settings):
|
|
mock_getJob.return_value = {"pipeline": {"project_id": 42, "source": "push"}, "ref": "branchA"}
|
|
mock_getProject.return_value = {"path_with_namespace": "namespaceA/projectA"}
|
|
mock_projects_settings.exists.return_value = True
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|
|
|
|
mock_getJob.assert_called_once_with(auth, "job-token")
|
|
mock_getProject.assert_called_once_with(auth, 42)
|
|
mock_projects_settings.exists.assert_called_once_with("namespaceA/projectA", "branchA")
|
|
|
|
def test_authenticate_no_job(self, mock_getJob):
|
|
mock_getJob.return_value = None
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(AuthenticationFailed, match="Failed to retrieve job"):
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|
|
|
|
def test_authenticate_job_without_project_id(self, mock_getJob):
|
|
mock_getJob.return_value = {"ref": "branchA"}
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(AuthenticationFailed, match="pipeline.project_id"):
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|
|
|
|
def test_authenticate_no_project(self, mock_getJob, mock_getProject):
|
|
mock_getJob.return_value = {"pipeline": {"project_id": 42, "source": "push"}, "ref": "branchA"}
|
|
mock_getProject.return_value = None
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(AuthenticationFailed, match="Failed to retrieve project"):
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|
|
|
|
def test_authenticate_job_without_ref(self, mock_getJob, mock_getProject):
|
|
mock_getJob.return_value = {"pipeline": {"project_id": 42, "source": "push"}}
|
|
mock_getProject.return_value = {}
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(AuthenticationFailed, match="does not match the job's branch"):
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|
|
|
|
def test_authenticate_job_with_wrong_branch(self, mock_getJob, mock_getProject):
|
|
mock_getJob.return_value = {"pipeline": {"project_id": 42, "source": "push"}, "ref": "wrongBranch"}
|
|
mock_getProject.return_value = {}
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(AuthenticationFailed, match="does not match the job's branch"):
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|
|
|
|
def test_authenticate_merge_request_job(self, mock_getJob, mock_getProject, mock_projects_settings):
|
|
mock_getJob.return_value = {"pipeline": {"project_id": 42, "source": "merge_request_event"}, "ref": "branchA"}
|
|
mock_getProject.return_value = {"path_with_namespace": "namespaceA/projectA"}
|
|
mock_projects_settings.exists.return_value = True
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(AuthenticationFailed, match="Merge requests are not cleared for this request"):
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|
|
|
|
def test_authenticate_project_without_path(self, mock_getJob, mock_getProject):
|
|
mock_getJob.return_value = {"pipeline": {"project_id": 42, "source": "push"}, "ref": "branchA"}
|
|
mock_getProject.return_value = {}
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(AuthenticationFailed, match="does not match the job's project path"):
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|
|
|
|
def test_authenticate_project_with_wrong_path(self, mock_getJob, mock_getProject):
|
|
mock_getJob.return_value = {"pipeline": {"project_id": 42, "source": "push"}, "ref": "branchA"}
|
|
mock_getProject.return_value = {"path_with_namespace": "wrong/path"}
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(AuthenticationFailed, match="does not match the job's project path"):
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|
|
|
|
def test_authenticate_disallowed_project_branch(self, mock_getJob, mock_getProject, mock_projects_settings):
|
|
mock_getJob.return_value = {"pipeline": {"project_id": 42, "source": "push"}, "ref": "branchA"}
|
|
mock_getProject.return_value = {"path_with_namespace": "namespaceA/projectA"}
|
|
mock_projects_settings.exists.return_value = False
|
|
|
|
auth = GitLabJobTokenAuthentication("https://gitlab.example.net/api/v4")
|
|
with pytest.raises(AuthenticationFailed, match="not cleared"):
|
|
auth.authenticate(request=Request(token="job-token", branch="branchA", projectPath="namespaceA/projectA"))
|