Coverage for lobster/tools/codebeamer/codebeamer.py: 53%

1#!/usr/bin/env python3

3# lobster_codebeamer - Extract codebeamer items for LOBSTER

6# This program is free software: you can redistribute it and/or modify

7# it under the terms of the GNU Affero General Public License as

8# published by the Free Software Foundation, either version 3 of the

9# License, or (at your option) any later version.

10#

11# This program is distributed in the hope that it will be useful, but

12# WITHOUT ANY WARRANTY; without even the implied warranty of

13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU

14# Affero General Public License for more details.

15#

16# You should have received a copy of the GNU Affero General Public

17# License along with this program. If not, see

18# <https://www.gnu.org/licenses/>.

20# This tool is based on the codebeamer Rest API v3, as documented here:

21# https://codebeamer.com/cb/wiki/11631738

22#

23# There are some assumptions encoded here that are not clearly

24# documented, in that items have a type and the type has a name.

25#

26#

27# Main limitations:

28# * Item descriptions are ignored right now

29# * Branches (if they exist) are ignored

30# * We only ever fetch the HEAD item

31#

32# However you _can_ import all the items referenced from another

33# lobster artefact.

35import os

36import sys

37import argparse

38import netrc

39from typing import Dict, Iterable, List, Optional, Sequence, TextIO, Union

40from urllib.parse import quote, urlparse

41from enum import Enum

42from http import HTTPStatus

44import requests

45from requests.adapters import HTTPAdapter

46from requests.exceptions import (

47 Timeout,

48 ConnectionError as RequestsConnectionError,

49 RequestException,

50)

51import yaml

52from urllib3.util.retry import Retry

54from lobster.common.items import Tracing_Tag, Requirement, Implementation, Activity

55from lobster.common.location import Codebeamer_Reference

56from lobster.common.errors import Message_Handler, LOBSTER_Error

57from lobster.common.io import lobster_read, lobster_write, ensure_output_directory

58from lobster.common.meta_data_tool_base import MetaDataToolBase

59from lobster.tools.codebeamer.bearer_auth import BearerAuth

60from lobster.tools.codebeamer.config import AuthenticationConfig, Config

61from lobster.tools.codebeamer.exceptions import (

62 MismatchException, NotFileException, QueryException,

63)

66TOOL_NAME = "lobster-codebeamer"

69class SupportedConfigKeys(Enum):

70 """Helper class to define supported configuration keys."""

71 NUM_REQUEST_RETRY = "num_request_retry"

72 RETRY_ERROR_CODES = "retry_error_codes"

73 IMPORT_TAGGED = "import_tagged"

74 IMPORT_QUERY = "import_query"

75 BASELINE_ID = "baseline_id"

76 VERIFY_SSL = "verify_ssl"

77 PAGE_SIZE = "page_size"

78 REFS = "refs"

79 SCHEMA = "schema"

80 CB_TOKEN = "token"

81 CB_ROOT = "root"

82 CB_USER = "user"

83 CB_PASS = "pass"

84 TIMEOUT = "timeout"

85 OUT = "out"

87 @classmethod

88 def as_set(cls) -> set:

89 return {parameter.value for parameter in cls}

92def get_authentication(cb_auth_config: AuthenticationConfig) -> requests.auth.AuthBase:

93 if cb_auth_config.token:

94 return BearerAuth(cb_auth_config.token)

95 return requests.auth.HTTPBasicAuth(cb_auth_config.user,

96 cb_auth_config.password)

99def _get_response_message(response: requests.Response) -> str:

100 try:

101 data = response.json()

102 if isinstance(data, dict) and "message" in data:

103 return data["message"]

104 except ValueError:

105 pass

106

107 return response.text.strip() or "Unknown error"

108

109

110def _get_http_reason(response: requests.Response) -> str:

111 if response.reason:

112 return response.reason

113 try:

114 return HTTPStatus(response.status_code).phrase

115 except ValueError:

116 return "Unknown Status"

117

118

119def query_cb_single(cb_config: Config, url: str):

120 if cb_config.num_request_retry <= 0:

121 raise ValueError("Retry is disabled (num_request_retry is set to 0). "

122 "Cannot proceed with retries.")

123

124 # Set up a Retry object with exponential backoff

125 retry_strategy = Retry(

126 total=cb_config.num_request_retry,

127 backoff_factor=1, # Exponential backoff: 1s, 2s, 4s, etc.

128 status_forcelist=cb_config.retry_error_codes,

129 allowed_methods=["GET"],

130 raise_on_status=False,

131 )

