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

Merge pull request #39 from Qlean/master

Browse Source 
Add deleting images by age in hours
This commit is contained in:
andrey-pohilko
2018-02-06 01:56:05 +06:00
committed by GitHub
parent 617e3e86e5 de4a6e97bc
commit 975edfaefa
3 changed files with 444 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
README.md
View File

@@ -1,4 +1,4 @@
[![CircleCI](https://circleci.com/gh/andrey-pohilko/registry-cli/tree/master.svg?style=svg&circle-token=5216bf89763aec24bbcd6d15494ea32ffc53d66d)](https://circleci.com/gh/andrey-pohilko/registry-cli/tree/master)
[![CircleCI](https://circleci.com/gh/andrey-pohilko/registry-cli/tree/master.svg?style=svg&circle-token=5216bf89763aec24bbcd6d15494ea32ffc53d66d)](https://circleci.com/gh/andrey-pohilko/registry-cli/tree/master)
# registry-cli
registry.py is a script for easy manipulation of docker-registry from command line (and from scripts)
@@ -14,15 +14,15 @@ You can download ready-made docker image with the script and all python dependen
docker pull anoxis/registry-cli
```
In this case, in replace
In this case, in replace
```
registry.py
```
with
with
```
docker run --rm anoxis/registry-cli
docker run --rm anoxis/registry-cli
```
in all commands below, e.g.
in all commands below, e.g.
```
docker run --rm anoxis/registry-cli -r http://example.com:5000
```
@@ -61,28 +61,28 @@ List particular image(s) or image:tag (all tags of ubuntu and alpine in this exa
```
registry.py -l user:pass -r https://example.com:5000 -i ubuntu alpine
```
Same as above but with layers
```
registry.py -l user:pass -r https://example.com:5000 -i ubuntu alpine --layers
```
## Username and password
It is optional, you can omit it in case if you use insecure registry without authentication (up to you,
It is optional, you can omit it in case if you use insecure registry without authentication (up to you,
but its really insecure; make sure you protect your entire registry from anyone)
username and password pair can be provided in the following forms
```
```
-l username:password
-l 'username':'password'
-l "username":"password"
```
Username cannot contain colon (':') (I don't think it will contain ever, but anyway I warned you).
Password, in its turn, can contain as many colons as you wish.
## Deleting images
## Deleting images
Keep only last 10 versions (useful for CI):
Delete all tags of all images but keep last 10 tags (you can put this command to your build script
@@ -118,25 +118,29 @@ Delete all tags for particular image (e.g. delete all ubuntu tags):
```
registry.py -l user:pass -r https://example.com:5000 -i ubuntu --delete-all
```
Delete all tags for all images (do you really want to do it?):
```
registry.py -l user:pass -r https://example.com:5000 --delete-all
registry.py -l user:pass -r https://example.com:5000 --delete-all --dry-run
```
Delete all tags by age in hours for the particular image (e.g. older than 24 hours, with --keep-tags and --keep-tags-like options, --dry-run for safe).
```
registry.py -r https://example.com:5000 -i api-docs-origin/master --dry-run --delete-by-hours 24 --keep-tags c59c02c25f023263fd4b5d43fc1ff653f08b3d4x --keep-tags-like late
```
## Disable ssl verification
If you are using docker registry with a self signed ssl certificate, you can disable ssl verification:
```
registry.py -l user:pass -r https://example.com:5000 --no-validate-ssl
registry.py -l user:pass -r https://example.com:5000 --no-validate-ssl
```
## Important notes:
### garbage-collection in docker-registry
1. docker registry API does not actually delete tags or images, it marks them for later
garbage collection. So, make sure you run something like below
## Important notes:
### garbage-collection in docker-registry
1. docker registry API does not actually delete tags or images, it marks them for later
garbage collection. So, make sure you run something like below
(or put them in your crontab):
```
cd [path-where-your-docker-compose.yml]
@@ -145,30 +149,30 @@ garbage collection. So, make sure you run something like below
registry bin/registry garbage-collect \
/etc/docker/registry/config.yml
docker-compose up -d registry
```
```
or (if you are not using docker-compose):
```
docker stop registry:2
docker run --rm registry:2 bin/registry garbage-collect \
/etc/docker/registry/config.yml
docker start registry:2
```
```
for more detail on garbage collection read here:
https://docs.docker.com/registry/garbage-collection/
### enable image deletion in docker-registry
Make sure to enable it by either creating environment variable
Make sure to enable it by either creating environment variable
`REGISTRY_STORAGE_DELETE_ENABLED: "true"`
or adding relevant configuration option to the docker-registry's config.yml.
For more on docker-registry configuration, read here:
https://docs.docker.com/registry/configuration/
You may get `Error 405` message from script (`Functionality not supported`) when this option is not enabled.
You may get `Error 405` message from script (`Functionality not supported`) when this option is not enabled.
## Contribution
You are very welcome to contribute to this script. Of course, when making changes,
please include your changes into `test.py` and run tests to check that your changes
You are very welcome to contribute to this script. Of course, when making changes,
please include your changes into `test.py` and run tests to check that your changes
do not break existing functionality.
For tests to work, install `mock` library
@@ -187,9 +191,9 @@ Testing started at 9:31 AM ...
tag digest not found: 400
error 400
```
this is ok, because test simulates invalid inputs also.
this is ok, because test simulates invalid inputs also.
# Contact
Please feel free to contact me at anoxis@gmail.com if you wish to add more functionality
Please feel free to contact me at anoxis@gmail.com if you wish to add more functionality
or want to contribute.

202
registry.py
View File

@@ -6,39 +6,44 @@ from requests.packages.urllib3.exceptions import InsecureRequestWarning
import json
import re
import argparse
from datetime import timedelta, datetime as dt
## this is a registry manipulator, can do following:
## - list all images (including layers)
## - delete images
## - all except last N images
## - all images and/or tags
##
## run
## registry.py -h
## to get more help
##
## important: after removing the tags, run the garbage collector
## on your registry host:
## docker-compose -f [path_to_your_docker_compose_file] run \
## registry bin/registry garbage-collect \
## /etc/docker/registry/config.yml
##
## or if you are not using docker-compose:
## docker run registry:2 bin/registry garbage-collect \
## /etc/docker/registry/config.yml
##
## for more detail on garbage collection read here:
## https://docs.docker.com/registry/garbage-collection/
# this is a registry manipulator, can do following:
# - list all images (including layers)
# - delete images
# - all except last N images
# - all images and/or tags
#
# run
# registry.py -h
# to get more help
#
# important: after removing the tags, run the garbage collector
# on your registry host:
# docker-compose -f [path_to_your_docker_compose_file] run \
# registry bin/registry garbage-collect \
# /etc/docker/registry/config.yml
#
# or if you are not using docker-compose:
# docker run registry:2 bin/registry garbage-collect \
# /etc/docker/registry/config.yml
#
# for more detail on garbage collection read here:
# https://docs.docker.com/registry/garbage-collection/
# number of image versions to keep
CONST_KEEP_LAST_VERSIONS = 10
# this class is created for testing
class Requests:
def request(self, method, url, **kwargs):
return requests.request(method, url, **kwargs)
def natural_keys(text):
'''
alist.sort(key=natural_keys) sorts in human order
@@ -49,7 +54,7 @@ def natural_keys(text):
def __atoi(text):
return int(text) if text.isdigit() else text
return [ __atoi(c) for c in re.split('(\d+)', text) ]
return [__atoi(c) for c in re.split('(\d+)', text)]
# class to manipulate registry
@@ -82,7 +87,6 @@ class Registry:
return (None, None)
@staticmethod
def create(host, login, no_validate_ssl):
r = Registry()
@@ -97,15 +101,14 @@ class Registry:
r.http = Requests()
return r
def send(self, path, method="GET"):
# try:
result = self.http.request(
method, "{0}{1}".format(self.hostname, path),
headers = self.HEADERS,
headers=self.HEADERS,
auth=(None if self.username == ""
else (self.username, self.password)),
verify = not self.no_validate_ssl)
verify=not self.no_validate_ssl)
# except Exception as error:
# print("cannot connect to {0}\nerror {1}".format(
@@ -116,11 +119,11 @@ class Registry:
self.last_error = None
return result
self.last_error=result.status_code
self.last_error = result.status_code
return None
def list_images(self):
result = self.send('/v2/_catalog')
result = self.send('/v2/_catalog?n=10000')
if result == None:
return []
@@ -169,7 +172,8 @@ class Registry:
tag_digest = self.get_tag_digest(image_name, tag)
if tag_digest in tag_digests_to_ignore:
print("Digest {0} for tag {1} is referenced by another tag or has already been deleted and will be ignored".format(tag_digest, tag))
print("Digest {0} for tag {1} is referenced by another tag or has already been deleted and will be ignored".format(
tag_digest, tag))
return True
if tag_digest == None:
@@ -205,7 +209,6 @@ class Registry:
# print("done")
# return True
def list_tag_layers(self, image_name, tag):
layers_result = self.send("/v2/{0}/manifests/{1}".format(
image_name, tag))
@@ -222,7 +225,45 @@ class Registry:
return layers
def parse_args(args = None):
def get_tag_config(self, image_name, tag):
config_result = self.send(
"/v2/{0}/manifests/{1}".format(image_name, tag))
if config_result == None:
print(" tag digest not found: {0}".format(self.last_error))
return []
json_result = json.loads(config_result.text)
if json_result['schemaVersion'] == 1:
print("Docker schemaVersion 1 isn't supported for deleting by age now")
exit(1)
else:
tag_config = json_result['config']
return tag_config
def get_image_age(self, image_name, image_config):
container_header = {"Accept": "{0}".format(
image_config['mediaType'])}
response = self.http.request("GET", "{0}{1}".format(self.hostname, "/v2/{0}/blobs/{1}".format(
image_name, image_config['digest'])),
headers=container_header,
auth=(None if self.username == ""
else (self.username, self.password)),
verify=not self.no_validate_ssl)
if str(response.status_code)[0] == '2':
self.last_error = None
image_age = json.loads(response.text)
return image_age['created']
else:
print(" blob not found: {0}".format(self.last_error))
self.last_error = response.status_code
return []
def parse_args(args=None):
parser = argparse.ArgumentParser(
description="List or delete images from Docker registry",
formatter_class=argparse.RawDescriptionHelpFormatter,
@@ -243,19 +284,19 @@ for more detail on garbage collection read here:
https://docs.docker.com/registry/garbage-collection/
"""))
parser.add_argument(
'-l','--login',
'-l', '--login',
help="Login and password to access to docker registry",
required=False,
metavar="USER:PASSWORD")
parser.add_argument(
'-r','--host',
'-r', '--host',
help="Hostname for registry server, e.g. https://example.com:5000",
required=True,
metavar="URL")
parser.add_argument(
'-d','--delete',
'-d', '--delete',
help=('If specified, delete all but last {0} tags '
'of all images').format(CONST_KEEP_LAST_VERSIONS),
action='store_const',
@@ -263,7 +304,7 @@ for more detail on garbage collection read here:
const=True)
parser.add_argument(
'-n','--num',
'-n', '--num',
help=('Set the number of tags to keep'
'({0} if not set)').format(CONST_KEEP_LAST_VERSIONS),
default=CONST_KEEP_LAST_VERSIONS,
@@ -279,10 +320,10 @@ for more detail on garbage collection read here:
const=True)
parser.add_argument(
'-i','--image',
'-i', '--image',
help='Specify images and tags to list/delete',
nargs='+',
metavar="IMAGE:[TAG]")
metavar="IMAGE:[TAG]")
parser.add_argument(
'--keep-tags',
@@ -307,7 +348,7 @@ for more detail on garbage collection read here:
parser.add_argument(
'--no-validate-ssl',
help="Disable ssl validation",
help="Disable ssl validation",
action='store_const',
default=False,
const=True)
@@ -326,12 +367,18 @@ for more detail on garbage collection read here:
default=False,
const=True)
parser.add_argument(
'--delete-by-hours',
help=('Will delete all tags that older than specified hours. Be careful!'),
default=False,
nargs='?',
metavar='Hours')
return parser.parse_args(args)
def delete_tags(
registry, image_name, dry_run, tags_to_delete, tags_to_keep):
registry, image_name, dry_run, tags_to_delete, tags_to_keep):
keep_tag_digests = []
@@ -341,8 +388,9 @@ def delete_tags(
print("Getting digest for tag {0}".format(tag))
digest = registry.get_tag_digest(image_name, tag)
if digest is None:
print("Tag {0} does not exist for image {1}. Ignore here.".format(tag, image_name))
if digest is None:
print("Tag {0} does not exist for image {1}. Ignore here.".format(
tag, image_name))
continue
print("Keep digest {0} for tag {1}".format(digest, tag))
@@ -355,15 +403,16 @@ def delete_tags(
print(" deleting tag {0}".format(tag))
## deleting layers is disabled because
## it also deletes shared layers
##
## for layer in registry.list_tag_layers(image_name, tag):
## layer_digest = layer['digest']
## registry.delete_tag_layer(image_name, layer_digest, dry_run)
# deleting layers is disabled because
# it also deletes shared layers
##
# for layer in registry.list_tag_layers(image_name, tag):
# layer_digest = layer['digest']
# registry.delete_tag_layer(image_name, layer_digest, dry_run)
registry.delete_tag(image_name, tag, dry_run, keep_tag_digests)
def get_tags_like(args_tags_like, tags_list):
result = set()
for tag_like in args_tags_like:
@@ -374,6 +423,7 @@ def get_tags_like(args_tags_like, tags_list):
result.add(tag)
return result
def get_tags(all_tags_list, image_name, tags_like):
# check if there are args for special tags
result = set()
@@ -389,11 +439,38 @@ def get_tags(all_tags_list, image_name, tags_like):
return result
def delete_tags_by_age(registry, image_name, dry_run, hours, tags_to_keep):
image_tags = registry.list_tags(image_name)
tags_to_delete = []
print('---------------------------------')
for tag in image_tags:
image_config = registry.get_tag_config(image_name, tag)
if image_config == []:
print("tag not found")
continue
image_age = registry.get_image_age(image_name, image_config)
if image_age == []:
print("timestamp not found")
continue
if dt.strptime(image_age[:-4], "%Y-%m-%dT%H:%M:%S.%f") < dt.now() - timedelta(hours=int(hours)):
print("will be deleted tag: {0} timestamp: {1}".format(
tag, image_age))
tags_to_delete.append(tag)
print('------------deleting-------------')
delete_tags(registry, image_name, dry_run, tags_to_delete, tags_to_keep)
def main_loop(args):
keep_last_versions = int(args.num)
if args.no_validate_ssl:
if args.no_validate_ssl:
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
registry = Registry.create(args.host, args.login, args.no_validate_ssl)
@@ -414,12 +491,12 @@ def main_loop(args):
all_tags_list = registry.list_tags(image_name)
if not all_tags_list:
print(" no tags!")
continue
print(" no tags!")
continue
tags_list = get_tags(all_tags_list, image_name, args.tags_like)
# print(tags and optionally layers
# print(tags and optionally layers
for tag in tags_list:
print(" tag: {0}".format(tag))
if args.layers:
@@ -431,29 +508,37 @@ def main_loop(args):
print(" layer: {0}".format(
layer['blobSum']))
# add tags to "tags_to_keep" list, if we have regexp "tags_to_keep" entries:
keep_tags=[]
# add tags to "tags_to_keep" list, if we have regexp "tags_to_keep"
# entries:
keep_tags = []
if args.keep_tags_like:
keep_tags.extend(get_tags_like(args.keep_tags_like, tags_list))
# delete tags if told so
if args.delete or args.delete_all:
if args.delete_all:
tags_list_to_delete = list(tags_list)
else:
tags_list_to_delete = sorted(tags_list, key=natural_keys)[:-keep_last_versions]
tags_list_to_delete = sorted(tags_list, key=natural_keys)[
:-keep_last_versions]
# A manifest might be shared between different tags. Explicitly add those
# tags that we want to preserve to the keep_tags list, to prevent
# any manifest they are using from being deleted.
tags_list_to_keep = [tag for tag in tags_list if tag not in tags_list_to_delete]
tags_list_to_keep = [
tag for tag in tags_list if tag not in tags_list_to_delete]
keep_tags.extend(tags_list_to_keep)
delete_tags(
registry, image_name, args.dry_run,
tags_list_to_delete, keep_tags)
# delete tags by age in hours
if args.delete_by_hours:
keep_tags.extend(args.keep_tags)
delete_tags_by_age(registry, image_name, args.dry_run,
args.delete_by_hours, keep_tags)
if __name__ == "__main__":
args = parse_args()
try:
@@ -461,4 +546,3 @@ if __name__ == "__main__":
except KeyboardInterrupt:
print("Ctrl-C pressed, quitting")
exit(1)

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()