linode_api4-python/test/integration/models/image/test_image.py at 2d3027728e6e637e140db40caba545b7d9b1ca2b · linode/linode_api4-python · GitHub

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
117
118
119
120
121
122
123
124
125
from io import BytesIO
from test.integration.conftest import get_regions
from test.integration.helpers import get_test_label

import polling
import pytest

from linode_api4 import LinodeClient
from linode_api4.objects import Image

DISALLOWED_IMAGE_REGIONS = {
    "gb-lon",
    "au-mel",
    "sg-sin-2",
    "jp-tyo-3",
    "no-osl-1",
}


def get_image_upload_regions(client: LinodeClient):
    """
    This is necessary because the API does not currently expose
    a capability for regions that allow custom image uploads.

    In the future, we should remove this if the API exposes a custom images capability or
    if all Object Storage regions support custom images.
    """

    return [
        region
        for region in get_regions(
            client,
            capabilities={"Linodes", "Object Storage"},
            site_type="core",
        )
        if region.id not in DISALLOWED_IMAGE_REGIONS
    ]


@pytest.fixture(scope="session")
def image_upload_url(test_linode_client):
    label = get_test_label() + "_image"

    region = get_image_upload_regions(test_linode_client)[0]

    test_linode_client.image_create_upload(
        label, region.id, "integration test image upload"
    )

    image = test_linode_client.images()[0]

    yield image

    image.delete()


@pytest.fixture(scope="session")
def test_uploaded_image(test_linode_client):
    test_image_content = (
        b"\x1f\x8b\x08\x08\xbd\x5c\x91\x60\x00\x03\x74\x65\x73\x74\x2e\x69"
        b"\x6d\x67\x00\x03\x00\x00\x00\x00\x00\x00\x00\x00\x00"
    )

    label = get_test_label() + "_image"

    regions = get_image_upload_regions(test_linode_client)

    image = test_linode_client.image_upload(
        label,
        regions[1].id,
        BytesIO(test_image_content),
        description="integration test image upload",
        tags=["tests"],
    )

    yield image, regions

    image.delete()


@pytest.mark.smoke
def test_get_image(test_linode_client, image_upload_url):
    image = test_linode_client.load(Image, image_upload_url.id)

    assert image.label == image_upload_url.label


def test_image_create_upload(test_linode_client, test_uploaded_image):
    uploaded_image, _ = test_uploaded_image

    image = test_linode_client.load(Image, uploaded_image.id)

    assert image.label == uploaded_image.label
    assert image.description == "integration test image upload"
    assert image.tags[0] == "tests"


@pytest.mark.smoke
@pytest.mark.flaky(reruns=3, reruns_delay=2)
def test_image_replication(test_linode_client, test_uploaded_image):
    uploaded_image, regions = test_uploaded_image

    image = test_linode_client.load(Image, uploaded_image.id)

    # wait for image to be available for replication
    def poll_func() -> bool:
        image._api_get()
        return image.status in {"available"}

    try:
        polling.poll(
            poll_func,
            step=10,
            timeout=250,
        )
    except polling.TimeoutException:
        print("failed to wait for image status: timeout period expired.")

    replicate_regions = [r.id for r in regions[:2]]
    image.replicate(replicate_regions)

    assert image.label == uploaded_image.label
    assert len(image.regions) == 2
    assert image.regions[0].region in replicate_regions
    assert image.regions[1].region in replicate_regions