File: test_http.py

package info (click to toggle)
python-azure 20251118%2Bgit-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 783,356 kB
  • sloc: python: 6,474,533; ansic: 804; javascript: 287; sh: 205; makefile: 198; xml: 109
file content (47 lines) | stat: -rw-r--r-- 2,021 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------

import base64
from types import SimpleNamespace
from azure.core.credentials import AzureKeyCredential
from azure.core.pipeline.transport import HttpRequest
from azure.core.pipeline import PipelineRequest
from azure.appconfiguration._azure_appconfiguration_requests import (
    AppConfigRequestsCredentialsPolicy,
)
from azure.appconfiguration._utils import parse_connection_string


def test_parse_connection_string_returns_http_endpoint():
    # An http (not https) endpoint to ensure scheme is preserved
    conn_str = "Endpoint=http://localhost:8483;Id=test-id;Secret=test-secret"
    endpoint, id_, secret = parse_connection_string(conn_str)
    assert endpoint == "http://localhost:8483"
    assert id_ == "test-id"
    assert secret == "test-secret"


def test_signed_request_with_http_endpoint():
    endpoint = "http://localhost:8483"
    request = HttpRequest("GET", endpoint + "/kv/foo?label=env")
    # Create pipeline request with placeholder context, then replace with a namespace that has a transport attr.
    pipeline_request = PipelineRequest(request, {})
    pipeline_request.context = SimpleNamespace(transport=object())

    key = base64.b64encode(b"secret").decode("utf-8")
    credential = AzureKeyCredential(key)
    policy = AppConfigRequestsCredentialsPolicy(credential, endpoint, "test-id")

    policy._signed_request(pipeline_request)  # pylint: disable=protected-access

    headers = pipeline_request.http_request.headers

    assert pipeline_request.http_request.url.startswith("http://")
    assert "x-ms-date" in headers
    assert "x-ms-content-sha256" in headers
    auth = headers.get("Authorization")
    assert auth is not None
    assert auth.startswith("HMAC-SHA256 ")