Repository: getmoto/moto
Pull request #8915 (2025-05-21 03:32:06)

[Cloud Directory] added schema endpoints
Maintainers and Reviewers @bblommers  and @bpandola.
This PR adds some endpoints to Cloud Directory Service. To create a cloud directory, a [published schema ARN](https://docs.aws.amazon.com/clouddirectory/latest/APIReference/API_CreateDirectory.html) needs to be passed. When a schema is created, it is in the development state, until it is published. Only then can it be applied, during this process, the schema arn changes. [see](https://docs.aws.amazon.com/clouddirectory/latest/APIReference/arns.html)

Hence the flow:

```mermaid
flowchart TD
    A[Create Schema] -->|development| C{what state is schema}
    C -->|development| D[Publish Schema]
    C -->|published| E[Apply Schema]
    C -->|applied| F[Nothing to do]
    D --> B(Create Directory)
```

- [create_schema](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/clouddirectory/client/create_schema.html)
- [delete_schema](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/clouddirectory/client/delete_schema.html)
- [list_development_schema_arns](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/clouddirectory/client/list_development_schema_arns.html)
- [list_published_schema_arns](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/clouddirectory/client/list_published_schema_arns.html)
- [apply_schema](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/clouddirectory/client/apply_schema.html)
- [publish_schema](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/clouddirectory/client/publish_schema.html)



codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8915?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 92.85%. Comparing base [(`c04116b`)](https://app.codecov.io/gh/getmoto/moto/commit/c04116b3556f8c5d309934ec322ca3ae93aaa34f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`c965479`)](https://app.codecov.io/gh/getmoto/moto/commit/c96547958a5073c36af6705fb9e427b959bb5032?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).
> Report is 7 commits behind head on master.

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##           master    #8915   +/-   ##
=======================================
  Coverage   92.85%   92.85%           
=======================================
  Files        1282     1282           
  Lines      112028   112035    +7     
=======================================
+ Hits       104027   104035    +8     
+ Misses       8001     8000    -1     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8915/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/8915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `27.95% <12.50%> (-0.01%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8915/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `92.83% <100.00%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/8915?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

<details><summary> :rocket: New features to boost your workflow: </summary>

- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>

Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 3d35c5c87..fe4b9b236 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -1012,7 +1012,7 @@
 <summary>10% implemented</summary>
 
 - [ ] add_facet_to_object
-- [ ] apply_schema
+- [X] apply_schema
 - [ ] attach_object
 - [ ] attach_policy
 - [ ] attach_to_index
@@ -1023,12 +1023,12 @@
 - [ ] create_facet
 - [ ] create_index
 - [ ] create_object
-- [ ] create_schema
+- [X] create_schema
 - [ ] create_typed_link_facet
 - [X] delete_directory
 - [ ] delete_facet
 - [ ] delete_object
-- [ ] delete_schema
+- [X] delete_schema
 - [ ] delete_typed_link_facet
 - [ ] detach_from_index
 - [ ] detach_object
@@ -1046,7 +1046,7 @@
 - [ ] get_typed_link_facet_information
 - [ ] list_applied_schema_arns
 - [ ] list_attached_indices
-- [ ] list_development_schema_arns
+- [X] list_development_schema_arns
 - [X] list_directories
 - [ ] list_facet_attributes
 - [ ] list_facet_names
@@ -1060,12 +1060,12 @@
 - [ ] list_object_policies
 - [ ] list_outgoing_typed_links
 - [ ] list_policy_attachments
-- [ ] list_published_schema_arns
+- [X] list_published_schema_arns
 - [X] list_tags_for_resource
 - [ ] list_typed_link_facet_attributes
 - [ ] list_typed_link_facet_names
 - [ ] lookup_policy
-- [ ] publish_schema
+- [X] publish_schema
 - [ ] put_schema_from_json
 - [ ] remove_facet_from_object
 - [X] tag_resource
diff --git a/moto/clouddirectory/exceptions.py b/moto/clouddirectory/exceptions.py
index 7b2d582ec..8e3892849 100644
--- a/moto/clouddirectory/exceptions.py
+++ b/moto/clouddirectory/exceptions.py
@@ -18,3 +18,33 @@ class InvalidArnException(JsonRESTError):
             "Message": "Invalid Arn",
         }
         self.description = json.dumps(body)
+
+
+class ResourceNotFoundException(JsonRESTError):
+    def __init__(self, resource_id: str):
+        super().__init__("ResourceNotFoundException", "Resource not found")
+        body = {
+            "ResourceId": resource_id,
+            "Message": "Resource not found",
+        }
+        self.description = json.dumps(body)
+
+
+class SchemaAlreadyPublishedException(JsonRESTError):
+    def __init__(self, schema_arn: str):
+        super().__init__("SchemaAlreadyPublishedException", "Schema already published")
+        body = {
+            "SchemaArn": schema_arn,
+            "Message": "Schema already published",
+        }
+        self.description = json.dumps(body)
+
+
+class SchemaAlreadyExistsException(JsonRESTError):
+    def __init__(self, schema_arn: str):
+        super().__init__("SchemaAlreadyExistsException", "Schema already exists")
+        body = {
+            "SchemaArn": schema_arn,
+            "Message": "Schema already exists",
+        }
+        self.description = json.dumps(body)
diff --git a/moto/clouddirectory/models.py b/moto/clouddirectory/models.py
index 40142e42b..f75ef4282 100644
--- a/moto/clouddirectory/models.py
+++ b/moto/clouddirectory/models.py
@@ -8,7 +8,11 @@ from moto.core.common_models import BaseModel
 from moto.utilities.paginator import paginate
 from moto.utilities.tagging_service import TaggingService
 
-from .exceptions import InvalidArnException
+from .exceptions import (
+    InvalidArnException,
+    ResourceNotFoundException,
+    SchemaAlreadyPublishedException,
+)
 
 PAGINATION_MODEL = {
     "list_directories": {
@@ -16,7 +20,19 @@ PAGINATION_MODEL = {
         "limit_key": "max_results",
         "limit_default": 100,
         "unique_attribute": "directory_arn",
-    }
+    },
+    "list_development_schema_arns": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": "schema_arn",
+    },
+    "list_published_schema_arns": {
+        "input_token": "next_token",
+        "limit_key": "max_results",
+        "limit_default": 100,
+        "unique_attribute": "schema_arn",
+    },
 }
 
 
