generic_steps.py

"""Steps for features tests."""

# pylint: disable=function-redefined, missing-function-docstring, not-callable
# pyright: reportRedeclaration=false, reportAttributeAccessIssue=false, reportCallIssue=false

import difflib
import json
import os
import pathlib
import re
import shutil
from contextlib import contextmanager
from itertools import zip_longest
from typing import Iterable, List, Optional, Pattern, Tuple, Union

from behave import given, then, when  # pylint: disable=no-name-in-module
from behave.runner import Context

from dfetch.__main__ import DfetchFatalException, run
from dfetch.log import DLogger
from dfetch.reporting.sbom_reporter import INFER_LICENSE_VERSION
from dfetch.util.util import in_directory

ansi_escape = re.compile(r"\[/?[a-z\_ ]+\]")
dfetch_title = re.compile(r"Dfetch \(\d+.\d+.\d+\)")
timestamp = re.compile(r"\d+\/\d+\/\d+, \d+:\d+:\d+")
git_hash = re.compile(r"(\s?)[a-f0-9]{40}(\s?)")
git_timestamp = re.compile(
    r"[A-Za-z]{3},\s+\d{2}\s+[A-Za-z]{3}\s+\d{4}\s+\d{2}:\d{2}:\d{2}\s+\+0000\n?"
)
git_user_and_mail = re.compile(r"[^\s<]+(?: [^\s<]+)* <[^\s]+@[^\s]+>")
iso_timestamp = re.compile(r'"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{6}\+\d{2}:\d{2}')
urn_uuid = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")
bom_ref = re.compile(r"BomRef\.[0-9]+\.[0-9]+")
svn_error = re.compile(r"svn: E\d{6}: .+")
abs_path = re.compile(r"/tmp/[\w_]+")


@contextmanager
def temporary_env(key: str, value: str):
    """Temporarily set an environment variable inside a context."""
    old_value = os.environ.get(key)
    os.environ[key] = value
    try:
        yield
    finally:
        if old_value is None:
            del os.environ[key]
        else:
            os.environ[key] = old_value


def remote_server_path(context) -> str:
    """Get the path to the remote dir as a POSIX path string."""
    return pathlib.Path(context.remotes_dir_path).as_uri()


def call_command(context: Context, args: list[str], path: Optional[str] = ".") -> None:
    before = context.console.export_text()

    DLogger.reset_projects()

    with temporary_env("CI", "true"):
        with in_directory(path or "."):
            try:
                run(args, context.console)
                context.cmd_returncode = 0
            except DfetchFatalException:
                context.cmd_returncode = 1

    after = context.console.export_text()
    context.cmd_output = after[len(before) :].strip("\n")


def check_file(path, content, strict=False):
    """Check a file."""
    with open(path, "r", encoding="UTF-8", newline="") as file_to_check:
        check_content(
            content.splitlines(True), file_to_check.readlines(), strict=strict
        )


def check_file_exists(path):
    """Check a file."""
    assert os.path.isfile(path), f"Expected {path} to exist, but it didn't!"


def apply_archive_substitutions(text: str, context) -> str:
    """Replace archive-related dynamic placeholders with values stored on *context*."""
    if hasattr(context, "archive_sha256"):
        text = text.replace("<archive-sha256>", context.archive_sha256)
    if hasattr(context, "archive_sha384"):
        text = text.replace("<archive-sha384>", context.archive_sha384)
    if hasattr(context, "archive_sha512"):
        text = text.replace("<archive-sha512>", context.archive_sha512)
    if hasattr(context, "archive_url"):
        text = text.replace("<archive-url>", context.archive_url)
    if hasattr(context, "license_base64"):
        text = text.replace("<license-base64>", context.license_base64)
    text = text.replace(
        "<infer-license-version>", f"infer-license {INFER_LICENSE_VERSION}"
    )
    return text


