238 lines
11 KiB
Python
238 lines
11 KiB
Python
"""Unit-Tests für die gemeinsame Python-HTTP- und CLI-Schicht."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
from urllib.error import HTTPError, URLError
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
TOOLS = ROOT / "tools" / "python"
|
|
sys.path.insert(0, str(TOOLS))
|
|
|
|
import common # noqa: E402
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, status=200, body=b"", content_type="application/json"):
|
|
self.status = status
|
|
self._body = body
|
|
self.headers = {"Content-Type": content_type}
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def read(self):
|
|
return self._body
|
|
|
|
|
|
class CommonTests(unittest.TestCase):
|
|
def test_environment_credentials_are_trimmed_and_require_both_values(self):
|
|
with patch.dict(os.environ, {common.API_USER_ENV: " user ", common.API_PASSWORD_ENV: " secret "}, clear=True):
|
|
self.assertEqual(common.credentials_from_environment(), common.Credentials("user", "secret"))
|
|
with patch.dict(os.environ, {common.API_USER_ENV: "user", common.API_PASSWORD_ENV: " "}, clear=True):
|
|
self.assertIsNone(common.credentials_from_environment())
|
|
|
|
def test_prompt_credentials_prefers_environment_then_interactive_values(self):
|
|
client = common.ApiClient("http://localhost:47145")
|
|
with patch.dict(os.environ, {common.API_USER_ENV: " env-user ", common.API_PASSWORD_ENV: " env-pass "}, clear=True):
|
|
client.prompt_credentials()
|
|
self.assertEqual(client.credentials, common.Credentials("env-user", "env-pass"))
|
|
|
|
client = common.ApiClient("http://localhost:47145", "default", "old")
|
|
with patch.dict(os.environ, {}, clear=True), patch("builtins.input", return_value=""), patch(
|
|
"getpass.getpass", return_value=" new-pass "
|
|
):
|
|
client.prompt_credentials()
|
|
self.assertEqual(client.credentials, common.Credentials("default", "new-pass"))
|
|
|
|
with patch.dict(os.environ, {}, clear=True), patch("builtins.input", return_value=""):
|
|
with self.assertRaises(common.ApiError):
|
|
common.ApiClient("http://localhost:47145").prompt_credentials()
|
|
with patch.dict(os.environ, {}, clear=True), patch("builtins.input", return_value="user"), patch(
|
|
"getpass.getpass", return_value=" "
|
|
):
|
|
with self.assertRaises(common.ApiError):
|
|
common.ApiClient("http://localhost:47145").prompt_credentials()
|
|
|
|
def test_payload_generation_and_multipart_encoding(self):
|
|
data, headers = common._request_payload({"ä": "ö"}, None, None)
|
|
self.assertEqual(json.loads(data.decode("utf-8")), {"ä": "ö"})
|
|
self.assertEqual(headers["Content-Type"], common.APPLICATION_JSON)
|
|
|
|
data, headers = common._request_payload(None, {"value": 7}, None)
|
|
self.assertEqual(json.loads(data), {"value": 7})
|
|
self.assertEqual(headers["Accept"], common.APPLICATION_JSON)
|
|
|
|
data, headers = common._request_payload(None, None, None)
|
|
self.assertIsNone(data)
|
|
self.assertNotIn("Content-Type", headers)
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
file_path = Path(directory) / 'a"b.txt'
|
|
file_path.write_bytes(b"content")
|
|
data, headers = common._request_payload(None, {"skip": None, "caption": "Text"}, {"file": file_path})
|
|
self.assertIn(b'filename="a_b.txt"', data)
|
|
self.assertIn(b'name="caption"', data)
|
|
self.assertNotIn(b'name="skip"', data)
|
|
self.assertTrue(headers["Content-Type"].startswith("multipart/form-data; boundary="))
|
|
|
|
with self.assertRaises(FileNotFoundError):
|
|
common.encode_multipart({}, {"file": Path(directory) / "missing.bin"})
|
|
|
|
def test_response_parsing_and_error_decoding(self):
|
|
self.assertIsNone(common.parse_json(b""))
|
|
self.assertEqual(common.parse_json(b'{"ok":true}'), {"ok": True})
|
|
self.assertIsNone(common.parse_json(b"\xff"))
|
|
self.assertIsNone(common.parse_json(b"{"))
|
|
|
|
self.assertIsNone(common.parse_response(b"", common.APPLICATION_JSON))
|
|
self.assertEqual(common.parse_response(b'{"ok":true}', "application/problem+json"), {"ok": True})
|
|
self.assertEqual(common.parse_response(b"plain", "text/plain"), "plain")
|
|
self.assertEqual(common.parse_response(b"not-json", common.APPLICATION_JSON), "not-json")
|
|
|
|
self.assertEqual(common.decode_error(b'{"message":"bad"}'), "bad")
|
|
self.assertEqual(common.decode_error(b'{"error":"wrong"}'), "wrong")
|
|
self.assertIn("value", common.decode_error(b'{"value":1}'))
|
|
self.assertEqual(common.decode_error(b" text "), "text")
|
|
self.assertEqual(common.decode_error(b""), "Unbekannter API-Fehler")
|
|
|
|
def test_client_request_success_retry_and_failures(self):
|
|
anonymous = common.ApiClient("http://localhost:47145")
|
|
self.assertIsNone(anonymous._authorization_header())
|
|
self.assertEqual(anonymous._request_headers({"Accept": common.APPLICATION_JSON}), {"Accept": common.APPLICATION_JSON})
|
|
|
|
client = common.ApiClient("http://localhost:47145", " user ", " pass ")
|
|
self.assertEqual(client.base_url, "http://localhost:47145/")
|
|
self.assertTrue(client._authorization_header().startswith("Basic "))
|
|
self.assertIn("Authorization", client._request_headers({"Accept": common.APPLICATION_JSON}))
|
|
|
|
with patch.object(common, "urlopen", return_value=FakeResponse(200, b'{"ok":true}')):
|
|
self.assertEqual(client.request("GET", "/api/health"), {"ok": True})
|
|
|
|
with patch.object(common, "urlopen", return_value=FakeResponse(204, b"", "text/plain")):
|
|
self.assertIsNone(client.request("DELETE", "/api/item", expected=(204,)))
|
|
|
|
with patch.object(common, "urlopen", return_value=FakeResponse(202, b'{"queued":true}')):
|
|
with self.assertRaises(common.ApiError) as context:
|
|
client.request("POST", "/api/item")
|
|
self.assertEqual(context.exception.status, 202)
|
|
|
|
unauthorized = HTTPError(
|
|
"http://localhost:47145/api/item",
|
|
401,
|
|
"Unauthorized",
|
|
{},
|
|
io.BytesIO(b'{"message":"login"}'),
|
|
)
|
|
with patch.object(common, "urlopen", side_effect=[unauthorized, FakeResponse(200, b'{"ok":true}')]), patch.object(
|
|
client, "prompt_credentials"
|
|
) as prompt:
|
|
self.assertEqual(client.request("POST", "/api/item"), {"ok": True})
|
|
prompt.assert_called_once_with()
|
|
|
|
forbidden = HTTPError(
|
|
"http://localhost:47145/api/item",
|
|
403,
|
|
"Forbidden",
|
|
{},
|
|
io.BytesIO(b'{"error":"denied"}'),
|
|
)
|
|
with patch.object(common, "urlopen", side_effect=forbidden):
|
|
with self.assertRaises(common.ApiError) as context:
|
|
client.request("POST", "/api/item")
|
|
self.assertEqual(context.exception.status, 403)
|
|
self.assertEqual(context.exception.payload, {"error": "denied"})
|
|
|
|
with patch.object(common, "urlopen", side_effect=URLError("offline")):
|
|
with self.assertRaisesRegex(RuntimeError, "offline"):
|
|
client.request("GET", "/api/health")
|
|
|
|
def test_argument_and_prompt_helpers(self):
|
|
parser = argparse.ArgumentParser()
|
|
common.add_common_arguments(parser)
|
|
args = parser.parse_args(["--base-url", "http://example.test", "--username", "alice"])
|
|
client = common.client_from_args(args)
|
|
self.assertEqual(client.base_url, "http://example.test/")
|
|
self.assertIsNone(client.credentials)
|
|
|
|
env_args = argparse.Namespace(base_url=None, username=None)
|
|
with patch.dict(os.environ, {"WEGWICHTEL_URL": "http://env.test"}, clear=True):
|
|
self.assertEqual(common.client_from_args(env_args).base_url, "http://env.test/")
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
self.assertEqual(common.client_from_args(env_args).base_url, common.DEFAULT_BASE_URL + "/")
|
|
|
|
self.assertEqual(common.ask("Zahl", 4, cast=int), 4)
|
|
stderr = io.StringIO()
|
|
with patch("builtins.input", side_effect=["bad", "5"]), redirect_stderr(stderr):
|
|
self.assertEqual(common.ask("Zahl", required=True, cast=int), 5)
|
|
self.assertIn("Ungültiger Wert", stderr.getvalue())
|
|
with patch("builtins.input", return_value=""):
|
|
self.assertIsNone(common.ask("Optional"))
|
|
with patch("builtins.input", side_effect=["", "value"]), redirect_stderr(io.StringIO()):
|
|
self.assertEqual(common.ask("Pflicht", required=True), "value")
|
|
|
|
with patch("builtins.input", return_value=""):
|
|
self.assertEqual(common.ask_with_default("Wert", 3, int), 3)
|
|
with patch("builtins.input", side_effect=["bad", "4"]), redirect_stderr(io.StringIO()):
|
|
self.assertEqual(common.ask_with_default("Wert", 3, int), 4)
|
|
|
|
def test_path_confirmation_output_and_run_error_tracebacks(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
existing = Path(directory) / "file.txt"
|
|
existing.write_text("ok", encoding="utf-8")
|
|
self.assertEqual(common.ask_path("Datei", str(existing)), existing.resolve())
|
|
with self.assertRaises(FileNotFoundError):
|
|
common.ask_path("Datei", str(Path(directory) / "missing"))
|
|
with patch("builtins.input", return_value=""):
|
|
self.assertIsNone(common.ask_path("Optional", required=False))
|
|
with patch("builtins.input", side_effect=["", str(Path(directory) / "missing"), str(existing)]), redirect_stderr(
|
|
io.StringIO()
|
|
):
|
|
self.assertEqual(common.ask_path("Pflicht"), existing.resolve())
|
|
|
|
with patch("builtins.input", return_value="Ja"):
|
|
self.assertTrue(common.confirm("Fortfahren?"))
|
|
with patch("builtins.input", return_value="nein"):
|
|
self.assertFalse(common.confirm("Fortfahren?"))
|
|
|
|
stdout = io.StringIO()
|
|
with redirect_stdout(stdout):
|
|
common.print_result({"ok": True})
|
|
common.print_result("text")
|
|
self.assertIn('"ok": true', stdout.getvalue())
|
|
self.assertTrue(stdout.getvalue().rstrip().endswith("text"))
|
|
|
|
stdout = io.StringIO()
|
|
with redirect_stdout(stdout):
|
|
common.run(lambda: {"ok": True})
|
|
self.assertIn('"ok": true', stdout.getvalue())
|
|
|
|
stderr = io.StringIO()
|
|
with redirect_stderr(stderr), self.assertRaises(SystemExit) as context:
|
|
common.run(lambda: (_ for _ in ()).throw(ValueError("kaputt")))
|
|
self.assertEqual(context.exception.code, 1)
|
|
self.assertIn("Traceback", stderr.getvalue())
|
|
self.assertIn("ValueError: kaputt", stderr.getvalue())
|
|
|
|
stderr = io.StringIO()
|
|
with redirect_stderr(stderr), self.assertRaises(SystemExit) as context:
|
|
common.run(lambda: (_ for _ in ()).throw(KeyboardInterrupt()))
|
|
self.assertEqual(context.exception.code, 130)
|
|
self.assertIn("KeyboardInterrupt", stderr.getvalue())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|