Upstream/harbor
1
0
Fork 0
mirror of https://github.com/goharbor/harbor.git synced 2025-02-23 07:11:36 +01:00
Code Issues Projects Releases Wiki Activity

Merge pull request #14445 from danfengliu/fix-api-python-test-log-missing-issue

Browse Source 
Fix API python tests log missing issue
This commit is contained in:
danfengliu 2021-03-16 12:14:27 +08:00 committed by GitHub
commit ac7a0f8841
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 100 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
tests/apitests/python/library/artifact.py
View File

@ -10,11 +10,9 @@ class Artifact(base.Base, object):
super(Artifact,self).__init__(api_type = "artifact")
def list_artifacts(self, project_name, repo_name, **kwargs):
client = self._get_client(**kwargs)
return client.list_artifacts(project_name, repo_name)
return self._get_client(**kwargs).list_artifacts(project_name, repo_name)
def get_reference_info(self, project_name, repo_name, reference, expect_status_code = 200, ignore_not_found = False,**kwargs):
client = self._get_client(**kwargs)
params = {}
if "with_signature" in kwargs:
params["with_signature"] = kwargs["with_signature"]
@ -26,7 +24,7 @@ class Artifact(base.Base, object):
params["with_immutable_status"] = kwargs["with_immutable_status"]
try:
data, status_code, _ = client.get_artifact_with_http_info(project_name, repo_name, reference, **params)
data, status_code, _ = self._get_client(**kwargs).get_artifact_with_http_info(project_name, repo_name, reference, **params)
return data
except ApiException as e:
if e.status == 404 and ignore_not_found == True:
@ -39,10 +37,8 @@ class Artifact(base.Base, object):
return None
def delete_artifact(self, project_name, repo_name, reference, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
_, status_code, _ = client.delete_artifact_with_http_info(project_name, repo_name, reference)
_, status_code, _ = self._get_client(**kwargs).delete_artifact_with_http_info(project_name, repo_name, reference)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
@ -53,14 +49,12 @@ class Artifact(base.Base, object):
base._assert_status_code(200, status_code)
def get_addition(self, project_name, repo_name, reference, addition, **kwargs):
client = self._get_client(**kwargs)
return client.get_addition_with_http_info(project_name, repo_name, reference, addition)
return self._get_client(**kwargs).get_addition_with_http_info(project_name, repo_name, reference, addition)
def add_label_to_reference(self, project_name, repo_name, reference, label_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
label = v2_swagger_client.Label(id = label_id)
try:
body, status_code, _ = client.add_label_with_http_info(project_name, repo_name, reference, label)
body, status_code, _ = self._get_client(**kwargs).add_label_with_http_info(project_name, repo_name, reference, label)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:
@ -69,10 +63,8 @@ class Artifact(base.Base, object):
return body
def copy_artifact(self, project_name, repo_name, _from, expect_status_code = 201, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
data, status_code, _ = client.copy_artifact_with_http_info(project_name, repo_name, _from)
data, status_code, _ = self._get_client(**kwargs).copy_artifact_with_http_info(project_name, repo_name, _from)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
@ -84,10 +76,9 @@ class Artifact(base.Base, object):
return data
def create_tag(self, project_name, repo_name, reference, tag_name, expect_status_code = 201, ignore_conflict = False, **kwargs):
client = self._get_client(**kwargs)
tag = v2_swagger_client.Tag(name = tag_name)
try:
_, status_code, _ = client.create_tag_with_http_info(project_name, repo_name, reference, tag)
_, status_code, _ = self._get_client(**kwargs).create_tag_with_http_info(project_name, repo_name, reference, tag)
except ApiException as e:
if e.status == 409 and ignore_conflict == True:
return
@ -98,9 +89,8 @@ class Artifact(base.Base, object):
base._assert_status_code(201, status_code)
def delete_tag(self, project_name, repo_name, reference, tag_name, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
try:
_, status_code, _ = client.delete_tag_with_http_info(project_name, repo_name, reference, tag_name)
_, status_code, _ = self._get_client(**kwargs).delete_tag_with_http_info(project_name, repo_name, reference, tag_name)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:

2
tests/apitests/python/library/base.py
View File

@ -121,7 +121,7 @@ def restart_process(process):
def run_command_with_popen(command):
print("Command: ", subprocess.list2cmdline(command))
print("Command: ", command)
try:
proc = subprocess.Popen(command, universal_newlines=True, shell=True,

20
tests/apitests/python/library/gc.py
View File

@ -11,10 +11,8 @@ class GC(base.Base, object):
super(GC,self).__init__(api_type = "gc")
def get_gc_history(self, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
data, status_code, _ = client.get_gc_history_with_http_info()
data, status_code, _ = self._get_client(**kwargs).get_gc_history_with_http_info()
except ApiException as e:
if e.status == expect_status_code:
if expect_response_body is not None and e.body.strip() != expect_response_body.strip():
@ -27,10 +25,8 @@ class GC(base.Base, object):
return data
def get_gc_status_by_id(self, job_id, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
data, status_code, _ = client.get_gc_with_http_info(job_id)
data, status_code, _ = self._get_client(**kwargs).get_gc_with_http_info(job_id)
except ApiException as e:
if e.status == expect_status_code:
if expect_response_body is not None and e.body.strip() != expect_response_body.strip():
@ -43,10 +39,8 @@ class GC(base.Base, object):
return data
def get_gc_log_by_id(self, job_id, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
data, status_code, _ = client.get_gc_log_with_http_info(job_id)
data, status_code, _ = self._get_client(**kwargs).get_gc_log_with_http_info(job_id)
except ApiException as e:
if e.status == expect_status_code:
if expect_response_body is not None and e.body.strip() != expect_response_body.strip():
@ -59,10 +53,8 @@ class GC(base.Base, object):
return data
def get_gc_schedule(self, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
data, status_code, _ = client.get_gc_schedule_with_http_info()
data, status_code, _ = self._get_client(**kwargs).get_gc_schedule_with_http_info()
except ApiException as e:
if e.status == expect_status_code:
if expect_response_body is not None and e.body.strip() != expect_response_body.strip():
@ -75,8 +67,6 @@ class GC(base.Base, object):
return data
def create_gc_schedule(self, schedule_type, is_delete_untagged, cron = None, expect_status_code = 201, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
gc_parameters = {'delete_untagged':is_delete_untagged}
gc_schedule = v2_swagger_client.ScheduleObj()
@ -89,7 +79,7 @@ class GC(base.Base, object):
gc_job.parameters = gc_parameters
try:
_, status_code, header = client.create_gc_schedule_with_http_info(gc_job)
_, status_code, header = self._get_client(**kwargs).create_gc_schedule_with_http_info(gc_job)
except ApiException as e:
if e.status == expect_status_code:
if expect_response_body is not None and e.body.strip() != expect_response_body.strip():

18
tests/apitests/python/library/preheat.py
View File

@ -14,12 +14,11 @@ class Preheat(base.Base, object):
expect_status_code = 201, expect_response_body = None, **kwargs):
if name is None:
name = base._random_name("instance")
client = self._get_client(**kwargs)
instance = v2_swagger_client.Instance(name=name, description=description,vendor=vendor,
endpoint=endpoint_url, auth_mode=auth_mode, enabled=enabled)
print("instance:",instance)
try:
_, status_code, header = client.create_instance_with_http_info(instance)
_, status_code, header = self._get_client(**kwargs).create_instance_with_http_info(instance)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
@ -34,13 +33,12 @@ class Preheat(base.Base, object):
expect_status_code = 201, expect_response_body = None, **kwargs):
if name is None:
name = base._random_name("policy")
client = self._get_client(**kwargs)
policy = v2_swagger_client.PreheatPolicy(name=name, project_id=project_id, provider_id=provider_id,
description=description,filters=filters,
trigger=trigger, enabled=enabled)
print("policy:",policy)
try:
data, status_code, header = client.create_policy_with_http_info(project_name, policy)
data, status_code, header = self._get_client(**kwargs).create_policy_with_http_info(project_name, policy)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
@ -51,21 +49,17 @@ class Preheat(base.Base, object):
return base._get_id_from_header(header), name
def get_instance(self, instance_name, **kwargs):
client = self._get_client(**kwargs)
return client.get_instance(instance_name)
return self._get_client(**kwargs).get_instance(instance_name)
def get_policy(self, project_name, preheat_policy_name, **kwargs):
client = self._get_client(**kwargs)
return client.get_policy(project_name, preheat_policy_name)
return self._get_client(**kwargs).get_policy(project_name, preheat_policy_name)
def update_policy(self, project_name, preheat_policy_name, policy, **kwargs):
client = self._get_client(**kwargs)
return client.update_policy(project_name, preheat_policy_name, policy)
return self._get_client(**kwargs).update_policy(project_name, preheat_policy_name, policy)
def delete_instance(self, preheat_instance_name, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
_, status_code, header = _, status_code, _ = client.delete_instance_with_http_info(preheat_instance_name)
_, status_code, header = _, status_code, _ = self._get_client(**kwargs).delete_instance_with_http_info(preheat_instance_name)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:

39
tests/apitests/python/library/project.py
View File

@ -37,11 +37,9 @@ class Project(base.Base):
metadata = {}
if registry_id is None:
registry_id = registry_id
client = self._get_client(**kwargs)
project = v2_swagger_client.ProjectReq(project_name=name, registry_id = registry_id, metadata=metadata)
try:
_, status_code, header = client.create_project_with_http_info(v2_swagger_client.ProjectReq(project_name=name, registry_id = registry_id, metadata=metadata))
_, status_code, header = self._get_client(**kwargs).create_project_with_http_info(project)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
@ -52,9 +50,8 @@ class Project(base.Base):
return base._get_id_from_header(header), name
def get_projects(self, params, **kwargs):
client = self._get_client(**kwargs)
data = []
data, status_code, _ = client.list_projects_with_http_info(**params)
data, status_code, _ = self._get_client(**kwargs).list_projects_with_http_info(**params)
base._assert_status_code(200, status_code)
return data
@ -75,9 +72,8 @@ class Project(base.Base):
raise Exception(r"Project-id check failed, expect {} but got {}, please check this test case.".format(str(expected_project_id), str(project_data[0].project_id)))
def check_project_name_exist(self, name=None, **kwargs):
client = self._get_client(**kwargs)
try:
_, status_code, _ = client.head_project_with_http_info(name)
_, status_code, _ = self._get_client(**kwargs).head_project_with_http_info(name)
except ApiException as e:
status_code = -1
return {
@ -86,9 +82,8 @@ class Project(base.Base):
}.get(status_code,False)
def get_project(self, project_id, expect_status_code = 200, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
data, status_code, _ = client.get_project_with_http_info(project_id)
data, status_code, _ = self._get_client(**kwargs).get_project_with_http_info(project_id)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
@ -101,23 +96,20 @@ class Project(base.Base):
return data
def update_project(self, project_id, expect_status_code=200, metadata=None, cve_allowlist=None, **kwargs):
client = self._get_client(**kwargs)
project = v2_swagger_client.ProjectReq(metadata=metadata, cve_allowlist=cve_allowlist)
try:
_, sc, _ = client.update_project_with_http_info(project_id, project)
_, sc, _ = self._get_client(**kwargs).update_project_with_http_info(project_id, project)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:
base._assert_status_code(expect_status_code, sc)
def delete_project(self, project_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.delete_project_with_http_info(project_id)
_, status_code, _ = self._get_client(**kwargs).delete_project_with_http_info(project_id)
base._assert_status_code(expect_status_code, status_code)
def get_project_log(self, project_name, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
body, status_code, _ = client.get_logs_with_http_info(project_name)
body, status_code, _ = self._get_client(**kwargs).get_logs_with_http_info(project_name)
base._assert_status_code(expect_status_code, status_code)
return body
@ -134,16 +126,14 @@ class Project(base.Base):
def get_project_members(self, project_id, **kwargs):
kwargs['api_type'] = 'products'
client = self._get_client(**kwargs)
return client.projects_project_id_members_get(project_id)
return self._get_client(**kwargs).projects_project_id_members_get(project_id)
def get_project_member(self, project_id, member_id, expect_status_code = 200, expect_response_body = None, **kwargs):
from swagger_client.rest import ApiException
kwargs['api_type'] = 'products'
client = self._get_client(**kwargs)
data = []
try:
data, status_code, _ = client.projects_project_id_members_mid_get_with_http_info(project_id, member_id,)
data, status_code, _ = self._get_client(**kwargs).projects_project_id_members_mid_get_with_http_info(project_id, member_id,)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
@ -179,17 +169,15 @@ class Project(base.Base):
def update_project_member_role(self, project_id, member_id, member_role_id, expect_status_code = 200, **kwargs):
kwargs['api_type'] = 'products'
client = self._get_client(**kwargs)
role = swagger_client.Role(role_id = member_role_id)
data, status_code, _ = client.projects_project_id_members_mid_put_with_http_info(project_id, member_id, role = role)
data, status_code, _ = self._get_client(**kwargs).projects_project_id_members_mid_put_with_http_info(project_id, member_id, role = role)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
return data
def delete_project_member(self, project_id, member_id, expect_status_code = 200, **kwargs):
kwargs['api_type'] = 'products'
client = self._get_client(**kwargs)
_, status_code, _ = client.projects_project_id_members_mid_delete_with_http_info(project_id, member_id)
_, status_code, _ = self._get_client(**kwargs).projects_project_id_members_mid_delete_with_http_info(project_id, member_id)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
@ -205,10 +193,9 @@ class Project(base.Base):
if _ldap_group_dn is not None:
projectMember.member_group = swagger_client.UserGroup(ldap_group_dn=_ldap_group_dn)
client = self._get_client(**kwargs)
data = []
try:
data, status_code, header = client.projects_project_id_members_post_with_http_info(project_id, project_member = projectMember)
data, status_code, header = self._get_client(**kwargs).projects_project_id_members_post_with_http_info(project_id, project_member = projectMember)
except swagger_client.rest.ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:

3
tests/apitests/python/library/projectV2.py
View File

@ -10,8 +10,7 @@ class ProjectV2(base.Base, object):
super(ProjectV2,self).__init__(api_type = "projectv2")
def get_project_log(self, project_name, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
body, status_code, _ = client.get_logs_with_http_info(project_name)
body, status_code, _ = self._get_client(**kwargs).get_logs_with_http_info(project_name)
base._assert_status_code(expect_status_code, status_code)
return body

6
tests/apitests/python/library/replication_v2.py
View File

@ -23,13 +23,11 @@ class ReplicationV2(base.Base, object):
raise Exception("The jobs not Succeed")
def trigger_replication_executions(self, rule_id, expect_status_code = 201, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.start_replication_with_http_info({"policy_id":rule_id})
_, status_code, _ = self._get_client(**kwargs).start_replication_with_http_info({"policy_id":rule_id})
base._assert_status_code(expect_status_code, status_code)
def get_replication_executions(self, rule_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
data, status_code, _ = client.list_replication_executions_with_http_info(policy_id=rule_id)
data, status_code, _ = self._get_client(**kwargs).list_replication_executions_with_http_info(policy_id=rule_id)
base._assert_status_code(expect_status_code, status_code)
return data

97
tests/apitests/python/library/retention.py
View File

@ -1,23 +1,22 @@
# -*- coding: utf-8 -*-
import base
import v2_swagger_client as swagger_client
import v2_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()
metadatas, status_code, _ = self._get_client(**kwargs).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(
policy=v2_swagger_client.RetentionPolicy(
algorithm='or',
rules=[
swagger_client.RetentionRule(
v2_swagger_client.RetentionRule(
disabled=False,
action="retain",
template="always",
@ -55,23 +54,21 @@ class Retention(base.Base):
"ref": project_id
},
)
client = self._get_client(**kwargs)
_, status_code, header = client.create_retention_with_http_info(policy)
_, status_code, header = self._get_client(**kwargs).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)
policy, status_code, _ = self._get_client(**kwargs).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(
policy=v2_swagger_client.RetentionPolicy(
id=retention_id,
algorithm='or',
rules=[
swagger_client.RetentionRule(
v2_swagger_client.RetentionRule(
disabled=False,
action="retain",
template="always",
@ -109,79 +106,67 @@ class Retention(base.Base):
"ref": project_id
},
)
client = self._get_client(**kwargs)
_, status_code, _ = client.update_retention_with_http_info(retention_id, policy)
_, status_code, _ = self._get_client(**kwargs).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):
retention_rule = v2_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
}
]
)
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
}
]
))
policy.rules.append(retention_rule)
_, 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})
_, status_code, _ = self._get_client(**kwargs).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"})
r, status_code, _ = self._get_client(**kwargs).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)
r, status_code, _ = self._get_client(**kwargs).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)
r, status_code, _ = self._get_client(**kwargs).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)
r, status_code, _ = self._get_client(**kwargs).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()
r, status_code, _ = self._get_client(**kwargs).get_rentenition_metadata_with_http_info()
base._assert_status_code(expect_status_code, status_code)
return r

24
tests/apitests/python/library/robot.py
View File

@ -11,9 +11,8 @@ class Robot(base.Base, object):
super(Robot,self).__init__(api_type = "robot")
def list_robot(self, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
try:
body, status_code, _ = client.list_robot_with_http_info()
body, status_code, _ = self._get_client(**kwargs).list_robot_with_http_info()
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
return []
@ -83,10 +82,9 @@ class Robot(base.Base, object):
permission_list.append(robotaccountPermissions)
robotAccountCreate = v2_swagger_client.RobotCreate(name=robot_name, description=robot_desc, duration=duration, level="project", permissions = permission_list)
client = self._get_client(**kwargs)
data = []
try:
data, status_code, header = client.create_robot_with_http_info(robotAccountCreate)
data, status_code, header = self._get_client(**kwargs).create_robot_with_http_info(robotAccountCreate)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:
@ -97,22 +95,19 @@ class Robot(base.Base, object):
return base._get_id_from_header(header), data
def get_robot_account_by_id(self, robot_id, **kwargs):
client = self._get_client(**kwargs)
data, status_code, _ = client.get_robot_by_id_with_http_info(robot_id)
data, status_code, _ = self._get_client(**kwargs).get_robot_by_id_with_http_info(robot_id)
return data
def disable_robot_account(self, robot_id, disable, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
data = self.get_robot_account_by_id(robot_id, **kwargs)
robotAccountUpdate = v2_swagger_client.RobotCreate(name=data.name, description=data.description, duration=data.duration, level=data.level, permissions = data.permissions, disable = disable)
_, status_code, _ = client.update_robot_with_http_info(robot_id, robotAccountUpdate)
_, status_code, _ = self._get_client(**kwargs).update_robot_with_http_info(robot_id, robotAccountUpdate)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
def delete_robot_account(self, robot_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.delete_robot_with_http_info(robot_id)
_, status_code, _ = self._get_client(**kwargs).delete_robot_with_http_info(robot_id)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
@ -124,16 +119,14 @@ class Robot(base.Base, object):
robotAccountCreate = v2_swagger_client.RobotCreate(name=robot_name, description=robot_desc, duration=duration, level="system", disable = False, permissions = permission_list)
print("robotAccountCreate:", robotAccountCreate)
client = self._get_client(**kwargs)
data = []
data, status_code, header = client.create_robot_with_http_info(robotAccountCreate)
data, status_code, header = self._get_client(**kwargs).create_robot_with_http_info(robotAccountCreate)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(201, status_code)
return base._get_id_from_header(header), data
def update_robot_account(self, robot_id, robot, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.update_robot_with_http_info(robot_id, robot)
_, status_code, _ = self._get_client(**kwargs).update_robot_with_http_info(robot_id, robot)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
@ -145,8 +138,7 @@ class Robot(base.Base, object):
def refresh_robot_account_secret(self, robot_id, robot_new_sec, expect_status_code = 200, **kwargs):
robot_sec = v2_swagger_client.RobotSec(secret = robot_new_sec)
client = self._get_client(**kwargs)
data, status_code, _ = client.refresh_sec_with_http_info(robot_id, robot_sec)
data, status_code, _ = self._get_client(**kwargs).refresh_sec_with_http_info(robot_id, robot_sec)
base._assert_status_code(expect_status_code, status_code)
base._assert_status_code(200, status_code)
print("Refresh new secret:", data)

3
tests/apitests/python/library/scan.py
View File

@ -10,9 +10,8 @@ class Scan(base.Base, object):
super(Scan,self).__init__(api_type = "scan")
def scan_artifact(self, project_name, repo_name, reference, expect_status_code = 202, expect_response_body = None, **kwargs):
client = self._get_client(**kwargs)
try:
data, status_code, _ = client.scan_artifact_with_http_info(project_name, repo_name, reference)
data, status_code, _ = self._get_client(**kwargs).scan_artifact_with_http_info(project_name, repo_name, reference)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:

4
tests/apitests/python/library/scan_all.py
View File

@ -10,8 +10,6 @@ class ScanAll(base.Base):
super(ScanAll,self).__init__(api_type="scanall")
def create_scan_all_schedule(self, schedule_type, cron=None, expect_status_code=201, expect_response_body=None, **kwargs):
client = self._get_client(**kwargs)
schedule_obj = v2_swagger_client.ScheduleObj()
schedule_obj.type = schedule_type
if cron is not None:
@ -21,7 +19,7 @@ class ScanAll(base.Base):
schedule.schedule = schedule_obj
try:
_, status_code, _ = client.create_scan_all_schedule_with_http_info(schedule)
_, status_code, _ = self._get_client(**kwargs).create_scan_all_schedule_with_http_info(schedule)
except ApiException as e:
if e.status == expect_status_code:
if expect_response_body is not None and e.body.strip() != expect_response_body.strip():

6
tests/apitests/python/library/system_cve_allowlist.py
View File

@ -9,16 +9,14 @@ class SystemCVEAllowlist(base.Base, object):
super(SystemCVEAllowlist, self).__init__(api_type = "system_cve_allowlist")
def set_cve_allowlist(self, expires_at=None, expected_status_code=200, *cve_ids, **kwargs):
client = self._get_client(**kwargs)
cve_list = [v2_swagger_client.CVEAllowlistItem(cve_id=c) for c in cve_ids]
allowlist = v2_swagger_client.CVEAllowlist(expires_at=expires_at, items=cve_list)
try:
r = client.put_system_cve_allowlist_with_http_info(allowlist=allowlist, _preload_content=False)
r = self._get_client(**kwargs).put_system_cve_allowlist_with_http_info(allowlist=allowlist, _preload_content=False)
except ApiException as e:
base._assert_status_code(expected_status_code, e.status)
else:
base._assert_status_code(expected_status_code, r.status)
def get_cve_allowlist(self, **kwargs):
client = self._get_client(**kwargs)
return client.get_system_cve_allowlist()
return self._get_client(**kwargs).get_system_cve_allowlist()

9
tests/apitests/python/library/tag_immutability.py
View File

@ -12,7 +12,6 @@ class Tag_Immutability(base.Base, object):
selector_repository="**", selector_tag_decoration = "matches",
selector_tag="**", expect_status_code = 201, **kwargs):
#repoExcludes,excludes
client = self._get_client(**kwargs)
immutable_rule = v2_swagger_client.ImmutableRule(
action="immutable",
template="immutable_template",
@ -35,7 +34,7 @@ class Tag_Immutability(base.Base, object):
]
)
try:
_, status_code, header = client.create_immu_rule_with_http_info(project_id, immutable_rule)
_, status_code, header = self._get_client(**kwargs).create_immu_rule_with_http_info(project_id, immutable_rule)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:
@ -44,8 +43,7 @@ class Tag_Immutability(base.Base, object):
return base._get_id_from_header(header)
def list_tag_immutability_policy_rules(self, project_id, **kwargs):
client = self._get_client(**kwargs)
return client.list_immu_rules_with_http_info(project_id)
return self._get_client(**kwargs).list_immu_rules_with_http_info(project_id)
def get_rule(self, project_id, rule_id, **kwargs):
rules = self.list_tag_immutability_policy_rules(project_id, **kwargs)
@ -70,9 +68,8 @@ class Tag_Immutability(base.Base, object):
rule.tag_selectors[0].pattern = selector_tag
if disabled is not None:
rule.disabled = disabled
client = self._get_client(**kwargs)
try:
_, status_code, header = client.update_immu_rule_with_http_info(project_id, rule_id, rule)
_, status_code, header = self._get_client(**kwargs).update_immu_rule_with_http_info(project_id, rule_id, rule)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
if expect_response_body is not None:

24
tests/apitests/python/library/user.py
View File

@ -19,11 +19,10 @@ class User(base.Base):
if role_id is None:
role_id = 0
client = self._get_client(**kwargs)
user = swagger_client.User(username = name, email = email, password = user_password, realname = realname, role_id = role_id)
try:
_, status_code, header = client.users_post_with_http_info(user)
_, status_code, header = self._get_client(**kwargs).users_post_with_http_info(user)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:
@ -31,7 +30,6 @@ class User(base.Base):
return base._get_id_from_header(header), name
def get_users(self, user_name=None, email=None, page=None, page_size=None, expect_status_code=200, **kwargs):
client = self._get_client(**kwargs)
params={}
if user_name is not None:
params["username"] = user_name
@ -42,7 +40,7 @@ class User(base.Base):
if page_size is not None:
params["page_size"] = page_size
try:
data, status_code, _ = client.users_get_with_http_info(**params)
data, status_code, _ = self._get_client(**kwargs).users_get_with_http_info(**params)
except ApiException as e:
base._assert_status_code(expect_status_code, e.status)
else:
@ -50,8 +48,7 @@ class User(base.Base):
return data
def get_user_by_id(self, user_id, **kwargs):
client = self._get_client(**kwargs)
data, status_code, _ = client.users_user_id_get_with_http_info(user_id)
data, status_code, _ = self._get_client(**kwargs).users_user_id_get_with_http_info(user_id)
base._assert_status_code(200, status_code)
return data
@ -64,14 +61,12 @@ class User(base.Base):
def get_user_current(self, **kwargs):
client = self._get_client(**kwargs)
data, status_code, _ = client.users_current_get_with_http_info()
data, status_code, _ = self._get_client(**kwargs).users_current_get_with_http_info()
base._assert_status_code(200, status_code)
return data
def delete_user(self, user_id, expect_status_code = 200, **kwargs):
client = self._get_client(**kwargs)
_, status_code, _ = client.users_user_id_delete_with_http_info(user_id)
_, status_code, _ = self._get_client(**kwargs).users_user_id_delete_with_http_info(user_id)
base._assert_status_code(expect_status_code, status_code)
return user_id
@ -79,21 +74,18 @@ class User(base.Base):
if old_password is None:
old_password = ""
password = swagger_client.Password(old_password, new_password)
client = self._get_client(**kwargs)
_, status_code, _ = client.users_user_id_password_put_with_http_info(user_id, password)
_, status_code, _ = self._get_client(**kwargs).users_user_id_password_put_with_http_info(user_id, password)
base._assert_status_code(200, status_code)
return user_id
def update_user_profile(self, user_id, email=None, realname=None, comment=None, **kwargs):
client = self._get_client(**kwargs)
user_rofile = swagger_client.UserProfile(email, realname, comment)
_, status_code, _ = client.users_user_id_put_with_http_info(user_id, user_rofile)
_, status_code, _ = self._get_client(**kwargs).users_user_id_put_with_http_info(user_id, user_rofile)
base._assert_status_code(200, status_code)
return user_id
def update_user_role_as_sysadmin(self, user_id, IsAdmin, **kwargs):
client = self._get_client(**kwargs)
sysadmin_flag = swagger_client.SysAdminFlag(IsAdmin)
_, status_code, _ = client.users_user_id_sysadmin_put_with_http_info(user_id, sysadmin_flag)
_, status_code, _ = self._get_client(**kwargs).users_user_id_sysadmin_put_with_http_info(user_id, sysadmin_flag)
base._assert_status_code(200, status_code)
return user_id