API Reference

@dataclass(frozen=True)
class Context:
    """A basic context to hold"""

    github_token: Optional[str] = None
    dashverse_token: Optional[str] = None

@dataclass
class CheckResult:
    """
    Datatype for indicator check results.
    """

    process: str = "Undefined process"
    status_id: str = "missing"
    output: str = "missing"
    evidence: str = "missing"
    success: bool = False

    def __bool__(self):
        return self.success

class Summary:
    """
    Summary of the software quality assessment.
    """

    def __init__(
        self,
        author,
        email,
        project_name,
        repo_url,
        software_version,
        branch_hash_or_tag,
    ):
        self.author = author
        self.email = email
        self.project_name = project_name
        self.repo_url = repo_url
        self.software_version = software_version
        self.branch_hash_or_tag = branch_hash_or_tag
        self.checks = []

    def add_indicator_result(self, indicator, checking_software, result):
        self.checks.append(
            {
                "@type": "CheckResult",
                "assessesIndicator": {"@id": indicator["@id"]},
                "checkingSoftware": {
                    "name": checking_software.name,
                    "version": checking_software.version,
                },
                "process": result.process,
                "status": {"@id": result.status_id},
                "output": result.output,
                "evidence": result.evidence,
            }
        )

    def to_json(self):
        return json.dumps(
            {
                "@context": "https://w3id.org/everse/rsqa/0.0.1/",
                "@type": "SoftwareQualityAssessment",
                "dateCreated": str(datetime.now()),
                "license": "CC0-1.0",
                "author": {"@type": "Person", "name": "Quality Pipeline"},
                "assessedSoftware": {
                    "@type": "SoftwareApplication",
                    "name": self.project_name,
                    "softwareVersion": self.software_version,
                    "url": self.repo_url,
                },
                "checks": self.checks,
            },
            sort_keys=True,
            indent=4,
        )

    def write(self, filename):
        with open(filename, "w") as f:
            f.write(self.to_json())

    def upload(self, dashverse_token=None):
        api = APIClient(dashverse_token)
        api.post(self.to_json())

class Configuration:
    """
    A basic wrapper for the configuration.
    """

    def __init__(self, filepath=None):
        if filepath is None:
            print("Loading default configuration.")
            self._cfg = DEFAULT_CONFIG
        else:
            print(f"Loading configuration from '{filepath}'.")
            with open(filepath) as f:
                self._cfg = json.load(f)

class IndicatorPlugin:
    """Skeleton for an Indicator Plugin"""

    name = None
    version = None
    id = None
    indicators = []

class PluginInitError(Exception):
    """Thrown if the initialisation of a plugin fails (e.g. missing GITHUB token)"""

    pass

class PythonExecutor:
    """A Python executor which uses a temporary virtual environment.

    The `packages` should be a list of package names with optional
    requirement specifiers as accepted by `pip`.

    More information:
    https://packaging.python.org/en/latest/glossary/#term-Requirement-Specifier

    """

    def __init__(self, packages=None, environment=None):
        """Instantiates a virtual environment in a temporary folder."""
        self.temp_dir = tempfile.mkdtemp()
        self.environment = environment if environment is not None else {}
        try:
            venv.create(self.temp_dir, with_pip=True)
            if packages is None:
                return
            for package in packages:
                self.install(package)
        except (FileNotFoundError, subprocess.CalledProcessError) as e:
            raise ExecutorInitError(f"failed to initialise Python executor: {e}")

    def install(self, package):
        try:
            subprocess.run(
                [f"{self.temp_dir}/bin/pip", "install", package],
                check=True,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
        except (FileNotFoundError, subprocess.CalledProcessError) as e:
            raise ExecutorInitError(f"failed to initialise Python executor: {e}")

    def is_installed(self, package_name, version=None):
        out = self.execute(
            normalized(
                """
            from importlib.metadata import distributions
            for dist in distributions():
                name = dist.metadata.get('Name', '<unknown>')
                version = getattr(dist, 'version', '<unknown>')
                print(f"{name}=={version}")
        """
            )
        )
        package = package_name + "" if version is None else f"=={version}"
        return package in out.stdout

    def execute(self, script):
        """Run the script in the virtual environment."""
        env = os.environ.copy()
        env.update(self.environment)
        return subprocess.run(
            [f"{self.temp_dir}/bin/python", "-c", script],
            capture_output=True,
            text=True,
            env=env,
        )

    def __del__(self):
        """Cleanup the temporary virtual environment on destruction."""
        try:
            if os.path.exists(self.temp_dir):
                shutil.rmtree(self.temp_dir)
        except Exception as e:
            print(f"Failed to remove virtualenv at {self.temp_dir}: {e}")

class DockerExecutor:
    """A Docker executor."""

    def __init__(self, image_url, pull_args=None):
        self.url = image_url
        if pull_args is None:
            pull_args = []
        command = ["docker", "pull"] + pull_args + [self.url]
        try:
            subprocess.run(
                command,
                check=True,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
        except subprocess.CalledProcessError:
            raise ExecutorInitError(
                f"failed to initialise Docker executor: '{' '.join(command)}'"
            )

    def run(self, command, run_args=None):
        """
        Run command (popenargs) inside a Docker container and return a
        CompletedProcess instance from the subprocess Python module.

        Extra arguments to the command can be passed as a list of
        strings via `run_args`.
        """
        if run_args is None:
            run_args = []
        cmd = ["docker", "run"] + run_args + [self.url] + command
        return subprocess.run(cmd, capture_output=True, text=True)

class ExecutorInitError(Exception):
    """Thrown if the initialisation of an execution fails (e.g. Docker not installed)"""

    pass