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

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 VERIFY_SSL = "verify_ssl"

76 PAGE_SIZE = "page_size"

77 REFS = "refs"

78 SCHEMA = "schema"

79 CB_TOKEN = "token"

80 CB_ROOT = "root"

81 CB_USER = "user"

82 CB_PASS = "pass"

83 TIMEOUT = "timeout"

84 OUT = "out"

86 @classmethod

87 def as_set(cls) -> set:

88 return {parameter.value for parameter in cls}

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

92 if cb_auth_config.token: 92 ↛ 94line 92 didn't jump to line 94 because the condition on line 92 was always true

93 return BearerAuth(cb_auth_config.token)

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

95 cb_auth_config.password)

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

99 try:

100 data = response.json()

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

102 return data["message"]

103 except ValueError:

104 pass

105

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

107

108

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

110 if response.reason: 110 ↛ 112line 110 didn't jump to line 112 because the condition on line 110 was always true

111 return response.reason

112 try:

113 return HTTPStatus(response.status_code).phrase

114 except ValueError:

115 return "Unknown Status"

116

117

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

119 if cb_config.num_request_retry <= 0: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true

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

121 "Cannot proceed with retries.")

122

123 # Set up a Retry object with exponential backoff

124 retry_strategy = Retry(

125 total=cb_config.num_request_retry,

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

127 status_forcelist=cb_config.retry_error_codes,

128 allowed_methods=["GET"],

129 raise_on_status=False,

130 )

131

132 adapter = HTTPAdapter(max_retries=retry_strategy)

133 session = requests.Session()

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

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

136

137 try:

138 response = session.get(

139 url,

140 auth=get_authentication(cb_config.cb_auth_conf),

141 timeout=cb_config.timeout,

142 verify=cb_config.verify_ssl,

143 )

144 except Timeout as ex:

145 raise QueryException(

146 "Connection timed out while contacting Codebeamer\n"

147 f"URL: {url}\n"

148 f"Reason: {ex}\n"

149 "\nPossible actions:\n"

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

151 ) from ex

152

153 except RequestsConnectionError as ex:

154 raise QueryException(

155 "Unable to connect to Codebeamer\n"

156 f"URL: {url}\n"

157 f"Reason: {ex}\n"

158 "\nPossible actions:\n"

159 "• Check internet connection\n"

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

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

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

163 ) from ex

164

165 except RequestException as ex:

166 raise QueryException(

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

168 f"URL: {url}\n"

169 f"Reason: {ex}"

170 "\nPossible actions:\n"

171 "• Check network stability\n"

172 ) from ex

173

174 if response.status_code == 200:

175 return response.json()

176

177 error_message = _get_response_message(response)

178 reason = _get_http_reason(response)

179

180 raise QueryException(

181 "Codebeamer request failed:\n"

182 f" URL: {url}\n"

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

184 f"Reason: {error_message}"

185 )

186

187

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

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

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

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

192 return query_cb_single(cb_config, url)

193

194

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

196 rv = []

197

198 page_id = 1

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

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

201

202 while True:

203 base_url = "%s/items/query?page=%u&pageSize=%u&queryString=%s"\

204 % (cb_config.base, page_id,

205 cb_config.page_size, query_string)

206 data = query_cb_single(cb_config, base_url)

207 rv += data["items"]

208 if len(rv) == data["total"]:

209 break

210 page_id += 1

211

212 return rv

213

214

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

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

217 raise ValueError(

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

219 )

220

221 rv = []

222 url = ""

223 page_id = 1

224 total_items = None

225

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

227 print("Fetching page %u of query..." % page_id)

228 if isinstance(query, int):

229 url = ("%s/reports/%u/items?page=%u&pageSize=%u" %

230 (cb_config.base,

231 query,

232 page_id,

233 cb_config.page_size))

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

235 url = ("%s/items/query?page=%u&pageSize=%u&queryString=%s" %

236 (cb_config.base,

237 page_id,

238 cb_config.page_size,

239 query))

240 data = query_cb_single(cb_config, url)

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

242 raise MismatchException(

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

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

245 )

246

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

248 # lobster-trace: codebeamer_req.Get_Query_Zero_Items_Message

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

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

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

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

253

254 if page_id != data["page"]: 254 ↛ 255line 254 didn't jump to line 255 because the condition on line 254 was never true

255 raise MismatchException(

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

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

258 )

259

260 if page_id == 1:

261 total_items = data["total"]

262 elif total_items != data["total"]: 262 ↛ 263line 262 didn't jump to line 263 because the condition on line 262 was never true

263 raise MismatchException(

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

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

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

267 f"items in total."

268 )

269

270 if isinstance(query, int):

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

272 for cb_item in data["items"]]

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

