vodorod/Dorod-Sky
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

add orgnization ID to build_xxxx_uri methods + make methods require named args + add basic tests (#2628)

Browse Source
This commit is contained in:
Asher Foa
2025-06-10 13:29:10 -04:00
committed by GitHub
parent cd0b5e25f7
commit e30a4cf258
8 changed files with 361 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/ci.yml vendored
View File

@@ -107,6 +107,8 @@ jobs:
env:
ENABLE_OPENAI: "true"
OPENAI_API_KEY: "sk-dummy"
AWS_ACCESS_KEY_ID: "dummy"
AWS_SECRET_ACCESS_KEY: "dummy"
run: poetry run pytest
fe-lint-build:
name: Frontend Lint and Build

35
skyvern/forge/sdk/artifact/manager.py
View File

@@ -100,7 +100,12 @@ class ArtifactManager:
path: str | None = None,
) -> str:
artifact_id = generate_artifact_id()
uri = app.STORAGE.build_log_uri(log_entity_type, log_entity_id, artifact_type)
uri = app.STORAGE.build_log_uri(
organization_id=organization_id,
log_entity_type=log_entity_type,
log_entity_id=log_entity_id,
artifact_type=artifact_type,
)
return await self._create_artifact(
aio_task_primary_key=log_entity_id,
artifact_id=artifact_id,
@@ -123,7 +128,12 @@ class ArtifactManager:
path: str | None = None,
) -> str:
artifact_id = generate_artifact_id()
uri = app.STORAGE.build_thought_uri(artifact_id, thought, artifact_type)
uri = app.STORAGE.build_thought_uri(
organization_id=thought.organization_id,
artifact_id=artifact_id,
thought=thought,
artifact_type=artifact_type,
)
return await self._create_artifact(
aio_task_primary_key=thought.observer_cruise_id,
artifact_id=artifact_id,
@@ -144,7 +154,12 @@ class ArtifactManager:
path: str | None = None,
) -> str:
artifact_id = generate_artifact_id()
uri = app.STORAGE.build_task_v2_uri(artifact_id, task_v2, artifact_type)
uri = app.STORAGE.build_task_v2_uri(
organization_id=task_v2.organization_id,
artifact_id=artifact_id,
task_v2=task_v2,
artifact_type=artifact_type,
)
return await self._create_artifact(
aio_task_primary_key=task_v2.observer_cruise_id,
artifact_id=artifact_id,
@@ -164,7 +179,12 @@ class ArtifactManager:
path: str | None = None,
) -> str:
artifact_id = generate_artifact_id()
uri = app.STORAGE.build_workflow_run_block_uri(artifact_id, workflow_run_block, artifact_type)
uri = app.STORAGE.build_workflow_run_block_uri(
organization_id=workflow_run_block.organization_id,
artifact_id=artifact_id,
workflow_run_block=workflow_run_block,
artifact_type=artifact_type,
)
return await self._create_artifact(
aio_task_primary_key=workflow_run_block.workflow_run_block_id,
artifact_id=artifact_id,
@@ -185,7 +205,12 @@ class ArtifactManager:
path: str | None = None,
) -> str:
artifact_id = generate_artifact_id()
uri = app.STORAGE.build_ai_suggestion_uri(artifact_id, ai_suggestion, artifact_type)
uri = app.STORAGE.build_ai_suggestion_uri(
organization_id=ai_suggestion.organization_id,
artifact_id=artifact_id,
ai_suggestion=ai_suggestion,
artifact_type=artifact_type,
)
return await self._create_artifact(
aio_task_primary_key=ai_suggestion.ai_suggestion_id,
artifact_id=artifact_id,

21
skyvern/forge/sdk/artifact/storage/base.py
View File

@@ -47,26 +47,37 @@ class BaseStorage(ABC):
pass
@abstractmethod
def build_log_uri(self, log_entity_type: LogEntityType, log_entity_id: str, artifact_type: ArtifactType) -> str:
def build_log_uri(
self, *, organization_id: str, log_entity_type: LogEntityType, log_entity_id: str, artifact_type: ArtifactType
) -> str:
pass
@abstractmethod
def build_thought_uri(self, artifact_id: str, thought: Thought, artifact_type: ArtifactType) -> str:
def build_thought_uri(
self, *, organization_id: str, artifact_id: str, thought: Thought, artifact_type: ArtifactType
) -> str:
pass
@abstractmethod
def build_task_v2_uri(self, artifact_id: str, task_v2: TaskV2, artifact_type: ArtifactType) -> str:
def build_task_v2_uri(
self, *, organization_id: str, artifact_id: str, task_v2: TaskV2, artifact_type: ArtifactType
) -> str:
pass
@abstractmethod
def build_workflow_run_block_uri(
self, artifact_id: str, workflow_run_block: WorkflowRunBlock, artifact_type: ArtifactType
self,
*,
organization_id: str,
artifact_id: str,
workflow_run_block: WorkflowRunBlock,
artifact_type: ArtifactType,
) -> str:
pass
@abstractmethod
def build_ai_suggestion_uri(
self, artifact_id: str, ai_suggestion: AISuggestion, artifact_type: ArtifactType
self, *, organization_id: str, artifact_id: str, ai_suggestion: AISuggestion, artifact_type: ArtifactType
) -> str:
pass

29
skyvern/forge/sdk/artifact/storage/local.py
View File

@@ -43,29 +43,40 @@ class LocalStorage(BaseStorage):
except Exception:
return []
def build_log_uri(self, log_entity_type: LogEntityType, log_entity_id: str, artifact_type: ArtifactType) -> str:
def build_log_uri(
self, *, organization_id: str, log_entity_type: LogEntityType, log_entity_id: str, artifact_type: ArtifactType
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"file://{self.artifact_path}/logs/{log_entity_type}/{log_entity_id}/{datetime.utcnow().isoformat()}_{artifact_type}.{file_ext}"
def build_thought_uri(self, artifact_id: str, thought: Thought, artifact_type: ArtifactType) -> str:
def build_thought_uri(
self, *, organization_id: str, artifact_id: str, thought: Thought, artifact_type: ArtifactType
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"file://{self.artifact_path}/{settings.ENV}/tasks/{thought.observer_cruise_id}/{thought.observer_thought_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
return f"file://{self.artifact_path}/{settings.ENV}/{organization_id}/tasks/{thought.observer_cruise_id}/{thought.observer_thought_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
def build_task_v2_uri(self, artifact_id: str, task_v2: TaskV2, artifact_type: ArtifactType) -> str:
def build_task_v2_uri(
self, *, organization_id: str, artifact_id: str, task_v2: TaskV2, artifact_type: ArtifactType
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"file://{self.artifact_path}/{settings.ENV}/observers/{task_v2.observer_cruise_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
return f"file://{self.artifact_path}/{settings.ENV}/{organization_id}/observers/{task_v2.observer_cruise_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
def build_workflow_run_block_uri(
self, artifact_id: str, workflow_run_block: WorkflowRunBlock, artifact_type: ArtifactType
self,
*,
organization_id: str,
artifact_id: str,
workflow_run_block: WorkflowRunBlock,
artifact_type: ArtifactType,
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"file://{self.artifact_path}/{settings.ENV}/workflow_runs/{workflow_run_block.workflow_run_id}/{workflow_run_block.workflow_run_block_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
return f"file://{self.artifact_path}/{settings.ENV}/{organization_id}/workflow_runs/{workflow_run_block.workflow_run_id}/{workflow_run_block.workflow_run_block_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
def build_ai_suggestion_uri(
self, artifact_id: str, ai_suggestion: AISuggestion, artifact_type: ArtifactType
self, *, organization_id: str, artifact_id: str, ai_suggestion: AISuggestion, artifact_type: ArtifactType
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"file://{self.artifact_path}/{settings.ENV}/ai_suggestions/{ai_suggestion.ai_suggestion_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
return f"file://{self.artifact_path}/{settings.ENV}/{organization_id}/ai_suggestions/{ai_suggestion.ai_suggestion_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
async def store_artifact(self, artifact: Artifact, data: bytes) -> None:
file_path = None

35
skyvern/forge/sdk/artifact/storage/s3.py
View File

@@ -29,13 +29,15 @@ LOG = structlog.get_logger()
class S3Storage(BaseStorage):
_PATH_VERSION = "v1"
def __init__(self, bucket: str | None = None) -> None:
self.async_client = AsyncAWSClient()
self.bucket = bucket or settings.AWS_S3_BUCKET_ARTIFACTS
def build_uri(self, artifact_id: str, step: Step, artifact_type: ArtifactType) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"s3://{self.bucket}/{settings.ENV}/{step.task_id}/{step.order:02d}_{step.retry_index}_{step.step_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
return f"s3://{self.bucket}/{self._PATH_VERSION}/{settings.ENV}/{step.task_id}/{step.order:02d}_{step.retry_index}_{step.step_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
async def retrieve_global_workflows(self) -> list[str]:
uri = f"s3://{self.bucket}/{settings.ENV}/global_workflows.txt"
@@ -44,29 +46,40 @@ class S3Storage(BaseStorage):
return []
return [line.strip() for line in data.decode("utf-8").split("\n") if line.strip()]
def build_log_uri(self, log_entity_type: LogEntityType, log_entity_id: str, artifact_type: ArtifactType) -> str:
def build_log_uri(
self, *, organization_id: str, log_entity_type: LogEntityType, log_entity_id: str, artifact_type: ArtifactType
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"s3://{self.bucket}/{settings.ENV}/logs/{log_entity_type}/{log_entity_id}/{datetime.utcnow().isoformat()}_{artifact_type}.{file_ext}"
return f"s3://{self.bucket}/{self._PATH_VERSION}/{settings.ENV}/{organization_id}/logs/{log_entity_type}/{log_entity_id}/{datetime.utcnow().isoformat()}_{artifact_type}.{file_ext}"
def build_thought_uri(self, artifact_id: str, thought: Thought, artifact_type: ArtifactType) -> str:
def build_thought_uri(
self, *, organization_id: str, artifact_id: str, thought: Thought, artifact_type: ArtifactType
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"s3://{self.bucket}/{settings.ENV}/observers/{thought.observer_cruise_id}/{thought.observer_thought_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
return f"s3://{self.bucket}/{self._PATH_VERSION}/{settings.ENV}/{organization_id}/observers/{thought.observer_cruise_id}/{thought.observer_thought_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
def build_task_v2_uri(self, artifact_id: str, task_v2: TaskV2, artifact_type: ArtifactType) -> str:
def build_task_v2_uri(
self, *, organization_id: str, artifact_id: str, task_v2: TaskV2, artifact_type: ArtifactType
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"s3://{self.bucket}/{settings.ENV}/observers/{task_v2.observer_cruise_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
return f"s3://{self.bucket}/{self._PATH_VERSION}/{settings.ENV}/{organization_id}/observers/{task_v2.observer_cruise_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
def build_workflow_run_block_uri(
self, artifact_id: str, workflow_run_block: WorkflowRunBlock, artifact_type: ArtifactType
self,
*,
organization_id: str,
artifact_id: str,
workflow_run_block: WorkflowRunBlock,
artifact_type: ArtifactType,
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"s3://{self.bucket}/{settings.ENV}/workflow_runs/{workflow_run_block.workflow_run_id}/{workflow_run_block.workflow_run_block_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
return f"s3://{self.bucket}/{self._PATH_VERSION}/{settings.ENV}/{organization_id}/workflow_runs/{workflow_run_block.workflow_run_id}/{workflow_run_block.workflow_run_block_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
def build_ai_suggestion_uri(
self, artifact_id: str, ai_suggestion: AISuggestion, artifact_type: ArtifactType
self, *, organization_id: str, artifact_id: str, ai_suggestion: AISuggestion, artifact_type: ArtifactType
) -> str:
file_ext = FILE_EXTENTSION_MAP[artifact_type]
return f"s3://{self.bucket}/{settings.ENV}/ai_suggestions/{ai_suggestion.ai_suggestion_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
return f"s3://{self.bucket}/{self._PATH_VERSION}/{settings.ENV}/{organization_id}/ai_suggestions/{ai_suggestion.ai_suggestion_id}/{datetime.utcnow().isoformat()}_{artifact_id}_{artifact_type}.{file_ext}"
async def store_artifact(self, artifact: Artifact, data: bytes) -> None:
sc = await self._get_storage_class_for_org(artifact.organization_id)

65
skyvern/forge/sdk/artifact/storage/test_helpers.py Normal file
View File

@@ -0,0 +1,65 @@
from datetime import datetime
from skyvern.forge.sdk.models import Step, StepStatus
from skyvern.forge.sdk.schemas.ai_suggestions import AISuggestion
from skyvern.forge.sdk.schemas.task_v2 import TaskV2, Thought
from skyvern.forge.sdk.schemas.workflow_runs import BlockType, WorkflowRunBlock
# Constants
TEST_ORGANIZATION_ID = "test-org-123"
TEST_TASK_ID = "tsk_123456789"
def create_fake_for_ai_suggestion(ai_suggestion_id: str) -> AISuggestion:
return AISuggestion(
ai_suggestion_id=ai_suggestion_id,
organization_id=TEST_ORGANIZATION_ID,
ai_suggestion_type="test_suggestion_type",
created_at=datetime.utcnow(),
modified_at=datetime.utcnow(),
)
def create_fake_thought(cruise_id: str, thought_id: str) -> Thought:
return Thought(
observer_cruise_id=cruise_id,
observer_thought_id=thought_id,
created_at=datetime.utcnow(),
modified_at=datetime.utcnow(),
organization_id=TEST_ORGANIZATION_ID,
)
def create_fake_step(step_id: str) -> Step:
return Step(
task_id=TEST_TASK_ID,
order=1,
retry_index=0,
step_id=step_id,
created_at=datetime.utcnow(),
modified_at=datetime.utcnow(),
status=StepStatus.created,
is_last=False,
organization_id=TEST_ORGANIZATION_ID,
)
def create_fake_task_v2(observer_cruise_id: str) -> TaskV2:
return TaskV2(
observer_cruise_id=observer_cruise_id,
created_at=datetime.utcnow(),
modified_at=datetime.utcnow(),
status=StepStatus.created,
organization_id=TEST_ORGANIZATION_ID,
)
def create_fake_workflow_run_block(workflow_run_id: str, workflow_run_block_id: str) -> WorkflowRunBlock:
return WorkflowRunBlock(
workflow_run_id=workflow_run_id,
workflow_run_block_id=workflow_run_block_id,
created_at=datetime.utcnow(),
modified_at=datetime.utcnow(),
organization_id=TEST_ORGANIZATION_ID,
block_type=BlockType.TASK,
)

102
skyvern/forge/sdk/artifact/storage/test_local_storage.py Normal file
View File

@@ -0,0 +1,102 @@
import pytest
from freezegun import freeze_time
from skyvern.config import settings
from skyvern.forge.sdk.artifact.models import ArtifactType, LogEntityType
from skyvern.forge.sdk.artifact.storage.local import LocalStorage
from skyvern.forge.sdk.artifact.storage.test_helpers import (
create_fake_for_ai_suggestion,
create_fake_step,
create_fake_task_v2,
create_fake_thought,
create_fake_workflow_run_block,
)
# Test constants
TEST_BUCKET = "test-skyvern-bucket"
TEST_ORGANIZATION_ID = "test-org-123"
TEST_TASK_ID = "tsk_123456789"
TEST_STEP_ID = "step_123456789"
TEST_WORKFLOW_RUN_ID = "wfr_123456789"
TEST_BLOCK_ID = "block_123456789"
TEST_AI_SUGGESTION_ID = "ai_sugg_test_123"
@pytest.fixture
def local_storage() -> LocalStorage:
return LocalStorage()
@freeze_time("2025-06-09T12:00:00")
class TestLocalStorageBuildURIs:
def test_build_uri(self, local_storage: LocalStorage) -> None:
step = create_fake_step(TEST_STEP_ID)
uri = local_storage.build_uri("artifact123", step, ArtifactType.LLM_PROMPT)
assert (
uri
== f"file://{local_storage.artifact_path}/{TEST_TASK_ID}/01_0_{TEST_STEP_ID}/2025-06-09T12:00:00_artifact123_llm_prompt.txt"
)
def test_build_log_uri(self, local_storage: LocalStorage) -> None:
uri = local_storage.build_log_uri(
organization_id=TEST_ORGANIZATION_ID,
log_entity_type=LogEntityType.WORKFLOW_RUN_BLOCK,
log_entity_id="log_id",
artifact_type=ArtifactType.SKYVERN_LOG,
)
assert (
uri
== f"file://{local_storage.artifact_path}/logs/workflow_run_block/log_id/2025-06-09T12:00:00_skyvern_log.log"
)
def test_build_thought_uri(self, local_storage: LocalStorage) -> None:
thought = create_fake_thought("cruise123", "thought123")
uri = local_storage.build_thought_uri(
organization_id=TEST_ORGANIZATION_ID,
artifact_id="artifact123",
thought=thought,
artifact_type=ArtifactType.VISIBLE_ELEMENTS_TREE,
)
assert (
uri
== f"file://{local_storage.artifact_path}/{settings.ENV}/{TEST_ORGANIZATION_ID}/tasks/cruise123/thought123/2025-06-09T12:00:00_artifact123_visible_elements_tree.json"
)
def test_build_task_v2_uri(self, local_storage: LocalStorage) -> None:
task_v2 = create_fake_task_v2("cruise123")
uri = local_storage.build_task_v2_uri(
organization_id=TEST_ORGANIZATION_ID,
artifact_id="artifact123",
task_v2=task_v2,
artifact_type=ArtifactType.HTML_ACTION,
)
assert (
uri
== f"file://{local_storage.artifact_path}/{settings.ENV}/{TEST_ORGANIZATION_ID}/observers/cruise123/2025-06-09T12:00:00_artifact123_html_action.html"
)
def test_build_workflow_run_block_uri(self, local_storage: LocalStorage) -> None:
workflow_run_block = create_fake_workflow_run_block(TEST_WORKFLOW_RUN_ID, TEST_BLOCK_ID)
uri = local_storage.build_workflow_run_block_uri(
organization_id=TEST_ORGANIZATION_ID,
artifact_id="artifact123",
workflow_run_block=workflow_run_block,
artifact_type=ArtifactType.HAR,
)
assert (
uri
== f"file://{local_storage.artifact_path}/{settings.ENV}/{TEST_ORGANIZATION_ID}/workflow_runs/{TEST_WORKFLOW_RUN_ID}/{TEST_BLOCK_ID}/2025-06-09T12:00:00_artifact123_har.har"
)
def test_build_ai_suggestion_uri(self, local_storage: LocalStorage) -> None:
ai_suggestion = create_fake_for_ai_suggestion(TEST_AI_SUGGESTION_ID)
uri = local_storage.build_ai_suggestion_uri(
organization_id=TEST_ORGANIZATION_ID,
artifact_id="artifact123",
ai_suggestion=ai_suggestion,
artifact_type=ArtifactType.SCREENSHOT_LLM,
)
assert (
uri
== f"file://{local_storage.artifact_path}/{settings.ENV}/{TEST_ORGANIZATION_ID}/ai_suggestions/{TEST_AI_SUGGESTION_ID}/2025-06-09T12:00:00_artifact123_screenshot_llm.png"
)

102
skyvern/forge/sdk/artifact/storage/test_s3_storage.py Normal file
View File

@@ -0,0 +1,102 @@
import pytest
from freezegun import freeze_time
from skyvern.config import settings
from skyvern.forge.sdk.artifact.models import ArtifactType, LogEntityType
from skyvern.forge.sdk.artifact.storage.s3 import S3Storage
from skyvern.forge.sdk.artifact.storage.test_helpers import (
create_fake_for_ai_suggestion,
create_fake_step,
create_fake_task_v2,
create_fake_thought,
create_fake_workflow_run_block,
)
# Test constants
TEST_BUCKET = "test-skyvern-bucket"
TEST_ORGANIZATION_ID = "test-org-123"
TEST_TASK_ID = "tsk_123456789"
TEST_STEP_ID = "step_123456789"
TEST_WORKFLOW_RUN_ID = "wfr_123456789"
TEST_BLOCK_ID = "block_123456789"
TEST_AI_SUGGESTION_ID = "ai_sugg_test_123"
@pytest.fixture
def s3_storage() -> S3Storage:
return S3Storage(bucket=TEST_BUCKET)
@freeze_time("2025-06-09T12:00:00")
class TestS3StorageBuildURIs:
def test_build_uri(self, s3_storage: S3Storage) -> None:
step = create_fake_step(TEST_STEP_ID)
uri = s3_storage.build_uri("artifact123", step, ArtifactType.LLM_PROMPT)
assert (
uri
== f"s3://{TEST_BUCKET}/v1/{settings.ENV}/{TEST_TASK_ID}/01_0_{TEST_STEP_ID}/2025-06-09T12:00:00_artifact123_llm_prompt.txt"
)
def test_build_log_uri(self, s3_storage: S3Storage) -> None:
uri = s3_storage.build_log_uri(
organization_id=TEST_ORGANIZATION_ID,
log_entity_type=LogEntityType.WORKFLOW_RUN_BLOCK,
log_entity_id="log_id",
artifact_type=ArtifactType.SKYVERN_LOG,
)
assert (
uri
== f"s3://{TEST_BUCKET}/v1/{settings.ENV}/{TEST_ORGANIZATION_ID}/logs/workflow_run_block/log_id/2025-06-09T12:00:00_skyvern_log.log"
)
def test_build_thought_uri(self, s3_storage: S3Storage) -> None:
thought = create_fake_thought("cruise123", "thought123")
uri = s3_storage.build_thought_uri(
organization_id=TEST_ORGANIZATION_ID,
artifact_id="artifact123",
thought=thought,
artifact_type=ArtifactType.VISIBLE_ELEMENTS_TREE,
)
assert (
uri
== f"s3://{TEST_BUCKET}/v1/{settings.ENV}/{TEST_ORGANIZATION_ID}/observers/cruise123/thought123/2025-06-09T12:00:00_artifact123_visible_elements_tree.json"
)
def test_build_task_v2_uri(self, s3_storage: S3Storage) -> None:
task_v2 = create_fake_task_v2("cruise123")
uri = s3_storage.build_task_v2_uri(
organization_id=TEST_ORGANIZATION_ID,
artifact_id="artifact123",
task_v2=task_v2,
artifact_type=ArtifactType.HTML_ACTION,
)
assert (
uri
== f"s3://{TEST_BUCKET}/v1/{settings.ENV}/{TEST_ORGANIZATION_ID}/observers/cruise123/2025-06-09T12:00:00_artifact123_html_action.html"
)
def test_build_workflow_run_block_uri(self, s3_storage: S3Storage) -> None:
workflow_run_block = create_fake_workflow_run_block(TEST_WORKFLOW_RUN_ID, TEST_BLOCK_ID)
uri = s3_storage.build_workflow_run_block_uri(
organization_id=TEST_ORGANIZATION_ID,
artifact_id="artifact123",
workflow_run_block=workflow_run_block,
artifact_type=ArtifactType.HAR,
)
assert (
uri
== f"s3://{TEST_BUCKET}/v1/{settings.ENV}/{TEST_ORGANIZATION_ID}/workflow_runs/{TEST_WORKFLOW_RUN_ID}/{TEST_BLOCK_ID}/2025-06-09T12:00:00_artifact123_har.har"
)
def test_build_ai_suggestion_uri(self, s3_storage: S3Storage) -> None:
ai_suggestion = create_fake_for_ai_suggestion(TEST_AI_SUGGESTION_ID)
uri = s3_storage.build_ai_suggestion_uri(
organization_id=TEST_ORGANIZATION_ID,
artifact_id="artifact123",
ai_suggestion=ai_suggestion,
artifact_type=ArtifactType.SCREENSHOT_LLM,
)
assert (
uri
== f"s3://{TEST_BUCKET}/v1/{settings.ENV}/{TEST_ORGANIZATION_ID}/ai_suggestions/{TEST_AI_SUGGESTION_ID}/2025-06-09T12:00:00_artifact123_screenshot_llm.png"
)