mirror of
https://github.com/goharbor/harbor.git
synced 2024-11-23 02:35:17 +01:00
be58c1d3ff
Signed-off-by: Ziming Zhang <zziming@vmware.com>
188 lines
7.9 KiB
Python
188 lines
7.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import base
|
|
import v2_swagger_client as swagger_client
|
|
|
|
class Retention(base.Base):
|
|
def __init__(self):
|
|
super(Retention,self).__init__(api_type="retention")
|
|
|
|
def get_metadatas(self, expect_status_code = 200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
metadatas, status_code, _ = client.get_rentenition_metadata_with_http_info()
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return metadatas
|
|
|
|
def create_retention_policy(self, project_id, selector_repository="**", selector_tag="**", expect_status_code = 201, **kwargs):
|
|
policy=swagger_client.RetentionPolicy(
|
|
algorithm='or',
|
|
rules=[
|
|
swagger_client.RetentionRule(
|
|
disabled=False,
|
|
action="retain",
|
|
template="always",
|
|
params= {
|
|
|
|
},
|
|
scope_selectors={
|
|
"repository": [
|
|
{
|
|
"kind": "doublestar",
|
|
"decoration": "repoMatches",
|
|
"pattern": selector_repository
|
|
}
|
|
]
|
|
},
|
|
tag_selectors=[
|
|
{
|
|
"kind": "doublestar",
|
|
"decoration": "matches",
|
|
"pattern": selector_tag
|
|
}
|
|
]
|
|
)
|
|
],
|
|
trigger= {
|
|
"kind": "Schedule",
|
|
"settings": {
|
|
"cron": ""
|
|
},
|
|
"references": {
|
|
}
|
|
},
|
|
scope= {
|
|
"level": "project",
|
|
"ref": project_id
|
|
},
|
|
)
|
|
client = self._get_client(**kwargs)
|
|
_, status_code, header = client.create_retention_with_http_info(policy)
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return base._get_id_from_header(header)
|
|
|
|
def get_retention_policy(self, retention_id, expect_status_code = 200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
policy, status_code, _ = client.get_retention_with_http_info(retention_id)
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return policy
|
|
|
|
def update_retention_policy(self, retention_id, selector_repository="**", selector_tag="**", expect_status_code = 200, **kwargs):
|
|
policy=swagger_client.RetentionPolicy(
|
|
id=retention_id,
|
|
algorithm='or',
|
|
rules=[
|
|
swagger_client.RetentionRule(
|
|
disabled=False,
|
|
action="retain",
|
|
template="always",
|
|
params= {
|
|
|
|
},
|
|
scope_selectors={
|
|
"repository": [
|
|
{
|
|
"kind": "doublestar",
|
|
"decoration": "repoMatches",
|
|
"pattern": selector_repository
|
|
}
|
|
]
|
|
},
|
|
tag_selectors=[
|
|
{
|
|
"kind": "doublestar",
|
|
"decoration": "matches",
|
|
"pattern": selector_tag
|
|
}
|
|
]
|
|
)
|
|
],
|
|
trigger= {
|
|
"kind": "Schedule",
|
|
"settings": {
|
|
"cron": ""
|
|
},
|
|
"references": {
|
|
}
|
|
},
|
|
scope= {
|
|
"level": "project",
|
|
"ref": project_id
|
|
},
|
|
)
|
|
client = self._get_client(**kwargs)
|
|
_, status_code, _ = client.update_retention_with_http_info(retention_id, policy)
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
|
|
def update_retention_add_rule(self, retention_id, selector_repository="**", selector_tag="**", with_untag="True", expect_status_code = 200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
policy, status_code, _ = client.get_retention_with_http_info(retention_id)
|
|
base._assert_status_code(200, status_code)
|
|
policy.rules.append(swagger_client.RetentionRule(
|
|
disabled=False,
|
|
action="retain",
|
|
template="always",
|
|
params= {
|
|
|
|
},
|
|
scope_selectors={
|
|
"repository": [
|
|
{
|
|
"kind": "doublestar",
|
|
"decoration": "repoMatches",
|
|
"pattern": selector_repository
|
|
}
|
|
]
|
|
},
|
|
tag_selectors=[
|
|
{
|
|
"kind": "doublestar",
|
|
"decoration": "matches",
|
|
"extras":'["untagged":'+with_untag+']',
|
|
"pattern": selector_tag
|
|
}
|
|
]
|
|
))
|
|
_, status_code, _ = client.update_retention_with_http_info(retention_id, policy)
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
|
|
def trigger_retention_policy(self, retention_id, dry_run=False, expect_status_code = 201, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
|
|
_, status_code, _ = client.trigger_retention_execution_with_http_info(retention_id, {"dry_run":dry_run})
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
|
|
def stop_retention_execution(self, retention_id, exec_id, expect_status_code = 200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
|
|
r, status_code, _ = client.operate_retention_execution_with_http_info(retention_id, exec_id, {"action":"stop"})
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return r
|
|
|
|
def get_retention_executions(self, retention_id, expect_status_code = 200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
|
|
r, status_code, _ = client.list_retention_executions_with_http_info(retention_id)
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return r
|
|
|
|
def get_retention_exec_tasks(self, retention_id, exec_id, expect_status_code = 200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
|
|
r, status_code, _ = client.list_retention_tasks_with_http_info(retention_id, exec_id)
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return r
|
|
|
|
def get_retention_exec_task_log(self, retention_id, exec_id, task_id, expect_status_code = 200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
|
|
r, status_code, _ = client.get_retention_task_log_with_http_info(retention_id, exec_id, task_id)
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return r
|
|
|
|
def get_retention_metadatas(self, expect_status_code = 200, **kwargs):
|
|
client = self._get_client(**kwargs)
|
|
|
|
r, status_code, _ = client.get_rentenition_metadata_with_http_info()
|
|
base._assert_status_code(expect_status_code, status_code)
|
|
return r
|