def check_content(
    expected_content: Iterable[str], actual_content: Iterable[str], strict=False
) -> None:
    """Compare two texts as list of strings."""

    for line_nr, (actual, expected) in enumerate(
        zip_longest(actual_content, expected_content, fillvalue=""), start=1
    ):
        expected = multisub(
            patterns=[
                (git_hash, r"\1[commit-hash]\2"),
                (iso_timestamp, "[timestamp]"),
                (git_timestamp, "[git-timestamp]"),
                (git_user_and_mail, "[user <email>]"),
                (urn_uuid, "[urn-uuid]"),
                (bom_ref, "[bom-ref]"),
            ],
            text=expected,
        )

        actual = multisub(
            patterns=[
                (git_hash, r"\1[commit-hash]\2"),
                (iso_timestamp, "[timestamp]"),
                (git_timestamp, "[git-timestamp]"),
                (git_user_and_mail, "[user <email>]"),
                (urn_uuid, "[urn-uuid]"),
                (bom_ref, "[bom-ref]"),
            ],
            text=actual,
        )

        if not strict:
            expected = expected.strip()
            actual = actual.strip()

        assert actual == expected, (
            f"Line {line_nr}: Actual >>{actual}<< != Expected >>{expected}<<\n"
            f"ACTUAL:\n{''.join(actual_content)}"
        )


def generate_file(path, content, encoding="UTF-8"):
    opt_dir = path.rsplit("/", maxsplit=1)

    if len(opt_dir) > 1:
        pathlib.Path(opt_dir[0]).mkdir(parents=True, exist_ok=True)

    with open(path, "w", encoding=encoding) as new_file:
        for line in content.splitlines():
            print(line, file=new_file)


def extend_file(path, content):
    with open(path, "a", encoding="UTF-8") as existing_file:
        for line in content.splitlines():
            print(line, file=existing_file)


def replace_in_file(path: str, old: str, new: str) -> None:
    with open(path, "r", encoding="utf-8") as f:
        content = f.read()

    content = content.replace(old, new)

    with open(path, "w", encoding="utf-8") as f:
        f.write(content)


def list_dir(path):
    # Get list of all nodes
    nodes = [os.path.normpath(path).split(os.sep)]
    for root, dirs, files in os.walk(path, followlinks=False):
        root = os.path.relpath(root)
        for name in dirs:
            nodes += [os.path.normpath(os.path.join(root, name)).split(os.sep)]
        for name in files:
            nodes += [os.path.normpath(os.path.join(root, name)).split(os.sep)]

    result = ""
    prev_node = []
    for node in list(sorted(nodes)) + [""]:
        if prev_node:
            end = ""
            if "".join(node).startswith("".join(prev_node)):
                end = "/"
            result += "    " * (len(prev_node) - 1) + prev_node[-1] + end + os.linesep
        prev_node = node

    return result


def normalize_lines(text: str) -> list[str]:
    """Normalize text for diffing."""
    lines = text.splitlines()
    return [line.rstrip() for line in lines if line.strip() != ""]


def check_output(context, line_count=None):
    """Check command output against expected text.

    Args:
        context: Behave context with cmd_output and expected text
        line_count: If set, compare only the first N lines of actual output
    """
    expected_raw = apply_archive_substitutions(context.text, context)

    expected_text = multisub(
        patterns=[
            (dfetch_title, "Dfetch (x.x.x)"),
            (git_hash, r"\1[commit-hash]\2"),
            (timestamp, "[timestamp]"),
            (ansi_escape, ""),
            (svn_error, "svn: EXXXXXX: <some error text>"),
        ],
        text=expected_raw,
    )

    actual_text = multisub(
        patterns=[
            (dfetch_title, "Dfetch (x.x.x)"),
            (git_hash, r"\1[commit-hash]\2"),
            (timestamp, "[timestamp]"),
            (ansi_escape, ""),
            (
                re.compile(remote_server_path(context)),
                "some-remote-server",
            ),
            (svn_error, "svn: EXXXXXX: <some error text>"),
            (abs_path, "/some/path"),
        ],
        text=context.cmd_output,
    )

    actual_lines = normalize_lines(actual_text)[:line_count]
    expected_lines = normalize_lines(expected_text)

    diff = difflib.ndiff(actual_lines, expected_lines)

    diffs = [x for x in diff if x[0] in ("+", "-")]
    if diffs:
        comp = "\n".join(diffs)
        print(actual_text)
        print(comp)
        assert False, "Output not as expected!"


@given('"{old}" is replaced with "{new}" in "{path}"')
def step_impl(_, old: str, new: str, path: str):
    replace_in_file(path, old, new)


@given('"{old_path}" in {directory} is renamed as "{new_path}"')
def step_impl(_, old_path: str, new_path: str, directory: str):
    with in_directory(directory):
        shutil.move(old_path, new_path)


@given("the patch file '{name}'")
@given("the patch file '{name}' in {directory}")
@given("the patch file '{name}' with '{encoding}' encoding")
def step_impl(context, name, encoding="UTF-8", directory="."):
    generate_file(os.path.join(directory, name), context.text, encoding)


