This commit is contained in:
@@ -0,0 +1,73 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from urllib.error import HTTPError
|
||||
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
||||
|
||||
import trigger_project_refresh as refresh # noqa: E402
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, status: int = 204) -> None:
|
||||
self.status = status
|
||||
|
||||
def __enter__(self) -> "FakeResponse":
|
||||
return self
|
||||
|
||||
def __exit__(self, *args: object) -> None:
|
||||
return None
|
||||
|
||||
def getcode(self) -> int:
|
||||
return self.status
|
||||
|
||||
|
||||
class TriggerProjectRefreshTests(unittest.TestCase):
|
||||
def test_dispatch_posts_main_without_putting_token_in_url_or_body(self) -> None:
|
||||
with patch.object(refresh, "urlopen", return_value=FakeResponse()) as opener:
|
||||
refresh.dispatch_refresh("secret-token", timeout=7)
|
||||
|
||||
request = opener.call_args.args[0]
|
||||
self.assertEqual(request.full_url, refresh.DISPATCH_URL)
|
||||
self.assertEqual(request.method, "POST")
|
||||
self.assertEqual(request.data, b'{"ref": "main"}')
|
||||
self.assertEqual(request.get_header("Authorization"), "token secret-token")
|
||||
self.assertNotIn(b"secret-token", request.data)
|
||||
self.assertEqual(opener.call_args.kwargs["timeout"], 7)
|
||||
|
||||
def test_dispatch_reports_http_status_without_response_or_token(self) -> None:
|
||||
error = HTTPError(refresh.DISPATCH_URL, 403, "Forbidden", {}, None)
|
||||
with patch.object(refresh, "urlopen", side_effect=error):
|
||||
with self.assertRaisesRegex(refresh.RefreshError, r"HTTP 403") as raised:
|
||||
refresh.dispatch_refresh("secret-token")
|
||||
self.assertNotIn("secret-token", str(raised.exception))
|
||||
|
||||
def test_main_requires_environment_token(self) -> None:
|
||||
stderr = io.StringIO()
|
||||
with patch.dict(os.environ, {}, clear=True), contextlib.redirect_stderr(stderr):
|
||||
result = refresh.main([])
|
||||
self.assertEqual(result, 2)
|
||||
self.assertIn(refresh.TOKEN_ENVIRONMENT_VARIABLE, stderr.getvalue())
|
||||
|
||||
def test_dry_run_does_not_read_or_send_token(self) -> None:
|
||||
stdout = io.StringIO()
|
||||
with (
|
||||
patch.dict(os.environ, {}, clear=True),
|
||||
patch.object(refresh, "urlopen") as opener,
|
||||
contextlib.redirect_stdout(stdout),
|
||||
):
|
||||
result = refresh.main(["--dry-run"])
|
||||
self.assertEqual(result, 0)
|
||||
self.assertIn(refresh.DISPATCH_URL, stdout.getvalue())
|
||||
opener.assert_not_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user