274 rv += [to_lobster(cb_config, cb_item)

275 for cb_item in data["items"]]

276

277 page_id += 1

278

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

280 raise MismatchException(

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

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

283 )

284

285 return rv

286

287

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

289 """

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

291 in the cb_config dictionary.

292

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

294

295 Positional arguments:

296 cb_config -- configuration object containing the schema.

297

298 Returns:

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

300

301 Raises:

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

303 """

304 schema_map = {

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

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

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

308 }

309 schema = cb_config.schema.lower()

310

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

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

313

314 return schema_map[schema]

315

316

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

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

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

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

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

322

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

324 # configurable?

325

326 categories = cb_item.get("categories")

327 if categories: 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true

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

329 else:

330 kind = "codebeamer item"

331

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

333

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

335 # come up with one.

336 if "name" in cb_item: 336 ↛ 339line 336 didn't jump to line 339 because the condition on line 336 was always true

337 item_name = cb_item["name"]

338 else:

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

340

341 schema_config = get_schema_config(cb_config)

342

343 # Construct the appropriate object based on 'kind'

344 common_params = _create_common_params(

345 schema_config["namespace"], cb_item,

346 cb_config.cb_auth_conf.root, item_name, kind)

347 item = _create_lobster_item(

348 schema_config["class"],

349 common_params, item_name, status)

350

351 if cb_config.references:

352 for displayed_name in cb_config.references:

353 if cb_item.get(displayed_name): 353 ↛ 358line 353 didn't jump to line 358 because the condition on line 353 was always true

354 item_references = cb_item.get(displayed_name) if (

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

356 else [cb_item.get(displayed_name)]

357 else:

358 item_references = [value for custom_field

359 in cb_item["customFields"]

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

361 custom_field.get("values")

362 for value in custom_field["values"]]

363

364 for value in item_references:

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

366

367 return item

368

369

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

371 item_name: str, kind: str):

372 """

373 Creates and returns common parameters for a Codebeamer item.

374 Args:

375 namespace (str): Namespace for the tag.

376 cb_item (dict): Codebeamer item dictionary.

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

378 item_name (str): Name of the item.

379 kind (str): Type of the item.

380 Returns:

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

382 """

383 return {

384 'tag': Tracing_Tag(

385 namespace=namespace,

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

387 version=cb_item["version"]

388 ),

389 'location': Codebeamer_Reference(

390 cb_root=cb_root,

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

392 item=cb_item["id"],

393 version=cb_item["version"],

394 name=item_name

395 ),

396 'kind': kind

397 }

398

399

400def _create_lobster_item(schema_class, common_params, item_name, status):

401 """

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

403 Args:

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

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

406 item_name (str): Name of the item.

407 status (str): Status of the item.

408 Returns:

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

410 """

411 if schema_class is Requirement: 411 ↛ 420line 411 didn't jump to line 420 because the condition on line 411 was always true

412 return Requirement(

413 **common_params,

414 framework="codebeamer",

415 text=None,

416 status=status,

417 name= item_name

418 )

419

420 if schema_class is Implementation:

421 return Implementation(

422 **common_params,

423 language="python",

424 name= item_name,

425 )

426

427 if schema_class is Activity:

428 return Activity(

429 **common_params,

430 framework="codebeamer",

431 status=status

432 )

433

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

435

436

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

438 rv = []

439

440 cb_items = get_many_items(cb_config, items_to_import)

441 for cb_item in cb_items:

442 l_item = to_lobster(cb_config, cb_item)

443 rv.append(l_item)

444

445 return rv

446

447

448def ensure_list(instance) -> List:

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

450 return instance

451 return [instance]

452

453

454def update_authentication_parameters(

455 auth_conf: AuthenticationConfig,

456 netrc_path: Optional[str] = None):

457 if (auth_conf.token is None and 457 ↛ 459line 457 didn't jump to line 459 because the condition on line 457 was never true

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

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

460 ".netrc")

461 if os.path.isfile(netrc_file):

462 netrc_config = netrc.netrc(netrc_file)

463 machine = urlparse(auth_conf.root).hostname

464 auth = netrc_config.authenticators(machine)

465 if auth is not None:

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

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

468 else:

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

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

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

472

