Coverage for lobster/tools/codebeamer/codebeamer.py: 63%
309 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-15 06:41 +0000
1#!/usr/bin/env python3
2#
3# lobster_codebeamer - Extract codebeamer items for LOBSTER
4# Copyright (C) 2023-2025 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
5#
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: 93 ↛ 95line 93 didn't jump to line 95 because the condition on line 93 was always true
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
107 return response.text.strip() or "Unknown error"
110def _get_http_reason(response: requests.Response) -> str:
111 if response.reason: 111 ↛ 113line 111 didn't jump to line 113 because the condition on line 111 was always true
112 return response.reason
113 try:
114 return HTTPStatus(response.status_code).phrase
115 except ValueError:
116 return "Unknown Status"
119def query_cb_single(cb_config: Config, url: str):
120 if cb_config.num_request_retry <= 0: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true
121 raise ValueError("Retry is disabled (num_request_retry is set to 0). "
122 "Cannot proceed with retries.")
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 )
133 adapter = HTTPAdapter(max_retries=retry_strategy)
134 session = requests.Session()
135 session.mount("https://", adapter)
136 session.mount("http://", adapter)
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
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
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
175 if response.status_code == 200:
176 return response.json()
178 error_message = _get_response_message(response)
179 reason = _get_http_reason(response)
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 )
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)
196def get_many_items(cb_config: Config, item_ids: Iterable[int]):
197 rv = []
199 page_id = 1
200 query_string = quote(f"item.id IN "
201 f"({','.join(str(item_id) for item_id in item_ids)})")
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"]:
210 break
211 page_id += 1
213 return rv
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 )
222 rv = []
223 url = ""
224 page_id = 1
225 total_items = None
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 )
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.")
251 if page_id != data["page"]: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 raise MismatchException(
253 f"Page mismatch in query result: expected page "
254 f"{page_id} from codebeamer, but got {data['page']}"
255 )
257 if page_id == 1:
258 total_items = data["total"]
259 elif total_items != data["total"]: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
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 )
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"]]
274 page_id += 1
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 )
282 return rv
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.
290 If there is no match, it raises a KeyError.
292 Positional arguments:
293 cb_config -- configuration object containing the schema.
295 Returns:
296 A dictionary containing the namespace and class associated with the schema.
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()
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.")
311 return schema_map[schema]
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!")
320 # This looks like it's business logic, maybe we should make this
321 # configurable?
323 categories = cb_item.get("categories")
324 if categories: 324 ↛ 325line 324 didn't jump to line 325 because the condition on line 324 was never true
325 kind = categories[0].get("name", "codebeamer item")
326 else:
327 kind = "codebeamer item"
329 status = cb_item["status"].get("name", None) if "status" in cb_item else None
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: 333 ↛ 336line 333 didn't jump to line 336 because the condition on line 333 was always true
334 item_name = cb_item["name"]
335 else:
336 item_name = f"Unnamed item {cb_item['id']}"
338 schema_config = get_schema_config(cb_config)
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)
348 if cb_config.references:
349 for displayed_name in cb_config.references:
350 if cb_item.get(displayed_name): 350 ↛ 355line 350 didn't jump to line 355 because the condition on line 350 was always true
351 item_references = cb_item.get(displayed_name) if (
352 isinstance(cb_item.get(displayed_name), list)) \
353 else [cb_item.get(displayed_name)]
354 else:
355 item_references = [value for custom_field
356 in cb_item["customFields"]
357 if custom_field["name"] == displayed_name and
358 custom_field.get("values")
359 for value in custom_field["values"]]
361 for value in item_references:
362 item.add_tracing_target(Tracing_Tag("req", str(value["id"])))
364 return item
367def _create_common_params(namespace: str, cb_item: dict, cb_root: str,
368 item_name: str, kind: str):
369 """
370 Creates and returns common parameters for a Codebeamer item.
371 Args:
372 namespace (str): Namespace for the tag.
373 cb_item (dict): Codebeamer item dictionary.
374 cb_root (str): Root URL or path of Codebeamer.
375 item_name (str): Name of the item.
376 kind (str): Type of the item.
377 Returns:
378 dict: Common parameters including tag, location, and kind.
379 """
380 return {
381 'tag': Tracing_Tag(
382 namespace=namespace,
383 tag=str(cb_item["id"]),
384 version=cb_item["version"]
385 ),
386 'location': Codebeamer_Reference(
387 cb_root=cb_root,
388 tracker=cb_item["tracker"]["id"],
389 item=cb_item["id"],
390 version=cb_item["version"],
391 name=item_name
392 ),
393 'kind': kind
394 }
397def _create_lobster_item(schema_class, common_params, item_name, status):
398 """
399 Creates and returns a Lobster item based on the schema class.
400 Args:
401 schema_class: Class of the schema (Requirement, Implementation, Activity).
402 common_params (dict): Common parameters for the item.
403 item_name (str): Name of the item.
404 status (str): Status of the item.
405 Returns:
406 Object: An instance of the schema class with the appropriate parameters.
407 """
408 if schema_class is Requirement: 408 ↛ 417line 408 didn't jump to line 417 because the condition on line 408 was always true
409 return Requirement(
410 **common_params,
411 framework="codebeamer",
412 text=None,
413 status=status,
414 name= item_name
415 )
417 if schema_class is Implementation:
418 return Implementation(
419 **common_params,
420 language="python",
421 name= item_name,
422 )
424 if schema_class is Activity:
425 return Activity(
426 **common_params,
427 framework="codebeamer",
428 status=status
429 )
431 raise KeyError(f"Unsupported schema class '{schema_class}'!")
434def import_tagged(cb_config: Config, items_to_import: Iterable[int]):
435 rv = []
437 cb_items = get_many_items(cb_config, items_to_import)
438 for cb_item in cb_items:
439 l_item = to_lobster(cb_config, cb_item)
440 rv.append(l_item)
442 return rv
445def ensure_list(instance) -> List:
446 if isinstance(instance, list): 446 ↛ 448line 446 didn't jump to line 448 because the condition on line 446 was always true
447 return instance
448 return [instance]
451def update_authentication_parameters(
452 auth_conf: AuthenticationConfig,
453 netrc_path: Optional[str] = None):
454 if (auth_conf.token is None and 454 ↛ 456line 454 didn't jump to line 456 because the condition on line 454 was never true
455 (auth_conf.user is None or auth_conf.password is None)):
456 netrc_file = netrc_path or os.path.join(os.path.expanduser("~"),
457 ".netrc")
458 if os.path.isfile(netrc_file):
459 netrc_config = netrc.netrc(netrc_file)
460 machine = urlparse(auth_conf.root).hostname
461 auth = netrc_config.authenticators(machine)
462 if auth is not None:
463 print(f"Using .netrc login for {auth_conf.root}")
464 auth_conf.user, _, auth_conf.password = auth
465 else:
466 provided_machine = ", ".join(netrc_config.hosts.keys()) or "None"
467 raise KeyError(f"Error parsing .netrc file."
468 f"\nExpected '{machine}', but got '{provided_machine}'.")
470 if (auth_conf.token is None and 470 ↛ 472line 470 didn't jump to line 472 because the condition on line 470 was never true
471 (auth_conf.user is None or auth_conf.password is None)):
472 raise KeyError("Please add your token to the config file, "
473 "or use user and pass in the config file, "
474 "or configure credentials in the .netrc file.")
477def load_config(file_name: str) -> Config:
478 """
479 Parses a YAML configuration file and returns a validated configuration object.
481 Args:
482 file_name (str): Path to the YAML configuration file.
484 Returns:
485 Config: validated configuration.
487 Raises:
488 ValueError: If `file_name` is not a string.
489 FileNotFoundError: If the file does not exist.
490 KeyError: If required fields are missing or unsupported keys are present.
491 """
492 with open(file_name, encoding='utf-8') as file:
493 return parse_config_data(yaml.safe_load(file) or {})
496def parse_config_data(data: dict) -> Config:
497 # Validate supported keys
498 provided_config_keys = set(data.keys())
499 unsupported_keys = provided_config_keys - SupportedConfigKeys.as_set()
500 if unsupported_keys: 500 ↛ 501line 500 didn't jump to line 501 because the condition on line 500 was never true
501 raise KeyError(
502 f"Unsupported config keys: {', '.join(unsupported_keys)}. "
503 f"Supported keys are: {', '.join(SupportedConfigKeys.as_set())}."
504 )
506 # create config object
507 config = Config(
508 references=ensure_list(data.get(SupportedConfigKeys.REFS.value, [])),
509 import_tagged=data.get(SupportedConfigKeys.IMPORT_TAGGED.value),
510 import_query=data.get(SupportedConfigKeys.IMPORT_QUERY.value),
511 baseline_id=data.get(SupportedConfigKeys.BASELINE_ID.value),
512 verify_ssl=data.get(SupportedConfigKeys.VERIFY_SSL.value, True),
513 page_size=data.get(SupportedConfigKeys.PAGE_SIZE.value, 100),
514 schema=data.get(SupportedConfigKeys.SCHEMA.value, "Requirement"),
515 timeout=data.get(SupportedConfigKeys.TIMEOUT.value, 30),
516 out=data.get(SupportedConfigKeys.OUT.value),
517 num_request_retry=data.get(SupportedConfigKeys.NUM_REQUEST_RETRY.value, 5),
518 retry_error_codes=data.get(SupportedConfigKeys.RETRY_ERROR_CODES.value, []),
519 cb_auth_conf=AuthenticationConfig(
520 token=data.get(SupportedConfigKeys.CB_TOKEN.value),
521 user=data.get(SupportedConfigKeys.CB_USER.value),
522 password=data.get(SupportedConfigKeys.CB_PASS.value),
523 root=data.get(SupportedConfigKeys.CB_ROOT.value)
524 ),
525 )
527 # Ensure consistency of the configuration
528 if (not config.import_tagged) and (not config.import_query): 528 ↛ 529line 528 didn't jump to line 529 because the condition on line 528 was never true
529 raise KeyError(f"Either {SupportedConfigKeys.IMPORT_TAGGED.value} or "
530 f"{SupportedConfigKeys.IMPORT_QUERY.value} must be provided!")
532 if config.cb_auth_conf.root is None: 532 ↛ 533line 532 didn't jump to line 533 because the condition on line 532 was never true
533 raise KeyError(f"{SupportedConfigKeys.CB_ROOT.value} must be provided!")
535 if not config.cb_auth_conf.root.startswith("https://"): 535 ↛ 536line 535 didn't jump to line 536 because the condition on line 535 was never true
536 raise KeyError(f"{SupportedConfigKeys.CB_ROOT.value} must start with https://, "
537 f"but value is {config.cb_auth_conf.root}.")
539 if config.baseline_id is not None:
540 if config.import_tagged:
541 raise KeyError(
542 f"The keys {SupportedConfigKeys.BASELINE_ID.value} and "
543 f"{SupportedConfigKeys.IMPORT_TAGGED.value} are both present "
544 f"in the configuration, but they are mutually exclusive!"
545 )
546 if config.import_query and not isinstance(config.import_query, str):
547 raise KeyError(
548 f"The key {SupportedConfigKeys.BASELINE_ID.value} is only "
549 f"allowed if {SupportedConfigKeys.IMPORT_QUERY.value} is a "
550 f"cbQL query string, not a numeric report ID!"
551 )
552 try:
553 config.baseline_id = int(config.baseline_id)
554 except (TypeError, ValueError) as exc:
555 raise ValueError(
556 f"{SupportedConfigKeys.BASELINE_ID.value} must be a positive integer."
557 ) from exc
558 if config.baseline_id <= 0:
559 raise ValueError(
560 f"{SupportedConfigKeys.BASELINE_ID.value} must be a positive integer."
561 )
563 return config
566class CodebeamerTool(MetaDataToolBase):
567 def __init__(self):
568 super().__init__(
569 name="codebeamer",
570 description="Extract codebeamer items for LOBSTER",
571 official=True,
572 )
573 self._argument_parser.add_argument(
574 "--config",
575 help=(f"Path to YAML file with arguments, "
576 f"by default (codebeamer-config.yaml) "
577 f"supported references: '{', '.join(SupportedConfigKeys.as_set())}'"),
578 default=os.path.join(os.getcwd(), "codebeamer-config.yaml"))
580 self._argument_parser.add_argument(
581 "--out",
582 help=("Name of output file"),
583 default="codebeamer.lobster",
584 )
586 def _run_impl(self, options: argparse.Namespace) -> int:
587 try:
588 self._execute(options)
589 return 0
590 except NotFileException as ex:
591 print(ex)
592 except QueryException as query_ex:
593 print(query_ex)
594 except FileNotFoundError as file_ex:
595 self._print_error(f"File '{file_ex.filename}' not found.")
596 except IsADirectoryError as isdir_ex:
597 self._print_error(
598 f"Path '{isdir_ex.filename}' is a directory, but a file was expected.",
599 )
600 except ValueError as value_error:
601 self._print_error(value_error)
602 except KeyError as key_error:
603 self._print_error(key_error)
604 except LOBSTER_Error as lobster_error:
605 self._print_error(lobster_error)
607 return 1
609 @staticmethod
610 def _print_error(error: Union[Exception, str]):
611 print(f"{TOOL_NAME}: {error}", file=sys.stderr)
613 def _execute(self, options: argparse.Namespace) -> None:
614 mh = Message_Handler()
616 cb_config = load_config(options.config)
618 if cb_config.out is None: 618 ↛ 619line 618 didn't jump to line 619 because the condition on line 618 was never true
619 cb_config.out = options.out
621 update_authentication_parameters(cb_config.cb_auth_conf)
623 items_to_import = set()
625 if cb_config.import_tagged: 625 ↛ 626line 625 didn't jump to line 626 because the condition on line 625 was never true
626 source_items = {}
627 lobster_read(
628 mh = mh,
629 filename = cb_config.import_tagged,
630 level = "N/A",
631 items = source_items,
632 )
634 for item in source_items.values():
635 for tag in item.unresolved_references:
636 if tag.namespace != "req":
637 continue
638 try:
639 item_id = int(tag.tag, 10)
640 if item_id > 0:
641 items_to_import.add(item_id)
642 else:
643 mh.warning(item.location,
644 f"invalid codebeamer reference to {item_id}")
645 except ValueError:
646 mh.warning(
647 item.location,
648 f"cannot convert reference '{tag.tag}' to integer "
649 f"Codebeamer ID",
650 )
652 items = import_tagged(cb_config, items_to_import)
654 elif cb_config.import_query is not None: 654 ↛ 674line 654 didn't jump to line 674 because the condition on line 654 was always true
655 try:
656 if isinstance(cb_config.import_query, str):
657 if (cb_config.import_query.startswith("-") and 657 ↛ 659line 657 didn't jump to line 659 because the condition on line 657 was never true
658 cb_config.import_query[1:].isdigit()):
659 self._argument_parser.error(
660 "import_query must be a positive integer")
661 elif cb_config.import_query.startswith("-"): 661 ↛ 662line 661 didn't jump to line 662 because the condition on line 661 was never true
662 self._argument_parser.error(
663 "import_query must be a valid cbQL query")
664 elif cb_config.import_query == "": 664 ↛ 665line 664 didn't jump to line 665 because the condition on line 664 was never true
665 self._argument_parser.error(
666 "import_query must either be a query string or a query ID")
667 elif cb_config.import_query.isdigit(): 667 ↛ 668line 667 didn't jump to line 668 because the condition on line 667 was never true
668 cb_config.import_query = int(cb_config.import_query)
669 except ValueError as e:
670 self._argument_parser.error(str(e))
672 items = get_query(cb_config, cb_config.import_query)
673 else:
674 raise ValueError(
675 f"Unclear what to do, because neither "
676 f"'{SupportedConfigKeys.IMPORT_QUERY.value}' nor "
677 f"'{SupportedConfigKeys.IMPORT_TAGGED.value}' is specified!",
678 )
680 with _get_out_stream(cb_config.out) as out_stream:
681 _cb_items_to_lobster(items, cb_config, out_stream)
682 if cb_config.out: 682 ↛ exitline 682 didn't return from function '_execute' because the condition on line 682 was always true
683 print(f"Written {len(items)} requirements to {cb_config.out}")
686def _get_out_stream(config_out: Optional[str]) -> TextIO:
687 if config_out: 687 ↛ 690line 687 didn't jump to line 690 because the condition on line 687 was always true
688 ensure_output_directory(config_out)
689 return open(config_out, "w", encoding="UTF-8")
690 return sys.stdout
693def _cb_items_to_lobster(items: List[Dict], config: Config, out_file: TextIO) -> None:
694 schema_config = get_schema_config(config)
695 lobster_write(out_file, schema_config["class"], TOOL_NAME.replace("-", "_"), items)
698def lobster_codebeamer(config: Config, out_file: str) -> None:
699 """Loads items from codebeamer and serializes them in the LOBSTER interchange
700 format to the given file.
701 """
702 # This is an API function.
703 items = get_query(config, config.import_query)
704 ensure_output_directory(out_file)
705 with open(out_file, "w", encoding="UTF-8") as fd:
706 _cb_items_to_lobster(items, config, fd)
709def main(args: Optional[Sequence[str]] = None) -> int:
710 return CodebeamerTool().run(args)