1
diff --git a/src/provisioningserver/drivers/power/proxmox.py b/src/provisioningserver/drivers/power/proxmox.py
2
index c0e6ca1..7e141e9 100644
3
--- a/src/provisioningserver/drivers/power/proxmox.py
4
+++ b/src/provisioningserver/drivers/power/proxmox.py
5
@@ -5,11 +5,13 @@
6
5
5
7
6
from io import BytesIO
6
from io import BytesIO
8
7
import json
7
import json
9
8
from urllib.parse import urlencode, urlparse
10
8
9
11
9
from twisted.internet.defer import inlineCallbacks, succeed
10
from twisted.internet.defer import inlineCallbacks, succeed
12
10
from twisted.web.client import FileBodyProducer
11
from twisted.web.client import FileBodyProducer
13
11
12
14
12
from provisioningserver.drivers import (
13
from provisioningserver.drivers import (
15
14
    IP_EXTRACTOR_PATTERNS,
16
13
    make_ip_extractor,
15
    make_ip_extractor,
17
14
    make_setting_field,
16
    make_setting_field,
18
15
    SETTING_SCOPE,
17
    SETTING_SCOPE,
19
@@ -62,16 +64,30 @@ class ProxmoxPowerDriver(WebhookPowerDriver):
20
62
        ),
64
        ),
21
63
    ]
65
    ]
22
64
66
24
65
    ip_extractor = make_ip_extractor("power_address")
67
    ip_extractor = make_ip_extractor(
25
68
        "power_address", IP_EXTRACTOR_PATTERNS.URL
26
69
    )
27
66
70
28
67
    def _get_url(self, context, endpoint, params=None):
71
    def _get_url(self, context, endpoint, params=None):
30
68
        uri = f"https://{context['power_address']}:8006/api2/json/{endpoint}"
72
        url = urlparse(context["power_address"])
31
73
        if not url.scheme:
32
74
            # When the scheme is not included in the power address
33
75
            # urlparse puts the url into path.
34
76
            url = url._replace(scheme="https", netloc=url.path, path="")
35
77
        if not url.port:
36
78
            if url.netloc:
37
79
                url = url._replace(netloc="%s:8006" % url.netloc)
38
80
            else:
39
81
                # Similar to above, we need to swap netloc and path.
40
82
                url = url._replace(netloc="%s:8006" % url.path, path="")
41
69
        if params:
83
        if params:
47
70
            uri = "%s?%s" % (
84
            query = urlencode(params)
48
71
                uri,
85
        else:
49
72
                "&".join([f"{key}={value}" for key, value in params.items()]),
86
            query = ""
50
73
            )
87
        url = url._replace(
51
74
        return uri.encode()
88
            path="/api2/json/%s" % endpoint, query=query, fragment=""
52
89
        )
53
90
        return url.geturl().encode()
54
75
91
55
76
    def _login(self, system_id, context):
92
    def _login(self, system_id, context):
56
77
        power_token_name = context.get("power_token_name")
93
        power_token_name = context.get("power_token_name")
57
diff --git a/src/provisioningserver/drivers/power/tests/test_proxmox.py b/src/provisioningserver/drivers/power/tests/test_proxmox.py
58
index 7c32b78..881ac5f 100644
59
--- a/src/provisioningserver/drivers/power/tests/test_proxmox.py
60
+++ b/src/provisioningserver/drivers/power/tests/test_proxmox.py
61
@@ -52,6 +52,26 @@ class TestProxmoxPowerDriver(MAASTestCase):
62
52
            ),
52
            ),
63
53
        )
53
        )
64
54
54
65
55
    def test_get_url_funky_params(self):
66
56
        power_address = factory.make_name("power_address")
67
57
        endpoint = factory.make_name("endpoint")
68
58
        params = {"test": "? /"}
69
59
        params_str = "test=%3F+%2F"
70
60
        self.assertEqual(
71
61
            f"https://{power_address}:8006/api2/json/{endpoint}?{params_str}".encode(),
72
62
            self.proxmox._get_url(
73
63
                {"power_address": power_address}, endpoint, params
74
64
            ),
75
65
        )
76
66
77
67
    def test_get_url_allows_custom_port(self):
78
68
        power_address = "%s:443" % factory.make_name("power_address")
79
69
        endpoint = factory.make_name("endpoint")
80
70
        self.assertEqual(
81
71
            f"https://{power_address}/api2/json/{endpoint}".encode(),
82
72
            self.proxmox._get_url({"power_address": power_address}, endpoint),
83
73
        )
84
74
85
55
    @inlineCallbacks
75
    @inlineCallbacks
86
56
    def test_login(self):
76
    def test_login(self):
87
57
        system_id = factory.make_name("system_id")
77
        system_id = factory.make_name("system_id")
Reviewer	Date Requested	Status
Adam Collard (community)	2021-02-02	Approve on 2021-02-03
MAAS Lander		Approve on 2021-02-03
Review via email: mp+397333@code.launchpad.net