@@ -50,13 +66,62 @@ class CloudDirectoryBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str) -> None:
         super().__init__(region_name, account_id)
         self.directories: Dict[str, Directory] = {}
+        self.schemas_states: Dict[str, List[str]] = {
+            "development": [],
+            "published": [],
+            "applied": [],
+        }
         self.tagger = TaggingService()
 
+    def apply_schema(self, directory_arn: str, published_schema_arn: str) -> None:
+        directory = self.directories.get(directory_arn)
+        if not directory:
+            raise ResourceNotFoundException(directory_arn)
+        if published_schema_arn not in self.schemas_states["published"]:
+            raise ResourceNotFoundException(published_schema_arn)
+        directory.schema_arn = published_schema_arn
+        return
+
+    def publish_schema(
+        self, name: str, version: str, development_schema_arn: str, minor_version: str
+    ) -> str:
+        schema_arn = f"arn:aws:clouddirectory:{self.region_name}:{self.account_id}:schema/published/{name}/{version}/{minor_version}"
+        if development_schema_arn in self.schemas_states["published"]:
+            raise SchemaAlreadyPublishedException(development_schema_arn)
+        if development_schema_arn in self.schemas_states["development"]:
+            self.schemas_states["development"].remove(development_schema_arn)
+            self.schemas_states["published"].append(schema_arn)
+        else:
+            raise ResourceNotFoundException(development_schema_arn)
+        return schema_arn
+
     def create_directory(self, name: str, schema_arn: str) -> Directory:
         directory = Directory(self.account_id, self.region_name, name, schema_arn)
         self.directories[directory.directory_arn] = directory
         return directory
 