473 if (auth_conf.token is None and 473 ↛ 475line 473 didn't jump to line 475 because the condition on line 473 was never true

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

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

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

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

478

479

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

481 """

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

483

484 Args:

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

486

487 Returns:

488 Config: validated configuration.

489

490 Raises:

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

492 FileNotFoundError: If the file does not exist.

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

494 """

495 with open(file_name, "r", encoding='utf-8') as file:

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

497

498

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

500 # Validate supported keys

501 provided_config_keys = set(data.keys())

502 unsupported_keys = provided_config_keys - SupportedConfigKeys.as_set()

503 if unsupported_keys: 503 ↛ 504line 503 didn't jump to line 504 because the condition on line 503 was never true

504 raise KeyError(

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

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

507 )

508

509 # create config object

510 config = Config(

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

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

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

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

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

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

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

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

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

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

521 cb_auth_conf=AuthenticationConfig(

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

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

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

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

526 ),

527 )

528

529 # Ensure consistency of the configuration

530 if (not config.import_tagged) and (not config.import_query): 530 ↛ 531line 530 didn't jump to line 531 because the condition on line 530 was never true

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

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

533

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

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

536

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

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

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

540

541 return config

542

543

544class CodebeamerTool(MetaDataToolBase):

545 def __init__(self):

546 super().__init__(

547 name="codebeamer",

548 description="Extract codebeamer items for LOBSTER",

549 official=True,

550 )

551 self._argument_parser.add_argument(

552 "--config",

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

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

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

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

557

558 self._argument_parser.add_argument(

559 "--out",

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

561 default="codebeamer.lobster",

562 )

563

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

565 try:

566 self._execute(options)

567 return 0

568 except NotFileException as ex:

569 print(ex)

570 except QueryException as query_ex:

571 print(query_ex)

572 except FileNotFoundError as file_ex:

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

574 except IsADirectoryError as isdir_ex:

575 self._print_error(

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

577 )

578 except ValueError as value_error:

579 self._print_error(value_error)

580 except LOBSTER_Error as lobster_error:

581 self._print_error(lobster_error)

582

583 return 1

584

585 @staticmethod

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

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

588

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

590 mh = Message_Handler()

591

592 cb_config = load_config(options.config)

593

594 if cb_config.out is None: 594 ↛ 595line 594 didn't jump to line 595 because the condition on line 594 was never true

595 cb_config.out = options.out

596

597 update_authentication_parameters(cb_config.cb_auth_conf)

598

599 items_to_import = set()

600

601 if cb_config.import_tagged: 601 ↛ 602line 601 didn't jump to line 602 because the condition on line 601 was never true

602 source_items = {}

603 lobster_read(

604 mh = mh,

605 filename = cb_config.import_tagged,

606 level = "N/A",

607 items = source_items,

608 )

609

610 for item in source_items.values():

611 for tag in item.unresolved_references:

612 if tag.namespace != "req":

613 continue

614 try:

615 item_id = int(tag.tag, 10)

616 if item_id > 0:

617 items_to_import.add(item_id)

618 else:

619 mh.warning(item.location,

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

621 except ValueError:

622 mh.warning(

623 item.location,

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

625 f"Codebeamer ID",

626 )

627

628 items = import_tagged(cb_config, items_to_import)

629

630 elif cb_config.import_query is not None: 630 ↛ 650line 630 didn't jump to line 650 because the condition on line 630 was always true

631 try:

632 if isinstance(cb_config.import_query, str):

633 if (cb_config.import_query.startswith("-") and 633 ↛ 635line 633 didn't jump to line 635 because the condition on line 633 was never true

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

635 self._argument_parser.error(

636 "import_query must be a positive integer")

637 elif cb_config.import_query.startswith("-"): 637 ↛ 638line 637 didn't jump to line 638 because the condition on line 637 was never true

638 self._argument_parser.error(

639 "import_query must be a valid cbQL query")

640 elif cb_config.import_query == "": 640 ↛ 641line 640 didn't jump to line 641 because the condition on line 640 was never true

641 self._argument_parser.error(

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

643 elif cb_config.import_query.isdigit(): 643 ↛ 644line 643 didn't jump to line 644 because the condition on line 643 was never true

644 cb_config.import_query = int(cb_config.import_query)

645 except ValueError as e:

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

647

648 items = get_query(cb_config, cb_config.import_query)

649 else:

650 raise ValueError(

651 f"Unclear what to do, because neither "

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

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

654 )

655

656 with _get_out_stream(cb_config.out) as out_stream:

657 _cb_items_to_lobster(items, cb_config, out_stream)

658 if cb_config.out: 658 ↛ exitline 658 didn't return from function '_execute' because the condition on line 658 was always true

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

660

661

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

663 if config_out: 663 ↛ 666line 663 didn't jump to line 666 because the condition on line 663 was always true

664 ensure_output_directory(config_out)

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

666 return sys.stdout

667

668

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

670 schema_config = get_schema_config(config)

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

672

673

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

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

676 format to the given file.

677 """

678 # This is an API function.

679 items = get_query(config, config.import_query)

680 ensure_output_directory(out_file)

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

682 _cb_items_to_lobster(items, config, fd)

683

684

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

686 return CodebeamerTool().run(args)