|
26 | 26 | from __future__ import annotations |
27 | 27 |
|
28 | 28 | import re |
29 | | -from typing import Optional |
| 29 | +from typing import Optional, Union, Any |
30 | 30 | import json |
31 | 31 | from http import HTTPStatus |
32 | 32 | from threading import Lock |
|
37 | 37 | import sonar.util.constants as c |
38 | 38 |
|
39 | 39 | from sonar import aggregations, exceptions, applications, app_branches |
| 40 | +from sonar.projects import Project |
| 41 | +from sonar.branches import Branch |
| 42 | + |
40 | 43 | import sonar.permissions.permissions as perms |
41 | 44 | import sonar.permissions.portfolio_permissions as pperms |
42 | 45 | import sonar.sqobject as sq |
@@ -100,17 +103,17 @@ class Portfolio(aggregations.Aggregation): |
100 | 103 | def __init__(self, endpoint: pf.Platform, key: str, name: Optional[str] = None) -> None: |
101 | 104 | """Constructor, don't use - use class methods instead""" |
102 | 105 | super().__init__(endpoint=endpoint, key=key) |
103 | | - self.name = name if name is not None else key |
104 | | - self._selection_mode = {_SELECTION_MODE_NONE: True} #: Portfolio project selection mode |
105 | | - self._tags = [] #: Portfolio tags when selection mode is TAGS |
| 106 | + self.name: str = name if name is not None else key |
| 107 | + self._selection_mode: dict[str, Any] = {_SELECTION_MODE_NONE: True} #: Portfolio project selection mode |
| 108 | + self._tags: list[str] = [] #: Portfolio tags when selection mode is TAGS |
106 | 109 | self._description: Optional[str] = None #: Portfolio description |
107 | 110 | self._visibility: Optional[str] = None #: Portfolio visibility |
108 | | - self._applications = {} #: applications |
109 | | - self._permissions: Optional[object] = None #: Permissions |
110 | | - |
| 111 | + self._applications: dict[str, Any] = {} #: applications |
| 112 | + self._permissions: Optional[dict[str, list[str]]] = None #: Permissions |
| 113 | + self._projects: Optional[dict[str, set[str]]] = None #: Projects and branches in portfolio |
111 | 114 | self.parent_portfolio: Optional[Portfolio] = None #: Ref to parent portfolio object, if any |
112 | 115 | self.root_portfolio: Optional[Portfolio] = None #: Ref to root portfolio, if any |
113 | | - self._sub_portfolios = {} #: Subportfolios |
| 116 | + self._sub_portfolios: dict[str, Portfolio] = {} #: Subportfolios |
114 | 117 | Portfolio.CACHE.put(self) |
115 | 118 | log.debug("Created portfolio object name '%s'", name) |
116 | 119 |
|
@@ -168,15 +171,15 @@ def __str__(self) -> str: |
168 | 171 | """Returns string representation of object""" |
169 | 172 | return ( |
170 | 173 | f"subportfolio '{self.key}'" |
171 | | - if self.sq_json and self.sq_json.get("qualifier", _PORTFOLIO_QUALIFIER) == _SUBPORTFOLIO_QUALIFIER |
| 174 | + if self.sq_json.get("qualifier", _PORTFOLIO_QUALIFIER) == _SUBPORTFOLIO_QUALIFIER |
172 | 175 | else f"portfolio '{self.key}'" |
173 | 176 | ) |
174 | 177 |
|
175 | 178 | def reload(self, data: types.ApiPayload) -> None: |
176 | 179 | """Reloads a portfolio with returned API data""" |
177 | 180 | super().reload(data) |
178 | 181 | if "originalKey" not in data and data["qualifier"] == _PORTFOLIO_QUALIFIER: |
179 | | - self.parent_portfolio: Optional[object] = None |
| 182 | + self.parent_portfolio = None |
180 | 183 | self.root_portfolio = self |
181 | 184 | self.load_selection_mode() |
182 | 185 | self.reload_sub_portfolios() |
@@ -226,12 +229,25 @@ def url(self) -> str: |
226 | 229 | return f"{self.base_url(local=False)}/portfolio?id={self.key}" |
227 | 230 |
|
228 | 231 | def projects(self) -> Optional[dict[str, str]]: |
229 | | - """Returns list of projects and their branches if selection mode is manual, None otherwise""" |
| 232 | + """Returns list of projects and their branches if selection mode is manual, or the list of projects for other modes""" |
230 | 233 | if not self._selection_mode or _SELECTION_MODE_MANUAL not in self._selection_mode: |
231 | | - log.debug("%s: Not manual mode, no projects", str(self)) |
232 | | - return None |
| 234 | + if self._projects is None: |
| 235 | + data = json.loads(self.get("api/views/projects_status", params={"portfolio": self.key}).text) |
| 236 | + self._projects = {p["refKey"]: {c.DEFAULT_BRANCH} for p in data["projects"]} |
| 237 | + return self._projects |
233 | 238 | return self._selection_mode[_SELECTION_MODE_MANUAL] |
234 | 239 |
|
| 240 | + def components(self) -> list[Union[Project, Branch]]: |
| 241 | + """Returns the list of components objects (projects/branches) in the portfolio""" |
| 242 | + log.debug("Collecting portfolio components from %s", util.json_dump(self.sq_json)) |
| 243 | + new_comps = [] |
| 244 | + for p_key, p_branches in self.projects().items(): |
| 245 | + proj = Project.get_object(self.endpoint, p_key) |
| 246 | + for br in p_branches: |
| 247 | + # If br is DEFAULT_BRANCH, next() will find nothing and we'll append the project itself |
| 248 | + new_comps.append(next((b for b in proj.branches().values() if b.name == br), proj)) |
| 249 | + return new_comps |
| 250 | + |
235 | 251 | def applications(self) -> Optional[dict[str, str]]: |
236 | 252 | log.debug("Collecting portfolios applications from %s", util.json_dump(self.sq_json)) |
237 | 253 | if "subViews" not in self.sq_json: |
@@ -522,7 +538,7 @@ def add_application_branch(self, app_key: str, branch: str = c.DEFAULT_BRANCH) - |
522 | 538 | self._applications[app_key].append(branch) |
523 | 539 | return True |
524 | 540 |
|
525 | | - def add_subportfolio(self, key: str, name: Optional[str] = None, by_ref: bool = False) -> Portfolio: |
| 541 | + def add_subportfolio(self, key: str, name: str = None, by_ref: bool = False) -> Portfolio: |
526 | 542 | """Adds a subportfolio to a portfolio, defined by key, name and by reference option""" |
527 | 543 |
|
528 | 544 | log.info("Adding sub-portfolios to %s", str(self)) |
|
0 commit comments