+    def create_schema(self, name: str) -> str:
+        self.schema_arn = f"arn:aws:clouddirectory:{self.region_name}:{self.account_id}:schema/development/{name}"
+        self.schemas_states["development"].append(self.schema_arn)
+        return self.schema_arn
+
+    def delete_schema(self, schema_arn: str) -> None:
+        if schema_arn in self.schemas_states["development"]:
+            self.schemas_states["development"].remove(schema_arn)
+        elif schema_arn in self.schemas_states["published"]:
+            self.schemas_states["published"].remove(schema_arn)
+        else:
+            raise ResourceNotFoundException(schema_arn)
+        return
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_development_schema_arns(self) -> List[str]:
+        return self.schemas_states["development"]
+
+    @paginate(pagination_model=PAGINATION_MODEL)
+    def list_published_schema_arns(self) -> List[str]:
+        return self.schemas_states["published"]
+
     @paginate(pagination_model=PAGINATION_MODEL)
     def list_directories(self, state: str) -> List[Directory]:
         directories = list(self.directories.values())
diff --git a/moto/clouddirectory/responses.py b/moto/clouddirectory/responses.py
index c967791de..3038ee26c 100644
--- a/moto/clouddirectory/responses.py
+++ b/moto/clouddirectory/responses.py
@@ -18,9 +18,36 @@ class CloudDirectoryResponse(BaseResponse):
         """Return backend instance specific for this region."""
         return clouddirectory_backends[self.current_account][self.region]
 
+    def apply_schema(self) -> str:
+        directory_arn = self.headers.get("x-amz-data-partition")
+        published_schema_arn = self._get_param("PublishedSchemaArn")
+        self.clouddirectory_backend.apply_schema(
+            directory_arn=directory_arn,
+            published_schema_arn=published_schema_arn,
+        )
+        return json.dumps(
+            {
+                "AppliedSchemaArn": published_schema_arn,
+                "DirectoryArn": directory_arn,
+            }
+        )
+
+    def publish_schema(self) -> str:
+        development_schema_arn = self.headers.get("x-amz-data-partition")
+        version = self._get_param("Version")
+        minor_version = self._get_param("MinorVersion")
+        name = self._get_param("Name")
+        schema = self.clouddirectory_backend.publish_schema(
+            name=name,
+            version=version,
+            minor_version=minor_version,
+            development_schema_arn=development_schema_arn,
+        )
+        return json.dumps({"PublishedSchemaArn": schema})
+
     def create_directory(self) -> str:
         name = self._get_param("Name")
-        schema_arn = self._get_param("SchemaArn")
+        schema_arn = self.headers.get("x-amz-data-partition")
         directory = self.clouddirectory_backend.create_directory(
             name=name,
             schema_arn=schema_arn,
@@ -35,6 +62,13 @@ class CloudDirectoryResponse(BaseResponse):
             )
         )
 
+    def create_schema(self) -> str:
+        name = self._get_param("Name")
+        schema = self.clouddirectory_backend.create_schema(
+            name=name,
+        )
+        return json.dumps(dict(SchemaArn=schema))
+
     def list_directories(self) -> str:
         next_token = self._get_param("NextToken")
         max_results = self._get_param("MaxResults")
@@ -72,6 +106,33 @@ class CloudDirectoryResponse(BaseResponse):
         )
         return json.dumps(dict(DirectoryArn=directory_arn))
 
