Coverage for lobster/tools/codebeamer/codebeamer.py: 63%
309 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-08-27 12:15 +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,
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)
352 if cb_config.references:
353 for displayed_name in cb_config.references:
354 if cb_item.get(displayed_name): 354 ↛ 359line 354 didn't jump to line 359 because the condition on line 354 was always true
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"]]
365 for value in item_references:
366 item.add_tracing_target(Tracing_Tag("req", str(value["id"])))
368 return item
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 }
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: 415 ↛ 424line 415 didn't jump to line 424 because the condition on line 415 was always true
416 return Requirement(
417 **common_params,
418 framework="codebeamer",
419 text=text,
420 status=status,
421 name= item_name
422 )
424 if schema_class is Implementation:
425 return Implementation(
426 **common_params,
427 language="python",
428 name= item_name,
429 )
431 if schema_class is Activity:
432 return Activity(
433 **common_params,
434 framework="codebeamer",
435 text=text,
436 status=status
437 )
439 raise KeyError(f"Unsupported schema class '{schema_class}'!")
442def import_tagged(cb_config: Config, items_to_import: Iterable[int]):
443 rv = []
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)
450 return rv
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]
459def update_authentication_parameters(
460 auth_conf: AuthenticationConfig,
461 netrc_path: Optional[str] = None):
462 if (auth_conf.token is None and 462 ↛ 464line 462 didn't jump to line 464 because the condition on line 462 was never true
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}'.")
478 if (auth_conf.token is None and 478 ↛ 480line 478 didn't jump to line 480 because the condition on line 478 was never true
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.")
485def load_config(file_name: str) -> Config:
486 """
487 Parses a YAML configuration file and returns a validated configuration object.
489 Args:
490 file_name (str): Path to the YAML configuration file.
492 Returns:
493 Config: validated configuration.
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 {})
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: 508 ↛ 509line 508 didn't jump to line 509 because the condition on line 508 was never true
509 raise KeyError(
510 f"Unsupported config keys: {', '.join(unsupported_keys)}. "
511 f"Supported keys are: {', '.join(SupportedConfigKeys.as_set())}."
512 )
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 )
535 # Ensure consistency of the configuration
536 if (not config.import_tagged) and (not config.import_query): 536 ↛ 537line 536 didn't jump to line 537 because the condition on line 536 was never true
537 raise KeyError(f"Either {SupportedConfigKeys.IMPORT_TAGGED.value} or "
538 f"{SupportedConfigKeys.IMPORT_QUERY.value} must be provided!")
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!")
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}.")
547 if config.baseline_id is not None:
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 )
571 return config
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"))
588 self._argument_parser.add_argument(
589 "--out",
590 help=("Name of output file"),
591 default="codebeamer.lobster",
592 )
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)
615 return 1
617 @staticmethod
618 def _print_error(error: Union[Exception, str]):
619 print(f"{TOOL_NAME}: {error}", file=sys.stderr)
621 def _execute(self, options: argparse.Namespace) -> None:
622 mh = Message_Handler()
624 cb_config = load_config(options.config)
626 if cb_config.out is None: 626 ↛ 627line 626 didn't jump to line 627 because the condition on line 626 was never true
627 cb_config.out = options.out
629 update_authentication_parameters(cb_config.cb_auth_conf)
631 items_to_import = set()
633 if cb_config.import_tagged: 633 ↛ 634line 633 didn't jump to line 634 because the condition on line 633 was never true
634 source_items = {}
635 lobster_read(
636 mh = mh,
637 filename = cb_config.import_tagged,
638 level = "N/A",
639 items = source_items,
640 )
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 )
660 items = import_tagged(cb_config, items_to_import)
662 elif cb_config.import_query is not None: 662 ↛ 682line 662 didn't jump to line 682 because the condition on line 662 was always true
663 try:
664 if isinstance(cb_config.import_query, str):
665 if (cb_config.import_query.startswith("-") and 665 ↛ 667line 665 didn't jump to line 667 because the condition on line 665 was never true
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("-"): 669 ↛ 670line 669 didn't jump to line 670 because the condition on line 669 was never true
670 self._argument_parser.error(
671 "import_query must be a valid cbQL query")
672 elif cb_config.import_query == "": 672 ↛ 673line 672 didn't jump to line 673 because the condition on line 672 was never true
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(): 675 ↛ 676line 675 didn't jump to line 676 because the condition on line 675 was never true
676 cb_config.import_query = int(cb_config.import_query)
677 except ValueError as e:
678 self._argument_parser.error(str(e))
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 )
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: 690 ↛ exitline 690 didn't return from function '_execute' because the condition on line 690 was always true
691 print(f"Written {len(items)} requirements to {cb_config.out}")
694def _get_out_stream(config_out: Optional[str]) -> TextIO:
695 if config_out: 695 ↛ 698line 695 didn't jump to line 698 because the condition on line 695 was always true
696 ensure_output_directory(config_out)
697 return open(config_out, "w", encoding="UTF-8")
698 return sys.stdout
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)
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)
717def main(args: Optional[Sequence[str]] = None) -> int:
718 return CodebeamerTool().run(args)