132

133 adapter = HTTPAdapter(max_retries=retry_strategy)

134 session = requests.Session()

135 session.mount("https://", adapter)

136 session.mount("http://", adapter)

137

138 try:

139 response = session.get(

140 url,

141 auth=get_authentication(cb_config.cb_auth_conf),

142 timeout=cb_config.timeout,

143 verify=cb_config.verify_ssl,

144 )

145 except Timeout as ex:

146 raise QueryException(

147 "Connection timed out while contacting Codebeamer\n"

148 f"URL: {url}\n"

149 f"Reason: {ex}\n"

150 "\nPossible actions:\n"

151 "• Increase the timeout using the 'timeout' parameter"

152 ) from ex

153

154 except RequestsConnectionError as ex:

155 raise QueryException(

156 "Unable to connect to Codebeamer\n"

157 f"URL: {url}\n"

158 f"Reason: {ex}\n"

159 "\nPossible actions:\n"

160 "• Check internet connection\n"

161 "• Increase retries using 'num_request_retry'\n"

162 "• Check SSL certificates or disable verification by setting "

163 f"'{SupportedConfigKeys.VERIFY_SSL.value}' to false"

164 ) from ex

165

166 except RequestException as ex:

167 raise QueryException(

168 "Unexpected network error while connecting to Codebeamer\n"

169 f"URL: {url}\n"

170 f"Reason: {ex}"

171 "\nPossible actions:\n"

172 "• Check network stability\n"

173 ) from ex

174

175 if response.status_code == 200:

176 return response.json()

177

178 error_message = _get_response_message(response)

179 reason = _get_http_reason(response)

180

181 raise QueryException(

182 "Codebeamer request failed:\n"

183 f" URL: {url}\n"

184 f" HTTP Status: {response.status_code} ({reason})\n"

185 f"Reason: {error_message}"

186 )

187

188

189def get_single_item(cb_config: Config, item_id: int):

190 if not isinstance(item_id, int) or (item_id <= 0):

191 raise ValueError("item_id must be a positive integer")

192 url = f"{cb_config.base}/items/{item_id}"

193 return query_cb_single(cb_config, url)

194

195

196def get_many_items(cb_config: Config, item_ids: Iterable[int]):

197 rv = []

198

199 page_id = 1

200 query_string = quote(f"item.id IN "

201 f"({','.join(str(item_id) for item_id in item_ids)})")

202

203 while True:

204 base_url = (f"{cb_config.base}/items/query?page={page_id}"

205 f"&pageSize={cb_config.page_size}"

206 f"&queryString={query_string}")

207 data = query_cb_single(cb_config, base_url)

208 rv += data["items"]

209 if len(rv) == data["total"]: 209 ↛ 211line 209 didn't jump to line 211 because the condition on line 209 was always true

210 break

211 page_id += 1

212

213 return rv

214

215

216def get_query(cb_config: Config, query: Union[int, str]):

217 if (not query) or (not isinstance(query, (int, str))): 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true

218 raise ValueError(

219 "The query must either be a real positive integer or a non-empty string!",

220 )

221

222 rv = []

223 url = ""

224 page_id = 1

225 total_items = None

226

227 while total_items is None or len(rv) < total_items:

228 print(f"Fetching page {page_id} of query...")

229 if isinstance(query, int):

230 url = (f"{cb_config.base}/reports/{query}/items"

231 f"?page={page_id}&pageSize={cb_config.page_size}")

232 elif isinstance(query, str): 232 ↛ 237line 232 didn't jump to line 237 because the condition on line 232 was always true

233 url = (f"{cb_config.base}/items/query?page={page_id}"

234 f"&pageSize={cb_config.page_size}&queryString={query}")

235 if cb_config.baseline_id is not None:

236 url += f"&baselineId={cb_config.baseline_id}"

237 data = query_cb_single(cb_config, url)

238 if len(data) != 4: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true

239 raise MismatchException(

240 f"Expected codebeamer response with 4 data entries, but instead "

241 f"received {len(data)}!",

242 )

243

244 if page_id == 1 and len(data["items"]) == 0:

245 # lobster-trace: codebeamer_req.Get_Query_Zero_Items_Message

246 print("This query doesn't generate items. Please check:")

247 print(" * is the number actually correct?")

248 print(" * do you have permissions to access it?")

