trestle.tasks.csv_to_oscal_mc

class CsvColumn:
    """CsvColumn."""

    columns_required = [
        f'{L_SOURCE_RESOURCE}',
        f'{L_TARGET_RESOURCE}',
        f'{L_MAP_SOURCE_ID_REF_LIST}',
        f'{L_MAP_TARGET_ID_REF_LIST}',
        f'{L_MAP_RELATIONSHIP}',
    ]

    columns_optional = [f'{L_MAP_CONFIDENCE_SCORE}', f'{L_MAP_COVERAGE}']

    @staticmethod
    def get_required_column_names() -> List[str]:
        """Get required column names."""
        rval = []
        rval += CsvColumn.columns_required
        return rval

    @staticmethod
    def get_optional_column_names() -> List[str]:
        """Get optional column names."""
        rval = []
        rval += CsvColumn.columns_optional
        return rval

class CsvToOscalMappingCollection(TaskBase):
    """
    Task to create OSCAL mc.json.

    Attributes:
        name: Name of the task.
    """

    name = 'csv-to-oscal-mc'

    def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
        """
        Initialize trestle task csv-to-oscal-mc.

        Args:
            config_object: Config section associated with the task.
        """
        super().__init__(config_object)

    def print_info(self) -> None:
        """Print the help string."""
        name = self.name
        oscal_name = 'mapping-collection'
        #
        logger.info(f'Help information for {name} task.')
        logger.info('')
        logger.info(f'Purpose: From csv produce OSCAL {oscal_name} file.')
        logger.info('')
        logger.info('')
        logger.info(f'Configuration flags sit under [task.{name}]:')
        text1 = '  title                = '
        text2 = '(required) the mapping collection title.'
        logger.info(text1 + text2)
        text1 = '  version              = '
        text2 = '(required) the mapping collection version.'
        logger.info(text1 + text2)
        text1 = '  csv-file             = '
        text2 = '(required) the path of the csv file.'
        text3 = ' [1st row are column headings; 2nd row are column descriptions; 3rd row and beyond is data]'
        logger.info(text1 + text2 + text3)
        text1 = '  required columns:      '
        for text2 in CsvColumn.columns_required:
            logger.info(text1 + text2)
            text1 = '                         '
        text1 = '  optional columns:      '
        for text2 in CsvColumn.columns_optional:
            logger.info(text1 + text2)
            text1 = '                         '
        text1 = '  output-dir           = '
        text2 = '(required) the path of the output directory for synthesized OSCAL .json files.'
        logger.info(text1 + text2)
        text1 = '  output-overwrite     = '
        text2 = '(optional) true [default] or false; replace existing output when true.'
        logger.info(text1 + text2)

    def configure(self) -> bool:
        """Configure."""
        self._timestamp = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat()
        # config verbosity
        self._quiet = self._config.get('quiet', False)
        self._verbose = not self._quiet
        # title
        self._title = self._config.get('title')
        if self._title is None:
            logger.warning('config missing "title"')
            return False
        # version
        self._version = self._config.get('version')
        if self._version is None:
            logger.warning('config missing "version"')
            return False
        # config csv
        self._csv_file = self._config.get('csv-file')
        if self._csv_file is None:
            logger.warning('config missing "csv-file"')
            return False
        self._csv_path = pathlib.Path(self._csv_file)
        if not self._csv_path.exists():
            logger.warning('"csv-file" not found')
            return False
        # announce csv
        if self._verbose:
            logger.info(f'input: {self._csv_file}')
        # workspace
        self._workspace = os.getcwd()
        return True

    def simulate(self) -> TaskOutcome:
        """Provide a simulated outcome."""
        return TaskOutcome('simulated-success')

    def execute(self) -> TaskOutcome:
        """Provide an executed outcome."""
        try:
            return self._execute()
        except Exception:
            logger.error(traceback.format_exc())
            return TaskOutcome('failure')

    def _execute(self) -> TaskOutcome:
        """Execute path core."""
        if not self.configure():
            return TaskOutcome('failure')
        # config output
        odir = self._config.get('output-dir')
        opth = pathlib.Path(odir)
        self._overwrite = self._config.getboolean('output-overwrite', True)
        # insure output dir exists
        opth.mkdir(exist_ok=True, parents=True)
        # calculate output file name & check writability
        oname = 'mapping-collection.json'
        ofile = opth / oname
        if not self._overwrite and pathlib.Path(ofile).exists():
            logger.warning(f'output: {ofile} already exists')
            return TaskOutcome('failure')
        # title
        title = self._config.get('title')
        # version
        version = self._config.get('version')
        # fetch csv
        self._csv_mgr = _CsvMgr(self._csv_path)
        # Mapping Collection manager
        self._mc_mgr = _McMgr(ofile, title, version)
        # href manager
        href_manager = HrefManager()
        # accumulators for missing ids
        ctl_missing_src = []
        ctl_missing_tgt = []
        # process each row of mapping csv
        for row in self._csv_mgr.row_generator():
            # src
            src_resource_href = self._csv_mgr.get_cell(row, L_SOURCE_RESOURCE)
            href_manager.add(src_resource_href)
            src_id_list = self._csv_mgr.get_cell(row, L_MAP_SOURCE_ID_REF_LIST).split()
            src_id_ref_list = href_manager.get_id_list(src_resource_href, src_id_list)
            # tgt
            tgt_resource_href = self._csv_mgr.get_cell(row, L_TARGET_RESOURCE)
            href_manager.add(tgt_resource_href)
            tgt_id_list = self._csv_mgr.get_cell(row, L_MAP_TARGET_ID_REF_LIST).split()
            tgt_id_ref_list = href_manager.get_id_list(tgt_resource_href, tgt_id_list)
            # Get relationship type for all rows
            relationship_type = self._csv_mgr.get_cell(row, L_MAP_RELATIONSHIP)
            # relationship mapping
            if src_id_ref_list and tgt_id_ref_list:
                src_resource_type = href_manager.get_type(src_resource_href)
                tgt_resource_type = href_manager.get_type(tgt_resource_href)
                # Get confidence score and coverage
                confidence_score_str = self._csv_mgr.get_cell(row, L_MAP_CONFIDENCE_SCORE)
                coverage_str = self._csv_mgr.get_cell(row, L_MAP_COVERAGE)
                # Get other user columns as properties
                kvp_set = {}
                for user_column_name in self._csv_mgr.get_user_column_names():
                    value = self._csv_mgr.get_cell(row, user_column_name)
                    kvp_set[user_column_name] = value
                self._mc_mgr.add_mapping(
                    src_resource_type,
                    src_resource_href,
                    tgt_resource_type,
                    tgt_resource_href,
                    src_id_ref_list,
                    tgt_id_ref_list,
                    relationship_type,
                    confidence_score_str,
                    coverage_str,
                    kvp_set,
                )
            else:
                # Track unmapped controls for gap summary - only when relationship is empty
                # If relationship is specified but controls are missing, that's an error
                if relationship_type:
                    # Relationship specified - missing controls are errors
                    if not src_id_ref_list:
                        for src_id in src_id_list:
                            if src_id not in ctl_missing_src:
                                ctl_missing_src.append(src_id)
                    if not tgt_id_ref_list:
                        for tgt_id in tgt_id_list:
                            if tgt_id not in ctl_missing_tgt:
                                ctl_missing_tgt.append(tgt_id)
                else:
                    # No relationship - track as unmapped for gap summary
                    if src_id_ref_list and not tgt_id_ref_list:
                        # Source has controls but no target = unmapped source controls
                        for src_id in src_id_list:
                            self._mc_mgr.add_unmapped_control(src_id, is_source=True)
                    elif tgt_id_ref_list and not src_id_ref_list:
                        # Target has controls but no source = unmapped target controls
                        for tgt_id in tgt_id_list:
                            self._mc_mgr.add_unmapped_control(tgt_id, is_source=False)
                    else:
                        # Both empty - keep track of ids not found in specified corresponding href
                        if not src_id_ref_list:
                            for src_id in src_id_list:
                                if src_id not in ctl_missing_src:
                                    ctl_missing_src.append(src_id)
                        if not tgt_id_ref_list:
                            for tgt_id in tgt_id_list:
                                if tgt_id not in ctl_missing_tgt:
                                    ctl_missing_tgt.append(tgt_id)
        # issues?
        text = ''
        if ctl_missing_src:
            text += f'source(s) missing from href: {ctl_missing_src}\n'
        if ctl_missing_tgt:
            text += f'target(s) missing from href: {ctl_missing_tgt}\n'
        if text:
            raise RuntimeError(text)
        # fetch synthesized mapping collection
        mapping_collection = self._mc_mgr.get_mapping_collection()
        # write OSCAL mc to file
        if self._verbose:
            logger.info(f'output: {ofile}')
        mapping_collection.oscal_write(pathlib.Path(ofile))
        return TaskOutcome('success')