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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
|
import json
import os
import random
import subprocess
import sys
import time
import unittest
import requests
from .podman import Podman
class ArtifactFile:
__test__: bool = False
name: str | None
size: int | None
sig: bytes | None
def __init__(
self, name: str | None = None, size: int | None = None, sig: bytes | None = None
) -> None:
self.name = name
self.size = size
self.sig = sig
self.render_test_file()
def render_test_file(self) -> None:
if self.name is None:
self.name = "test_file_1"
if self.size is None:
self.size = 1048576
file_data = None
if self.sig is not None:
random_bytes = random.randbytes(self.size - len(self.sig))
file_data = bytearray(self.sig)
file_data.extend(random_bytes)
else:
file_data = os.urandom(self.size)
try:
with open(self.name, "wb") as f:
_ = f.write(file_data)
except Exception as e:
print(f"File write error for {self.name}: {e}")
raise
class Artifact:
__test__: bool = False
uri: str
name: str
parameters: dict[str, str | list[str]]
file: ArtifactFile
def __init__(
self,
uri: str,
name: str,
parameters: dict[str, str | list[str]],
file: ArtifactFile,
) -> None:
self.uri = uri
self.name = name
self.parameters = parameters
self.file = file
def add(self) -> requests.Response:
try:
with open(self.file.name, "rb") as file_to_upload:
file_content = file_to_upload.read()
r = requests.post(
self.uri + "/artifacts/add",
data=file_content,
params=self.parameters,
)
except Exception:
pass
os.remove(self.file.name)
return r
def do_artifact_inspect_request(self) -> requests.Response:
r = requests.get(
self.uri + "/artifacts/" + self.name + "/json",
)
return r
class APITestCase(unittest.TestCase):
PODMAN_URL = "http://localhost:8080"
podman = None # initialized podman configuration for tests
service = None # podman service instance
@classmethod
def setUpClass(cls):
super().setUpClass()
APITestCase.podman = Podman()
APITestCase.service = APITestCase.podman.open(
"system", "service", "tcp://localhost:8080", "--time=0"
)
# give the service some time to be ready...
time.sleep(2)
returncode = APITestCase.service.poll()
if returncode is not None:
raise subprocess.CalledProcessError(returncode, "podman system service")
r = requests.post(
APITestCase.uri("/images/pull?reference=quay.io%2Flibpod%2Falpine%3Alatest")
)
if r.status_code != 200:
raise subprocess.CalledProcessError(
r.status_code, f"podman images pull quay.io/libpod/alpine:latest {r.text}"
)
@classmethod
def tearDownClass(cls):
APITestCase.service.terminate()
stdout, stderr = APITestCase.service.communicate(timeout=1)
if stdout:
sys.stdout.write("\nService Stdout:\n" + stdout.decode("utf-8"))
if stderr:
sys.stderr.write("\nService Stderr:\n" + stderr.decode("utf-8"))
return super().tearDownClass()
def setUp(self):
super().setUp()
APITestCase.podman.run("run", "-d", "alpine", "top", check=True)
def tearDown(self) -> None:
APITestCase.podman.run("pod", "rm", "--all", "--force", check=True)
APITestCase.podman.run("rm", "--all", "--force", check=True)
super().tearDown()
@property
def podman_url(self):
return "http://localhost:8080"
@staticmethod
def uri(path: str) -> str:
return APITestCase.PODMAN_URL + "/v2.0.0/libpod" + path
@staticmethod
def compat_uri(path):
return APITestCase.PODMAN_URL + "/v3.0.0/" + path
def resolve_container(self, path):
"""Find 'first' container and return 'Id' formatted into given URI path."""
try:
r = requests.get(self.uri("/containers/json?all=true"))
containers = r.json()
except Exception as e:
msg = f"Bad container response: {e}"
if r is not None:
msg += ": " + r.text
raise self.failureException(msg)
return path.format(containers[0]["Id"])
def assertContainerExists(self, member, msg=None): # pylint: disable=invalid-name
r = requests.get(self.uri(f"/containers/{member}/exists"))
if r.status_code == 404:
if msg is None:
msg = f"Container '{member}' does not exist."
self.failureException(msg)
def assertContainerNotExists(self, member, msg=None): # pylint: disable=invalid-name
r = requests.get(self.uri(f"/containers/{member}/exists"))
if r.status_code == 204:
if msg is None:
msg = f"Container '{member}' exists."
self.failureException(msg)
def assertId(self, content): # pylint: disable=invalid-name
objects = json.loads(content)
try:
if isinstance(objects, dict):
_ = objects["Id"]
else:
for item in objects:
_ = item["Id"]
except KeyError:
self.failureException("Failed in find 'Id' in return value.")
|