249 print(f"You can try to access '{url}' manually to check.")

250

251 if page_id != data["page"]:

252 raise MismatchException(

253 f"Page mismatch in query result: expected page "

254 f"{page_id} from codebeamer, but got {data['page']}"

255 )

256

257 if page_id == 1: 257 ↛ 259line 257 didn't jump to line 259 because the condition on line 257 was always true

258 total_items = data["total"]

259 elif total_items != data["total"]:

260 raise MismatchException(

261 f"Item count mismatch in query result: expected "

262 f"{total_items} items so far, but page "

263 f"{data['page']} claims to have sent {data['total']} "

264 f"items in total."

265 )

266

267 if isinstance(query, int):

268 rv += [to_lobster(cb_config, cb_item["item"])

269 for cb_item in data["items"]]

270 elif isinstance(query, str): 270 ↛ 274line 270 didn't jump to line 274 because the condition on line 270 was always true

271 rv += [to_lobster(cb_config, cb_item)

272 for cb_item in data["items"]]

273

274 page_id += 1

275

276 if total_items != len(rv): 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true

277 raise MismatchException(

278 f"Expected to receive {total_items} items in total from codebeamer, "

279 f"but actually received {len(rv)}!",

280 )

281

282 return rv

283

284

285def get_schema_config(cb_config: Config) -> dict:

286 """

287 The function returns a schema map based on the schema mentioned

288 in the cb_config dictionary.

289

290 If there is no match, it raises a KeyError.

291

292 Positional arguments:

293 cb_config -- configuration object containing the schema.

294

295 Returns:

296 A dictionary containing the namespace and class associated with the schema.

297

298 Raises:

299 KeyError -- if the provided schema is not supported.

300 """

301 schema_map = {

302 'requirement': {"namespace": "req", "class": Requirement},

303 'implementation': {"namespace": "imp", "class": Implementation},

304 'activity': {"namespace": "act", "class": Activity},

305 }

306 schema = cb_config.schema.lower()

307

308 if schema not in schema_map: 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true

309 raise KeyError(f"Unsupported SCHEMA '{schema}' provided in configuration.")

310

311 return schema_map[schema]

312

313

314def to_lobster(cb_config: Config, cb_item: dict):

315 if not isinstance(cb_item, dict): 315 ↛ 316line 315 didn't jump to line 316 because the condition on line 315 was never true

316 raise ValueError("'cb_item' must be of type 'dict'!")

317 if "id" not in cb_item: 317 ↛ 318line 317 didn't jump to line 318 because the condition on line 317 was never true

318 raise KeyError("Codebeamer item does not contain ID!")

319

320 # This looks like it's business logic, maybe we should make this

321 # configurable?

322

323 categories = cb_item.get("categories")

324 if categories:

325 kind = categories[0].get("name", "codebeamer item")

326 else:

327 kind = "codebeamer item"

328

329 status = cb_item["status"].get("name", None) if "status" in cb_item else None

330

331 # Get item name. Sometimes items do not have one, in which case we

332 # come up with one.

333 if "name" in cb_item:

334 item_name = cb_item["name"]

335 else:

336 item_name = f"Unnamed item {cb_item['id']}"

337

338 schema_config = get_schema_config(cb_config)

339

340 # Construct the appropriate object based on 'kind'

341 common_params = _create_common_params(

342 schema_config["namespace"], cb_item,

343 cb_config.cb_auth_conf.root, item_name, kind)

344 item = _create_lobster_item(

345 schema_config["class"],

346 common_params, item_name, status,

347 cb_config.item_to_text(cb_item)

348 if cb_config.item_to_text is not None and

349 schema_config["class"] in (Requirement, Activity)

350 else None)

351

352 if cb_config.references: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true

353 for displayed_name in cb_config.references:

354 if cb_item.get(displayed_name):

355 item_references = cb_item.get(displayed_name) if (

356 isinstance(cb_item.get(displayed_name), list)) \

357 else [cb_item.get(displayed_name)]

358 else:

359 item_references = [value for custom_field

360 in cb_item["customFields"]

361 if custom_field["name"] == displayed_name and

362 custom_field.get("values")

363 for value in custom_field["values"]]

364

365 for value in item_references:

366 item.add_tracing_target(Tracing_Tag("req", str(value["id"])))

367

368 return item

369

370

371def _create_common_params(namespace: str, cb_item: dict, cb_root: str,

372 item_name: str, kind: str):

