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
|
import re
from azurelinuxagent.common.utils import restutil
class HttpRequestPredicates(object):
"""
Utility functions to check the urls used by tests
"""
@staticmethod
def is_goal_state_request(url):
return url.lower() == 'http://{0}/machine/?comp=goalstate'.format(restutil.KNOWN_WIRESERVER_IP)
@staticmethod
def is_certificates_request(url):
return re.match(r'http://{0}(:80)?/machine/.*?comp=certificates'.format(restutil.KNOWN_WIRESERVER_IP), url, re.IGNORECASE)
@staticmethod
def is_extensions_config_request(url):
return re.match(r'http://{0}(:80)?/machine/.*?comp=config&type=extensionsConfig'.format(restutil.KNOWN_WIRESERVER_IP), url, re.IGNORECASE)
@staticmethod
def is_hosting_environment_config_request(url):
return re.match(r'http://{0}(:80)?/machine/.*?comp=config&type=hostingEnvironmentConfig'.format(restutil.KNOWN_WIRESERVER_IP), url, re.IGNORECASE)
@staticmethod
def is_shared_config_request(url):
return re.match(r'http://{0}(:80)?/machine/.*?comp=config&type=sharedConfig'.format(restutil.KNOWN_WIRESERVER_IP), url, re.IGNORECASE)
@staticmethod
def is_telemetry_request(url):
return url.lower() == 'http://{0}/machine?comp=telemetrydata'.format(restutil.KNOWN_WIRESERVER_IP)
@staticmethod
def is_health_service_request(url):
return url.lower() == 'http://{0}:80/healthservice'.format(restutil.KNOWN_WIRESERVER_IP)
@staticmethod
def is_in_vm_artifacts_profile_request(url):
return re.match(r'https://.+\.blob\.core\.windows\.net/\$system/.+\.(vmSettings|settings)\?.+', url) is not None
@staticmethod
def _get_host_plugin_request_artifact_location(url, request_kwargs):
if 'headers' not in request_kwargs:
raise ValueError('Host plugin request is missing HTTP headers ({0})'.format(url))
headers = request_kwargs['headers']
if 'x-ms-artifact-location' not in headers:
raise ValueError('Host plugin request is missing the x-ms-artifact-location header ({0})'.format(url))
return headers['x-ms-artifact-location']
@staticmethod
def is_host_plugin_vm_settings_request(url):
return url.lower() == 'http://{0}:{1}/vmsettings'.format(restutil.KNOWN_WIRESERVER_IP, restutil.HOST_PLUGIN_PORT)
@staticmethod
def is_host_plugin_health_request(url):
return url.lower() == 'http://{0}:{1}/health'.format(restutil.KNOWN_WIRESERVER_IP, restutil.HOST_PLUGIN_PORT)
@staticmethod
def is_host_plugin_extension_artifact_request(url):
return url.lower() == 'http://{0}:{1}/extensionartifact'.format(restutil.KNOWN_WIRESERVER_IP, restutil.HOST_PLUGIN_PORT)
@staticmethod
def is_status_request(url):
return HttpRequestPredicates.is_storage_status_request(url) or HttpRequestPredicates.is_host_plugin_status_request(url)
@staticmethod
def is_storage_status_request(url):
# e.g. 'https://test.blob.core.windows.net/vhds/test-cs12.test-cs12.test-cs12.status?sr=b&sp=rw&se=9999-01-01&sk=key1&sv=2014-02-14&sig=hfRh7gzUE7sUtYwke78IOlZOrTRCYvkec4hGZ9zZzXo'
return re.match(r'^https://.+/.*\.status\?[^/]+$', url, re.IGNORECASE)
@staticmethod
def is_host_plugin_status_request(url):
return url.lower() == 'http://{0}:{1}/status'.format(restutil.KNOWN_WIRESERVER_IP, restutil.HOST_PLUGIN_PORT)
@staticmethod
def is_host_plugin_extension_request(request_url, request_kwargs, extension_url):
if not HttpRequestPredicates.is_host_plugin_extension_artifact_request(request_url):
return False
artifact_location = HttpRequestPredicates._get_host_plugin_request_artifact_location(request_url, request_kwargs)
return artifact_location == extension_url
@staticmethod
def is_host_plugin_in_vm_artifacts_profile_request(url, request_kwargs):
if not HttpRequestPredicates.is_host_plugin_extension_artifact_request(url):
return False
artifact_location = HttpRequestPredicates._get_host_plugin_request_artifact_location(url, request_kwargs)
return HttpRequestPredicates.is_in_vm_artifacts_profile_request(artifact_location)
@staticmethod
def is_host_plugin_put_logs_request(url):
return url.lower() == 'http://{0}:{1}/vmagentlog'.format(restutil.KNOWN_WIRESERVER_IP,
restutil.HOST_PLUGIN_PORT)
@staticmethod
def is_agent_package_request(url):
return re.match(r"^http://mock-goal-state/ga-manifests/OSTCExtensions.WALinuxAgent__([\d.]+)$", url) is not None
@staticmethod
def is_ga_manifest_request(url):
return "manifest_of_ga.xml" in url
|
