import base64 import json import subprocess import pytest from datadog_tools.utils import languages from datadog_tools.utils.languages import Language @pytest.fixture() def mock_syft_success(mocker): # Mock the subprocess call to Syft to return a predefined SBOM. return mocker.patch( "subprocess.run", return_value=mocker.Mock( stdout=json.dumps( { "artifacts": [ { "name": "python", "version": "3.13.1", } ] } ), returncode=0, ), ) @pytest.fixture() def mock_syft_failure(mocker): # Mock the subprocess call to Syft to simulate a failure. return mocker.patch( "subprocess.run", side_effect=subprocess.CalledProcessError( returncode=1, cmd=["syft"], output="", stderr="Error: failed to generate SBOM", ), ) @pytest.fixture() def mock_syft_non_json_output(mocker): # Mock the subprocess call to Syft to return non-JSON output. return mocker.patch( "subprocess.run", return_value=mocker.Mock(stdout="Not JSON output", returncode=0), ) @pytest.fixture() def mock_ecr_login(mocker): # Mock the boto3 ECR client to return a predefined authorization token. mock_ecr_client = mocker.Mock() mock_ecr_client.get_authorization_token.return_value = { "authorizationData": [{"authorizationToken": base64.b64encode(b"AWS:fakepassword").decode("utf-8")}] } return mocker.patch("boto3.client", return_value=mock_ecr_client) @pytest.mark.parametrize( "project_file, expected_language", [ pytest.param("requirements.txt", Language.PYTHON), pytest.param("pyproject.toml", Language.PYTHON), pytest.param("Pipfile", Language.PYTHON), pytest.param("setup.py", Language.PYTHON), pytest.param("package.json", Language.JAVASCRIPT), pytest.param("composer.json", Language.PHP), pytest.param("Cargo.toml", Language.RUST), pytest.param("pom.xml", None), ], ) def test_determine_language_from_manifest(project_file: str, expected_language: Language, tmp_path): """Test determine_language with various project files.""" project_file_path = tmp_path / project_file project_file_path.write_text("") language = languages.determine_language_from_manifest(tmp_path) assert language == expected_language @pytest.mark.parametrize( "project_files, expected_package_manager", [ pytest.param(["pyproject.toml", "uv.lock"], "uv"), pytest.param(["pyproject.toml", "poetry.lock"], "poetry"), pytest.param(["Pipfile", "Pipfile.lock"], "pipenv"), pytest.param(["requirements.txt"], "pip"), pytest.param(["setup.py"], "pip"), pytest.param(["package.json", "yarn.lock"], "yarn"), pytest.param(["package.json", "pnpm-lock.yaml"], "pnpm"), pytest.param(["package.json"], "npm"), pytest.param(["composer.json"], "composer"), pytest.param(["Cargo.toml"], "cargo"), pytest.param([], "unknown"), ], ) def test_determine_package_manager(project_files: list[str], expected_package_manager: str, tmp_path): """Test determine_package_manager with various project files.""" for project_file in project_files: (tmp_path / project_file).write_text("") package_manager = languages.determine_package_manager(tmp_path) assert package_manager == expected_package_manager @pytest.mark.parametrize( "image_repo, image_tag, expected_metadata", [ pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "python313", (Language.PYTHON, "3.13") ), pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "python313-blah", (Language.PYTHON, "3.13"), ), pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "lambda-python313", (Language.PYTHON, "3.13"), ), pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "lambda-python313-b", (Language.PYTHON, "3.13"), ), pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "node20", (Language.JAVASCRIPT, "20") ), pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "node20-build", (Language.JAVASCRIPT, "20"), ), pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "debian-php82", (Language.PHP, "8.2") ), pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "kafka-connect", (Language.JAVA, "11") ), pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "kafka-connect77", (Language.JAVA, "17"), ), pytest.param( "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "kafka-connect81", (Language.JAVA, "21"), ), pytest.param("amazon/aws-lambda-python", "3.13", (Language.PYTHON, "3.13")), pytest.param("amazon/aws-lambda-python", "3.13-x", (Language.PYTHON, "3.13")), pytest.param("public.ecr.aws/lambda/python", "3.13", (Language.PYTHON, "3.13")), pytest.param("public.ecr.aws/lambda/python", "3.13-x", (Language.PYTHON, "3.13")), pytest.param("amazon/aws-lambda-nodejs", "22", (Language.JAVASCRIPT, "22")), pytest.param("amazon/aws-lambda-nodejs", "22-x", (Language.JAVASCRIPT, "22")), pytest.param("public.ecr.aws/lambda/nodejs", "22", (Language.JAVASCRIPT, "22")), pytest.param("public.ecr.aws/lambda/nodejs", "22-x", (Language.JAVASCRIPT, "22")), pytest.param("python", "3.13", (Language.PYTHON, "3.13")), pytest.param("python", "3.13-x", (Language.PYTHON, "3.13")), pytest.param("node", "22", (Language.JAVASCRIPT, "22")), pytest.param("node", "22-x", (Language.JAVASCRIPT, "22")), pytest.param("amazoncorretto", "21", (Language.JAVA, "21")), pytest.param("amazoncorretto", "21-alpine3.21", (Language.JAVA, "21")), pytest.param("amazoncorretto/amazoncorretto", "17", (Language.JAVA, "17")), pytest.param("public.ecr.aws/amazoncorretto/amazoncorretto", "17", (Language.JAVA, "17")), pytest.param("public.ecr.aws/amazoncorretto/amazoncorretto", "17-al2023", (Language.JAVA, "17")), pytest.param("alpine", "latest", None), ], ) def test_determine_metadata_from_image_name(image_repo, image_tag, expected_metadata: tuple[Language, str] | None): metadata = languages.determine_metadata_from_image_name(image_repo, image_tag) assert metadata == expected_metadata @pytest.mark.parametrize( "sbom_data, expected_metadata", [ pytest.param( { "artifacts": [ { "name": "stdlib", "version": "go1.20.0", } ] }, (Language.GO, "1.20"), ), pytest.param( { "artifacts": [ { "name": "python", "version": "3.13.1", } ] }, (Language.PYTHON, "3.13"), ), pytest.param( { "artifacts": [ { "name": "php", "version": "8.2.1", } ] }, (Language.PHP, "8.2"), ), pytest.param( { "artifacts": [ { "name": "node", "version": "20.3.1", } ] }, (Language.JAVASCRIPT, "20"), ), pytest.param( { "artifacts": [ { "name": "ruby", "version": "3.1.2", } ] }, (Language.RUBY, "3.1"), ), pytest.param( { "artifacts": [ { "name": "rust", "version": "1.70.0", } ] }, (Language.RUST, "1.70"), ), pytest.param( { "artifacts": [ { "name": "openjdk", "version": "17.0.5", } ] }, (Language.JAVA, "17"), ), pytest.param( { "artifacts": [ { "name": "some-other-package", "version": "1.0.0", } ] }, None, ), ], ) def test_determine_metadata_from_sbom(sbom_data, expected_metadata, mocker): mocker.patch("datadog_tools.utils.languages._generate_sbom", return_value=sbom_data) metadata = languages.determine_metadata_from_sbom("some_repo", "some_tag") assert metadata == expected_metadata def test_generate_sbom_public_image(mock_syft_success): sbom = languages._generate_sbom("alpine", "latest") assert sbom == { "artifacts": [ { "name": "python", "version": "3.13.1", } ] } def test_generate_sbom_ecr_image(mock_syft_success, mock_ecr_login): sbom = languages._generate_sbom("086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "python313") assert sbom == { "artifacts": [ { "name": "python", "version": "3.13.1", } ] } assert mock_syft_success.call_count == 1 syft_env_vars = mock_syft_success.call_args.kwargs.get("env", {}) assert syft_env_vars.get("SYFT_REGISTRY_AUTH_USERNAME") == "AWS" assert syft_env_vars.get("SYFT_REGISTRY_AUTH_PASSWORD") == "fakepassword" assert syft_env_vars.get("SYFT_REGISTRY_AUTH_AUTHORITY") == "086679231553.dkr.ecr.us-east-1.amazonaws.com" def test_generate_sbom_syft_failure(mock_syft_failure): with pytest.raises(subprocess.CalledProcessError): languages._generate_sbom("alpine", "latest") def test_generate_sbom_syft_output_not_json(mock_syft_non_json_output): with pytest.raises(json.JSONDecodeError): languages._generate_sbom("alpine", "latest") @pytest.mark.parametrize( "runtime, expected_metadata", [ pytest.param("python3.11", (Language.PYTHON, "3.11")), pytest.param("python3.12", (Language.PYTHON, "3.12")), pytest.param("python3.9", (Language.PYTHON, "3.9")), pytest.param("python3.13", (Language.PYTHON, "3.13")), pytest.param("nodejs20.x", (Language.JAVASCRIPT, "20")), pytest.param("nodejs18.x", (Language.JAVASCRIPT, "18")), pytest.param("nodejs16.x", (Language.JAVASCRIPT, "16")), pytest.param("nodejs8.10", (Language.JAVASCRIPT, "8")), pytest.param("nodejs10.x", (Language.JAVASCRIPT, "10")), pytest.param("nodejs12.x", (Language.JAVASCRIPT, "12")), pytest.param("java11", None), pytest.param("dotnet6", None), pytest.param("ruby3.2", None), pytest.param("go1.x", None), pytest.param("", None), ], ) def test_determine_metadata_from_lambda_runtime(runtime, expected_metadata): """Test determine_metadata_from_lambda_runtime with various runtime strings.""" metadata = languages.determine_metadata_from_lambda_runtime(runtime) assert metadata == expected_metadata @pytest.mark.parametrize( "nvmrc_content, expected_metadata", [ pytest.param("20", (Language.JAVASCRIPT, "20")), pytest.param("v20", (Language.JAVASCRIPT, "20")), pytest.param("20.11.0", (Language.JAVASCRIPT, "20")), pytest.param("v20.11.0", (Language.JAVASCRIPT, "20")), pytest.param("18.16.1", (Language.JAVASCRIPT, "18")), pytest.param("v18.16.1", (Language.JAVASCRIPT, "18")), pytest.param(" 20.11.0 ", (Language.JAVASCRIPT, "20")), pytest.param(" v20 \n", (Language.JAVASCRIPT, "20")), ], ) def test_determine_metadata_from_nvmrc(nvmrc_content, expected_metadata, tmp_path): """Test determine_metadata_from_nvmrc with various .nvmrc file contents.""" nvmrc_path = tmp_path / ".nvmrc" nvmrc_path.write_text(nvmrc_content) metadata = languages.determine_metadata_from_nvmrc(tmp_path) assert metadata == expected_metadata @pytest.mark.parametrize( "engines_node, expected_metadata", [ pytest.param("20.x", (Language.JAVASCRIPT, "20")), pytest.param("22.x", (Language.JAVASCRIPT, "22")), pytest.param("24.x", (Language.JAVASCRIPT, "24")), pytest.param("^20.0.0", (Language.JAVASCRIPT, "20")), pytest.param("^22.0.0", (Language.JAVASCRIPT, "22")), pytest.param(">=20.0.0", (Language.JAVASCRIPT, "20")), ], ) def test_determine_metadata_from_package_json_engines(engines_node, expected_metadata, tmp_path): """Test determine_metadata_from_package_json_engines with various engines.node strings.""" (tmp_path / "package.json").write_text(json.dumps({"engines": {"node": engines_node}})) assert languages.determine_metadata_from_package_json_engines(tmp_path) == expected_metadata def test_determine_metadata_from_package_json_engines_no_engines_field(tmp_path): (tmp_path / "package.json").write_text(json.dumps({"name": "test"})) assert languages.determine_metadata_from_package_json_engines(tmp_path) is None def test_determine_metadata_from_package_json_engines_no_file(tmp_path): assert languages.determine_metadata_from_package_json_engines(tmp_path) is None