mirror of
https://github.com/openSUSE/osc.git
synced 2024-11-10 06:46:15 +01:00
Merge pull request #1550 from dmach/xmlmodel-tokens
Migrate 'token' command to obs_api.Token
This commit is contained in:
GitHub
commit
cc9f23faa0
@ -5,7 +5,6 @@ Scenario: Run `osc token` with no arguments
|
||||
When I execute osc with args "token"
|
||||
Then stdout is
|
||||
"""
|
||||
<directory count="0"/>
|
||||
"""
|
||||
|
||||
|
||||
@ -24,11 +23,15 @@ Scenario: Run `osc token --operation rebuild`
|
||||
Given I execute osc with args "token"
|
||||
And stdout matches
|
||||
"""
|
||||
<directory count="1">
|
||||
<entry id="1" string=".*" kind="rebuild" description="" triggered_at="" project="test:factory" package="test-pkgA"/>
|
||||
</directory>
|
||||
ID : 1
|
||||
String : .*
|
||||
Operation : rebuild
|
||||
Description :
|
||||
Project : test:factory
|
||||
Package : test-pkgA
|
||||
Triggered at :
|
||||
"""
|
||||
And I search 'string="(?P<token>[^"]+)' in stdout and store named groups in 'tokens'
|
||||
And I search 'String *: *(?P<token>.+)\n' in stdout and store named groups in 'tokens'
|
||||
When I execute osc with args "token --trigger {context.tokens[0][token]}"
|
||||
Then stdout is
|
||||
"""
|
||||
|
||||
@ -1688,6 +1688,7 @@ class Osc(cmdln.Cmdln):
|
||||
osc token --delete <TOKENID>
|
||||
osc token --trigger <TOKENSTRING> [--operation <OPERATION>] [<PROJECT> <PACKAGE>]
|
||||
"""
|
||||
from . import obs_api
|
||||
|
||||
args = slash_split(args)
|
||||
|
||||
@ -1696,60 +1697,51 @@ class Osc(cmdln.Cmdln):
|
||||
raise oscerr.WrongOptions(msg)
|
||||
|
||||
apiurl = self.get_api_url()
|
||||
url_path = ['person', conf.get_apiurl_usr(apiurl), 'token']
|
||||
user = conf.get_apiurl_usr(apiurl)
|
||||
|
||||
if len(args) > 1:
|
||||
project = args[0]
|
||||
package = args[1]
|
||||
else:
|
||||
project = None
|
||||
package = None
|
||||
|
||||
if opts.create:
|
||||
if not opts.operation:
|
||||
self.argparser.error("Please specify --operation")
|
||||
|
||||
if opts.operation == 'workflow' and not opts.scm_token:
|
||||
msg = 'The --operation=workflow option requires a --scm-token=<token> option'
|
||||
raise oscerr.WrongOptions(msg)
|
||||
print("Create a new token")
|
||||
query = {'cmd': 'create'}
|
||||
if opts.operation:
|
||||
query['operation'] = opts.operation
|
||||
if opts.scm_token:
|
||||
query['scm_token'] = opts.scm_token
|
||||
if len(args) > 1:
|
||||
query['project'] = args[0]
|
||||
query['package'] = args[1]
|
||||
|
||||
url = makeurl(apiurl, url_path, query)
|
||||
f = http_POST(url)
|
||||
while True:
|
||||
data = f.read(16384)
|
||||
if not data:
|
||||
break
|
||||
sys.stdout.buffer.write(data)
|
||||
print("Create a new token")
|
||||
status = obs_api.Token.cmd_create(
|
||||
apiurl,
|
||||
user,
|
||||
operation=opts.operation,
|
||||
project=project,
|
||||
package=package,
|
||||
scm_token=opts.scm_token,
|
||||
)
|
||||
print(status.to_string())
|
||||
|
||||
elif opts.delete:
|
||||
print("Delete token")
|
||||
url_path.append(opts.delete)
|
||||
url = makeurl(apiurl, url_path)
|
||||
http_DELETE(url)
|
||||
status = obs_api.Token.do_delete(apiurl, user, token=opts.delete)
|
||||
print(status.to_string())
|
||||
elif opts.trigger:
|
||||
print("Trigger token")
|
||||
query = {}
|
||||
if len(args) > 1:
|
||||
query['project'] = args[0]
|
||||
query['package'] = args[1]
|
||||
if opts.operation:
|
||||
url = makeurl(apiurl, ["trigger", opts.operation], query)
|
||||
else:
|
||||
url = makeurl(apiurl, ["trigger"], query)
|
||||
headers = {
|
||||
'Content-Type': 'application/octet-stream',
|
||||
'Authorization': "Token " + opts.trigger,
|
||||
}
|
||||
fd = http_POST(url, headers=headers)
|
||||
print(decode_it(fd.read()))
|
||||
status = obs_api.Token.do_trigger(apiurl, token=opts.trigger, project=project, package=package)
|
||||
print(status.to_string())
|
||||
else:
|
||||
if args and args[0] in ['create', 'delete', 'trigger']:
|
||||
raise oscerr.WrongArgs("Did you mean --" + args[0] + "?")
|
||||
# just list token
|
||||
url = makeurl(apiurl, url_path)
|
||||
for data in streamfile(url, http_GET):
|
||||
sys.stdout.buffer.write(data)
|
||||
|
||||
# just list tokens
|
||||
token_list = obs_api.Token.do_list(apiurl, user)
|
||||
for obj in token_list:
|
||||
print(obj.to_human_readable_string())
|
||||
print()
|
||||
|
||||
@cmdln.option('-a', '--attribute', metavar='ATTRIBUTE',
|
||||
help='affect only a given attribute')
|
||||
|
||||
@ -4,3 +4,4 @@ from .package_sources import PackageSources
|
||||
from .person import Person
|
||||
from .project import Project
|
||||
from .request import Request
|
||||
from .token import Token
|
||||
|
||||
@ -11,6 +11,8 @@ class StatusData(XmlModel):
|
||||
SOURCEPACKAGE = "sourcepackage"
|
||||
TARGETPROJECT = "targetproject"
|
||||
TARGETPACKAGE = "targetpackage"
|
||||
TOKEN = "token"
|
||||
ID = "id"
|
||||
|
||||
name: NameEnum = Field(
|
||||
xml_attribute=True,
|
||||
|
||||
179
osc/obs_api/token.py
Normal file
179
osc/obs_api/token.py
Normal file
@ -0,0 +1,179 @@
|
||||
import textwrap
|
||||
|
||||
from ..util.models import * # pylint: disable=wildcard-import,unused-wildcard-import
|
||||
from .status import Status
|
||||
|
||||
|
||||
class Token(XmlModel):
|
||||
XML_TAG = "entry"
|
||||
|
||||
id: int = Field(
|
||||
xml_attribute=True,
|
||||
description=textwrap.dedent(
|
||||
"""
|
||||
The unique id of this token.
|
||||
"""
|
||||
),
|
||||
)
|
||||
|
||||
string: str = Field(
|
||||
xml_attribute=True,
|
||||
description=textwrap.dedent(
|
||||
"""
|
||||
The token secret. This string can be used instead of the password to
|
||||
authenticate the user or to trigger service runs via the
|
||||
`POST /trigger/runservice` route.
|
||||
"""
|
||||
),
|
||||
)
|
||||
|
||||
description: Optional[str] = Field(
|
||||
xml_attribute=True,
|
||||
description=textwrap.dedent(
|
||||
"""
|
||||
This attribute can be used to identify a token from the list of tokens
|
||||
of a user.
|
||||
"""
|
||||
),
|
||||
)
|
||||
|
||||
project: Optional[str] = Field(
|
||||
xml_attribute=True,
|
||||
description=textwrap.dedent(
|
||||
"""
|
||||
If this token is bound to a specific package, then the packages'
|
||||
project is available in this attribute.
|
||||
"""
|
||||
),
|
||||
)
|
||||
|
||||
package: Optional[str] = Field(
|
||||
xml_attribute=True,
|
||||
description=textwrap.dedent(
|
||||
"""
|
||||
The package name to which this token is bound, if it has been created
|
||||
for a specific package. Otherwise this attribute and the project
|
||||
attribute are omitted.
|
||||
"""
|
||||
),
|
||||
)
|
||||
|
||||
class Kind(str, Enum):
|
||||
RSS = "rss"
|
||||
REBUILD = "rebuild"
|
||||
RELEASE = "release"
|
||||
RUNSERVICE = "runservice"
|
||||
WORKFLOW = "workflow"
|
||||
|
||||
kind: Kind = Field(
|
||||
xml_attribute=True,
|
||||
description=textwrap.dedent(
|
||||
"""
|
||||
This attribute specifies which actions can be performed via this token.
|
||||
- rss: used to retrieve the notification RSS feed
|
||||
- rebuild: trigger rebuilds of packages
|
||||
- release: trigger project releases
|
||||
- runservice: run a service via the POST /trigger/runservice route
|
||||
- workflow: trigger SCM/CI workflows, see https://openbuildservice.org/help/manuals/obs-user-guide/cha.obs.scm_ci_workflow_integration.html
|
||||
"""
|
||||
),
|
||||
)
|
||||
|
||||
triggered_at: str = Field(
|
||||
xml_attribute=True,
|
||||
description=textwrap.dedent(
|
||||
"""
|
||||
The date and time a token got triggered the last time.
|
||||
"""
|
||||
),
|
||||
)
|
||||
|
||||
def to_human_readable_string(self) -> str:
|
||||
"""
|
||||
Render the object as a human readable string.
|
||||
"""
|
||||
from ..output import KeyValueTable
|
||||
|
||||
table = KeyValueTable()
|
||||
table.add("ID", str(self.id))
|
||||
table.add("String", self.string, color="bold")
|
||||
table.add("Operation", self.kind)
|
||||
table.add("Description", self.description)
|
||||
table.add("Project", self.project)
|
||||
table.add("Package", self.package)
|
||||
table.add("Triggered at", self.triggered_at)
|
||||
return f"{table}"
|
||||
|
||||
@classmethod
|
||||
def do_list(cls, apiurl: str, user: str):
|
||||
from ..util.xml import ET
|
||||
|
||||
url_path = ["person", user, "token"]
|
||||
url_query = {}
|
||||
response = cls.xml_request("GET", apiurl, url_path, url_query)
|
||||
root = ET.parse(response).getroot()
|
||||
assert root.tag == "directory"
|
||||
result = []
|
||||
for node in root:
|
||||
result.append(cls.from_xml(node, apiurl=apiurl))
|
||||
return result
|
||||
|
||||
@classmethod
|
||||
def cmd_create(
|
||||
cls,
|
||||
apiurl: str,
|
||||
user: str,
|
||||
*,
|
||||
operation: Optional[str] = None,
|
||||
project: Optional[str] = None,
|
||||
package: Optional[str] = None,
|
||||
scm_token: Optional[str] = None,
|
||||
):
|
||||
if operation == "workflow" and not scm_token:
|
||||
raise ValueError('``operation`` = "workflow" requires ``scm_token``')
|
||||
|
||||
url_path = ["person", user, "token"]
|
||||
url_query = {
|
||||
"cmd": "create",
|
||||
"operation": operation,
|
||||
"project": project,
|
||||
"package": package,
|
||||
"scm_token": scm_token,
|
||||
}
|
||||
response = cls.xml_request("POST", apiurl, url_path, url_query)
|
||||
return Status.from_file(response, apiurl=apiurl)
|
||||
|
||||
@classmethod
|
||||
def do_delete(cls, apiurl: str, user: str, token: str):
|
||||
url_path = ["person", user, "token", token]
|
||||
url_query = {}
|
||||
response = cls.xml_request("DELETE", apiurl, url_path, url_query)
|
||||
return Status.from_file(response, apiurl=apiurl)
|
||||
|
||||
@classmethod
|
||||
def do_trigger(
|
||||
cls,
|
||||
apiurl: str,
|
||||
token: str,
|
||||
*,
|
||||
operation: Optional[str] = None,
|
||||
project: Optional[str] = None,
|
||||
package: Optional[str] = None,
|
||||
):
|
||||
if operation:
|
||||
url_path = ["trigger", operation]
|
||||
else:
|
||||
url_path = ["trigger"]
|
||||
|
||||
url_query = {
|
||||
"project": project,
|
||||
"package": package,
|
||||
}
|
||||
|
||||
headers = {
|
||||
"Content-Type": "application/octet-stream",
|
||||
"Authorization": f"Token {token}",
|
||||
}
|
||||
|
||||
response = cls.xml_request("POST", apiurl, url_path, url_query, headers=headers)
|
||||
return Status.from_file(response, apiurl=apiurl)
|
||||
@ -756,12 +756,20 @@ class XmlModel(BaseModel):
|
||||
return obj
|
||||
|
||||
@classmethod
|
||||
def xml_request(cls, method: str, apiurl: str, path: List[str], query: Optional[dict] = None, data: Optional[str] = None) -> urllib3.response.HTTPResponse:
|
||||
def xml_request(
|
||||
cls,
|
||||
method: str,
|
||||
apiurl: str,
|
||||
path: List[str],
|
||||
query: Optional[dict] = None,
|
||||
headers: Optional[str] = None,
|
||||
data: Optional[str] = None,
|
||||
) -> urllib3.response.HTTPResponse:
|
||||
from ..connection import http_request
|
||||
from ..core import makeurl
|
||||
url = makeurl(apiurl, path, query)
|
||||
# TODO: catch HTTPError and return the wrapped response as XmlModel instance
|
||||
return http_request(method, url, data=data)
|
||||
return http_request(method, url, headers=headers, data=data)
|
||||
|
||||
def do_update(self, other: "XmlModel") -> None:
|
||||
"""
|
||||
|
||||
Loading…
Reference in New Issue
Block a user