+    def delete_schema(self) -> str:
+        # Retrieve arn from headers
+        # https://docs.aws.amazon.com/clouddirectory/latest/APIReference/API_DeleteSchema.html
+        arn = self.headers.get("x-amz-data-partition")
+        self.clouddirectory_backend.delete_schema(
+            schema_arn=arn,
+        )
+        return json.dumps(dict(SchemaArn=arn))
+
+    def list_development_schema_arns(self) -> str:
+        next_token = self._get_param("NextToken")
+        max_results = self._get_param("MaxResults")
+        schemas, next_token = self.clouddirectory_backend.list_development_schema_arns(
+            next_token=next_token,
+            max_results=max_results,
+        )
+        return json.dumps(dict(SchemaArns=schemas, NextToken=next_token))
+
+    def list_published_schema_arns(self) -> str:
+        next_token = self._get_param("NextToken")
+        max_results = self._get_param("MaxResults")
+        schemas, next_token = self.clouddirectory_backend.list_published_schema_arns(
+            next_token=next_token,
+            max_results=max_results,
+        )
+        return json.dumps(dict(SchemaArns=schemas, NextToken=next_token))
+
     def get_directory(self) -> str:
         # Retrieve arn from headers
         # https://docs.aws.amazon.com/clouddirectory/latest/APIReference/API_GetDirectory.html