373 """

374 Creates and returns common parameters for a Codebeamer item.

375 Args:

376 namespace (str): Namespace for the tag.

377 cb_item (dict): Codebeamer item dictionary.

378 cb_root (str): Root URL or path of Codebeamer.

379 item_name (str): Name of the item.

380 kind (str): Type of the item.

381 Returns:

382 dict: Common parameters including tag, location, and kind.

383 """

384 return {

385 'tag': Tracing_Tag(

386 namespace=namespace,

387 tag=str(cb_item["id"]),

388 version=cb_item["version"]

389 ),

390 'location': Codebeamer_Reference(

391 cb_root=cb_root,

392 tracker=cb_item["tracker"]["id"],

393 item=cb_item["id"],

394 version=cb_item["version"],

395 name=item_name

396 ),

397 'kind': kind

398 }

399

400

401def _create_lobster_item(schema_class, common_params, item_name, status,

402 text: Optional[str]):

403 """

404 Creates and returns a Lobster item based on the schema class.

405 Args:

406 schema_class: Class of the schema (Requirement, Implementation, Activity).

407 common_params (dict): Common parameters for the item.

408 item_name (str): Name of the item.

409 status (str): Status of the item.

410 text (str): Optional text generated from the Codebeamer item.

411 Is ignored for Implementation schema.

412 Returns:

413 Object: An instance of the schema class with the appropriate parameters.

414 """

415 if schema_class is Requirement:

416 return Requirement(

417 **common_params,

418 framework="codebeamer",

419 text=text,

420 status=status,

421 name= item_name

422 )

423

424 if schema_class is Implementation:

425 return Implementation(

426 **common_params,

427 language="python",

428 name= item_name,

429 )

430

431 if schema_class is Activity: 431 ↛ 439line 431 didn't jump to line 439 because the condition on line 431 was always true

432 return Activity(

433 **common_params,

434 framework="codebeamer",

435 text=text,

436 status=status

437 )

438

439 raise KeyError(f"Unsupported schema class '{schema_class}'!")

440

441

442def import_tagged(cb_config: Config, items_to_import: Iterable[int]):

443 rv = []

444

445 cb_items = get_many_items(cb_config, items_to_import)

446 for cb_item in cb_items:

447 l_item = to_lobster(cb_config, cb_item)

448 rv.append(l_item)

449

450 return rv

451

452

453def ensure_list(instance) -> List:

454 if isinstance(instance, list): 454 ↛ 456line 454 didn't jump to line 456 because the condition on line 454 was always true

455 return instance

456 return [instance]

457

458

459def update_authentication_parameters(

460 auth_conf: AuthenticationConfig,

461 netrc_path: Optional[str] = None):

462 if (auth_conf.token is None and

463 (auth_conf.user is None or auth_conf.password is None)):

464 netrc_file = netrc_path or os.path.join(os.path.expanduser("~"),

465 ".netrc")

466 if os.path.isfile(netrc_file):

467 netrc_config = netrc.netrc(netrc_file)

468 machine = urlparse(auth_conf.root).hostname

469 auth = netrc_config.authenticators(machine)

470 if auth is not None:

471 print(f"Using .netrc login for {auth_conf.root}")

472 auth_conf.user, _, auth_conf.password = auth

473 else:

474 provided_machine = ", ".join(netrc_config.hosts.keys()) or "None"

475 raise KeyError(f"Error parsing .netrc file."

476 f"\nExpected '{machine}', but got '{provided_machine}'.")

477

478 if (auth_conf.token is None and

479 (auth_conf.user is None or auth_conf.password is None)):

480 raise KeyError("Please add your token to the config file, "

481 "or use user and pass in the config file, "

482 "or configure credentials in the .netrc file.")

483

484

485def load_config(file_name: str) -> Config:

486 """

487 Parses a YAML configuration file and returns a validated configuration object.

488

489 Args:

490 file_name (str): Path to the YAML configuration file.

491

492 Returns:

493 Config: validated configuration.

494

495 Raises:

496 ValueError: If `file_name` is not a string.

497 FileNotFoundError: If the file does not exist.

498 KeyError: If required fields are missing or unsupported keys are present.

499 """

500 with open(file_name, encoding='utf-8') as file:

501 return parse_config_data(yaml.safe_load(file) or {})

502

503

504def parse_config_data(data: dict) -> Config:

505 # Validate supported keys

506 provided_config_keys = set(data.keys())

