moto 中 AWS DirectConnect Mock 实现解析:已支持 API 清单、内部机制与使用边界

• Mock
• 测试

本文基于 moto 仓库中 DirectConnect 服务的文档页与源码,系统介绍 moto 对 AWS DirectConnect 的模拟能力边界:当前已实现与未实现的 API 操作清单、通过mock_aws装饰器进行 DirectConnect 测试的完整代码示例,以及从请求路由、资源 ID 生成、MACsec 密钥管理到标签体系的底层实现细节。读完后你可以明确知道哪些 DirectConnect 场景可以直接用 moto 做单元测试,哪些场景仍需要真实云环境,并能看懂每个 API 背后的调用链与数据来源。

一、支持范围:10 个已实现操作与完整清单

moto 为每个 AWS 服务维护一份“已实现功能”文档,DirectConnect 对应的文档是 directconnect.rst,并在 IMPLEMENTATION_COVERAGE.md 中统计为15% implemented。当前已打勾([X])的操作共 10 个:

未实现的操作(文档中标记为[ ])完整清单如下,在写测试前可以先对照确认目标 API 是否可用:

accept_direct_connect_gateway_association_proposal allocate_connection_on_interconnect allocate_hosted_connection allocate_private_virtual_interface allocate_public_virtual_interface allocate_transit_virtual_interface associate_connection_with_lag associate_hosted_connection associate_virtual_interface confirm_connection confirm_customer_agreement confirm_private_virtual_interface confirm_public_virtual_interface confirm_transit_virtual_interface create_bgp_peer create_direct_connect_gateway create_direct_connect_gateway_association create_direct_connect_gateway_association_proposal create_interconnect create_private_virtual_interface create_public_virtual_interface create_transit_virtual_interface delete_bgp_peer delete_direct_connect_gateway delete_direct_connect_gateway_association delete_direct_connect_gateway_association_proposal delete_interconnect delete_lag delete_virtual_interface describe_connection_loa describe_connections_on_interconnect describe_customer_metadata describe_direct_connect_gateway_association_proposals describe_direct_connect_gateway_associations describe_direct_connect_gateway_attachments describe_direct_connect_gateways describe_hosted_connections describe_interconnect_loa describe_interconnects describe_loa describe_locations describe_router_configuration describe_tags describe_virtual_gateways describe_virtual_interfaces disassociate_connection_from_lag list_virtual_interface_routes list_virtual_interface_test_history start_bgp_failover_test stop_bgp_failover_test update_direct_connect_gateway update_direct_connect_gateway_association update_lag update_virtual_interface_attributes

可以看出,已实现部分集中在连接(LAG)生命周期 + MACsec 端口加密 + 标签这条主线上;Virtual Interface、Direct Connect Gateway、Interconnect、BGP 与 LOA 相关操作均尚未实现。此外,describe_tags虽未勾选,但后端方法list_tags_for_resources已实现,测试用例 test_directconnect.py 中确实调用了client.describe_tags,实际行为以源码为准。

二、快速上手:用 mock_aws 测试 DirectConnect

以下示例均取自仓库测试 tests/test_directconnect/test_directconnect.py,可原样复制到你的项目中使用。前置依赖是安装moto与boto3。

2.1 创建与查询连接

import boto3 from moto import mock_aws @mock_aws def test_connections(): client = boto3.client("directconnect", region_name="us-east-1") connection = client.create_connection( location="EqDC2", bandwidth="10Gbps", connectionName="TestConnection", ) # moto 生成的连接 ID 形如 # arn:aws:directconnect:us-east-1:<account>:dx-con/dx-moto-TestConnection-<时间戳> assert "dx-moto" in connection["connectionId"] assert connection["connectionState"] == "available" # 支持 requestMACSec=True 请求 MACsec 端口加密 client.create_connection( location="EqDC2", bandwidth="10Gbps", connectionName="TestConnection2", requestMACSec=True, ) resp = client.describe_connections() assert len(resp["connections"]) == 2 # requestMACSec=True 时 encryptionMode 为 must_encrypt,否则为 no_encrypt assert resp["connections"][0]["encryptionMode"] == "no_encrypt" assert resp["connections"][1]["encryptionMode"] == "must_encrypt" # 按 connectionId 过滤 resp = client.describe_connections( connectionId=resp["connections"][0]["connectionId"] ) assert len(resp["connections"]) == 1

2.2 更新与删除连接

@mock_aws def test_update_and_delete(): client = boto3.client("directconnect", region_name="us-east-1") conn = client.create_connection( location="EqDC2", bandwidth="10Gbps", connectionName="TestConnection1" ) # 更新连接名 updated = client.update_connection( connectionId=conn["connectionId"], connectionName="NewConnectionName", ) assert updated["connectionName"] == "NewConnectionName" # 更新加密模式(no_encrypt / should_encrypt / must_encrypt) client.update_connection( connectionId=conn["connectionId"], encryptionMode="should_encrypt", ) # 删除后状态变为 deleted deleted = client.delete_connection(connectionId=conn["connectionId"]) assert deleted["connectionState"] == "deleted"