@given('"{path}" in {directory} is created')
@when('"{path}" in {directory} is created')
def step_impl(context, path, directory="."):
    with in_directory(directory):
        generate_file(path, context.text or "Some content")


@given('the metadata file "{metadata_file}" of "{project_path}" is corrupt')
def step_impl(_, metadata_file, project_path):
    generate_file(
        os.path.join(os.getcwd(), project_path, metadata_file), "Corrupt metadata!"
    )


@given('the metadata file "{metadata_file}" of "{project_path}" is changed')
def step_impl(_, metadata_file, project_path):
    extend_file(
        os.path.join(os.getcwd(), project_path, metadata_file), "# Some comment"
    )


@given("all projects are updated in {path}")
@given("all projects are updated")
def step_impl(context, path=None):
    if path:
        context.execute_steps(f'When I run "dfetch update" in {path}')
    else:
        context.execute_steps('When I run "dfetch update"')
    assert not context.cmd_returncode, context.cmd_output


@when('I run "dfetch {args}" in {path}')
@when('I run "dfetch {args}"')
def step_impl(context, args, path=None):
    """Call a command."""
    resolved = args.replace("some-remote-server", remote_server_path(context))
    call_command(context, resolved.split(), path)


@given('"{path}" in {directory} is changed locally')
@when('"{path}" in {directory} is changed locally')
def step_impl(_, directory, path):
    with in_directory(directory):
        extend_file(path, "Some text")


@then("the patched '{name}' is")
def step_impl(context, name):
    """Check a manifest."""
    check_file(name, context.text)


@then("the first line of '{name}' is changed to")
def step_impl(context, name):
    """Check the first line of the file."""
    with open(name, "r", encoding="UTF-8") as file_to_check:
        check_content(context.text.strip(), file_to_check.readline().strip())


@then("the patch file '{name}' is generated")
@then("the patch file '{name}' is updated")
def step_impl(context, name):
    """Check a patch file."""
    if context.text:
        check_file(name, context.text)
    else:
        check_file_exists(name)


def check_json(path: Union[str, os.PathLike], content: str, context) -> None:
    """Check a JSON file for exact equality (after normalising formatting)."""
    content = apply_archive_substitutions(content, context)
    with open(path, "r", encoding="UTF-8") as file_to_check:
        actual_json = json.load(file_to_check)
    expected_json = json.loads(content)

    check_content(
        json.dumps(expected_json, indent=4, sort_keys=True).splitlines(),
        json.dumps(actual_json, indent=4, sort_keys=True).splitlines(),
    )


@then("the '{name}' file contains")
def step_impl(context, name):
    if name.endswith(".json"):
        check_json(name, context.text, context)
    else:
        check_file(name, context.text)


@then("'{name}' exists")
def step_impl(_, name):
    assert os.path.exists(name), f"Expected {name} to exist, but it didn't!"


@then("'{path}' is a symlink pointing to '{target}'")
def step_impl(_, path, target):
    assert os.path.islink(
        path
    ), f"Expected {path!r} to be a symbolic link, but it is not"
    actual = os.readlink(path)
    assert actual == target, f"Expected {path!r} to point to {target!r}, got {actual!r}"


def multisub(patterns: List[Tuple[Pattern[str], str]], text: str) -> str:
    """Apply a list of tuples that each contain a regex + replace string."""
    for pattern, replace in patterns:
        text = pattern.sub(replace, text)

    return text


@then("the output starts with:")
def step_impl(context):
    check_output(context, line_count=len(context.text.splitlines()))


@then("the output shows")
def step_impl(context):
    check_output(context)


@then("the following projects are fetched")
def step_impl(context):
    for project in context.table:
        assert os.path.exists(project["path"]), f"No project found at {project}"
        assert os.listdir(project["path"]), f"{project} is just an empty directory!"


@then("'{path}' looks like:")
def step_impl(context, path):
    expected_dir = context.text.strip()
    actual_dir = list_dir(path).strip()

    assert expected_dir == actual_dir, os.linesep.join(
        ["", "Expected:", expected_dir, "", "Actual:", actual_dir]
    )


@then("the directory '{path}' should be removed from disk")
def step_impl(_, path):
    """Check if a directory is removed."""
    assert not os.path.exists(path), f"Directory {path} still exists!"


@given("the directory '{path}' does not exist on disk")
def step_impl(_, path):
    """Assert that a directory has not been created."""
    assert not os.path.exists(path), f"Directory {path} exists but should not!"