Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
70 changes: 70 additions & 0 deletions tests/unit/vertexai/genai/replays/test_skills_revisions_get.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Tests the skills.revisions methods against the autopush endpoint."""

from tests.unit.vertexai.genai.replays import pytest_helper
from vertexai._genai import types


def test_get_skill_revision(client, tmp_path):
# Target the autopush sandbox endpoint for the Skill Registry API
client._api_client._http_options.base_url = (
"https://us-central1-autopush-aiplatform.sandbox.googleapis.com"
)

# 1. Create a fresh unique skill
with open(tmp_path / "SKILL.md", "w") as f:
f.write("# Replay Revision Test Skill\nThis is a test skill.")

created_skill = client.skills.create(
display_name="Replay Revision Test Skill",
description="A temporary skill to test revisions E2E",
config=types.CreateSkillConfig(
local_path=str(tmp_path), wait_for_completion=True
),
)

try:
assert created_skill.name is not None

# 2. List revisions to dynamically discover the revision ID
revisions_response = client.skills.revisions.list(name=created_skill.name)
revisions_list = revisions_response.skill_revisions

assert len(revisions_list) > 0
first_revision = revisions_list[0]
assert isinstance(first_revision, types.SkillRevision)
assert first_revision.name is not None

# 3. Explicitly GET the revision by its resource name
revision = client.skills.revisions.get(name=first_revision.name)

assert isinstance(revision, types.SkillRevision)
assert revision.name == first_revision.name
assert revision.state == types.SkillState.ACTIVE

finally:
# 4. Clean up the temporary skill
client.skills.delete(
name=created_skill.name,
config=types.DeleteSkillConfig(wait_for_completion=True),
)


pytestmark = pytest_helper.setup(
file=__file__,
globals_for_file=globals(),
test_method="skills.revisions.get",
)
64 changes: 64 additions & 0 deletions tests/unit/vertexai/genai/replays/test_skills_revisions_list.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Tests the skills.revisions.list() method against the autopush endpoint."""

from tests.unit.vertexai.genai.replays import pytest_helper
from vertexai._genai import types


def test_list_skill_revisions(client, tmp_path):
# Target the autopush sandbox endpoint for the Skill Registry API
client._api_client._http_options.base_url = (
"https://us-central1-autopush-aiplatform.sandbox.googleapis.com"
)

# 1. Create a fresh unique skill
with open(tmp_path / "SKILL.md", "w") as f:
f.write("# Replay List Revisions Test Skill\nThis is a test skill.")

created_skill = client.skills.create(
display_name="Replay List Revisions Test Skill",
description="A temporary skill to test list revisions E2E",
config=types.CreateSkillConfig(
local_path=str(tmp_path), wait_for_completion=True
),
)

try:
assert created_skill.name is not None

# 2. List revisions
revisions_response = client.skills.revisions.list(name=created_skill.name)
revisions_list = revisions_response.skill_revisions

assert len(revisions_list) > 0
first_revision = revisions_list[0]
assert isinstance(first_revision, types.SkillRevision)
assert first_revision.name is not None
assert first_revision.state == types.SkillState.ACTIVE

finally:
# 3. Clean up the temporary skill
client.skills.delete(
name=created_skill.name,
config=types.DeleteSkillConfig(wait_for_completion=True),
)


pytestmark = pytest_helper.setup(
file=__file__,
globals_for_file=globals(),
test_method="skills.revisions.list",
)
122 changes: 120 additions & 2 deletions tests/unit/vertexai/genai/test_genai_skills.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,6 @@ def async_skills_client():


class TestGenaiSkills:
"""Tests the Genai Skills client."""

mock_get_skill_response = {
"name": "projects/test-project/locations/test-location/skills/test-skill",
Expand Down Expand Up @@ -219,7 +218,8 @@ def test_create_skill(self, skills_client):
config={"local_path": tmpdir, "wait_for_completion": True},
)

# Verify requests using robust assert_has_calls matching mock.ANY for base64 zipped Filesystem
# Verify requests using robust assert_has_calls matching
# mock.ANY for base64 zipped Filesystem
request_mock.assert_has_calls(
[
mock.call(
Expand Down Expand Up @@ -896,3 +896,121 @@ async def test_delete_skill_async(self, async_skills_client):
),
]
)

def test_get_skill_revision(self, skills_client):
revision_name = "projects/test-project/locations/test-location/skills/test-skill/revisions/rev-1"
mock_response = {
"name": revision_name,
"state": "ACTIVE",
}

with mock.patch.object(
skills_client._api_client, "request", autospec=True
) as request_mock:
request_mock.return_value = genai_types.HttpResponse(
body=json.dumps(mock_response)
)

revision = skills_client.revisions.get(name=revision_name)

assert isinstance(revision, genai.types.SkillRevision)
assert revision.name == revision_name
assert revision.state == "ACTIVE"

request_mock.assert_called_once_with(
"get",
revision_name,
{"_url": {"name": revision_name}},
None,
)

@pytest.mark.asyncio
async def test_get_skill_revision_async(self, async_skills_client):
revision_name = "projects/test-project/locations/test-location/skills/test-skill/revisions/rev-1"
mock_response = {
"name": revision_name,
"state": "ACTIVE",
}

with mock.patch.object(
async_skills_client._api_client, "async_request", autospec=True
) as request_mock:
request_mock.return_value = genai_types.HttpResponse(
body=json.dumps(mock_response)
)

revision = await async_skills_client.revisions.get(name=revision_name)

assert isinstance(revision, genai.types.SkillRevision)
assert revision.name == revision_name
assert revision.state == "ACTIVE"

request_mock.assert_called_once_with(
"get",
revision_name,
{"_url": {"name": revision_name}},
None,
)

def test_list_skill_revisions(self, skills_client):
skill_name = "projects/test-project/locations/test-location/skills/test-skill"
mock_response = {
"skillRevisions": [
{
"name": f"{skill_name}/revisions/rev-1",
"state": "ACTIVE",
}
]
}

with mock.patch.object(
skills_client._api_client, "request", autospec=True
) as request_mock:
request_mock.return_value = genai_types.HttpResponse(
body=json.dumps(mock_response)
)

response = skills_client.revisions.list(name=skill_name)

assert isinstance(response, genai.types.ListSkillRevisionsResponse)
assert len(response.skill_revisions) == 1
assert response.skill_revisions[0].name == f"{skill_name}/revisions/rev-1"

request_mock.assert_called_once_with(
"get",
f"{skill_name}/revisions",
{"_url": {"name": skill_name}},
None,
)

@pytest.mark.asyncio
async def test_list_skill_revisions_async(self, async_skills_client):
skill_name = "projects/test-project/locations/test-location/skills/test-skill"
mock_response = {
"skillRevisions": [
{
"name": f"{skill_name}/revisions/rev-1",
"state": "ACTIVE",
}
]
}

with mock.patch.object(
async_skills_client._api_client, "async_request", autospec=True
) as request_mock:
request_mock.return_value = genai_types.HttpResponse(
body=json.dumps(mock_response)
)

response = await async_skills_client.revisions.list(name=skill_name)

assert isinstance(response, genai.types.ListSkillRevisionsResponse)
assert len(response.skill_revisions) == 1
assert response.skill_revisions[0].name == f"{skill_name}/revisions/rev-1"

request_mock.assert_called_once_with(
"get",
f"{skill_name}/revisions",
{"_url": {"name": skill_name}},
None,
)
Loading
Loading