507 unsupported_keys = provided_config_keys - SupportedConfigKeys.as_set()

508 if unsupported_keys:

509 raise KeyError(

510 f"Unsupported config keys: {', '.join(unsupported_keys)}. "

511 f"Supported keys are: {', '.join(SupportedConfigKeys.as_set())}."

512 )

513

514 # create config object

515 config = Config(

516 references=ensure_list(data.get(SupportedConfigKeys.REFS.value, [])),

517 import_tagged=data.get(SupportedConfigKeys.IMPORT_TAGGED.value),

518 import_query=data.get(SupportedConfigKeys.IMPORT_QUERY.value),

519 baseline_id=data.get(SupportedConfigKeys.BASELINE_ID.value),

520 verify_ssl=data.get(SupportedConfigKeys.VERIFY_SSL.value, True),

521 page_size=data.get(SupportedConfigKeys.PAGE_SIZE.value, 100),

522 schema=data.get(SupportedConfigKeys.SCHEMA.value, "Requirement"),

523 timeout=data.get(SupportedConfigKeys.TIMEOUT.value, 30),

524 out=data.get(SupportedConfigKeys.OUT.value),

525 num_request_retry=data.get(SupportedConfigKeys.NUM_REQUEST_RETRY.value, 5),

526 retry_error_codes=data.get(SupportedConfigKeys.RETRY_ERROR_CODES.value, []),

527 cb_auth_conf=AuthenticationConfig(

528 token=data.get(SupportedConfigKeys.CB_TOKEN.value),

529 user=data.get(SupportedConfigKeys.CB_USER.value),

530 password=data.get(SupportedConfigKeys.CB_PASS.value),

531 root=data.get(SupportedConfigKeys.CB_ROOT.value)

532 ),

533 )

534

535 # Ensure consistency of the configuration

536 if (not config.import_tagged) and (not config.import_query):

537 raise KeyError(f"Either {SupportedConfigKeys.IMPORT_TAGGED.value} or "

538 f"{SupportedConfigKeys.IMPORT_QUERY.value} must be provided!")

539

540 if config.cb_auth_conf.root is None: 540 ↛ 541line 540 didn't jump to line 541 because the condition on line 540 was never true

541 raise KeyError(f"{SupportedConfigKeys.CB_ROOT.value} must be provided!")

542

543 if not config.cb_auth_conf.root.startswith("https://"): 543 ↛ 544line 543 didn't jump to line 544 because the condition on line 543 was never true

544 raise KeyError(f"{SupportedConfigKeys.CB_ROOT.value} must start with https://, "

545 f"but value is {config.cb_auth_conf.root}.")

546

547 if config.baseline_id is not None: 547 ↛ 548line 547 didn't jump to line 548 because the condition on line 547 was never true

548 if config.import_tagged:

549 raise KeyError(

550 f"The keys {SupportedConfigKeys.BASELINE_ID.value} and "

551 f"{SupportedConfigKeys.IMPORT_TAGGED.value} are both present "

552 f"in the configuration, but they are mutually exclusive!"

553 )

554 if config.import_query and not isinstance(config.import_query, str):

555 raise KeyError(

556 f"The key {SupportedConfigKeys.BASELINE_ID.value} is only "

557 f"allowed if {SupportedConfigKeys.IMPORT_QUERY.value} is a "

558 f"cbQL query string, not a numeric report ID!"

559 )

560 try:

561 config.baseline_id = int(config.baseline_id)

562 except (TypeError, ValueError) as exc:

563 raise ValueError(

564 f"{SupportedConfigKeys.BASELINE_ID.value} must be a positive integer."

565 ) from exc

566 if config.baseline_id <= 0:

567 raise ValueError(

568 f"{SupportedConfigKeys.BASELINE_ID.value} must be a positive integer."

569 )

570

571 return config

572

573

574class CodebeamerTool(MetaDataToolBase):

575 def __init__(self):

576 super().__init__(

577 name="codebeamer",

578 description="Extract codebeamer items for LOBSTER",

579 official=True,

580 )

581 self._argument_parser.add_argument(

582 "--config",

583 help=(f"Path to YAML file with arguments, "

584 f"by default (codebeamer-config.yaml) "

585 f"supported references: '{', '.join(SupportedConfigKeys.as_set())}'"),

586 default=os.path.join(os.getcwd(), "codebeamer-config.yaml"))

587

