#!/usr/bin/env vpython3
# Copyright 2023 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import os
import sys
from typing import List
import unittest

_THIS_DIR = os.path.abspath(os.path.dirname(__file__))
# The repo's root directory.
_ROOT_DIR = os.path.abspath(os.path.join(_THIS_DIR, "..", ".."))

# Add the repo's root directory for clearer imports.
sys.path.insert(0, _ROOT_DIR)

import metadata.fields.known as known_fields  # noqa: E402
import metadata.fields.field_types as field_types  # noqa: E402
import metadata.validation_result as vr  # noqa: E402
import metadata.fields.custom.cpe_prefix as cpe_prefix_util  # noqa: E402
import metadata.fields.custom.mitigated  # noqa: E402
import metadata.fields.custom.update_mechanism  # noqa: E402
import metadata.fields.custom.license  # noqa: E402
import metadata.fields.custom.license_allowlist  # noqa: E402


class FieldValidationTest(unittest.TestCase):
    def _run_field_validation(
        self,
        field: field_types.MetadataField,
        valid_values: List[str],
        error_values: List[str],
        warning_values: List[str] = [],
        **kwargs,
    ):
        """Helper to run a field's validation for different values."""
        for value in valid_values:
            self.assertIsNone(field.validate(value, **kwargs), value)

        for value in error_values:
            self.assertIsInstance(
                field.validate(value, **kwargs), vr.ValidationError, value
            )

        for value in warning_values:
            self.assertIsInstance(
                field.validate(value, **kwargs), vr.ValidationWarning, value
            )

    def test_freeform_text_validation(self):
        # Check validation of a freeform text field that should be on
        # one line.
        self._run_field_validation(
            field=field_types.SingleLineTextField("Text single line"),
            valid_values=["Text on single line", "a", "1"],
            error_values=["", "\n", " "],
        )

        # Check validation of a freeform text field that can span
        # multiple lines.
        self._run_field_validation(
            field=field_types.FreeformTextField("Freeform multi"),
            valid_values=[
                "This is text spanning multiple lines:\n"
                "    * with this point\n"
                "    * and this other point",
                "Text on single line",
                "a",
                "1",
            ],
            error_values=["", "\n", " "],
        )

    def test_yes_no_field_validation(self):
        self._run_field_validation(
            field=field_types.YesNoField("Yes/No test"),
            valid_values=["yes", "no", "No", "YES"],
            error_values=["", "\n", "Probably yes"],
            warning_values=["Yes?", "not"],
        )

    def test_cpe_prefix_has_version_component(self):
        """Test has_version_component for CPE prefix."""
        # CPE 2.3 Formatted String
        self.assertTrue(
            cpe_prefix_util.has_version_component(
                "cpe:2.3:a:vendor:product:1.2.3:*:*:*:*:*:*:*"
            )
        )
        self.assertFalse(
            cpe_prefix_util.has_version_component(
                "cpe:2.3:a:vendor:product:*:*:*:*:*:*:*:*"
            )
        )
        self.assertFalse(
            cpe_prefix_util.has_version_component(
                "cpe:2.3:a:vendor:product:-:*:*:*:*:*:*:*"
            )
        )
        self.assertFalse(
            cpe_prefix_util.has_version_component("cpe:2.3:a:vendor:product")
        )

        # CPE 2.2 URN
        self.assertTrue(
            cpe_prefix_util.has_version_component("cpe:/a:vendor:product:1.2.3")
        )
        self.assertFalse(
            cpe_prefix_util.has_version_component("cpe:/a:vendor:product")
        )
        self.assertFalse(cpe_prefix_util.has_version_component("cpe:/a:vendor"))
        self.assertFalse(cpe_prefix_util.has_version_component("cpe:/a"))
        self.assertFalse(cpe_prefix_util.has_version_component("cpe:/"))

        # Invalid
        self.assertFalse(cpe_prefix_util.has_version_component("not a cpe"))

    def test_cpe_prefix_validation(self):
        self._run_field_validation(
            field=known_fields.CPE_PREFIX,
            valid_values=[
                "unknown",
                "cpe:2.3:a:sqlite:sqlite:3.0.0:*:*:*:*:*:*:*",
                "cpe:2.3:a:sqlite:sqlite:*:*:*:*:*:*:*:*",
                "cpe:/a:vendor:product:version:update:edition:lang",
                "cpe:/a::product:",
                "cpe:/:vendor::::edition",
                "cpe:/:vendor",
            ],
            error_values=[
                "",
                "\n",
                "cpe:2.3:a:sqlite:sqlite:3.0.0",
                "cpe:2.3:a:sqlite:sqlite::::::::",
                "cpe:/",
                "cpe:/a:vendor:product:version:update:edition:lang:",
            ],
        )

    def test_date_validation(self):
        self._run_field_validation(
            field=known_fields.DATE,
            valid_values=["2012-03-04"],
            error_values=[
                "",
                "\n",
                "N/A",
                "03-04-12",  # Ambiguous month and day.
                "04/03/2012",  # Ambiguous month and day.
            ],
            warning_values=[
                "2012-03-04 UTC",
                "2012-03-04 UTC+10:00",
                "2012/03/04 UTC+10:00",
                "20120304",
                "April 3, 2012",
                "3 Apr 2012",
                "30/12/2000",
                "20-03-2020",
                "Tue Apr 3 05:06:07 2012 +0800",
            ],
        )

    def test_revision_validation(self):
        self._run_field_validation(
            field=known_fields.REVISION,
            valid_values=[
                "DEPS",  # Sentinel.
                "a" * 64,
            ],
            error_values=[
                "a" * 6,
                "g" * 40,  # Right length, but not hex.
            ],
            warning_values=[
                "unknown",  # "unknown" is a soft failure, not a hard error.
            ],
        )

    def test_license_validation(self):
        self._run_field_validation(
            field=known_fields.LICENSE,
            valid_values=[
                "Apache-2.0 , MIT",
                "Apache-2.0",
                "BSD-2-Clause",
                "BSD-2-Clause-FreeBSD",
                "MIT",
                "Refer to additional_readme_paths.json",
                "LicenseRef-MIT",
                "LicenseRef-MIT, Apache-2.0",
            ],
            error_values=[
                "",
                "\n",
                ",",
                "Apache 2.0 ,",
                "Custom / MIT",
                "Apache-2.0 and MIT",
                "Apache-2.0; MIT; BSD-2-Clause",
            ],
            warning_values=[
                "Custom license",
                "Custom, MIT",
                "Refer to any_other_readme_paths.json",
                "APSL-2.0, MIT",
                "APSL-2.0 ,MIT",
            ],
        )

    def test_license_validation_open_source(self):
        # Reciprocal licenses are valid in open source projects.
        self._run_field_validation(
            field=known_fields.LICENSE,
            is_open_source_project=True,
            valid_values=[
                "APSL-2.0, MIT",
                "APSL-2.0 ,MIT",
            ],
            error_values=[],
        )

    def test_license_file_validation(self):
        self._run_field_validation(
            field=known_fields.LICENSE_FILE,
            valid_values=[
                "LICENSE",
                "src/LICENSE.txt",
                "LICENSE, //third_party_test/LICENSE-TEST",
                "src/MISSING_LICENSE",
            ],
            error_values=["", "\n", ","],
            warning_values=["NOT_SHIPPED"],
        )

        # Check relative path from README directory, and multiple
        # license files.
        result = known_fields.LICENSE_FILE.validate_on_disk(
            value="LICENSE, src/LICENSE.txt",
            source_file_dir=os.path.join(_THIS_DIR, "data"),
            repo_root_dir=_THIS_DIR,
        )
        self.assertIsNone(result)

        # Check relative path from Chromium src directory.
        result = known_fields.LICENSE_FILE.validate_on_disk(
            value="//data/LICENSE",
            source_file_dir=os.path.join(_THIS_DIR, "data"),
            repo_root_dir=_THIS_DIR,
        )
        self.assertIsNone(result)

        # Check missing file.
        result = known_fields.LICENSE_FILE.validate_on_disk(
            value="MISSING_LICENSE",
            source_file_dir=os.path.join(_THIS_DIR, "data"),
            repo_root_dir=_THIS_DIR,
        )
        self.assertIsInstance(result, vr.ValidationWarning)

        # Check deprecated NOT_SHIPPED.
        result = known_fields.LICENSE_FILE.validate_on_disk(
            value="NOT_SHIPPED",
            source_file_dir=os.path.join(_THIS_DIR, "data"),
            repo_root_dir=_THIS_DIR,
        )
        self.assertIsInstance(result, vr.ValidationWarning)

    def test_url_validation(self):
        self._run_field_validation(
            field=known_fields.URL,
            valid_values=[
                "https://www.example.com/a",
                "http://www.example.com/b",
                "ftp://www.example.com/c,git://www.example.com/d",
                "https://www.example.com/a\n  https://example.com/b",
                "This is the canonical public repository",
                "internal",
                "Internal.",
                "Google internal",
                "Google Internal.",
            ],
            warning_values=[
                # Scheme is case-insensitive, but should be lower case.
                "Https://www.example.com/g",
            ],
            error_values=[
                "",
                "\n",
                "rpc://project/project",
                "rpc://project.googlesource.com/project",
                "ghttps://www.example.com/e",
                "https://www.example.com/ f",
                "This is an unrecognized message for the URL",
            ],
        )

    def test_version_validation(self):
        self._run_field_validation(
            field=known_fields.VERSION,
            valid_values=["n / a", "123abc", "unknown forked version"],
            error_values=["", "\n"],
            warning_values=["0", "unknown"],
        )

    def test_local_modifications(self):
        # Checks local modifications field early terminates when we can reasonably infer there's no modification.
        _NO_MODIFICATION_VALUES = [
            "None",
            "None.",
            "N/A.",
            "(none).",
            "No modification",
            "\nNone.",
        ]
        for value in _NO_MODIFICATION_VALUES:
            self.assertTrue(
                known_fields.LOCAL_MODIFICATIONS.should_terminate_field(value)
            )

        # Checks ambiguous values won't early terminate the field.
        _MAY_CONTAIN_MODIFICATION_VALUES = [
            "None. Except doing something.",
            "Modify file X to include ....",
        ]
        for value in _MAY_CONTAIN_MODIFICATION_VALUES:
            self.assertFalse(
                known_fields.LOCAL_MODIFICATIONS.should_terminate_field(value)
            )

    def test_vulnerability_ids(self):
        valid_ids = [
            "CVE-2024-12345",
            "CVE-2024-1234567",
            "PYSEC-2024-1234",
            "OSV-2024-1234",
            "DSA-1234-1",
            "GHSA-1234-5678-90ab",
        ]

        invalid_ids = [
            "CVE-123-456",
            "GHSA-123-456",
            "PYSEC-2024",  # Missing ID part.
            "NOT-A-VALID-ID",  # Bad prefix.
            "CVE_2024_12345",  # Wrong separator.
            "",  # Empty.
            " ",  # Just space.
        ]

        test_ids = valid_ids + invalid_ids
        valid_result, invalid_result = (
            metadata.fields.custom.mitigated.validate_vuln_ids(
                ",".join(test_ids)
            )
        )

        self.assertListEqual(sorted(valid_result), sorted(valid_ids))
        self.assertListEqual(sorted(invalid_result), sorted(invalid_ids))

    def test_update_mechanism_validation(self):
        """Tests the validation logic for the Update Mechanism field."""
        self._run_field_validation(
            field=known_fields.UPDATE_MECHANISM,
            valid_values=[
                "Autoroll",
                "  Autoroll  ",
                "Manual (https://crbug.com/12345)",
                "Static (https://crbug.com/54321)",
                "Static.HardFork (https://crbug.com/98765)",
            ],
            error_values=[
                "",
                " ",
                "Invalid Value",
                "Custom (crbug.com/123)",
                "Custom (https://crbug.com/123)",
                "Manual (https://crbug.com/12345 )",
                "Manual (https://crbug.com/12345a)",
                "Manual (crbug.com/12345)",
                "Static (crbug.com/54321)",
                "Static (crbug/54321)",
                "Static (http://crbug/54321)",
                "Static (http://crbug.com/54321)",
                "Static (https://crbug/54321)",
                "Static.HardFork (crbug.com/98765)",
                "Static",
                "Static.HardFork",
            ],
        )

    def test_load_restrictive_license_approval_proto(self):
        path = os.path.join(
            _THIS_DIR, "data", "restrictive_license_approval.textproto"
        )
        result = metadata.fields.custom.license_allowlist.load_restrictive_license_approval_textproto(
            path
        )
        self.assertIn("testing-restrictive-license", result)

    def test_license_validation_with_rla(self):
        test_license = "LicenseRef-GUST-Font-License"

        field = metadata.fields.custom.license.LicenseField()
        # Validate WITHOUT the rla textproto: should return a ValidationWarning.
        res_without = field.validate(test_license, source_file_dir=_THIS_DIR)
        self.assertIsInstance(res_without, vr.ValidationWarning)

        # Validate WITH the rla textproto: should return None (approved / valid).
        res_with = field.validate(
            test_license, source_file_dir=os.path.join(_THIS_DIR, "data")
        )
        self.assertIsNone(res_with)

    def test_get_license_validation_status(self):
        get_status = metadata.fields.custom.license_allowlist.get_license_validation_status
        data_dir = os.path.join(_THIS_DIR, "data")

        # 1. Globally allowed
        self.assertEqual(get_status("MIT"), "ALLOWED")
        self.assertEqual(get_status("Apache-2.0"), "ALLOWED")
        self.assertEqual(get_status("MIT, Apache-2.0"), "ALLOWED")

        # 2. Restricted but approved
        self.assertEqual(
            get_status(
                "LicenseRef-GUST-Font-License", source_file_dir=data_dir
            ),
            "APPROVED[(GUST-Font-License, b/987654321)]",
        )
        self.assertEqual(
            get_status(
                "MIT, LicenseRef-GUST-Font-License", source_file_dir=data_dir
            ),
            "APPROVED[(GUST-Font-License, b/987654321)]",
        )

        # 3. Allowed-not-shipped (GPL-2.0)
        self.assertEqual(get_status("GPL-2.0", is_shipped=False), "ALLOWED")
        self.assertEqual(
            get_status("GPL-2.0", is_shipped=True),
            "NOT_ALLOWED_SHIPPED[GPL-2.0]",
        )
        self.assertEqual(get_status("GPL-2.0"), "ALLOWED")
        self.assertEqual(
            get_status("GPL-2.0", source_file_dir=data_dir, is_shipped=False),
            "ALLOWED",
        )
        self.assertEqual(
            get_status(
                "MIT, GPL-2.0", source_file_dir=data_dir, is_shipped=False
            ),
            "ALLOWED",
        )
        self.assertEqual(
            get_status(
                "LicenseRef-GUST-Font-License, GPL-2.0",
                source_file_dir=data_dir,
                is_shipped=False,
            ),
            "APPROVED[(GUST-Font-License, b/987654321)]",
        )
        self.assertEqual(
            get_status(
                "LicenseRef-GUST-Font-License, GPL-2.0",
                source_file_dir=data_dir,
                is_shipped=True,
            ),
            "NOT_ALLOWED_SHIPPED[GPL-2.0], APPROVED[(GUST-Font-License, b/987654321)]",
        )

        # 3b. Unknown and NOT approved
        self.assertEqual(
            get_status("My-Custom-License"), "UNKNOWN[My-Custom-License]"
        )
        self.assertEqual(
            get_status("My-Custom-License", source_file_dir=data_dir),
            "UNKNOWN[My-Custom-License]",
        )
        self.assertEqual(
            get_status("MIT, My-Custom-License", source_file_dir=data_dir),
            "UNKNOWN[My-Custom-License]",
        )
        self.assertEqual(
            get_status(
                "LicenseRef-GUST-Font-License, My-Custom-License",
                source_file_dir=data_dir,
            ),
            "UNKNOWN[My-Custom-License], APPROVED[(GUST-Font-License, b/987654321)]",
        )

        # 4. Empty or None
        self.assertEqual(get_status(""), "UNKNOWN")
        self.assertEqual(get_status(None), "UNKNOWN")

        # 5. Reciprocal licenses (only allowed in open source)
        self.assertEqual(get_status("CDDL-1.0"), "ALLOWED")
        self.assertEqual(
            get_status("CDDL-1.0", is_open_source_project=True), "ALLOWED"
        )
        self.assertEqual(
            get_status("CDDL-1.0", is_open_source_project=False),
            "RECIPROCAL_NOT_ALLOWED[CDDL-1.0]",
        )

    def test_get_license_validation_status_with_android_compatibility(self):
        get_status = metadata.fields.custom.license_allowlist.get_license_validation_status
        data_dir = os.path.join(_THIS_DIR, "data")

        self.assertEqual(
            get_status("MPL-2.0", is_open_source_project=True), "ALLOWED"
        )

        self.assertEqual(
            get_status(
                "MPL-2.0",
                is_open_source_project=False,
                android_compatible="yes",
            ),
            "ALLOWED",
        )

        self.assertEqual(
            get_status("MPL-2.0", is_open_source_project=False),
            "RECIPROCAL_NOT_ALLOWED[MPL-2.0]",
        )

        self.assertEqual(
            get_status(
                "MPL-2.0", is_open_source_project=False, android_compatible="no"
            ),
            "RECIPROCAL_NOT_ALLOWED[MPL-2.0]",
        )

        self.assertEqual(
            get_status(
                "MPL-2.0",
                is_open_source_project=False,
                source_file_dir=data_dir,
            ),
            "RECIPROCAL_NOT_ALLOWED[MPL-2.0]",
        )


if __name__ == "__main__":
    unittest.main()
