diff --git a/behave/features/token.feature b/behave/features/token.feature
index 372c222ab..413ba827f 100644
--- a/behave/features/token.feature
+++ b/behave/features/token.feature
@@ -5,7 +5,6 @@ Scenario: Run `osc token` with no arguments
When I execute osc with args "token"
Then stdout is
"""
-
"""
@@ -24,11 +23,15 @@ Scenario: Run `osc token --operation rebuild`
Given I execute osc with args "token"
And stdout matches
"""
-
-
-
+ ID : 1
+ String : .*
+ Operation : rebuild
+ Description :
+ Project : test:factory
+ Package : test-pkgA
+ Triggered at :
"""
- And I search 'string="(?P[^"]+)' in stdout and store named groups in 'tokens'
+ And I search 'String *: *(?P.+)\n' in stdout and store named groups in 'tokens'
When I execute osc with args "token --trigger {context.tokens[0][token]}"
Then stdout is
"""
diff --git a/osc/commandline.py b/osc/commandline.py
index 04f1868e9..d08c5973a 100644
--- a/osc/commandline.py
+++ b/osc/commandline.py
@@ -1688,6 +1688,7 @@ def do_token(self, subcmd, opts, *args):
osc token --delete
osc token --trigger [--operation ] [ ]
"""
+ from . import obs_api
args = slash_split(args)
@@ -1696,60 +1697,51 @@ def do_token(self, subcmd, opts, *args):
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= 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')
diff --git a/osc/obs_api/__init__.py b/osc/obs_api/__init__.py
index 0084bff9a..9010c732a 100644
--- a/osc/obs_api/__init__.py
+++ b/osc/obs_api/__init__.py
@@ -4,3 +4,4 @@
from .person import Person
from .project import Project
from .request import Request
+from .token import Token
diff --git a/osc/obs_api/status_data.py b/osc/obs_api/status_data.py
index 234e91b6f..6d5abef70 100644
--- a/osc/obs_api/status_data.py
+++ b/osc/obs_api/status_data.py
@@ -11,6 +11,8 @@ class NameEnum(str, Enum):
SOURCEPACKAGE = "sourcepackage"
TARGETPROJECT = "targetproject"
TARGETPACKAGE = "targetpackage"
+ TOKEN = "token"
+ ID = "id"
name: NameEnum = Field(
xml_attribute=True,
diff --git a/osc/obs_api/token.py b/osc/obs_api/token.py
new file mode 100644
index 000000000..2a405c37a
--- /dev/null
+++ b/osc/obs_api/token.py
@@ -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)
diff --git a/osc/util/models.py b/osc/util/models.py
index be41e5b92..868440e18 100644
--- a/osc/util/models.py
+++ b/osc/util/models.py
@@ -756,12 +756,20 @@ def from_xml(cls, root: ET.Element, *, apiurl: Optional[str] = None):
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:
"""