tests.py

# type: ignore (for mirror)
# ruff: noqa: S101

from _collections_abc import Callable
from datetime import datetime
from pathlib import Path
from types import ModuleType

from vdbpy.api.edits import get_cached_edits_by_entry_before_version_id
from vdbpy.api.entries import (
    get_cached_entry_version,
    get_cached_raw_entry_version,
    get_entry_link,
    is_entry_tagged,
)
from vdbpy.config import WEBSITE
from vdbpy.types.shared import EntryType
from vdbpy.utils.data import add_s
from vdbpy.utils.files import get_lines
from vdbpy.utils.logger import get_logger

from mod_types import (
    CorrectTestResults,
    RuleModuleResult,
    RuleModules,
    RuleTableRow,
)

logger = get_logger()


def run_tests(
    check_function: Callable[..., tuple[int, str, int, int]],
    rule_modules_by_rule_id: RuleModules,
    oldest_date_to_check: datetime,
    rule_id_to_check: int | None = None,
) -> None:
    for rule_id, (rule_name, rule_module) in rule_modules_by_rule_id.items():
        if rule_id_to_check:
            if rule_id != rule_id_to_check:
                logger.warning(f"Skipping rule tests for {rule_id}_{rule_name}")
                continue

        tests: CorrectTestResults = rule_module.test()
        edit_check_tests, entry_check_tests = tests
        assert edit_check_tests

        logger.info(f"\nRunning EDIT check tests for {rule_id}_{rule_name}")
        for correct_check_result, version_tuples in edit_check_tests.items():
            for entry_type, entry_id, version_id in version_tuples:
                version_ids: list[int] = [
                    edit.version_id
                    for edit in get_cached_edits_by_entry_before_version_id(
                        entry_type, entry_id, version_id, include_deleted=True
                    )
                ]
                assert version_id in version_ids

                logger.debug(
                    f"Testing v{version_id} for {get_entry_link(entry_type, entry_id)}"
                    f" (should be {correct_check_result})"
                )

                raw_entry = get_cached_raw_entry_version(entry_type, version_id)
                changed_fields = raw_entry["archivedVersion"]["changedFields"]
                version_number = int(raw_entry["archivedVersion"]["version"])
                if not set(changed_fields) & set(rule_module.FIELDS):
                    if version_number != 0:
                        logger.warning("Mismatching changed fields:")
                        raise Exception(
                            f"  Changed fields for {WEBSITE}/api/"
                            f"{add_s(str(entry_type).lower())}/versions/{version_id}:"
                            f" {changed_fields}\n"
                            f"  Rule check fields: {rule_module.FIELDS}"
                        )
                entry_version_data = get_cached_entry_version(entry_type, version_id)
                is_tagged = False
                if entry_version_data and hasattr(rule_module, "TAG_ID"):
                    logger.info(f"TAG ID found: {rule_module.TAG_ID}")
                    entry = (entry_type, entry_version_data.entry_id)
                    is_tagged = is_entry_tagged(entry, rule_module.TAG_ID)
                try:
                    test_check_result: RuleModuleResult = (
                        rule_module.check_entry_version_for_rule(
                            entry_version_data, is_tagged
                        )
                    )
                except TypeError:
                    test_check_result: RuleModuleResult = (
                        rule_module.check_entry_version_for_rule(entry_version_data)
                    )
                if test_check_result != correct_check_result:
                    logger.warning(
                        f"Test failed for for "
                        f"{WEBSITE}/api/{add_s(str(entry_type).lower())}/versions/{version_id}:\n"
                        f"  Expected: {correct_check_result}\n"
                        f"  Actual:   {test_check_result}"
                    )
                    raise Exception(f"  Entry version data: {entry_version_data}")

        if not entry_check_tests:
            logger.warning(f"Entry checks not found for {rule_id}_{rule_name}")
            continue

        logger.info(f"Running ENTRY check tests for {rule_id}_{rule_name}")
        for correct_check_result, correct_check_data in entry_check_tests.items():
            assert correct_check_result != "No autofix"
            assert correct_check_result != "No data"
            assert correct_check_result != "Reverted"
            assert correct_check_result != "Wrong entry type"
            for (
                relevant_version_id,
                relevant_user_id,
                version_tuple,
            ) in correct_check_data:
                entry_type, entry_id, version_id = version_tuple
                assert entry_id
                logger.debug(
                    f"Testing {get_entry_link(entry_type, entry_id)} edits since"
                    f" v{version_id} (should be {correct_check_result})"
                )
                version_ids: list[int] = [
                    edit.version_id
                    for edit in get_cached_edits_by_entry_before_version_id(
                        entry_type, entry_id, version_id, include_deleted=True
                    )
                ]
                assert version_id in version_ids

                correct = (
                    version_id,
                    correct_check_result,
                    relevant_version_id,
                    relevant_user_id,
                )
                result: tuple[int, str, int, int] = check_function(
                    entry=(entry_type, entry_id),
                    version_id=version_id,
                    rule_id=rule_id,
                    edit_age_thresold=oldest_date_to_check,
                    is_tagged=None,
                    check_all=False,
                    always_check_the_first_edit=False,
                )
                if correct != result:
                    raise Exception(
                        f"  Expected {correct} but got {result} for {version_tuple}"
                    )


