File: test_comprehend_integration.py

package info (click to toggle)
python-moto 5.1.18-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 116,520 kB
  • sloc: python: 636,725; javascript: 181; makefile: 39; sh: 3
file content (78 lines) | stat: -rw-r--r-- 2,447 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
import boto3

from moto import mock_aws
from moto.comprehend.models import comprehend_backends
from tests.test_comprehend.test_comprehend import (
    DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG,
)


@mock_aws
def test_tags_from_resourcegroupsapi():
    client = boto3.client("comprehend", region_name="ap-southeast-1")
    arn = client.create_document_classifier(
        DataAccessRoleArn="iam_role_with_20_chars",
        InputDataConfig=DOCUMENT_CLASSIFIER_INPUT_DATA_CONFIG,
        LanguageCode="en",
        DocumentClassifierName="tf-acc-test-1726651689102157637",
        Tags=[{"Key": "k1", "Value": "v1"}, {"Key": "k2", "Value": "v2"}],
    )["DocumentClassifierArn"]

    job_id = client.start_document_classification_job(
        JobName="test-job",
        DataAccessRoleArn="iam_role_with_20_chars",
        InputDataConfig={
            "S3Uri": "s3://input-bucket/input-prefix/",
            "InputFormat": "ONE_DOC_PER_FILE",
        },
        OutputDataConfig={"S3Uri": "s3://output-bucket/output-prefix/"},
        Tags=[{"Key": "jobkey", "Value": "jobvalue"}],
    )["JobId"]

    resource_groups_client = boto3.client(
        "resourcegroupstaggingapi", region_name="ap-southeast-1"
    )

    tags = resource_groups_client.get_resources(
        TagFilters=[{"Key": "k1", "Values": ["v1"]}],
    )["ResourceTagMappingList"]

    assert tags == [
        {
            "ResourceARN": arn,
            "Tags": [{"Key": "k1", "Value": "v1"}, {"Key": "k2", "Value": "v2"}],
        }
    ]

    tags = resource_groups_client.get_resources(
        ResourceTypeFilters=["comprehend:document-classification-job"],
    )["ResourceTagMappingList"]

    assert len(tags) == 1
    assert tags == [
        {
            "ResourceARN": f"arn:aws:comprehend:ap-southeast-1:123456789012:document-classification-job/{job_id}",
            "Tags": [{"Key": "jobkey", "Value": "jobvalue"}],
        }
    ]


@mock_aws
def test_tags_from_resourcegroupsapi_no_arn():
    resource_groups_client = boto3.client(
        "resourcegroupstaggingapi", region_name="ap-southeast-1"
    )
    account_id = "123456789012"

    backend = comprehend_backends[account_id]["ap-southeast-1"]

    class DummyResource:
        pass

    backend.jobs["fake-job"] = DummyResource()

    result = resource_groups_client.get_resources(
        ResourceTypeFilters=["comprehend:document-classification-job"]
    )["ResourceTagMappingList"]

    assert result == []