Upstream
/
harbor
mirror of https://github.com/goharbor/harbor.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Fix API python tests log missing issue 

After debugging for issue of missing some http message logs, we found out that swagger client configuration will be re-initiated by calling models in swagger client, so in API python tests, defination for models must be in front of swagger client defination.

Signed-off-by: danfengliu <danfengl@vmware.com>
This commit is contained in:
danfengliu 2021-03-15 16:08:48 +08:00
parent 68d7c91596
commit 068d1d46ca
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