2.3 LAG 的创建与查询

create_lag会根据numberOfConnections自动创建子连接,子连接名称遵循Requested Connection {i} for Lag {lagId}的格式:

@mock_aws def test_lag(): client = boto3.client("directconnect", region_name="us-east-1") lag = client.create_lag( numberOfConnections=1, location="eqDC2", connectionsBandwidth="10Gbps", lagName="TestLag0", ) assert "dxlag-moto" in lag["lagId"] assert lag["lagState"] == "available" assert len(lag["connections"]) == 1 assert "Requested Connection 1 for Lag" in lag["connections"][0]["connectionName"] lags = client.describe_lags() assert len(lags["lags"]) == 1

注意一个明确的使用边界:在 models.py 的 create_lag 中,如果传入connectionId(即把已有连接加入 LAG),后端会直接抛出NotImplementedError("creating a lag with a connection_id is not currently supported by moto"),这类场景目前无法用 moto 模拟。

2.4 MACsec 密钥关联与解除

MACsec 用于 Direct Connect 端口的链路层加密。moto 中密钥操作同时支持连接 ID 和 LAG ID(通过 ID 中是否包含dxlag-子串来区分路由):

@mock_aws def test_macsec(): client = boto3.client("directconnect", region_name="us-east-1") conn = client.create_connection( location="EqDC2", bandwidth="10Gbps", connectionName="TestConnection1" ) resp = client.associate_mac_sec_key( connectionId=conn["connectionId"], ckn="_fake_ckn_", cak="_fake_cak_", secretARN="_fake_secret_arn_", ) keys = resp["macSecKeys"] assert keys[0]["ckn"] == "_fake_ckn_" assert "cak" not in keys[0] # 响应中不回显 CAK assert keys[0]["secretARN"] == "_fake_secret_arn_" # 按 secretARN 解除关联,状态变为 disassociated resp = client.disassociate_mac_sec_key( connectionId=conn["connectionId"], secretARN="_fake_secret_arn_", ) assert resp["macSecKeys"][0]["state"] == "disassociated"

对 LAG 执行associate_mac_sec_key时,密钥会追加到 LAG 并同步到其所有子连接,这一行为由 test_associate_mac_sec_key_lag 覆盖。

2.5 标签操作

@mock_aws def test_tags(): client = boto3.client("directconnect", region_name="us-east-1") conn = client.create_connection( location="EqDC2", bandwidth="10Gbps", connectionName="TestConnection1", tags=[{"key": "t1", "value": "v1"}, {"key": "t2", "value": "v2"}], ) arn = conn["connectionId"] client.tag_resource( resourceArn=arn, tags=[{"key": "t1", "value": "v1"}, {"key": "t2", "value": "v2"}], ) # describe_tags 按 ARN 批量查询 tags = client.describe_tags(resourceArns=[arn])["resourceTags"][0]["tags"] assert tags == [{"key": "t1", "value": "v1"}, {"key": "t2", "value": "v2"}] client.untag_resource(resourceArn=arn, tagKeys=["t1", "t2"]) assert client.describe_tags(resourceArns=[arn])["resourceTags"][0]["tags"] == []

三、底层实现:请求如何被模拟

3.1 URL 路由与响应分发

DirectConnect 的请求拦截入口在 urls.py:

url_bases = [ r"https?://directconnect\.(.+)\.amazonaws\.com", ] url_paths = { "{0}/$": DirectConnectResponse.dispatch, }

所有请求都指向 region 化的directconnect.<region>.amazonaws.com端点,统一交给 DirectConnectResponse 的dispatch处理。responses.py中的每个方法负责从 JSON 请求体self.body中解析出参数,再调用后端同名方法,例如create_connection会提取location、bandwidth、connectionName、lagId、tags、providerName、requestMACSec七个参数传给后端。这是 moto 典型的“薄响应层 + 厚后端层”结构。

3.2 数据模型与资源 ID 生成

后端 DirectConnectBackend 用一个connections字典和lags字典在内存中保存全部状态,BackendDict按“账号 + 区域”维度隔离后端实例,即不同 region 或不同 AWS 账号之间数据互不可见,与真实 Direct Connect 的区域化行为一致。

