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
|
import datetime
import ipaddress
import json
import unittest
from dataclasses import dataclass
from typing import Any, Optional
from collections.abc import Iterable
from unittest import mock
from requests import Response
from podman import api
class ParseUtilsTestCase(unittest.TestCase):
def test_parse_repository(self):
@dataclass
class TestCase:
name: str
input: Any
expected: tuple[str, Optional[str]]
cases = [
TestCase(name="empty str", input="", expected=("", None)),
TestCase(
name="name",
input="quay.io/libpod/testimage",
expected=("quay.io/libpod/testimage", None),
),
TestCase(
name="@digest",
input="quay.io/libpod/testimage@71f1b47263fc",
expected=("quay.io/libpod/testimage", "71f1b47263fc"),
),
TestCase(
name=":tag",
input="quay.io/libpod/testimage:latest",
expected=("quay.io/libpod/testimage", "latest"),
),
TestCase(
name=":port",
input="quay.io:5000/libpod/testimage",
expected=("quay.io:5000/libpod/testimage", None),
),
TestCase(
name=":port@digest",
input="quay.io:5000/libpod/testimage@71f1b47263fc",
expected=("quay.io:5000/libpod/testimage", "71f1b47263fc"),
),
TestCase(
name=":port:tag",
input="quay.io:5000/libpod/testimage:latest",
expected=("quay.io:5000/libpod/testimage", "latest"),
),
]
for case in cases:
actual = api.parse_repository(case.input)
self.assertEqual(
case.expected,
actual,
f"failed test {case.name} expected {case.expected}, actual {actual}",
)
def test_decode_header(self):
actual = api.decode_header("eyJIZWFkZXIiOiJ1bml0dGVzdCJ9")
self.assertDictEqual(actual, {"Header": "unittest"})
self.assertDictEqual(api.decode_header(None), {})
def test_prepare_timestamp(self):
time = datetime.datetime(2022, 1, 24, 12, 0, 0)
self.assertEqual(api.prepare_timestamp(time), 1643025600)
self.assertEqual(api.prepare_timestamp(2), 2)
self.assertEqual(api.prepare_timestamp(None), None)
with self.assertRaises(ValueError):
api.prepare_timestamp("bad input") # type: ignore
def test_prepare_cidr(self):
net = ipaddress.IPv4Network("127.0.0.0/24")
self.assertEqual(api.prepare_cidr(net), ("127.0.0.0", "////AA=="))
def test_stream_helper(self):
streamed_results = [b'{"test":"val1"}', b'{"test":"val2"}']
mock_response = mock.Mock(spec=Response)
mock_response.iter_lines.return_value = iter(streamed_results)
streamable = api.stream_helper(mock_response)
self.assertIsInstance(streamable, Iterable)
for expected, actual in zip(streamed_results, streamable):
self.assertIsInstance(actual, bytes)
self.assertEqual(expected, actual)
def test_stream_helper_with_decode(self):
streamed_results = [b'{"test":"val1"}', b'{"test":"val2"}']
mock_response = mock.Mock(spec=Response)
mock_response.iter_lines.return_value = iter(streamed_results)
streamable = api.stream_helper(mock_response, decode_to_json=True)
self.assertIsInstance(streamable, Iterable)
for expected, actual in zip(streamed_results, streamable):
self.assertIsInstance(actual, dict)
self.assertDictEqual(json.loads(expected), actual)
if __name__ == '__main__':
unittest.main()
|