diff --git a/moto/clouddirectory/urls.py b/moto/clouddirectory/urls.py
index e82051ba4..ef70b8a98 100644
--- a/moto/clouddirectory/urls.py
+++ b/moto/clouddirectory/urls.py
@@ -8,10 +8,16 @@ url_bases = [
 
 url_paths = {
     "{0}/amazonclouddirectory/2017-01-11/directory/create$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/schema/create$": CloudDirectoryResponse.dispatch,
     "{0}/amazonclouddirectory/2017-01-11/directory/list$": CloudDirectoryResponse.dispatch,
     "{0}/amazonclouddirectory/2017-01-11/tags/add$": CloudDirectoryResponse.dispatch,
     "{0}/amazonclouddirectory/2017-01-11/tags/remove$": CloudDirectoryResponse.dispatch,
     "{0}/amazonclouddirectory/2017-01-11/directory$": CloudDirectoryResponse.dispatch,
     "{0}/amazonclouddirectory/2017-01-11/directory/get$": CloudDirectoryResponse.dispatch,
     "{0}/amazonclouddirectory/2017-01-11/tags$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/schema$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/schema/development$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/schema/published$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/schema/apply$": CloudDirectoryResponse.dispatch,
+    "{0}/amazonclouddirectory/2017-01-11/schema/publish$": CloudDirectoryResponse.dispatch,
 }


Test patch:
diff --git a/tests/test_clouddirectory/test_clouddirectory.py b/tests/test_clouddirectory/test_clouddirectory.py
index 9aa08fcc8..36cbe59dc 100644
--- a/tests/test_clouddirectory/test_clouddirectory.py
+++ b/tests/test_clouddirectory/test_clouddirectory.py
@@ -1,6 +1,7 @@
 """Unit tests for clouddirectory-supported APIs."""
 
 import boto3
+import pytest
 
 from moto import mock_aws
 from tests import DEFAULT_ACCOUNT_ID
@@ -10,9 +11,8 @@ from tests import DEFAULT_ACCOUNT_ID
 def test_create_directory():
     region = "us-west-2"
     client = boto3.client("clouddirectory", region_name=region)
-    schema_arn = (
-        f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:directory/test-schema/1"
-    )
+    schema = client.create_schema(Name="test-schema")
+    schema_arn = schema["SchemaArn"]
     resp = client.create_directory(SchemaArn=schema_arn, Name="test-directory")
     assert (
         resp["DirectoryArn"]
@@ -20,6 +20,109 @@ def test_create_directory():
     )
     assert resp["Name"] == "test-directory"
     assert "ObjectIdentifier" in resp
+    assert resp["AppliedSchemaArn"] == schema_arn
+
+
+@mock_aws
+def test_create_schema():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    resp = client.create_schema(Name="test-schema")
+    assert (
+        resp["SchemaArn"]
+        == f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:schema/development/test-schema"
+    )
+
+
+@mock_aws
+def test_apply_schema():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema = client.create_schema(Name="test-schema")
+    schema_arn = schema["SchemaArn"]
+    pub_schema = client.publish_schema(
+        DevelopmentSchemaArn=schema_arn,
+        Name="test-schema",
+        Version="1",
+        MinorVersion="0",
+    )
+    pub_schema_arn = pub_schema["PublishedSchemaArn"]
+    directory = client.create_directory(SchemaArn=pub_schema_arn, Name="test-directory")
+    directory_arn = directory["DirectoryArn"]
+    resp = client.apply_schema(
+        PublishedSchemaArn=pub_schema_arn, DirectoryArn=directory_arn
+    )
+    assert resp["AppliedSchemaArn"] == pub_schema_arn
+    assert resp["DirectoryArn"] == directory_arn
+
+
+@mock_aws
+def test_publish_schema():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema = client.create_schema(Name="test-schema")
+    schema_arn = schema["SchemaArn"]
+    resp = client.publish_schema(
+        DevelopmentSchemaArn=schema_arn,
+        Name="test-schema",
+        Version="1",
+        MinorVersion="0",
+    )
+    assert (
+        resp["PublishedSchemaArn"]
+        == f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:schema/published/test-schema/1/0"
+    )
+
+
+@mock_aws
+def test_list_development_schema_arns():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema_arn1 = client.create_schema(Name="test-schema1")["SchemaArn"]
+    schema_arn2 = client.create_schema(Name="test-schema2")["SchemaArn"]
+
+    resp = client.list_development_schema_arns()
+    assert len(resp["SchemaArns"]) == 2
+    assert schema_arn1 in resp["SchemaArns"]
+    assert schema_arn2 in resp["SchemaArns"]
+
+
+@mock_aws
+def test_list_published_schema_arns():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema_arn1 = client.create_schema(Name="test-schema1")["SchemaArn"]
+    schema_arn2 = client.create_schema(Name="test-schema2")["SchemaArn"]
+
+    published_arn1 = client.publish_schema(
+        DevelopmentSchemaArn=schema_arn1,
+        Name="test-schema1",
+        Version="1",
+        MinorVersion="0",
+    )["PublishedSchemaArn"]
+
+    published_arn2 = client.publish_schema(
+        DevelopmentSchemaArn=schema_arn2,
+        Name="test-schema2",
+        Version="1",
+        MinorVersion="0",
+    )["PublishedSchemaArn"]
+
+    resp = client.list_published_schema_arns()
+    assert len(resp["SchemaArns"]) == 2
+    assert published_arn1 in resp["SchemaArns"]
+    assert published_arn2 in resp["SchemaArns"]
+
+
+@mock_aws
+def test_delete_schema():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+    schema_arn = client.create_schema(Name="test-schema")["SchemaArn"]
+    client.create_directory(SchemaArn=schema_arn, Name="test-directory")
+    client.delete_schema(SchemaArn=schema_arn)
+    resp = client.list_development_schema_arns()
+    assert len(resp["SchemaArns"]) == 0
 
 
 @mock_aws
@@ -152,3 +255,139 @@ def test_get_directory():
     assert directory["DirectoryArn"] == directory_arns[0]
     assert directory["State"] == "ENABLED"
     assert "CreationDateTime" in directory
+
+
+@mock_aws
+def test_apply_schema_with_nonexistent_schema():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+
+    schema = client.create_schema(Name="test-schema")
+    schema_arn = schema["SchemaArn"]
+    pub_schema = client.publish_schema(
+        DevelopmentSchemaArn=schema_arn,
+        Name="test-schema",
+        Version="1",
+        MinorVersion="0",
+    )
+    pub_schema_arn = pub_schema["PublishedSchemaArn"]
+    directory = client.create_directory(SchemaArn=pub_schema_arn, Name="test-directory")
+    directory_arn = directory["DirectoryArn"]
+
+    non_existent_schema_arn = (
+        f"arn:aws:clouddirectory:{region}:123456789012:schema/published/nonexistent/1/0"
+    )
+
+    with pytest.raises(client.exceptions.ResourceNotFoundException):
+        client.apply_schema(
+            PublishedSchemaArn=non_existent_schema_arn, DirectoryArn=directory_arn
+        )
+
+
+@mock_aws
+def test_apply_schema_updates_directory_schema():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+
+    schema1 = client.create_schema(Name="test-schema1")
+    schema1_arn = schema1["SchemaArn"]
+    pub_schema1 = client.publish_schema(
+        DevelopmentSchemaArn=schema1_arn,
+        Name="test-schema1",
+        Version="1",
+        MinorVersion="0",
+    )
+    pub_schema1_arn = pub_schema1["PublishedSchemaArn"]
+
+    directory = client.create_directory(
+        SchemaArn=pub_schema1_arn, Name="test-directory"
+    )
+    directory_arn = directory["DirectoryArn"]
+
+    schema2 = client.create_schema(Name="test-schema2")
+    schema2_arn = schema2["SchemaArn"]
+    pub_schema2 = client.publish_schema(
+        DevelopmentSchemaArn=schema2_arn,
+        Name="test-schema2",
+        Version="1",
+        MinorVersion="0",
+    )
+    pub_schema2_arn = pub_schema2["PublishedSchemaArn"]
+
+    # Apply second schema to directory
+    resp = client.apply_schema(
+        PublishedSchemaArn=pub_schema2_arn, DirectoryArn=directory_arn
+    )
+
+    # Verify schema was updated
+    assert resp["AppliedSchemaArn"] == pub_schema2_arn
+
+
+@mock_aws
+def test_publish_nonexistent_schema():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+
+    non_existent_schema_arn = f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:schema/development/nonexistent"
+
+    with pytest.raises(client.exceptions.ResourceNotFoundException):
+        client.publish_schema(
+            DevelopmentSchemaArn=non_existent_schema_arn,
+            Name="nonexistent",
+            Version="1",
+            MinorVersion="0",
+        )
+
+
+@mock_aws
+def test_publish_schema_state_transition():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+
+    dev_schema = client.create_schema(Name="test-schema")
+    dev_schema_arn = dev_schema["SchemaArn"]
+
+    dev_schemas = client.list_development_schema_arns()
+    assert dev_schema_arn in dev_schemas["SchemaArns"]
+
+    published = client.publish_schema(
+        DevelopmentSchemaArn=dev_schema_arn,
+        Name="test-schema",
+        Version="1",
+        MinorVersion="0",
+    )
+    published_schema_arn = published["PublishedSchemaArn"]
+
+    dev_schemas = client.list_development_schema_arns()
+    pub_schemas = client.list_published_schema_arns()
+
+    assert dev_schema_arn not in dev_schemas["SchemaArns"]
+    assert published_schema_arn in pub_schemas["SchemaArns"]
+    assert (
+        published_schema_arn
+        == f"arn:aws:clouddirectory:{region}:{DEFAULT_ACCOUNT_ID}:schema/published/test-schema/1/0"
+    )
+
+
+@mock_aws
+def test_publish_already_published_schema():
+    region = "us-west-2"
+    client = boto3.client("clouddirectory", region_name=region)
+
+    schema = client.create_schema(Name="test-schema")
+    schema_arn = schema["SchemaArn"]
+    new_schema_arn = client.publish_schema(
+        DevelopmentSchemaArn=schema_arn,
+        Name="test-schema",
+        Version="1",
+        MinorVersion="0",
+    )
+
+    # Try to publish again with same development schema
+    with pytest.raises(client.exceptions.SchemaAlreadyPublishedException):
+        client.publish_schema(
+            DevelopmentSchemaArn=new_schema_arn["PublishedSchemaArn"],
+            Name="test-schema",
+            Version="1",
+            MinorVersion="0",
+        )