def verify_rule_fields(
    rule_id: int,
    rule_name: str,
    rule_module: ModuleType,
    rule_checks_dir: Path,
    rule_wiki_fields: RuleTableRow,
) -> None:
    def compare_entry_types(
        code_fields: list[EntryType], wiki_fields: list[str]
    ) -> bool:
        if wiki_fields[0] == "All":
            return not code_fields

        if "Events" in wiki_fields:
            wiki_fields.remove("Events")
            wiki_fields.append("ReleaseEvent")

        wiki_fields = [field.rstrip("s") for field in wiki_fields]
        return set(code_fields) == set(wiki_fields)

    code = get_lines(rule_checks_dir / f"{rule_id}_{rule_name}.py")
    rule_entry_status = "Draft"
    for line in code:
        if 'if version_data.status == "Draft":' in line:
            rule_entry_status = "Finished"
        elif 'if version_data.status != "Approved":' in line:
            rule_entry_status = "Approved"

    if not compare_entry_types(
        rule_module.ENTRY_TYPES, rule_wiki_fields.rule_entry_types_strings
    ):
        logger.warning(
            f"Rule {rule_id}-{rule_id} has entry types {rule_module.ENTRY_TYPES}"
            f" in code but {rule_wiki_fields.rule_entry_types_strings} in wiki"
        )
        _ = input("Press enter to continue...")

    wiki_rule_name = rule_wiki_fields.rule_name
    wiki_rule_entry_status = rule_wiki_fields.rule_entry_status
    wiki_rule_autofixed = rule_wiki_fields.rule_autofixed
    wiki_rule_complete = rule_wiki_fields.rule_complete
    wiki_rule_mikumodded = rule_wiki_fields.rule_mikumodded

    wiki_fields = {
        "name": (wiki_rule_name, rule_name),
        "entry status": (wiki_rule_entry_status, rule_entry_status),
        "autofixed": (wiki_rule_autofixed, rule_module.AUTOMATICALLY_FIXED),
        "complete": (wiki_rule_complete, rule_module.COMPLETE),
        "mikumodded": (wiki_rule_mikumodded, True),
    }

    for field_name, (wiki_field_value, code_field_value) in wiki_fields.items():
        if wiki_field_value != code_field_value:
            logger.warning(
                f"Rule {rule_id}-{rule_name}, {field_name}: '{code_field_value}' (CODE)"
                f" =/= '{wiki_field_value}' (WIKI)"
            )
            _ = input("Press enter to continue...")