import os import subprocess from unittest.mock import MagicMock import pytest import yaml from datadog_api_client.v1.model.dashboard_summary_definition import DashboardSummaryDefinition from datadog_api_client.v2.model.entity_attributes import EntityAttributes from datadog_api_client.v2.model.entity_data import EntityData from datadog_api_client.v2.model.entity_v3 import EntityV3 from datadog_tools.actions.software_catalog import SoftwareCatalog from datadog_tools.connectors.aws import AWS from datadog_tools.connectors.datadog import Datadog from datadog_tools.connectors.sentry import Sentry from datadog_tools.connectors.sonar import Sonar from datadog_tools.connectors.swagger import Swagger from datadog_tools.models.aws import AWSResource from datadog_tools.models.software_catalog import KindEnum from datadog_tools.utils.languages import Language @pytest.fixture() def mock_datadog_connector(): mock_datadog = MagicMock(spec=Datadog) mock_datadog.get_dashboards.return_value = [ DashboardSummaryDefinition(id="123", title="qa-test-service dashboard", url="/dashboard/123"), DashboardSummaryDefinition(id="456", title="prod-test-service dashboard", url="/dashboard/456"), DashboardSummaryDefinition(id="123", title="qa-test-service-other dashboard", url="/dashboard/123"), DashboardSummaryDefinition(id="234", title="shared-test-service dashboard", url="/dashboard/234"), DashboardSummaryDefinition(id="789", title="uat-test-service dashboard", url="/dashboard/789"), ] mock_datadog.list_entities.return_value = [ EntityV3( name="test-service", kind="service", owner="team1", extensions={ "sonymusic-pde.com/metadata-source": { "revision": "somerevision", } }, ).to_dict(), EntityV3( name="test-service-2", kind="service", owner="team1", ).to_dict(), ] return mock_datadog @pytest.fixture() def mock_sentry_connector(): mock_sentry = MagicMock(spec=Sentry) mock_sentry.organization = "test-org" mock_sentry.get_project.side_effect = lambda project: project return mock_sentry @pytest.fixture() def mock_sonar_connector(): mock_sonar = MagicMock(spec=Sonar) mock_sonar.get_project.return_value = {"key": "test-service"} mock_sonar.get_project_url.side_effect = lambda repo: f"https://sonar.example.com/dashboard?id={repo}" return mock_sonar @pytest.fixture() def mock_swagger_connector(): mock_swagger = MagicMock(spec=Swagger) mock_swagger.spec_exists.return_value = True mock_swagger.spec_url.side_effect = lambda service: f"https://swagger.example.com/service/{service}" return mock_swagger @pytest.fixture() def mock_aws_connector(): mock_aws = MagicMock(spec=AWS) mock_aws.find_resources_by_type_and_service_name.return_value = [ AWSResource( arn="arn:aws:ecs:us-east-1:123456789012:service/prod-test-service/prod-test-service", account_id="123456789012", application_family="devops", environment="prod", resource_type="ecs:service", region="us-east-1", terraform_github_repository="terraform-infra", terraform_github_path="prod/test-service", terraformed=True, ), AWSResource( arn="arn:aws:ecs:us-east-1:234567890123:service/qa-test-service/qa-test-service", account_id="234567890123", application_family="devops", environment="qa", resource_type="ecs:service", region="us-east-1", terraform_github_repository="terraform-infra", terraform_github_path="qa/test-service", terraformed=True, ), ] return mock_aws @pytest.fixture() def mock_aws_connector_lambda(): mock_aws = MagicMock(spec=AWS) mock_aws.find_resources_by_type_and_service_name.return_value = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:prod-test-lambda-function", account_id="123456789012", application_family="devops", environment="prod", resource_type="lambda:function", region="us-east-1", runtime="python3.11", terraformed=True, ), AWSResource( arn="arn:aws:ecs:us-east-1:234567890123:function:qa-test-lambda-function", account_id="234567890123", application_family="devops", environment="qa", resource_type="lambda:function", region="us-east-1", runtime="python3.11", terraformed=True, ), # Include a resource that is missing tags AWSResource( arn="arn:aws:ecs:us-east-1:234567890123:function:dev-test-lambda-function", account_id="234567890123", resource_type="lambda:function", region="us-east-1", ), ] return mock_aws @pytest.fixture() def mock_git(mocker): mocker.patch( "datadog_tools.actions.software_catalog.subprocess.run", return_value=MagicMock( stdout="abcdef", stderr="", returncode=0, ), ) @pytest.fixture() def mock_git_error(mocker): mocker.patch( "datadog_tools.actions.software_catalog.subprocess.run", side_effect=subprocess.CalledProcessError(1, "git"), ) @pytest.fixture() def valid_service_definition(): return { "apiVersion": "v3", "kind": "service", "metadata": { "name": "test-service", "description": "Test service", "owner": "test-team", "tags": [ "public:false", ], }, "spec": { "type": "web", }, } @pytest.fixture() def valid_service_definition_with_inherit_from(): return { "apiVersion": "v3", "kind": "service", "metadata": { "name": "test-service", "description": "Test service", "inheritFrom": "system:test-system", "tags": [ "public:false", ], }, "spec": { "type": "web", }, } @pytest.fixture() def valid_service_definition_with_languages_override(): return { "apiVersion": "v3", "kind": "service", "metadata": { "name": "test-service", "description": "Test service", "owner": "test-team", "tags": [ "public:false", ], }, "spec": { "type": "web", "languages": [ "php", "python", ], }, } @pytest.fixture() def valid_service_definition_with_tags_overrides(): return { "apiVersion": "v3", "kind": "service", "metadata": { "name": "test-service", "description": "Test service", "owner": "test-team", "tags": [ "application_family:test-app-family", "language_version:3.5", "package_manager:nuget", "parent_image_repo:alpine", "parent_image_tag:latest", "prod_aws_account_id:111222333444", "prod_aws_account_id:444333222111", "public:false", "qa_aws_account_id:555666777888", "qa_aws_account_id:888777666555", "terraformed:false", ], }, "spec": { "type": "web", }, } @pytest.fixture() def valid_lambda_service_definition(): return { "apiVersion": "v3", "kind": "service", "metadata": { "name": "test-lambda-function", "description": "Test lambda function", "owner": "test-team", "tags": [ "public:false", ], }, "spec": { "type": "function", }, } @pytest.fixture() def valid_library_definition(): return { "apiVersion": "v3", "kind": "library", "metadata": { "name": "test-library", "description": "Test library", "owner": "test-team", "tags": [ "public:false", ], }, "spec": { "type": "sdk", }, } @pytest.fixture() def valid_frontend_definition(): return { "apiVersion": "v3", "kind": "frontend", "metadata": { "name": "rum_application:00000000-1111-2222-3333-444444444444", "description": "Test frontend", "owner": "test-team", "tags": [ "public:false", ], }, "spec": { "type": "browser", }, } @pytest.fixture() def invalid_service_definition(): return {"apiVersion": "v2", "kind": "service"} @pytest.fixture() def valid_service_definition_file(valid_service_definition, tmp_path): return create_entity_definition(valid_service_definition, tmp_path) @pytest.fixture() def valid_service_definition_file_with_inherit_from(valid_service_definition_with_inherit_from, tmp_path): return create_entity_definition(valid_service_definition_with_inherit_from, tmp_path) @pytest.fixture() def valid_service_definition_file_in_subdir(valid_service_definition, tmp_path): return create_entity_definition(valid_service_definition, tmp_path, "some/subdir") @pytest.fixture() def valid_service_definition_file_with_languages_override(valid_service_definition_with_languages_override, tmp_path): return create_entity_definition(valid_service_definition_with_languages_override, tmp_path) @pytest.fixture() def valid_service_definition_file_with_tags_overrides(valid_service_definition_with_tags_overrides, tmp_path): return create_entity_definition(valid_service_definition_with_tags_overrides, tmp_path) @pytest.fixture() def valid_lambda_service_definition_file(valid_lambda_service_definition, tmp_path): return create_entity_definition(valid_lambda_service_definition, tmp_path) @pytest.fixture() def valid_library_definition_file(valid_library_definition, tmp_path): return create_entity_definition(valid_library_definition, tmp_path) @pytest.fixture() def valid_frontend_definition_file(valid_frontend_definition, tmp_path): return create_entity_definition(valid_frontend_definition, tmp_path) @pytest.fixture() def invalid_service_definition_file(invalid_service_definition, tmp_path): return create_entity_definition(invalid_service_definition, tmp_path) @pytest.fixture() def simple_dockerfile(tmp_path): dockerfile_path = os.path.join(tmp_path, "Dockerfile") with open(dockerfile_path, "w") as file: file.write("FROM 086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images:python313 AS base\n") return dockerfile_path @pytest.fixture() def multistage_dockerfile(tmp_path): dockerfile_path = os.path.join(tmp_path, "Dockerfile") with open(dockerfile_path, "w") as file: file.write("FROM alpine:latest AS builder\n") file.write("FROM 086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images:python313 AS base\n") file.write("FROM base AS deploy\n") return dockerfile_path @pytest.fixture() def multistage_dockerfile_with_additional_stages(tmp_path): dockerfile_path = os.path.join(tmp_path, "Dockerfile") with open(dockerfile_path, "w") as file: file.write("FROM 086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images:python313 AS deploy\n") file.write("FROM alpine:latest as unit-lint\n") return dockerfile_path @pytest.fixture() def dockerfile_with_args(tmp_path): dockerfile_path = os.path.join(tmp_path, "Dockerfile") with open(dockerfile_path, "w") as file: file.write("ARG IMAGE_TAG=python313\n") file.write("FROM 086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images:${IMAGE_TAG}\n") return dockerfile_path @pytest.fixture() def mock_sbom(mocker): mocker.patch("datadog_tools.utils.languages.determine_metadata_from_image_name", return_value=None) mocker.patch("datadog_tools.utils.languages.determine_metadata_from_sbom", return_value=(Language.RUST, "1.91")) def create_entity_definition(definition, tmp_path, subdir=""): """Create an entity definition file in the given temporary path.""" entity_dir = os.path.join(tmp_path, subdir) os.makedirs(entity_dir, exist_ok=True) entity_file = os.path.join(entity_dir, "software-catalog.yaml") with open(entity_file, "w") as file: yaml.dump(definition, file) return str(tmp_path), subdir def test_upsert_service_with_defaults( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, mock_swagger_connector, mock_aws_connector, valid_service_definition_file, mock_git, ): repository_path, service_path = valid_service_definition_file """Test upsert_service with default values.""" software_catalog = SoftwareCatalog( datadog=mock_datadog_connector, sentry=mock_sentry_connector, sonar=mock_sonar_connector, swagger=mock_swagger_connector, aws=mock_aws_connector, ) software_catalog.upsert_service( repository_url="https://github.com/theorchard/test-service", repository_path=repository_path, service_path=service_path, ) mock_datadog_connector.upsert_catalog_entity.assert_called_once_with( body={ "apiVersion": "v3", "kind": "service", "metadata": { "name": "test-service", "description": "Test service", "owner": "test-team", "tags": [ "application_family:devops", "aws_service:fargate", "dockerized:false", "metadata_origin:datadog-tools", "metadata_revision:abcdef", "package_manager:unknown", "prod_aws_account_id:123456789012", "public:false", "qa_aws_account_id:234567890123", "service_type:web", "terraformed:true", ], "links": [ { "name": "AWS Console (Prod - us-east-1)", "type": "other", "provider": "aws", "url": "https://us-east-1.console.aws.amazon.com/ecs/v2/clusters/prod-test-service/services/prod-test-service", }, { "name": "AWS Console (QA - us-east-1)", "type": "other", "provider": "aws", "url": "https://us-east-1.console.aws.amazon.com/ecs/v2/clusters/qa-test-service/services/qa-test-service", }, { "name": "GitHub Repository", "type": "repo", "provider": "github", "url": "https://github.com/theorchard/test-service", }, { "name": "Sentry Project (Prod)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/prod-test-service/", }, { "name": "Sentry Project (QA)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/qa-test-service/", }, { "name": "Sentry Project (Shared)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/shared-test-service/", }, { "name": "Sentry Project (UAT)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/uat-test-service/", }, { "name": "SonarQube Project", "type": "other", "provider": "sonarqube", "url": "https://sonar.example.com/dashboard?id=test-service", }, { "name": "Swagger Spec", "type": "other", "url": "https://swagger.example.com/service/test-service", }, { "name": "Terraform Infra (Prod - 123456789012)", "provider": "github", "type": "repo", "url": "https://github.com/theorchard/terraform-infra/tree/master/prod/test-service", }, { "name": "Terraform Infra (QA - 234567890123)", "provider": "github", "type": "repo", "url": "https://github.com/theorchard/terraform-infra/tree/master/qa/test-service", }, { "name": "prod-test-service dashboard", "type": "dashboard", "url": "https://sonymusic-pde.datadoghq.com/dashboard/456", }, { "name": "qa-test-service dashboard", "type": "dashboard", "url": "https://sonymusic-pde.datadoghq.com/dashboard/123", }, { "name": "shared-test-service dashboard", "type": "dashboard", "url": "https://sonymusic-pde.datadoghq.com/dashboard/234", }, { "name": "uat-test-service dashboard", "type": "dashboard", "url": "https://sonymusic-pde.datadoghq.com/dashboard/789", }, ], }, "datadog": { "codeLocations": [{"repositoryURL": "https://github.com/theorchard/test-service", "paths": ["**"]}] }, "spec": {"type": "web"}, "extensions": { "sonymusic-pde.com/metadata-source": { "repositoryPath": repository_path, "repositoryURL": "https://github.com/theorchard/test-service", "revision": "abcdef", "servicePath": service_path, } }, } ) def test_upsert_lambda_service_with_defaults( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, mock_swagger_connector, mock_aws_connector_lambda, valid_lambda_service_definition_file, mock_git, ): """Test upsert_service with default values for a lambda function.""" repository_path, service_path = valid_lambda_service_definition_file software_catalog = SoftwareCatalog( datadog=mock_datadog_connector, sentry=mock_sentry_connector, sonar=mock_sonar_connector, swagger=mock_swagger_connector, aws=mock_aws_connector_lambda, ) software_catalog.upsert_service( repository_url="https://github.com/theorchard/test-lambda-repo", repository_path=repository_path, service_path=service_path, ) mock_datadog_connector.upsert_catalog_entity.assert_called_once_with( body={ "apiVersion": "v3", "kind": "service", "metadata": { "name": "test-lambda-function", "description": "Test lambda function", "owner": "test-team", "tags": [ "application_family:devops", "aws_service:lambda", "dockerized:false", "language_version:3.11", "metadata_origin:datadog-tools", "metadata_revision:abcdef", "package_manager:unknown", "prod_aws_account_id:123456789012", "public:false", "qa_aws_account_id:234567890123", "service_type:function", "terraformed:true", ], "links": [ { "name": "AWS Console (Prod - us-east-1)", "type": "other", "provider": "aws", "url": "https://us-east-1.console.aws.amazon.com/lambda/home#/functions/prod-test-lambda-function", }, { "name": "AWS Console (QA - us-east-1)", "type": "other", "provider": "aws", "url": "https://us-east-1.console.aws.amazon.com/lambda/home#/functions/qa-test-lambda-function", }, { "name": "GitHub Repository", "type": "repo", "provider": "github", "url": "https://github.com/theorchard/test-lambda-repo", }, { "name": "Sentry Project (Prod)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/prod-test-lambda-function/", }, { "name": "Sentry Project (QA)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/qa-test-lambda-function/", }, { "name": "Sentry Project (Shared)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/shared-test-lambda-function/", }, { "name": "Sentry Project (UAT)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/uat-test-lambda-function/", }, { "name": "SonarQube Project", "type": "other", "provider": "sonarqube", "url": "https://sonar.example.com/dashboard?id=test-lambda-repo", }, ], }, "datadog": { "codeLocations": [{"repositoryURL": "https://github.com/theorchard/test-lambda-repo", "paths": ["**"]}] }, "spec": { "type": "function", "languages": ["python"], }, "extensions": { "sonymusic-pde.com/metadata-source": { "repositoryPath": repository_path, "repositoryURL": "https://github.com/theorchard/test-lambda-repo", "revision": "abcdef", "servicePath": service_path, } }, } ) def test_upsert_library_with_defaults( mock_datadog_connector, valid_library_definition_file, mock_git, ): """Test upsert_library with default values.""" repository_path, library_path = valid_library_definition_file software_catalog = SoftwareCatalog(datadog=mock_datadog_connector) software_catalog.upsert_library( repository_url="https://github.com/theorchard/test-library-repo", repository_path=repository_path, library_path=library_path, ) mock_datadog_connector.upsert_catalog_entity.assert_called_once_with( body={ "apiVersion": "v3", "kind": "library", "datadog": { "codeLocations": [{"repositoryURL": "https://github.com/theorchard/test-library-repo", "paths": ["**"]}] }, "metadata": { "name": "test-library", "description": "Test library", "owner": "test-team", "tags": [ "library_type:sdk", "metadata_origin:datadog-tools", "metadata_revision:abcdef", "package_manager:unknown", "public:false", ], "links": [ { "name": "GitHub Repository", "type": "repo", "provider": "github", "url": "https://github.com/theorchard/test-library-repo", }, ], }, "extensions": { "sonymusic-pde.com/metadata-source": { "repositoryPath": repository_path, "repositoryURL": "https://github.com/theorchard/test-library-repo", "revision": "abcdef", "libraryPath": library_path, } }, "spec": {"type": "sdk"}, } ) def test_upsert_frontend_with_defaults( mock_datadog_connector, valid_frontend_definition_file, mock_git, ): """Test upsert_frontend with default values.""" repository_path, frontend_path = valid_frontend_definition_file software_catalog = SoftwareCatalog(datadog=mock_datadog_connector) software_catalog.upsert_frontend( repository_url="https://github.com/theorchard/test-frontend-repo", repository_path=repository_path, frontend_path=frontend_path, ) mock_datadog_connector.upsert_catalog_entity.assert_called_once_with( body={ "apiVersion": "v3", "kind": "frontend", "datadog": { "codeLocations": [ {"repositoryURL": "https://github.com/theorchard/test-frontend-repo", "paths": ["**"]} ] }, "metadata": { "name": "rum_application:00000000-1111-2222-3333-444444444444", "description": "Test frontend", "owner": "test-team", "tags": [ "frontend_type:browser", "metadata_origin:datadog-tools", "metadata_revision:abcdef", "package_manager:unknown", "public:false", ], "links": [ { "name": "GitHub Repository", "type": "repo", "url": "https://github.com/theorchard/test-frontend-repo", "provider": "github", } ], }, "extensions": { "sonymusic-pde.com/metadata-source": { "repositoryURL": "https://github.com/theorchard/test-frontend-repo", "repositoryPath": repository_path, "revision": "abcdef", "frontendPath": frontend_path, } }, "spec": {"type": "browser"}, } ) def test_get_entity_schema(mock_datadog_connector): software_catalog = SoftwareCatalog(datadog=mock_datadog_connector) software_catalog.get_entity_schema("test-service", kind=KindEnum.service) mock_datadog_connector.get_entity_schema.assert_called_once_with( name="test-service", kind=KindEnum.service, ) def test_list_entities(mock_datadog_connector): software_catalog = SoftwareCatalog(datadog=mock_datadog_connector) entities = software_catalog.list_entities(kind=KindEnum.service) mock_datadog_connector.list_entities.assert_called_once_with(kind=KindEnum.service) assert len(entities) == 1 def test_list_entities_raw(mock_datadog_connector): software_catalog = SoftwareCatalog(datadog=mock_datadog_connector) entities = software_catalog.list_entities(kind=KindEnum.service, raw=True) mock_datadog_connector.list_entities.assert_called_once_with(kind=KindEnum.service) assert len(entities) == 2 def test_upsert_service_with_overrides( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, mock_swagger_connector, mock_aws_connector, valid_service_definition_file_in_subdir, mock_git, ): """Test upsert_service with overrides.""" repository_path, service_path = valid_service_definition_file_in_subdir software_catalog = SoftwareCatalog( datadog=mock_datadog_connector, sentry=mock_sentry_connector, sonar=mock_sonar_connector, swagger=mock_swagger_connector, aws=mock_aws_connector, ) software_catalog.upsert_service( repository_url="https://github.com/theorchard/test-service", repository_path=repository_path, service_path=service_path, pipeline_url="https://pipeline.theorchard.io/test-service", ) mock_datadog_connector.upsert_catalog_entity.assert_called_once_with( body={ "apiVersion": "v3", "kind": "service", "metadata": { "name": "test-service", "description": "Test service", "owner": "test-team", "tags": [ "application_family:devops", "aws_service:fargate", "dockerized:false", "metadata_origin:datadog-tools", "metadata_revision:abcdef", "package_manager:unknown", "prod_aws_account_id:123456789012", "public:false", "qa_aws_account_id:234567890123", "service_type:web", "terraformed:true", ], "links": [ { "name": "AWS Console (Prod - us-east-1)", "type": "other", "provider": "aws", "url": "https://us-east-1.console.aws.amazon.com/ecs/v2/clusters/prod-test-service/services/prod-test-service", }, { "name": "AWS Console (QA - us-east-1)", "type": "other", "provider": "aws", "url": "https://us-east-1.console.aws.amazon.com/ecs/v2/clusters/qa-test-service/services/qa-test-service", }, { "name": "CI/CD Pipeline", "type": "other", "provider": "jenkins", "url": "https://pipeline.theorchard.io/test-service", }, { "name": "GitHub Repository", "type": "repo", "provider": "github", "url": "https://github.com/theorchard/test-service", }, { "name": "Sentry Project (Prod)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/prod-test-service/", }, { "name": "Sentry Project (QA)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/qa-test-service/", }, { "name": "Sentry Project (Shared)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/shared-test-service/", }, { "name": "Sentry Project (UAT)", "type": "other", "provider": "sentry", "url": "https://test-org.sentry.io/projects/uat-test-service/", }, { "name": "SonarQube Project", "type": "other", "provider": "sonarqube", "url": "https://sonar.example.com/dashboard?id=test-service", }, { "name": "Swagger Spec", "type": "other", "url": "https://swagger.example.com/service/test-service", }, { "name": "Terraform Infra (Prod - 123456789012)", "provider": "github", "type": "repo", "url": "https://github.com/theorchard/terraform-infra/tree/master/prod/test-service", }, { "name": "Terraform Infra (QA - 234567890123)", "provider": "github", "type": "repo", "url": "https://github.com/theorchard/terraform-infra/tree/master/qa/test-service", }, { "name": "prod-test-service dashboard", "type": "dashboard", "url": "https://sonymusic-pde.datadoghq.com/dashboard/456", }, { "name": "qa-test-service dashboard", "type": "dashboard", "url": "https://sonymusic-pde.datadoghq.com/dashboard/123", }, { "name": "shared-test-service dashboard", "type": "dashboard", "url": "https://sonymusic-pde.datadoghq.com/dashboard/234", }, { "name": "uat-test-service dashboard", "type": "dashboard", "url": "https://sonymusic-pde.datadoghq.com/dashboard/789", }, ], }, "datadog": { "codeLocations": [ { "repositoryURL": "https://github.com/theorchard/test-service", "paths": ["some/subdir/**"], } ] }, "spec": {"type": "web"}, "extensions": { "sonymusic-pde.com/metadata-source": { "pipelineURL": "https://pipeline.theorchard.io/test-service", "repositoryPath": repository_path, "repositoryURL": "https://github.com/theorchard/test-service", "revision": "abcdef", "servicePath": service_path, } }, } ) def test_upsert_service_dry_run( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, mock_swagger_connector, valid_service_definition_file, mock_git, ): """Test upsert_service with default values.""" repository_path, service_path = valid_service_definition_file software_catalog = SoftwareCatalog( datadog=mock_datadog_connector, sentry=mock_sentry_connector, sonar=mock_sonar_connector, swagger=mock_swagger_connector, ) software_catalog.upsert_service( repository_url="https://github.com/theorchard/test-service", repository_path=repository_path, service_path=service_path, dry_run=True, ) mock_datadog_connector.upsert_catalog_entity.assert_not_called() def test_upsert_service_override_languages( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, mock_swagger_connector, valid_service_definition_file_with_languages_override, mock_git, ): """Test upsert_service with override for languages.""" repository_path, service_path = valid_service_definition_file_with_languages_override software_catalog = SoftwareCatalog( datadog=mock_datadog_connector, sentry=mock_sentry_connector, sonar=mock_sonar_connector, swagger=mock_swagger_connector, ) software_catalog.upsert_service( repository_url="https://github.com/theorchard/test-service", repository_path=repository_path, service_path=service_path, ) mock_datadog_connector.upsert_catalog_entity.assert_called_once() call_args = mock_datadog_connector.upsert_catalog_entity.call_args body = call_args[1]["body"] assert "languages" in body["spec"] assert set(body["spec"]["languages"]) == {"php", "python"} def test_upsert_service_override_tags( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, mock_swagger_connector, valid_service_definition_file_with_tags_overrides, mock_git, simple_dockerfile, ): """Test upsert_service with overrides for tags.""" repository_path, service_path = valid_service_definition_file_with_tags_overrides software_catalog = SoftwareCatalog( datadog=mock_datadog_connector, sentry=mock_sentry_connector, sonar=mock_sonar_connector, swagger=mock_swagger_connector, ) software_catalog.upsert_service( repository_url="https://github.com/theorchard/test-service", repository_path=repository_path, service_path=service_path, ) mock_datadog_connector.upsert_catalog_entity.assert_called_once() call_args = mock_datadog_connector.upsert_catalog_entity.call_args body = call_args[1]["body"] assert "tags" in body["metadata"] tags = body["metadata"]["tags"] assert tags == [ "application_family:test-app-family", "aws_service:fargate", "dockerized:true", "language_version:3.5", "metadata_origin:datadog-tools", "metadata_revision:abcdef", "package_manager:nuget", "parent_image_repo:alpine", "parent_image_tag:latest", "prod_aws_account_id:111222333444", "prod_aws_account_id:444333222111", "public:false", "qa_aws_account_id:555666777888", "qa_aws_account_id:888777666555", "service_type:web", "terraformed:false", ] @pytest.mark.parametrize( "dockerfile_fixture_name, expected_parent_image_repo, expected_parent_image_tag, expected_language_version", [ pytest.param( "simple_dockerfile", "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "python313", "3.13", id="Simple Dockerfile", ), pytest.param( "multistage_dockerfile", "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "python313", "3.13", id="Multi-stage Dockerfile", ), pytest.param( "multistage_dockerfile_with_additional_stages", "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "python313", "3.13", id="Multi-stage Dockerfile with additional non-deploy stages", ), pytest.param( "dockerfile_with_args", "086679231553.dkr.ecr.us-east-1.amazonaws.com/docker-parent-images", "python313", "3.13", id="Dockerfile with args in FROM statement", ), ], ) def test_upsert_service_with_dockerfile( dockerfile_fixture_name, expected_parent_image_repo, expected_parent_image_tag, expected_language_version, request, mock_datadog_connector, valid_service_definition_file, mock_git, ): """Test upsert_service with a Dockerfile present.""" repository_path, service_path = valid_service_definition_file request.getfixturevalue(dockerfile_fixture_name) software_catalog = SoftwareCatalog(datadog=mock_datadog_connector) body = software_catalog.upsert_service( repository_url="https://github.com/theorchard/test-service", repository_path=repository_path, service_path=service_path, ) assert "dockerized:true" in body["metadata"]["tags"] assert f"parent_image_repo:{expected_parent_image_repo}" in body["metadata"]["tags"] assert f"parent_image_tag:{expected_parent_image_tag}" in body["metadata"]["tags"] assert f"language_version:{expected_language_version}" in body["metadata"]["tags"] def test_upsert_service_with_git_error(mock_datadog_connector, valid_service_definition_file, mock_git_error): """Test upsert_service with a Git error.""" repository_path, service_path = valid_service_definition_file software_catalog = SoftwareCatalog(datadog=mock_datadog_connector) body = software_catalog.upsert_service( repository_url="https://github.com/theorchard/test-service", repository_path=repository_path, service_path=service_path, ) assert "metadata_revision:unknown" in body["metadata"]["tags"] def test_upsert_service_retrieves_language_metadata_from_sbom( mock_datadog_connector, simple_dockerfile, valid_service_definition_file, mock_sbom ): """Test upsert_service retrieves language metadata by analysing parent image SBOM if it can't otherwise be determined.""" repository_path, service_path = valid_service_definition_file software_catalog = SoftwareCatalog(datadog=mock_datadog_connector) body = software_catalog.upsert_service( repository_url="https://github.com/theorchard/test-service", repository_path=repository_path, service_path=service_path, ) assert ["rust"] == body["spec"]["languages"] assert "language_version:1.91" in body["metadata"]["tags"] def test_validate_service_valid( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, valid_service_definition_file ): """Test validate_service with a valid service definition.""" repository_path, service_path = valid_service_definition_file mock_datadog_connector.get_teams.return_value = ["test-team"] software_catalog = SoftwareCatalog(mock_datadog_connector, mock_sentry_connector, mock_sonar_connector) software_catalog.validate_service(repository_path=repository_path, service_path=service_path) def test_validate_service_with_inherit_from( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, valid_service_definition_file_with_inherit_from ): """Test validate_service with a valid service definition that inherits metadata from a system.""" repository_path, service_path = valid_service_definition_file_with_inherit_from mock_datadog_connector.get_teams.return_value = ["test-team"] mock_datadog_connector.get_system.return_value = EntityData( attributes=EntityAttributes( owner="test-team", tags=[ "application_family:devops", ], ) ) software_catalog = SoftwareCatalog(mock_datadog_connector, mock_sentry_connector, mock_sonar_connector) software_catalog.validate_service(repository_path=repository_path, service_path=service_path) def test_validate_service_with_inherit_from_non_existent_system( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, valid_service_definition_file_with_inherit_from ): """Test validate_service with a valid service definition that inherits metadata from a system.""" repository_path, service_path = valid_service_definition_file_with_inherit_from mock_datadog_connector.get_teams.return_value = ["test-team"] software_catalog = SoftwareCatalog(mock_datadog_connector, mock_sentry_connector, mock_sonar_connector) with pytest.raises(ValueError): software_catalog.validate_service(repository_path=repository_path, service_path=service_path) def test_validate_service_invalid( mock_datadog_connector, mock_sentry_connector, mock_sonar_connector, invalid_service_definition_file ): """Test validate_service with an invalid service definition.""" repository_path, service_path = invalid_service_definition_file software_catalog = SoftwareCatalog(mock_datadog_connector, mock_sentry_connector, mock_sonar_connector) with pytest.raises(ValueError): software_catalog.validate_service(repository_path=repository_path, service_path=service_path) @pytest.mark.parametrize( "service_type, expected_aws_service", [ pytest.param("web", "fargate"), pytest.param("graphql", "fargate"), pytest.param("task", "fargate"), pytest.param("daemon", "fargate"), pytest.param("connector", "fargate"), pytest.param("function", "lambda"), ], ) def test_get_aws_service(service_type: str, expected_aws_service: str): aws_service = SoftwareCatalog._get_aws_service(service_type) assert aws_service == expected_aws_service def test_extract_language_from_lambda_runtime_with_python_runtime(): """Test extraction of language metadata from Python Lambda runtime.""" aws_resources = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:prod-test-function", account_id="123456789012", environment="prod", resource_type="lambda:function", region="us-east-1", runtime="python3.11", ) ] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata == (Language.PYTHON, "3.11") def test_extract_language_from_lambda_runtime_with_nodejs_newer_format(): """Test extraction of language metadata from newer Node.js Lambda runtime format.""" aws_resources = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:prod-test-function", account_id="123456789012", environment="prod", resource_type="lambda:function", region="us-east-1", runtime="nodejs20.x", ) ] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata == (Language.JAVASCRIPT, "20") def test_extract_language_from_lambda_runtime_with_nodejs_older_format(): """Test extraction of language metadata from older Node.js Lambda runtime format.""" aws_resources = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:prod-test-function", account_id="123456789012", environment="prod", resource_type="lambda:function", region="us-east-1", runtime="nodejs8.10", ) ] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata == (Language.JAVASCRIPT, "8") def test_extract_language_from_lambda_runtime_with_no_prod_resource(): """Test that extraction returns None when no production resource exists.""" aws_resources = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:qa-test-function", account_id="123456789012", environment="qa", resource_type="lambda:function", region="us-east-1", runtime="python3.11", ) ] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata is None def test_extract_language_from_lambda_runtime_with_no_runtime_tag(): """Test that extraction returns None when prod resource has no runtime tag.""" aws_resources = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:prod-test-function", account_id="123456789012", environment="prod", resource_type="lambda:function", region="us-east-1", ) ] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata is None def test_extract_language_from_lambda_runtime_with_empty_resources(): """Test that extraction returns None when no resources are provided.""" aws_resources = [] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata is None def test_extract_language_from_lambda_runtime_with_unsupported_runtime(): """Test that extraction returns None for unsupported runtime.""" aws_resources = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:prod-test-function", account_id="123456789012", environment="prod", resource_type="lambda:function", region="us-east-1", runtime="java11", ) ] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata is None def test_extract_language_from_lambda_runtime_prioritizes_prod(): """Test that extraction prioritizes prod environment when multiple resources exist.""" aws_resources = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:qa-test-function", account_id="123456789012", environment="qa", resource_type="lambda:function", region="us-east-1", runtime="python3.9", ), AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:prod-test-function", account_id="123456789012", environment="prod", resource_type="lambda:function", region="us-east-1", runtime="python3.11", ), ] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata == (Language.PYTHON, "3.11") def test_extract_language_from_lambda_runtime_with_production_environment_tag(): """Test extraction of language metadata from resources with environment tagged as Production.""" aws_resources = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:prod-test-function", account_id="123456789012", environment="Production", resource_type="lambda:function", region="us-east-1", runtime="python3.11", ) ] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata == (Language.PYTHON, "3.11") def test_extract_language_from_lambda_runtime_with_missing_environment_tag(): """Test extraction of language metadata from resources with no environment tag.""" aws_resources = [ AWSResource( arn="arn:aws:lambda:us-east-1:123456789012:function:prod-test-function", account_id="123456789012", resource_type="lambda:function", region="us-east-1", runtime="python3.11", ) ] metadata = SoftwareCatalog._extract_language_from_lambda_runtime(aws_resources) assert metadata is None