Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions osc/conf.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@
import textwrap
from io import BytesIO
from io import StringIO
from typing import Optional, Tuple
from urllib.parse import urlsplit

from . import credentials
Expand Down Expand Up @@ -1571,7 +1572,10 @@ def sanitize_apiurl(apiurl):
return urljoin(*parse_apisrv_url(None, apiurl))


def parse_apisrv_url(scheme, apisrv):
def parse_apisrv_url(scheme: Optional[str], apisrv: str) -> Tuple[str, str, str]:
"""
Parses the api server url ``apisrv`` and returns the tuple (scheme, url, path).
"""
if apisrv.startswith('http://') or apisrv.startswith('https://'):
url = apisrv
elif scheme is not None:
Expand All @@ -1582,17 +1586,17 @@ def parse_apisrv_url(scheme, apisrv):
return scheme, url, path.rstrip('/')


def urljoin(scheme, apisrv, path=''):
return f"{scheme}://{apisrv}" + path
def urljoin(scheme: str, apisrv: str, path: str='') -> str:
return f"{scheme}://{apisrv}{path}"


def is_known_apiurl(url):
def is_known_apiurl(url: str) -> bool:
"""returns ``True`` if url is a known apiurl"""
apiurl = sanitize_apiurl(url)
return apiurl in config['api_host_options']


def extract_known_apiurl(url):
def extract_known_apiurl(url: str) -> Optional[str]:
"""
Return longest prefix of given url that is known apiurl,
None if there is no known apiurl that is prefix of given url.
Expand Down
78 changes: 69 additions & 9 deletions osc/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
from http.client import IncompleteRead
from io import StringIO
from pathlib import Path
from typing import Optional, Dict, Union, List, Iterable
from typing import TYPE_CHECKING, Any, Optional, Dict, Union, List, Iterable, Tuple, overload

Check notice

Code scanning / CodeQL

Unused import Note

Import of 'Any' is not used.
from urllib.parse import parse_qs, urlsplit, urlunsplit, urlparse, urlunparse, quote, urlencode, unquote
from urllib.error import HTTPError
from xml.etree import ElementTree as ET
Expand Down Expand Up @@ -1335,7 +1335,7 @@
raise


def show_package_meta(apiurl: str, prj: str, pac: str, meta=False, blame=None):
def show_package_meta(apiurl: str, prj: str, pac: str, meta: bool=False, blame=None) -> List[bytes]:
query: Dict[str, Union[str, int]] = {}
if meta:
query['meta'] = 1
Expand Down Expand Up @@ -1394,7 +1394,7 @@
return project_obj.scmsync


def show_devel_project(apiurl, prj, pac):
def show_devel_project(apiurl: str, prj: str, pac: str) -> Tuple[Optional[str], Optional[str]]:
from . import obs_api

package_obj = obs_api.Package.from_api(apiurl, prj, pac)
Expand Down Expand Up @@ -1521,7 +1521,7 @@
def __call__(self, **kwargs):
return self._delegate(**kwargs)

def __init__(self, url, input, change_is_required=False, file_ext='.xml', method=None):
def __init__(self, url, input: Union[List[str], str, bytes, List[bytes]], change_is_required=False, file_ext='.xml', method=None):
if isinstance(url, self._URLFactory):
self._url_factory = url
else:
Expand Down Expand Up @@ -1608,8 +1608,20 @@
os.unlink(self.filename)


if TYPE_CHECKING:
from typing import Literal, TypedDict

MetaType = Literal['prj', 'pkg', 'attribute', 'prjconf', 'user', 'group', 'pattern']

class _MetaType(TypedDict):
path: str
template: str
file_ext: str


# different types of metadata
metatypes = {'prj': {'path': 'source/%s/_meta',
metatypes: "Dict[MetaType, _MetaType]" = {
'prj': {'path': 'source/%s/_meta',
'template': new_project_templ,
'file_ext': '.xml'
},
Expand Down Expand Up @@ -1640,7 +1652,7 @@
}


def meta_exists(metatype: str, path_args=None, template_args=None, create_new=True, apiurl=None):
def meta_exists(metatype: "MetaType", path_args=None, template_args=None, create_new=True, apiurl=None):

global metatypes

Expand All @@ -1661,7 +1673,7 @@


def make_meta_url(
metatype: str,
metatype: "MetaType",
path_args=None,
apiurl: Optional[str] = None,
force=False,
Expand Down Expand Up @@ -1704,7 +1716,7 @@


def edit_meta(
metatype,
metatype: "MetaType",
path_args=None,
data: Optional[List[str]] = None,
template_args=None,
Expand Down Expand Up @@ -3684,6 +3696,7 @@
return r


@overload
def branch_pkg(
apiurl: str,
src_project: str,
Expand All @@ -3705,7 +3718,54 @@
maintenance=False,
newinstance=False,
disable_build=False,
):
) -> Tuple[False, Optional[str], Optional[str], Optional[str], Optional[str]]: ...

Check notice

Code scanning / CodeQL

Statement has no effect Note

This statement has no effect.

@overload
def branch_pkg(
apiurl: str,
src_project: str,
src_package: str,
nodevelproject=False,
rev=None,
linkrev=None,
target_project: Optional[str] = None,
target_package=None,
return_existing=True,
msg="",
force=False,
noaccess=False,
add_repositories=False,
add_repositories_block=None,
add_repositories_rebuild=None,
extend_package_names=False,
missingok=False,
maintenance=False,
newinstance=False,
disable_build=False,
) -> Union[Tuple[True, str, str, None, None], Tuple[False, Optional[str], Optional[str], Optional[str], Optional[str]]]: ...

Check notice

Code scanning / CodeQL

Statement has no effect Note

This statement has no effect.

def branch_pkg(
apiurl: str,
src_project: str,
src_package: str,
nodevelproject=False,
rev=None,
linkrev=None,
target_project: Optional[str] = None,
target_package=None,
return_existing=False,
msg="",
force=False,
noaccess=False,
add_repositories=False,
add_repositories_block=None,
add_repositories_rebuild=None,
extend_package_names=False,
missingok=False,
maintenance=False,
newinstance=False,
disable_build=False,
) -> Union[Tuple[True, str, str, None, None], Tuple[False, Optional[str], Optional[str], Optional[str], Optional[str]]]:
"""
Branch a package (via API call)
"""
Expand Down Expand Up @@ -6022,7 +6082,7 @@
safe_get_rpmlint_log(src_actions)
elif repl == 'bl' and src_actions:
get_build_log(src_actions)
elif repl == 'e' and editable_actions:
# this is only for editable actions
if not editprj:
editprj = clone_request(apiurl, request.reqid, 'osc editrequest')
Expand All @@ -6046,7 +6106,7 @@
footer = 'changing request from state \'%s\' to \'%s\'\n\n' \
% (request.state.name, state)
msg_template = change_request_state_template(request, state)
if tmpfile is None:
footer += str(request)
if tmpfile is not None:
tmpfile.seek(0)
Expand Down
Loading