588 self._argument_parser.add_argument(

589 "--out",

590 help=("Name of output file"),

591 default="codebeamer.lobster",

592 )

593

594 def _run_impl(self, options: argparse.Namespace) -> int:

595 try:

596 self._execute(options)

597 return 0

598 except NotFileException as ex:

599 print(ex)

600 except QueryException as query_ex:

601 print(query_ex)

602 except FileNotFoundError as file_ex:

603 self._print_error(f"File '{file_ex.filename}' not found.")

604 except IsADirectoryError as isdir_ex:

605 self._print_error(

606 f"Path '{isdir_ex.filename}' is a directory, but a file was expected.",

607 )

608 except ValueError as value_error:

609 self._print_error(value_error)

610 except KeyError as key_error:

611 self._print_error(key_error)

612 except LOBSTER_Error as lobster_error:

613 self._print_error(lobster_error)

614

615 return 1

616

617 @staticmethod

618 def _print_error(error: Union[Exception, str]):

619 print(f"{TOOL_NAME}: {error}", file=sys.stderr)

620

621 def _execute(self, options: argparse.Namespace) -> None:

622 mh = Message_Handler()

623

624 cb_config = load_config(options.config)

625

626 if cb_config.out is None:

627 cb_config.out = options.out

628

629 update_authentication_parameters(cb_config.cb_auth_conf)

630

631 items_to_import = set()

632

633 if cb_config.import_tagged:

634 source_items = {}

635 lobster_read(

636 mh = mh,

637 filename = cb_config.import_tagged,

638 level = "N/A",

639 items = source_items,

640 )

641

642 for item in source_items.values():

643 for tag in item.unresolved_references:

644 if tag.namespace != "req":

645 continue

646 try:

647 item_id = int(tag.tag, 10)

648 if item_id > 0:

649 items_to_import.add(item_id)

650 else:

651 mh.warning(item.location,

652 f"invalid codebeamer reference to {item_id}")

653 except ValueError:

654 mh.warning(

655 item.location,

656 f"cannot convert reference '{tag.tag}' to integer "

657 f"Codebeamer ID",

658 )

659

660 items = import_tagged(cb_config, items_to_import)

661

662 elif cb_config.import_query is not None:

663 try:

664 if isinstance(cb_config.import_query, str):

665 if (cb_config.import_query.startswith("-") and

666 cb_config.import_query[1:].isdigit()):

667 self._argument_parser.error(

668 "import_query must be a positive integer")

669 elif cb_config.import_query.startswith("-"):

670 self._argument_parser.error(

671 "import_query must be a valid cbQL query")

672 elif cb_config.import_query == "":

673 self._argument_parser.error(

674 "import_query must either be a query string or a query ID")

675 elif cb_config.import_query.isdigit():

676 cb_config.import_query = int(cb_config.import_query)

677 except ValueError as e:

678 self._argument_parser.error(str(e))

679

680 items = get_query(cb_config, cb_config.import_query)

681 else:

682 raise ValueError(

683 f"Unclear what to do, because neither "

684 f"'{SupportedConfigKeys.IMPORT_QUERY.value}' nor "

685 f"'{SupportedConfigKeys.IMPORT_TAGGED.value}' is specified!",

686 )

687

688 with _get_out_stream(cb_config.out) as out_stream:

689 _cb_items_to_lobster(items, cb_config, out_stream)

690 if cb_config.out:

691 print(f"Written {len(items)} requirements to {cb_config.out}")

692

693

694def _get_out_stream(config_out: Optional[str]) -> TextIO:

695 if config_out:

696 ensure_output_directory(config_out)

697 return open(config_out, "w", encoding="UTF-8")

698 return sys.stdout

699

700

701def _cb_items_to_lobster(items: List[Dict], config: Config, out_file: TextIO) -> None:

702 schema_config = get_schema_config(config)

703 lobster_write(out_file, schema_config["class"], TOOL_NAME.replace("-", "_"), items)

704

705

706def lobster_codebeamer(config: Config, out_file: str) -> None:

707 """Loads items from codebeamer and serializes them in the LOBSTER interchange

708 format to the given file.

709 """

710 # This is an API function.

711 items = get_query(config, config.import_query)

712 ensure_output_directory(out_file)

713 with open(out_file, "w", encoding="UTF-8") as fd:

714 _cb_items_to_lobster(items, config, fd)

715

716

717def main(args: Optional[Sequence[str]] = None) -> int:

718 return CodebeamerTool().run(args)