Skip to content

Upgrade moto dependency #40

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gchqdev227 merged 2 commits into from
Feb 20, 2026
Merged

Upgrade moto dependency #40

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
041ab8a
Replaced calls to moto internals with correct use of the moto interna…
gchqdev227 Feb 20, 2026
b24246b
Bumped test dependencies, including moto dependency
gchqdev227 Feb 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions requirements-tests.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,21 +1,21 @@
boto3==1.42.24
botocore==1.42.24
boto3==1.42.53
botocore==1.42.53
certifi==2026.1.4
cffi==2.0.0
charset-normalizer==3.4.4
cryptography==46.0.3
cryptography==46.0.5
idna==3.11
Jinja2==3.1.6
jmespath==1.0.1
jmespath==1.1.0
MarkupSafe==3.0.3
moto==5.0.12
pycparser==2.23
moto==5.1.21
pycparser==3.0
python-dateutil==2.9.0.post0
PyYAML==6.0.3
requests==2.32.5
responses==0.25.8
responses==0.26.0
s3transfer==0.16.0
six==1.17.0
urllib3==2.6.3
Werkzeug==3.1.5
xmltodict==1.0.2
Werkzeug==3.1.6
xmltodict==1.0.3
61 changes: 23 additions & 38 deletions tests/test_examples.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import os
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import TYPE_CHECKING, Any
from typing import Any
import unittest
import unittest.mock
import urllib.parse
Expand All @@ -18,7 +18,9 @@
import botocore # noqa: F401
from dateutil.tz import tzlocal
from moto import mock_aws
from moto.sagemaker.models import FakePipelineExecution
from moto.core import DEFAULT_ACCOUNT_ID
from moto.sagemaker.models import sagemaker_backends
from moto.sagemaker.utils import get_pipeline_execution_from_arn, get_pipeline_from_name
import requests # noqa: F401
import urllib3 # noqa: F401
except ImportError:
Expand All @@ -37,9 +39,6 @@
from examples.secrets import SecretVersion
from examples.xkcd import ComicVersion, XKCDResource

if TYPE_CHECKING:
from moto.sagemaker.responses import TYPE_RESPONSE, SageMakerResponse


def setUpModule() -> None:
"""Code to run before any single test in the module is run."""
Expand Down Expand Up @@ -448,6 +447,7 @@ def setUp(self) -> None:
pipeline_arn = "arn:aws:sagemaker:eu-west-1:<account>:pipeline:my-pipeline"
self.resource = PipelineResource(pipeline_arn)
client = self.resource._client
self._moto_backend = sagemaker_backends[DEFAULT_ACCOUNT_ID][client.meta.region_name]

pipeline_definition = {
"Version": "2020-12-01",
Expand Down Expand Up @@ -479,35 +479,29 @@ def setUp(self) -> None:

self.execution_arns = execution_arns[::-1] # reverse these to force moto to yield in descending order

def test_fetch_secret_no_version(self) -> None:
def test_fetch_no_version(self) -> None:
version, = self.resource.fetch_new_versions()
expected_version = ExecutionVersion(self.execution_arns[-1])
self.assertEqual(version, expected_version)

def test_fetch_secret_no_available_versions(self) -> None:
with unittest.mock.patch("moto.sagemaker.responses.SageMakerResponse.list_pipeline_executions",
_new_response_list_pipeline_executions_empty):
versions = self.resource.fetch_new_versions()
def test_fetch_no_available_versions(self) -> None:
pipeline = get_pipeline_from_name(self._moto_backend.pipelines, self.resource.pipeline_name)
pipeline.pipeline_executions.clear()
versions = self.resource.fetch_new_versions()
self.assertListEqual(versions, [])

def test_fetch_secret_no_version_with_pending(self) -> None:
fake_pipeline_executions: list[FakePipelineExecution] = FakePipelineExecution.instances
execution_mapping = {execution.pipeline_execution_arn: execution for execution in fake_pipeline_executions}

def test_fetch_no_version_with_pending(self) -> None:
for execution_arn in self.execution_arns[3:]:
execution = execution_mapping[execution_arn]
execution = get_pipeline_execution_from_arn(self._moto_backend.pipelines, execution_arn)
execution.pipeline_execution_status = "Executing"

version, = self.resource.fetch_new_versions()
expected_version = ExecutionVersion(self.execution_arns[2])
self.assertEqual(version, expected_version)

def test_fetch_secret_no_version_with_pending_and_all_statuses(self) -> None:
fake_pipeline_executions: list[FakePipelineExecution] = FakePipelineExecution.instances
execution_mapping = {execution.pipeline_execution_arn: execution for execution in fake_pipeline_executions}

def test_fetch_no_version_with_pending_and_all_statuses(self) -> None:
for execution_arn in self.execution_arns[3:]:
execution = execution_mapping[execution_arn]
execution = get_pipeline_execution_from_arn(self._moto_backend.pipelines, execution_arn)
execution.pipeline_execution_status = "Executing"

self.resource.statuses.append("Executing")
Expand All @@ -516,13 +510,13 @@ def test_fetch_secret_no_version_with_pending_and_all_statuses(self) -> None:
expected_version = ExecutionVersion(self.execution_arns[-1])
self.assertEqual(version, expected_version)

def test_fetch_secret_latest_version(self) -> None:
def test_fetch_latest_version(self) -> None:
previous_version, = self.resource.fetch_new_versions()
version, = self.resource.fetch_new_versions(previous_version)
expected_version = ExecutionVersion(self.execution_arns[-1])
self.assertEqual(version, expected_version)

def test_fetch_secret_old_version(self) -> None:
def test_fetch_old_version(self) -> None:
previous_version = ExecutionVersion(self.execution_arns[2])
versions = self.resource.fetch_new_versions(previous_version)
expected_versions = [ExecutionVersion(arn) for arn in self.execution_arns[2:]]
Expand All @@ -542,6 +536,7 @@ def setUp(self) -> None:
pipeline_name = "my-pipeline"
resource = PipelineResource("arn:aws:sagemaker:eu-west-1:<account>:pipeline:my-pipeline")
client = resource._client
self._moto_backend = sagemaker_backends[DEFAULT_ACCOUNT_ID][client.meta.region_name]

self.pipeline_definition = {
"Version": "2020-12-01",
Expand Down Expand Up @@ -611,11 +606,8 @@ def test_download_maximum_info(self) -> None:
self.assertDictEqual(metadata, expected_metadata)

def test_download_failed_execution(self) -> None:
fake_pipeline_executions: list[FakePipelineExecution] = FakePipelineExecution.instances
execution_mapping = {execution.pipeline_execution_arn: execution for execution in fake_pipeline_executions}

fake_execution = execution_mapping[self.version_minimum.execution_arn]
fake_execution.pipeline_execution_status = "Failed"
execution = get_pipeline_execution_from_arn(self._moto_backend.pipelines, self.version_minimum.execution_arn)
execution.pipeline_execution_status = "Failed"

resource = PipelineResource(pipeline="arn:aws:sagemaker:eu-west-1:<account>:pipeline:my-pipeline")
wrapper = SimpleTestResourceWrapper(resource)
Expand Down Expand Up @@ -675,20 +667,13 @@ def setUp(self) -> None:
)

def test_execution_creation(self) -> None:
fake_pipeline_executions: list[FakePipelineExecution] = FakePipelineExecution.instances
initial_execution_mapping = {execution.pipeline_execution_arn: execution for execution in fake_pipeline_executions}
intitial_executions = {version.execution_arn for version in self.resource._yield_potential_execution_versions()}

resource = PipelineResource(pipeline="arn:aws:sagemaker:eu-west-1:<account>:pipeline:my-pipeline")
wrapper = SimpleTestResourceWrapper(resource)
new_version, _ = wrapper.publish_new_version()

execution_mapping = {execution.pipeline_execution_arn: execution for execution in fake_pipeline_executions}
self.assertNotIn(new_version.execution_arn, initial_execution_mapping)
self.assertIn(new_version.execution_arn, execution_mapping)

new_executions = {version.execution_arn for version in self.resource._yield_potential_execution_versions()}

def _new_response_list_pipeline_executions_empty(self: SageMakerResponse) -> TYPE_RESPONSE:
response: dict[str, object] = {
"PipelineExecutionSummaries": [],
}
return 200, {}, json.dumps(response)
self.assertNotIn(new_version.execution_arn, intitial_executions)
self.assertIn(new_version.execution_arn, new_executions)