harbor/tests/apitests/python/library/retention.py

188 lines
7.9 KiB
Python
Raw Normal View History

# -*- 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