Connection、LAG、MacSecKey均为@dataclass(models.py#L28-L154)。两个值得注意的实现细节:

1. 资源 ID 自动生成:创建连接时若未生成 ID,__post_init__会按固定模板拼装 ARN(models.py#L71-L73):

   arn:aws:directconnect:{region}:{account}:dx-con/dx-moto-{connectionName}-{YYYYmmddHHMMSS}

   LAG 同理,前缀为dxlag/dxlag-moto-...。这意味着同一秒内创建两个同名连接理论上会冲突,且 ID 中嵌入了连接名——测试断言中常见的"dx-moto" in connectionId即来源于此。

2. 模拟硬件字段:aws_device、aws_logical_device_id、partner_name等真实环境中的物理属性统一返回mock_device、mock_logical_device_id、mock_partner等占位值(create_connection),新创建连接的状态直接置为available,LOA 签发时间为当前时间。

状态与枚举定义在 enums.py:ConnectionStateType(available/deleted/down/ordering/pending 等 9 种)、LagStateType、EncryptionModeType(no_encrypt/should_encrypt/must_encrypt)、MacSecKeyStateType(associating/associated/disassociating/disassociated)以及PortEncryptionStatusType(Encryption Up/Down)。

3.3requestMACSec对创建流程的影响

在 create_connection 中,requestMACSec=True会触发三件事:encryption_mode从NO提升为MUST;mac_sec_capable标记为 True;并自动附加一条 mock 密钥(secret_arn="mock_secret_arn",ckn="mock_ckn", 状态associated)。create_lag有相同逻辑,且会把 LAG 级密钥同步到每个子连接(models.py#L380-L383)。describe_connections/describe_lags返回的macSecKeys、encryptionMode、macSecCapable字段即由此驱动,对应测试 test_describe_connections 与 test_describe_lags 的断言。

3.4 MACsec 密钥的关联/解除算法

associate_mac_sec_key 的核心是一个字符串嗅探:

if "dxlag-" in connection_id: return self._associate_mac_sec_key_with_lag(...) return self._associate_mac_sec_key_with_connection(...)

从源码结构看,它是用 LAG ARN 中的dxlag-片段来区分资源类型,而不是显式查询两种资源表;disassociate_mac_sec_key(models.py#L399-L417)则是先按dxlag-前缀在lags或connections中查找密钥列表,再按secret_arn做大小写不敏感(casefold)匹配,命中后把状态改为DISASSOCIATED并弹出该密钥;未命中则抛出MacSecKeyNotFound。响应中不包含 CAK 明文,与真实 API 的安全行为一致(测试 test_associate_mac_sec_key_connection 显式断言了"cak" not in mac_sec_keys[0])。

3.5 标签体系与 Resource Groups Tagging API 集成

后端继承了TaggableResourcesMixin并使用TaggingService(models.py#L157-L166)。创建时传入的tags会写入 tagger;describe_connections/describe_lags的返回体中tags字段是通过self.backend.list_tags_for_resource(arn)实时聚合的(models.py#L98),而非创建时的快照,因此tag_resource/untag_resource的修改会立即反映在后续查询中。

更进一步的集成点是 iter_tagged_resources:它分别以directconnect:dxcon(连接)和directconnect:dxlag(LAG)两种资源类型产出TaggedResource迭代器,使 moto 的resourcegroupstaggingapi服务也能查到这些资源的标签。集成测试 test_directconnect_integration.py 验证了这条链路——通过resourcegroupstaggingapi客户端的get_resources能拿到 DirectConnect 连接和 LAG 的标签映射。如果你的业务代码依赖 Resource Groups Tagging API 统一管理 DirectConnect 资源标签,这条 mock 链路是可用且已覆盖的。

3.6 异常体系

错误响应集中在 exceptions.py,均基于JsonRESTError(HTTP 400):

• ConnectionIdMissing:删除/更新时缺少 connectionId;
• ConnectionNotFound:查询/操作不存在的连接,消息中会带上 region;
• LAGNotFound:操作不存在的 LAG;
• MacSecKeyNotFound:按 secretARN 解绑时未找到匹配密钥。

这意味着在测试中可以直接用botocore.exceptions.ClientError断言这些错误码(如ConnectionNotFound),与真实 API 行为一致。

四、使用边界与注意事项

1. 覆盖面仅约 15%:Virtual Interface、Gateway、Interconnect、BGP、LOA/locations 等全部未实现,涉及这些 API 的测试无法用 moto 完成,需真实环境或其他 mock 手段。
2. create_lag(connectionId=...)不受支持:会抛NotImplementedError,见 models.py#L332-L335。
3. 资源状态是简化的:连接创建后直接available,删除后为deleted,没有 pending/ordering/deleting 等中间态流转;portEncryptionStatus固定为Encryption Down。
4. ID 含时间戳:连接/LAG ID 中嵌入秒级时间戳,断言时建议用前缀匹配(如dx-moto、dxlag-moto)而非精确值。
5. 数据按账号+区域隔离:不同 region 的 client 看不到彼此的连接,查询不存在的 ID 会收到ConnectionNotFound/LagNotFound。

五、相关文件索引

总体而言,把 moto 用于 DirectConnect 测试时,建议把模拟场景锁定在“创建连接/LAG → 查询 → 更新/删除 → MACsec 密钥管理与标签”这一条已被测试充分覆盖的主线上;超出该清单的 API,在编写测试前务必先核对 directconnect.rst 中的勾选状态,避免依赖未实现的行为。

• Mock
• 测试