summaryrefslogtreecommitdiff
path: root/tests/unittests/sources/azure/test_errors.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/unittests/sources/azure/test_errors.py')
-rw-r--r--tests/unittests/sources/azure/test_errors.py65
1 files changed, 65 insertions, 0 deletions
diff --git a/tests/unittests/sources/azure/test_errors.py b/tests/unittests/sources/azure/test_errors.py
index e9d3e039..cf3e0e71 100644
--- a/tests/unittests/sources/azure/test_errors.py
+++ b/tests/unittests/sources/azure/test_errors.py
@@ -5,9 +5,11 @@ import datetime
from unittest import mock
import pytest
+import requests
from cloudinit import version
from cloudinit.sources.azure import errors
+from cloudinit.url_helper import UrlError
@pytest.fixture()
@@ -134,6 +136,69 @@ def test_dhcp_interface_not_found():
assert error.supporting_data["duration"] == 5.6
+@pytest.mark.parametrize(
+ "exception,reason",
+ [
+ (
+ UrlError(
+ requests.ConnectionError(),
+ ),
+ "connection error querying IMDS",
+ ),
+ (
+ UrlError(
+ requests.ConnectTimeout(),
+ ),
+ "connection timeout querying IMDS",
+ ),
+ (
+ UrlError(
+ requests.ReadTimeout(),
+ ),
+ "read timeout querying IMDS",
+ ),
+ (
+ UrlError(
+ Exception(),
+ code=500,
+ ),
+ "http error querying IMDS",
+ ),
+ (
+ UrlError(
+ requests.HTTPError(),
+ code=None,
+ ),
+ "unexpected error querying IMDS",
+ ),
+ ],
+)
+def test_imds_url_error(exception, reason):
+ duration = 123.4
+ fake_url = "fake://url"
+
+ exception.url = fake_url
+ error = errors.ReportableErrorImdsUrlError(
+ exception=exception, duration=duration
+ )
+
+ assert error.reason == reason
+ assert error.supporting_data["duration"] == duration
+ assert error.supporting_data["exception"] == repr(exception)
+ assert error.supporting_data["url"] == fake_url
+
+
+def test_imds_metadata_parsing_exception():
+ exception = ValueError("foobar")
+
+ error = errors.ReportableErrorImdsMetadataParsingException(
+ exception=exception
+ )
+
+ assert error.reason == "error parsing IMDS metadata"
+ assert error.supporting_data["exception"] == repr(exception)
+
+
def test_unhandled_exception():
source_error = None
try: