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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
import functools
import os
import tempfile
from devtools_testutils import (
AzureRecordedTestCase,
EnvironmentVariableLoader,
)
from azure.confidentialledger.certificate import (
ConfidentialLedgerCertificateClient,
)
from azure.confidentialledger.certificate.aio import (
ConfidentialLedgerCertificateClient as ConfidentialLedgerCertificateClientAsync
)
from .constants import USER_CERTIFICATE
ConfidentialLedgerPreparer = functools.partial(
EnvironmentVariableLoader,
"confidentialledger",
confidentialledger_id="fake",
confidentialledger_endpoint="https://fake.confidential-ledger.azure.com",
confidentialledger_resource_group="fakegroup",
)
class ConfidentialLedgerTestCase(AzureRecordedTestCase):
@classmethod
def setup_class(cls):
"""setup any state specific to the execution of the given class (which
usually contains tests).
"""
with tempfile.NamedTemporaryFile(
"w", suffix=".pem", delete=False
) as tls_cert_file:
cls.network_certificate_path = tls_cert_file.name
with tempfile.NamedTemporaryFile(
"w", suffix=".pem", delete=False
) as user_cert_file:
user_cert_file.write(USER_CERTIFICATE)
cls.user_certificate_path = user_cert_file.name
@classmethod
def teardown_class(cls):
"""teardown any state that was previously setup with a call to
setup_class.
"""
try:
os.remove(cls.user_certificate_path)
except FileNotFoundError:
pass
if cls.network_certificate_path:
try:
os.remove(cls.network_certificate_path)
except FileNotFoundError:
pass
def set_ledger_identity(self, confidentialledger_id: str) -> str:
"""Retrieves the Confidential Ledger's TLS certificate, saving it to the object's network
certificate path as well as returning it directly.
:param confidentialledger_id: Id of the Confidential Ledger.
:type confidentialledger_id: str
:return: The Confidential Ledger's TLS certificate.
:rtype: str
"""
client = self.create_client_from_credential(
ConfidentialLedgerCertificateClient,
credential=None,
)
network_identity = (
client.get_ledger_identity(
ledger_id=confidentialledger_id
)
)
with open(self.network_certificate_path, "w", encoding="utf-8") as outfile:
outfile.write(network_identity["ledgerTlsCertificate"])
return network_identity["ledgerTlsCertificate"]
async def set_ledger_identity_async(self, confidentialledger_id: str) -> str:
"""Retrieves the Confidential Ledger's TLS certificate, saving it to the object's network
certificate path as well as returning it directly.
An async version of this method is needed so that this request is recorded by async tests.
:param confidentialledger_id: Id of the Confidential Ledger.
:type confidentialledger_id: str
:return: The Confidential Ledger's TLS certificate.
:rtype: str
"""
client = self.create_client_from_credential(
ConfidentialLedgerCertificateClientAsync,
credential=None,
)
try:
network_identity = (
await client.get_ledger_identity(
ledger_id=confidentialledger_id
)
)
with open(self.network_certificate_path, "w", encoding="utf-8") as outfile:
outfile.write(network_identity["ledgerTlsCertificate"])
return network_identity["ledgerTlsCertificate"]
finally:
await client.close()
|
