admin/registry-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

core-329 add images delete by age in hours

Browse Source
This commit is contained in:
Ivan Pavlushin
2018-01-09 13:04:47 +03:00
parent 9f251722f7
commit e307c2fd2e
2 changed files with 411 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

326
test.py
View File

@@ -1,27 +1,29 @@
import unittest
from registry import Registry, Requests, get_tags, parse_args, delete_tags
from mock import MagicMock
from registry import Registry, Requests, get_tags, parse_args, delete_tags, delete_tags_by_age
from mock import MagicMock, patch
import requests
class ReturnValue:
def __init__(self, status_code = 200, text = ""):
def __init__(self, status_code=200, text=""):
self.status_code = status_code
self.text = text
class MockRequests:
def __init__(self, return_value = ReturnValue()):
def __init__(self, return_value=ReturnValue()):
self.return_value = return_value
self.request = MagicMock(return_value = self.return_value)
self.request = MagicMock(return_value=self.return_value)
def reset_return_value(self, status_code = 200, text = ""):
def reset_return_value(self, status_code=200, text=""):
self.return_value.status_code = status_code
self.return_value.text = text
class TestRequestsClass(unittest.TestCase):
def test_requests_created(self):
# simply create requests class and make sure it raises an exception
# from requests module
@@ -33,6 +35,7 @@ class TestRequestsClass(unittest.TestCase):
class TestCreateMethod(unittest.TestCase):
def test_create_nologin(self):
r = Registry.create("testhost", None, False)
self.assertTrue(isinstance(r.http, Requests))
@@ -57,6 +60,7 @@ class TestCreateMethod(unittest.TestCase):
with self.assertRaises(SystemExit):
Registry.create("testhost4", "invalid_login", False)
class TestParseLogin(unittest.TestCase):
def setUp(self):
@@ -79,7 +83,8 @@ class TestParseLogin(unittest.TestCase):
(username, password) = self.registry.parse_login("username/password")
self.assertEqual(username, None)
self.assertEqual(password, None)
self.assertEqual(self.registry.last_error, "Please provide -l in the form USER:PASSWORD")
self.assertEqual(self.registry.last_error,
"Please provide -l in the form USER:PASSWORD")
def test_login_args_singlequoted(self):
(username, password) = self.registry.parse_login("'username':password")
@@ -99,10 +104,12 @@ class TestParseLogin(unittest.TestCase):
then the result will be invalid in this case
and no error will be printed
"""
(username, password) = self.registry.parse_login("'user:name':'pass:word'")
(username, password) = self.registry.parse_login(
"'user:name':'pass:word'")
self.assertEqual(username, 'user')
self.assertEqual(password, "name':'pass:word")
class TestRegistrySend(unittest.TestCase):
def setUp(self):
@@ -117,10 +124,10 @@ class TestRegistrySend(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertEqual(self.registry.last_error, None)
self.registry.http.request.assert_called_with('GET',
'http://testdomain.com/test_string',
auth = (None, None),
headers = self.registry.HEADERS,
verify = True)
'http://testdomain.com/test_string',
auth=(None, None),
headers=self.registry.HEADERS,
verify=True)
def test_invalid_status_code(self):
self.registry.http.reset_return_value(400)
@@ -128,7 +135,6 @@ class TestRegistrySend(unittest.TestCase):
self.assertEqual(response, None)
self.assertEqual(self.registry.last_error, 400)
def test_login_pass(self):
self.registry.username = "test_login"
self.registry.password = "test_password"
@@ -137,10 +143,11 @@ class TestRegistrySend(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertEqual(self.registry.last_error, None)
self.registry.http.request.assert_called_with('GET',
'http://testdomain.com/v2/catalog',
auth = ("test_login", "test_password"),
headers = self.registry.HEADERS,
verify = True)
'http://testdomain.com/v2/catalog',
auth=("test_login",
"test_password"),
headers=self.registry.HEADERS,
verify=True)
class TestListImages(unittest.TestCase):
@@ -151,8 +158,8 @@ class TestListImages(unittest.TestCase):
self.registry.hostname = "http://testdomain.com"
def test_list_images_ok(self):
self.registry.http.reset_return_value(status_code = 200,
text = '{"repositories":["image1","image2"]}')
self.registry.http.reset_return_value(status_code=200,
text='{"repositories":["image1","image2"]}')
response = self.registry.list_images()
self.assertEqual(response, ["image1", "image2"])
self.assertEqual(self.registry.last_error, None)
@@ -165,6 +172,7 @@ class TestListImages(unittest.TestCase):
class TestListTags(unittest.TestCase):
def setUp(self):
self.registry = Registry()
self.registry.http = MockRequests()
@@ -172,16 +180,16 @@ class TestListTags(unittest.TestCase):
self.registry.http.reset_return_value(200)
def test_list_one_tag_ok(self):
self.registry.http.reset_return_value(status_code = 200,
text = u'{"name":"image1","tags":["0.1.306"]}')
self.registry.http.reset_return_value(status_code=200,
text=u'{"name":"image1","tags":["0.1.306"]}')
response = self.registry.list_tags('image1')
self.assertEqual(response, ["0.1.306"])
self.assertEqual(self.registry.last_error, None)
def test_list_tags_invalid_http_response(self):
self.registry.http.reset_return_value(status_code = 400,
text = "")
self.registry.http.reset_return_value(status_code=400,
text="")
response = self.registry.list_tags('image1')
self.assertEqual(response, [])
@@ -193,32 +201,36 @@ class TestListTags(unittest.TestCase):
response = self.registry.list_tags('image1')
self.assertEqual(response, [])
self.assertEqual(self.registry.last_error, "list_tags: invalid json response")
self.assertEqual(self.registry.last_error,
"list_tags: invalid json response")
def test_list_one_tag_sorted(self):
self.registry.http.reset_return_value(status_code=200,
text=u'{"name":"image1","tags":["0.1.306", "0.1.300", "0.1.290"]}')
text=u'{"name":"image1","tags":["0.1.306", "0.1.300", "0.1.290"]}')
response = self.registry.list_tags('image1')
self.assertEqual(response, ["0.1.290", "0.1.300", "0.1.306"])
self.assertEqual(self.registry.last_error, None)
def test_list_tags_like_various(self):
tags_list = set(['FINAL_0.1', 'SNAPSHOT_0.1', "0.1.SNAP", "1.0.0_FINAL"])
self.assertEqual(get_tags(tags_list, "", set(["FINAL"])), set(["FINAL_0.1", "1.0.0_FINAL"]))
self.assertEqual(get_tags(tags_list, "", set(["SNAPSHOT"])), set(['SNAPSHOT_0.1']))
tags_list = set(['FINAL_0.1', 'SNAPSHOT_0.1',
"0.1.SNAP", "1.0.0_FINAL"])
self.assertEqual(get_tags(tags_list, "", set(
["FINAL"])), set(["FINAL_0.1", "1.0.0_FINAL"]))
self.assertEqual(get_tags(tags_list, "", set(
["SNAPSHOT"])), set(['SNAPSHOT_0.1']))
self.assertEqual(get_tags(tags_list, "", set()),
set(['FINAL_0.1', 'SNAPSHOT_0.1', "0.1.SNAP", "1.0.0_FINAL"]))
self.assertEqual(get_tags(tags_list, "", set(["ABSENT"])), set())
self.assertEqual(get_tags(tags_list, "IMAGE:TAG00", ""), set(["TAG00"]))
self.assertEqual(get_tags(tags_list, "IMAGE:TAG00", set(["WILL_NOT_BE_CONSIDERED"])), set(["TAG00"]))
self.assertEqual(
get_tags(tags_list, "IMAGE:TAG00", ""), set(["TAG00"]))
self.assertEqual(get_tags(tags_list, "IMAGE:TAG00", set(
["WILL_NOT_BE_CONSIDERED"])), set(["TAG00"]))
class TestListDigest(unittest.TestCase):
def setUp(self):
self.registry = Registry()
self.registry.http = MockRequests()
@@ -226,12 +238,12 @@ class TestListDigest(unittest.TestCase):
self.registry.http.reset_return_value(200)
def test_get_digest_ok(self):
self.registry.http.reset_return_value(status_code = 200,
text = ('{'
'"schemaVersion": 2,\n '
'"mediaType": "application/vnd.docker.distribution.manifest.v2+json"'
'"digest": "sha256:357ea8c3d80bc25792e010facfc98aee5972ebc47e290eb0d5aea3671a901cab"'
))
self.registry.http.reset_return_value(status_code=200,
text=('{'
'"schemaVersion": 2,\n '
'"mediaType": "application/vnd.docker.distribution.manifest.v2+json"'
'"digest": "sha256:357ea8c3d80bc25792e010facfc98aee5972ebc47e290eb0d5aea3671a901cab"'
))
self.registry.http.return_value.headers = {
'Content-Length': '4935',
@@ -243,12 +255,13 @@ class TestListDigest(unittest.TestCase):
self.registry.http.request.assert_called_with(
"HEAD",
"http://testdomain.com/v2/image1/manifests/0.1.300",
auth = (None, None),
headers = self.registry.HEADERS,
verify = True
auth=(None, None),
headers=self.registry.HEADERS,
verify=True
)
self.assertEqual(response, 'sha256:85295b0e7456a8fbbc886722b483f87f2bff553fa0beeaf37f5d807aff7c1e52')
self.assertEqual(
response, 'sha256:85295b0e7456a8fbbc886722b483f87f2bff553fa0beeaf37f5d807aff7c1e52')
self.assertEqual(self.registry.last_error, None)
def test_invalid_status_code(self):
@@ -263,7 +276,139 @@ class TestListDigest(unittest.TestCase):
self.registry.get_tag_digest('image1', '0.1.300')
class TestTagConfig(unittest.TestCase):
def setUp(self):
self.registry = Registry()
self.registry.http = MockRequests()
self.registry.hostname = "http://testdomain.com"
self.registry.http.reset_return_value(200)
def test_get_tag_config_ok(self):
self.registry.http.reset_return_value(
200,
'''
{
"schemaVersion": 2,
"mediaType": "application/vnd.docker.distribution.manifest.v2+json",
"config": {
"mediaType": "application/vnd.docker.container.image.v1+json",
"size": 12953,
"digest": "sha256:8d71dfbf239c0015ad66993d55d3954cee2d52d86f829fdff9ccfb9f23b75aa8"
},
"layers": [
{
"mediaType": "application/vnd.docker.image.rootfs.diff.tar.gzip",
"size": 51480140,
"digest": "sha256:c6b13209f43b945816b7658a567720983ac5037e3805a779d5772c61599b4f73"
}
]
}
'''
)
response = self.registry.get_tag_config('image1', '0.1.300')
self.registry.http.request.assert_called_with(
"GET",
"http://testdomain.com/v2/image1/manifests/0.1.300",
auth=(None, None),
headers=self.registry.HEADERS,
verify=True)
self.assertEqual(
response, {'mediaType': 'application/vnd.docker.container.image.v1+json', 'size': 12953, 'digest': 'sha256:8d71dfbf239c0015ad66993d55d3954cee2d52d86f829fdff9ccfb9f23b75aa8'})
self.assertEqual(self.registry.last_error, None)
def test_tag_config_scheme_v1(self):
with self.assertRaises(SystemExit):
self.registry.http.reset_return_value(
200,
'''
{
"schemaVersion": 1
}
'''
)
response = self.registry.get_tag_config('image1', '0.1.300')
def test_invalid_status_code(self):
self.registry.http.reset_return_value(400, "whatever")
response = self.registry.get_tag_config('image1', '0.1.300')
self.registry.http.request.assert_called_with(
"GET",
"http://testdomain.com/v2/image1/manifests/0.1.300",
auth=(None, None),
headers=self.registry.HEADERS,
verify=True
)
self.assertEqual(response, [])
self.assertEqual(self.registry.last_error, 400)
class TestImageAge(unittest.TestCase):
def setUp(self):
self.registry = Registry()
self.registry.http = MockRequests()
self.registry.hostname = "http://testdomain.com"
self.registry.http.reset_return_value(200)
def test_image_age_ok(self):
self.registry.http.reset_return_value(
200,
'''
{
"architecture": "amd64",
"author": "Test",
"config": {},
"container": "c467822c5981cd446068eebafd81cb5cde60d4341a945f3fbf67e456dde5af51",
"container_config": {},
"created": "2017-12-27T12:47:33.511765448Z",
"docker_version": "1.11.2",
"history": []
}
'''
)
json_payload = {'mediaType': 'application/vnd.docker.container.image.v1+json', 'size': 12953,
'digest': 'sha256:8d71dfbf239c0015ad66993d55d3954cee2d52d86f829fdff9ccfb9f23b75aa8'}
header = {"Accept": "{0}".format(json_payload['mediaType'])}
response = self.registry.get_image_age('image1', json_payload)
self.registry.http.request.assert_called_with(
"GET",
"http://testdomain.com/v2/image1/blobs/sha256:8d71dfbf239c0015ad66993d55d3954cee2d52d86f829fdff9ccfb9f23b75aa8",
auth=(None, None),
headers=header,
verify=True)
self.assertEqual(
response, "2017-12-27T12:47:33.511765448Z")
self.assertEqual(self.registry.last_error, None)
def test_image_age_nok(self):
self.registry.http.reset_return_value(400, "err")
json_payload = {'mediaType': 'application/vnd.docker.container.image.v1+json', 'size': 12953,
'digest': 'sha256:8d71dfbf239c0015ad66993d55d3954cee2d52d86f829fdff9ccfb9f23b75aa8'}
header = {"Accept": "{0}".format(json_payload['mediaType'])}
response = self.registry.get_image_age('image1', json_payload)
self.registry.http.request.assert_called_with(
"GET",
"http://testdomain.com/v2/image1/blobs/sha256:8d71dfbf239c0015ad66993d55d3954cee2d52d86f829fdff9ccfb9f23b75aa8",
auth=(None, None),
headers=header,
verify=True)
self.assertEqual(response, [])
self.assertEqual(self.registry.last_error, 400)
class TestListLayers(unittest.TestCase):
def setUp(self):
self.registry = Registry()
self.registry.http = MockRequests()
@@ -286,15 +431,14 @@ class TestListLayers(unittest.TestCase):
self.registry.http.request.assert_called_with(
"GET",
"http://testdomain.com/v2/image1/manifests/0.1.300",
auth = (None, None),
headers = self.registry.HEADERS,
verify = True
auth=(None, None),
headers=self.registry.HEADERS,
verify=True
)
self.assertEqual(response, "layers_list")
self.assertEqual(self.registry.last_error, None)
def test_list_layers_schema_version_1_ok(self):
self.registry.http.reset_return_value(
200,
@@ -311,9 +455,9 @@ class TestListLayers(unittest.TestCase):
self.registry.http.request.assert_called_with(
"GET",
"http://testdomain.com/v2/image1/manifests/0.1.300",
auth = (None, None),
headers = self.registry.HEADERS,
verify = True
auth=(None, None),
headers=self.registry.HEADERS,
verify=True
)
self.assertEqual(response, "layers_list")
@@ -335,7 +479,9 @@ class TestListLayers(unittest.TestCase):
self.assertEqual(response, [])
self.assertEqual(self.registry.last_error, 400)
class TestDeletion(unittest.TestCase):
def setUp(self):
self.registry = Registry()
self.registry.http = MockRequests()
@@ -347,14 +493,14 @@ class TestDeletion(unittest.TestCase):
'X-Content-Type-Options': 'nosniff'
}
def test_delete_tag_dry_run(self):
response = self.registry.delete_tag("image1", 'test_tag', True, [])
self.assertFalse(response)
def test_delete_tag_ok(self):
keep_tag_digests = ['DIGEST1', 'DIGEST2']
response = self.registry.delete_tag('image1', 'test_tag', False, keep_tag_digests)
response = self.registry.delete_tag(
'image1', 'test_tag', False, keep_tag_digests)
self.assertEqual(response, True)
self.assertEqual(self.registry.http.request.call_count, 2)
self.registry.http.request.assert_called_with(
@@ -367,7 +513,8 @@ class TestDeletion(unittest.TestCase):
self.assertTrue("MOCK_DIGEST_HEADER" in keep_tag_digests)
def test_delete_tag_ignored(self):
response = self.registry.delete_tag('image1', 'test_tag', False, ['MOCK_DIGEST_HEADER'])
response = self.registry.delete_tag(
'image1', 'test_tag', False, ['MOCK_DIGEST_HEADER'])
self.assertEqual(response, True)
self.assertEqual(self.registry.http.request.call_count, 1)
self.registry.http.request.assert_called_with(
@@ -384,7 +531,9 @@ class TestDeletion(unittest.TestCase):
self.assertFalse(response)
self.assertEqual(self.registry.last_error, 400)
class TestDeleteTagsFunction(unittest.TestCase):
def setUp(self):
self.registry = Registry()
self.delete_mock = MagicMock()
@@ -408,10 +557,11 @@ class TestDeleteTagsFunction(unittest.TestCase):
)
def test_delete_tags_keep(self):
digest_mock = MagicMock(return_value = "DIGEST_MOCK")
digest_mock = MagicMock(return_value="DIGEST_MOCK")
self.registry.get_tag_digest = digest_mock
delete_tags(self.registry, "imagename", False, ["tag1", "tag2"], ["tag2"])
delete_tags(self.registry, "imagename",
False, ["tag1", "tag2"], ["tag2"])
digest_mock.assert_called_with("imagename", "tag2")
@@ -423,9 +573,10 @@ class TestDeleteTagsFunction(unittest.TestCase):
)
def test_delete_tags_digest_none(self):
digest_mock = MagicMock(return_value = None)
digest_mock = MagicMock(return_value=None)
self.registry.get_tag_digest = digest_mock
delete_tags(self.registry, "imagename", False, ["tag1", "tag2"], ["tag2"])
delete_tags(self.registry, "imagename",
False, ["tag1", "tag2"], ["tag2"])
digest_mock.assert_called_with("imagename", "tag2")
@@ -437,7 +588,62 @@ class TestDeleteTagsFunction(unittest.TestCase):
)
class TestDeleteTagsByAge(unittest.TestCase):
def setUp(self):
self.registry = Registry()
self.registry.http = MockRequests()
self.get_tag_config_mock = MagicMock(return_value={'mediaType': 'application/vnd.docker.container.image.v1+json', 'size': 12953,
'digest': 'sha256:8d71dfbf239c0015ad66993d55d3954cee2d52d86f829fdff9ccfb9f23b75aa8'})
self.registry.get_tag_config = self.get_tag_config_mock
self.get_image_age_mock = MagicMock(
return_value="2017-12-27T12:47:33.511765448Z")
self.registry.get_image_age = self.get_image_age_mock
self.list_tags_mock = MagicMock(return_value=["image"])
self.registry.list_tags = self.list_tags_mock
self.get_tag_digest_mock = MagicMock()
self.registry.get_tag_digest = self.get_tag_digest_mock
self.registry.http = MockRequests()
self.registry.hostname = "http://testdomain.com"
self.registry.http.reset_return_value(200, "MOCK_DIGEST")
@patch('registry.delete_tags')
def test_delete_tags_by_age_no_keep(self, delete_tags_patched):
delete_tags_by_age(self.registry, "imagename", False, 24, [])
self.list_tags_mock.assert_called_with(
"imagename"
)
self.list_tags_mock.assert_called_with("imagename")
self.get_tag_config_mock.assert_called_with("imagename", "image")
delete_tags_patched.assert_called_with(
self.registry, "imagename", False, ["image"], [])
@patch('registry.delete_tags')
def test_delete_tags_by_age_keep_tags(self, delete_tags_patched):
delete_tags_by_age(self.registry, "imagename", False, 24, ["latest"])
self.list_tags_mock.assert_called_with(
"imagename"
)
self.list_tags_mock.assert_called_with("imagename")
self.get_tag_config_mock.assert_called_with("imagename", "image")
delete_tags_patched.assert_called_with(
self.registry, "imagename", False, ["image"], ["latest"])
@patch('registry.delete_tags')
def test_delete_tags_by_age_dry_run(self, delete_tags_patched):
delete_tags_by_age(self.registry, "imagename", True, 24, ["latest"])
self.list_tags_mock.assert_called_with(
"imagename"
)
self.list_tags_mock.assert_called_with("imagename")
self.get_tag_config_mock.assert_called_with("imagename", "image")
delete_tags_patched.assert_called_with(
self.registry, "imagename", True, ["image"], ["latest"])
class TestArgParser(unittest.TestCase):
def test_no_args(self):
with self.assertRaises(SystemExit):
parse_args("")
@@ -453,7 +659,8 @@ class TestArgParser(unittest.TestCase):
"--tags-like", "tags_like_text",
"--no-validate-ssl",
"--delete-all",
"--layers"]
"--layers",
"--delete-by-hours", "24"]
args = parse_args(args_list)
self.assertTrue(args.delete)
self.assertTrue(args.layers)
@@ -466,7 +673,8 @@ class TestArgParser(unittest.TestCase):
self.assertEqual(args.tags_like, ["tags_like_text"])
self.assertEqual(args.host, "hostname")
self.assertEqual(args.keep_tags, ["keep1", "keep2"])
self.assertEqual(args.delete_by_hours, "24")
if __name__ == '__main__':
unittest.main()
unittest.main()