Coverage for lobster/tools/codebeamer/codebeamer.py: 55%
307 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 07:49 +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:
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:
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:
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_many_items(cb_config: Config, item_ids: Iterable[int]):
190 rv = []
192 page_id = 1
193 query_string = quote(f"item.id IN "
194 f"({','.join(str(item_id) for item_id in item_ids)})")
196 while True:
197 base_url = (f"{cb_config.base}/items/query?page={page_id}"
198 f"&pageSize={cb_config.page_size}"
199 f"&queryString={query_string}")
200 data = query_cb_single(cb_config, base_url)
201 rv += data["items"]
202 if len(rv) == data["total"]: 202 ↛ 204line 202 didn't jump to line 204 because the condition on line 202 was always true
203 break
204 page_id += 1
206 return rv
209def get_query(cb_config: Config, query: Union[int, str]):
210 if (not query) or (not isinstance(query, (int, str))): 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true
211 raise ValueError(
212 "The query must either be a real positive integer or a non-empty string!",
213 )
215 rv = []
216 url = ""
217 page_id = 1
218 total_items = None
220 while total_items is None or len(rv) < total_items:
221 print(f"Fetching page {page_id} of query...")
222 if isinstance(query, int):
223 url = (f"{cb_config.base}/reports/{query}/items"
224 f"?page={page_id}&pageSize={cb_config.page_size}")
225 elif isinstance(query, str): 225 ↛ 230line 225 didn't jump to line 230 because the condition on line 225 was always true
226 url = (f"{cb_config.base}/items/query?page={page_id}"
227 f"&pageSize={cb_config.page_size}&queryString={query}")
228 if cb_config.baseline_id is not None:
229 url += f"&baselineId={cb_config.baseline_id}"
230 data = query_cb_single(cb_config, url)
231 if len(data) != 4: 231 ↛ 232line 231 didn't jump to line 232 because the condition on line 231 was never true
232 raise MismatchException(
233 f"Expected codebeamer response with 4 data entries, but instead "
234 f"received {len(data)}!",
235 )
237 if page_id == 1 and len(data["items"]) == 0:
238 # lobster-trace: codebeamer_req.Get_Query_Zero_Items_Message
239 print("This query doesn't generate items. Please check:")
240 print(" * is the number actually correct?")
241 print(" * do you have permissions to access it?")
242 print(f"You can try to access '{url}' manually to check.")
244 if page_id != data["page"]:
245 raise MismatchException(
246 f"Page mismatch in query result: expected page "
247 f"{page_id} from codebeamer, but got {data['page']}"
248 )
250 if page_id == 1: 250 ↛ 252line 250 didn't jump to line 252 because the condition on line 250 was always true
251 total_items = data["total"]
252 elif total_items != data["total"]:
253 raise MismatchException(
254 f"Item count mismatch in query result: expected "
255 f"{total_items} items so far, but page "
256 f"{data['page']} claims to have sent {data['total']} "
257 f"items in total."
258 )
260 if isinstance(query, int):
261 rv += [to_lobster(cb_config, cb_item["item"])
262 for cb_item in data["items"]]
263 elif isinstance(query, str): 263 ↛ 267line 263 didn't jump to line 267 because the condition on line 263 was always true
264 rv += [to_lobster(cb_config, cb_item)
265 for cb_item in data["items"]]
267 page_id += 1
269 if total_items != len(rv): 269 ↛ 270line 269 didn't jump to line 270 because the condition on line 269 was never true
270 raise MismatchException(
271 f"Expected to receive {total_items} items in total from codebeamer, "
272 f"but actually received {len(rv)}!",
273 )
275 return rv
278def get_schema_config(cb_config: Config) -> dict:
279 """
280 The function returns a schema map based on the schema mentioned
281 in the cb_config dictionary.
283 If there is no match, it raises a KeyError.
285 Positional arguments:
286 cb_config -- configuration object containing the schema.
288 Returns:
289 A dictionary containing the namespace and class associated with the schema.
291 Raises:
292 KeyError -- if the provided schema is not supported.
293 """
294 schema_map = {
295 'requirement': {"namespace": "req", "class": Requirement},
296 'implementation': {"namespace": "imp", "class": Implementation},
297 'activity': {"namespace": "act", "class": Activity},
298 }
299 schema = cb_config.schema.lower()
301 if schema not in schema_map: 301 ↛ 302line 301 didn't jump to line 302 because the condition on line 301 was never true
302 raise KeyError(f"Unsupported SCHEMA '{schema}' provided in configuration.")
304 return schema_map[schema]
307def to_lobster(cb_config: Config, cb_item: dict):
308 if not isinstance(cb_item, dict): 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 raise ValueError("'cb_item' must be of type 'dict'!")
310 if "id" not in cb_item: 310 ↛ 311line 310 didn't jump to line 311 because the condition on line 310 was never true
311 raise KeyError("Codebeamer item does not contain ID!")
313 # This looks like it's business logic, maybe we should make this
314 # configurable?
316 categories = cb_item.get("categories")
317 if categories:
318 kind = categories[0].get("name", "codebeamer item")
319 else:
320 kind = "codebeamer item"
322 status = cb_item["status"].get("name", None) if "status" in cb_item else None
324 # Get item name. Sometimes items do not have one, in which case we
325 # come up with one.
326 if "name" in cb_item:
327 item_name = cb_item["name"]
328 else:
329 item_name = f"Unnamed item {cb_item['id']}"
331 schema_config = get_schema_config(cb_config)
333 # Construct the appropriate object based on 'kind'
334 common_params = _create_common_params(
335 schema_config["namespace"], cb_item,
336 cb_config.cb_auth_conf.root, item_name, kind)
337 item = _create_lobster_item(
338 schema_config["class"],
339 common_params, item_name, status,
340 cb_config.item_to_text(cb_item)
341 if cb_config.item_to_text is not None and
342 schema_config["class"] in (Requirement, Activity)
343 else None)
345 if cb_config.references:
346 for displayed_name in cb_config.references:
347 if cb_item.get(displayed_name):
348 item_references = cb_item.get(displayed_name) if (
349 isinstance(cb_item.get(displayed_name), list)) \
350 else [cb_item.get(displayed_name)]
351 else:
352 item_references = [value for custom_field
353 in cb_item["customFields"]
354 if custom_field["name"] == displayed_name and
355 custom_field.get("values")
356 for value in custom_field["values"]]
358 for value in item_references:
359 item.add_tracing_target(Tracing_Tag("req", str(value["id"])))
361 return item
364def _create_common_params(namespace: str, cb_item: dict, cb_root: str,
365 item_name: str, kind: str):
366 """
367 Creates and returns common parameters for a Codebeamer item.
368 Args:
369 namespace (str): Namespace for the tag.
370 cb_item (dict): Codebeamer item dictionary.
371 cb_root (str): Root URL or path of Codebeamer.
372 item_name (str): Name of the item.
373 kind (str): Type of the item.
374 Returns:
375 dict: Common parameters including tag, location, and kind.
376 """
377 return {
378 'tag': Tracing_Tag(
379 namespace=namespace,
380 tag=str(cb_item["id"]),
381 version=cb_item["version"]
382 ),
383 'location': Codebeamer_Reference(
384 cb_root=cb_root,
385 tracker=cb_item["tracker"]["id"],
386 item=cb_item["id"],
387 version=cb_item["version"],
388 name=item_name
389 ),
390 'kind': kind
391 }
394def _create_lobster_item(schema_class, common_params, item_name, status,
395 text: Optional[str]):
396 """
397 Creates and returns a Lobster item based on the schema class.
398 Args:
399 schema_class: Class of the schema (Requirement, Implementation, Activity).
400 common_params (dict): Common parameters for the item.
401 item_name (str): Name of the item.
402 status (str): Status of the item.
403 text (str): Optional text generated from the Codebeamer item.
404 Is ignored for Implementation schema.
405 Returns:
406 Object: An instance of the schema class with the appropriate parameters.
407 """
408 if schema_class is Requirement:
409 return Requirement(
410 **common_params,
411 framework="codebeamer",
412 text=text,
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: 424 ↛ 432line 424 didn't jump to line 432 because the condition on line 424 was always true
425 return Activity(
426 **common_params,
427 framework="codebeamer",
428 text=text,
429 status=status
430 )
432 raise KeyError(f"Unsupported schema class '{schema_class}'!")
435def import_tagged(cb_config: Config, items_to_import: Iterable[int]):
436 rv = []
438 cb_items = get_many_items(cb_config, items_to_import)
439 for cb_item in cb_items:
440 l_item = to_lobster(cb_config, cb_item)
441 rv.append(l_item)
443 return rv
446def ensure_list(instance) -> List:
447 if isinstance(instance, list): 447 ↛ 449line 447 didn't jump to line 449 because the condition on line 447 was always true
448 return instance
449 return [instance]
452def update_authentication_parameters(
453 auth_conf: AuthenticationConfig,
454 netrc_path: Optional[str] = None):
455 if (auth_conf.token is None and
456 (auth_conf.user is None or auth_conf.password is None)):
457 netrc_file = netrc_path or os.path.join(os.path.expanduser("~"),
458 ".netrc")
459 if os.path.isfile(netrc_file):
460 netrc_config = netrc.netrc(netrc_file)
461 machine = urlparse(auth_conf.root).hostname
462 auth = netrc_config.authenticators(machine)
463 if auth is not None:
464 print(f"Using .netrc login for {auth_conf.root}")
465 auth_conf.user, _, auth_conf.password = auth
466 else:
467 provided_machine = ", ".join(netrc_config.hosts.keys()) or "None"
468 raise KeyError(f"Error parsing .netrc file."
469 f"\nExpected '{machine}', but got '{provided_machine}'.")
471 if (auth_conf.token is None and
472 (auth_conf.user is None or auth_conf.password is None)):
473 raise KeyError("Please add your token to the config file, "
474 "or use user and pass in the config file, "
475 "or configure credentials in the .netrc file.")
478def load_config(file_name: str) -> Config:
479 """
480 Parses a YAML configuration file and returns a validated configuration object.
482 Args:
483 file_name (str): Path to the YAML configuration file.
485 Returns:
486 Config: validated configuration.
488 Raises:
489 ValueError: If `file_name` is not a string.
490 FileNotFoundError: If the file does not exist.
491 KeyError: If required fields are missing or unsupported keys are present.
492 """
493 with open(file_name, encoding='utf-8') as file:
494 return parse_config_data(yaml.safe_load(file) or {})
497def parse_config_data(data: dict) -> Config:
498 # Validate supported keys
499 provided_config_keys = set(data.keys())
500 unsupported_keys = provided_config_keys - SupportedConfigKeys.as_set()
501 if unsupported_keys:
502 raise KeyError(
503 f"Unsupported config keys: {', '.join(unsupported_keys)}. "
504 f"Supported keys are: {', '.join(SupportedConfigKeys.as_set())}."
505 )
507 # create config object
508 config = Config(
509 references=ensure_list(data.get(SupportedConfigKeys.REFS.value, [])),
510 import_tagged=data.get(SupportedConfigKeys.IMPORT_TAGGED.value),
511 import_query=data.get(SupportedConfigKeys.IMPORT_QUERY.value),
512 baseline_id=data.get(SupportedConfigKeys.BASELINE_ID.value),
513 verify_ssl=data.get(SupportedConfigKeys.VERIFY_SSL.value, True),
514 page_size=data.get(SupportedConfigKeys.PAGE_SIZE.value, 100),
515 schema=data.get(SupportedConfigKeys.SCHEMA.value, "Requirement"),
516 timeout=data.get(SupportedConfigKeys.TIMEOUT.value, 30),
517 out=data.get(SupportedConfigKeys.OUT.value),
518 num_request_retry=data.get(SupportedConfigKeys.NUM_REQUEST_RETRY.value, 5),
519 retry_error_codes=data.get(SupportedConfigKeys.RETRY_ERROR_CODES.value, []),
520 cb_auth_conf=AuthenticationConfig(
521 token=data.get(SupportedConfigKeys.CB_TOKEN.value),
522 user=data.get(SupportedConfigKeys.CB_USER.value),
523 password=data.get(SupportedConfigKeys.CB_PASS.value),
524 root=data.get(SupportedConfigKeys.CB_ROOT.value)
525 ),
526 )
528 # Ensure consistency of the configuration
529 if (not config.import_tagged) and (not config.import_query):
530 raise KeyError(f"Either {SupportedConfigKeys.IMPORT_TAGGED.value} or "
531 f"{SupportedConfigKeys.IMPORT_QUERY.value} must be provided!")
533 if config.cb_auth_conf.root is None: 533 ↛ 534line 533 didn't jump to line 534 because the condition on line 533 was never true
534 raise KeyError(f"{SupportedConfigKeys.CB_ROOT.value} must be provided!")
536 if not config.cb_auth_conf.root.startswith("https://"): 536 ↛ 537line 536 didn't jump to line 537 because the condition on line 536 was never true
537 raise KeyError(f"{SupportedConfigKeys.CB_ROOT.value} must start with https://, "
538 f"but value is {config.cb_auth_conf.root}.")
540 if config.baseline_id is not None: 540 ↛ 541line 540 didn't jump to line 541 because the condition on line 540 was never true
541 if config.import_tagged:
542 raise KeyError(
543 f"The keys {SupportedConfigKeys.BASELINE_ID.value} and "
544 f"{SupportedConfigKeys.IMPORT_TAGGED.value} are both present "
545 f"in the configuration, but they are mutually exclusive!"
546 )
547 if config.import_query and not isinstance(config.import_query, str):
548 raise KeyError(
549 f"The key {SupportedConfigKeys.BASELINE_ID.value} is only "
550 f"allowed if {SupportedConfigKeys.IMPORT_QUERY.value} is a "
551 f"cbQL query string, not a numeric report ID!"
552 )
553 try:
554 config.baseline_id = int(config.baseline_id)
555 except (TypeError, ValueError) as exc:
556 raise ValueError(
557 f"{SupportedConfigKeys.BASELINE_ID.value} must be a positive integer."
558 ) from exc
559 if config.baseline_id <= 0:
560 raise ValueError(
561 f"{SupportedConfigKeys.BASELINE_ID.value} must be a positive integer."
562 )
564 return config
567class CodebeamerTool(MetaDataToolBase):
568 def __init__(self):
569 super().__init__(
570 name="codebeamer",
571 description="Extract codebeamer items for LOBSTER",
572 official=True,
573 )
574 self._argument_parser.add_argument(
575 "--config",
576 help=(f"Path to YAML file with arguments, "
577 f"by default (codebeamer-config.yaml) "
578 f"supported references: '{', '.join(SupportedConfigKeys.as_set())}'"),
579 default=os.path.join(os.getcwd(), "codebeamer-config.yaml"))
581 self._argument_parser.add_argument(
582 "--out",
583 help=("Name of output file"),
584 default="codebeamer.lobster",
585 )
587 def _run_impl(self, options: argparse.Namespace) -> int:
588 try:
589 self._execute(options)
590 return 0
591 except NotFileException as ex:
592 print(ex)
593 except QueryException as query_ex:
594 print(query_ex)
595 except FileNotFoundError as file_ex:
596 self._print_error(f"File '{file_ex.filename}' not found.")
597 except IsADirectoryError as isdir_ex:
598 self._print_error(
599 f"Path '{isdir_ex.filename}' is a directory, but a file was expected.",
600 )
601 except PermissionError as perm_ex:
602 # Windows raises PermissionError instead of IsADirectoryError.
603 self._print_error(
604 f"Permission denied for '{perm_ex.filename}'.",
605 )
606 except ValueError as value_error:
607 self._print_error(value_error)
608 except KeyError as key_error:
609 self._print_error(key_error)
610 except LOBSTER_Error as lobster_error:
611 self._print_error(lobster_error)
613 return 1
615 @staticmethod
616 def _print_error(error: Union[Exception, str]):
617 print(f"{TOOL_NAME}: {error}", file=sys.stderr)
619 def _execute(self, options: argparse.Namespace) -> None:
620 mh = Message_Handler()
622 cb_config = load_config(options.config)
624 if cb_config.out is None:
625 cb_config.out = options.out
627 update_authentication_parameters(cb_config.cb_auth_conf)
629 items_to_import = set()
631 if cb_config.import_tagged:
632 source_items = {}
633 lobster_read(
634 mh = mh,
635 filename = cb_config.import_tagged,
636 level = "N/A",
637 items = source_items,
638 )
640 for item in source_items.values():
641 for tag in item.unresolved_references:
642 if tag.namespace != "req":
643 continue
644 try:
645 item_id = int(tag.tag, 10)
646 if item_id > 0:
647 items_to_import.add(item_id)
648 else:
649 mh.warning(item.location,
650 f"invalid codebeamer reference to {item_id}")
651 except ValueError:
652 mh.warning(
653 item.location,
654 f"cannot convert reference '{tag.tag}' to integer "
655 f"Codebeamer ID",
656 )
658 items = import_tagged(cb_config, items_to_import)
660 elif cb_config.import_query is not None:
661 try:
662 if isinstance(cb_config.import_query, str):
663 if (cb_config.import_query.startswith("-") and
664 cb_config.import_query[1:].isdigit()):
665 self._argument_parser.error(
666 "import_query must be a positive integer")
667 elif cb_config.import_query.startswith("-"):
668 self._argument_parser.error(
669 "import_query must be a valid cbQL query")
670 elif cb_config.import_query == "":
671 self._argument_parser.error(
672 "import_query must either be a query string or a query ID")
673 elif cb_config.import_query.isdigit():
674 cb_config.import_query = int(cb_config.import_query)
675 except ValueError as e:
676 self._argument_parser.error(str(e))
678 items = get_query(cb_config, cb_config.import_query)
679 else:
680 raise ValueError(
681 f"Unclear what to do, because neither "
682 f"'{SupportedConfigKeys.IMPORT_QUERY.value}' nor "
683 f"'{SupportedConfigKeys.IMPORT_TAGGED.value}' is specified!",
684 )
686 with _get_out_stream(cb_config.out) as out_stream:
687 _cb_items_to_lobster(items, cb_config, out_stream)
688 if cb_config.out:
689 print(f"Written {len(items)} requirements to {cb_config.out}")
692def _get_out_stream(config_out: Optional[str]) -> TextIO:
693 if config_out:
694 ensure_output_directory(config_out)
695 return open(config_out, "w", encoding="UTF-8")
696 return sys.stdout
699def _cb_items_to_lobster(items: List[Dict], config: Config, out_file: TextIO) -> None:
700 schema_config = get_schema_config(config)
701 lobster_write(out_file, schema_config["class"], TOOL_NAME.replace("-", "_"), items)
704def lobster_codebeamer(config: Config, out_file: str) -> None:
705 """Loads items from codebeamer and serializes them in the LOBSTER interchange
706 format to the given file.
707 """
708 # This is an API function.
709 items = get_query(config, config.import_query)
710 ensure_output_directory(out_file)
711 with open(out_file, "w", encoding="UTF-8") as fd:
712 _cb_items_to_lobster(items, config, fd)
715def main(args: Optional[Sequence[str]] = None) -> int:
716 return CodebeamerTool().run(args)