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
|
import random
import unittest
from podman import PodmanClient
from podman.errors import NotFound
from podman.tests.integration import base
class PodsIntegrationTest(base.IntegrationTest):
"""Pods Integration tests."""
def setUp(self):
super().setUp()
self.client = PodmanClient(base_url=self.socket_uri)
self.addCleanup(self.client.close)
self.alpine_image = self.client.images.pull("quay.io/libpod/alpine", tag="latest")
self.pod_name = f"pod_{random.getrandbits(160):x}"
for container in self.client.containers.list():
container.remove(force=True)
def tearDown(self):
if self.client.pods.exists(self.pod_name):
self.client.pods.remove(self.pod_name)
super().tearDown()
def test_pod_crud(self):
"""Test Pod CRUD."""
with self.subTest("Create no_infra"):
pod = self.client.pods.create(
self.pod_name,
labels={
"unittest": "true",
},
no_infra=True,
)
self.assertEqual(self.pod_name, pod.name)
self.assertTrue(self.client.pods.exists(pod.id))
with self.subTest("Inspect"):
actual = self.client.pods.get(pod.id)
self.assertEqual(actual.name, pod.name)
self.assertNotIn("Containers", actual.attrs)
with self.subTest("Exists"):
self.assertTrue(self.client.pods.exists(self.pod_name))
# TODO Need method for deterministic prune...
# with self.subTest("Prune"):
# report = self.client.pods.prune()
# self.assertIn("PodsDeleted", report)
# self.assertIn(actual.id, report["PodsDeleted"])
#
# with self.assertRaises(NotFound):
# pod.reload()
#
# For now, delete pod explicitly
with self.subTest("Delete"):
pod.remove(force=True)
with self.assertRaises(NotFound):
pod.reload()
@unittest.skipUnless(base.HAVE_FUSE, 'This test requires a working fuse setup.')
def test_pod_crud_infra(self):
"""Test Pod CRUD with infra container."""
with self.subTest("Create with infra"):
pod = self.client.pods.create(
self.pod_name,
labels={
"unittest": "true",
},
)
self.assertEqual(self.pod_name, pod.name)
with self.subTest("Inspect"):
actual = self.client.pods.get(pod.id)
self.assertEqual(actual.name, pod.name)
self.assertIn("Containers", actual.attrs)
self.assertEqual(actual.attrs["State"], "Created")
with self.subTest("Add container"):
container = self.client.containers.create(self.alpine_image, command=["ls"], pod=actual)
actual = self.client.pods.get(pod.id)
ids = {c["Id"] for c in actual.attrs["Containers"]}
self.assertIn(container.id, ids)
with self.subTest("List"):
pods = self.client.pods.list()
self.assertGreaterEqual(len(pods), 1)
ids = {p.id for p in pods}
self.assertIn(actual.id, ids)
with self.subTest("Delete"):
pod.remove(force=True)
with self.assertRaises(NotFound):
pod.reload()
@unittest.skipUnless(base.HAVE_FUSE, 'This test requires a working fuse setup.')
def test_ps(self):
pod = self.client.pods.create(
self.pod_name,
labels={
"unittest": "true",
},
no_infra=True,
)
self.assertTrue(self.client.pods.exists(pod.id))
self.client.containers.create(
self.alpine_image, command=["top"], detach=True, tty=True, pod=pod
)
pod.start()
pod.reload()
with self.subTest("top"):
# this is the API top call not the
# top command running in the container
procs = pod.top()
self.assertGreater(len(procs["Processes"]), 0)
self.assertGreater(len(procs["Titles"]), 0)
with self.subTest("stats"):
report = self.client.pods.stats(all=True)
self.assertGreaterEqual(len(report), 1)
with self.subTest("Stop/Start"):
pod.stop()
pod.reload()
self.assertIn(pod.attrs["State"], ("Stopped", "Exited"))
pod.start()
pod.reload()
self.assertEqual(pod.attrs["State"], "Running")
with self.subTest("Restart"):
pod.stop()
pod.restart()
pod.reload()
self.assertEqual(pod.attrs["State"], "Running")
with self.subTest("Pause/Unpause"):
pod.pause()
pod.reload()
self.assertEqual(pod.attrs["State"], "Paused")
pod.unpause()
pod.reload()
self.assertEqual(pod.attrs["State"], "Running")
pod.stop()
if __name__ == '__main__':
unittest.main()
|