Compare commits

...

6 Commits

Author SHA1 Message Date
Jonathon Broughton e99efe8105 Add workspaceId to test automation run and schema 2026-04-07 22:03:13 +01:00
Jonathon Broughton 30de4d207a Handle generic exceptions in AutomationContext project retrieval 2026-04-07 22:03:12 +01:00
Jonathon Broughton 69829266a4 extend AutomationRunData model to include workspace_id 2026-04-07 21:59:18 +01:00
Jonathon Broughton ca1b8c52ed add workspace_id property and resolution logic in AutomationContext 2026-04-07 21:59:17 +01:00
Jonathon Broughton aa16234e7f feat(automate): allow automation results with no affected objects (#488)
* allow empty affected objects

* adds unit tests for `attach_result_to_objects` method

Introduces tests for handling empty object lists and objects with IDs.

Enhances error handling for cases where objects lack IDs, ensuring robustness in the functionality.

Confirms that the method correctly appends results under various scenarios.

* line length
2026-02-24 20:21:59 +00:00
Jonathon Broughton c1f82fa0d2 fix(tests): Update broken test cases for StreamWrapper URLs (#489)
* Update test cases for StreamWrapper URLs

* Update branch name in StreamWrapper test

* Update project URLs in test_wrapper.py

* Uncomment URLs in test_to_string function

Uncommented specific URLs in the test case to enable testing.
2026-02-23 11:29:35 +01:00
5 changed files with 160 additions and 19 deletions
+36 -9
View File
@@ -97,6 +97,33 @@ class AutomationContext:
"""Get the current status message."""
return self._automation_result.status_message
@property
def workspace_id(self) -> Optional[str]:
"""Get the workspace id for the current automation run, if available."""
return self.automation_run_data.workspace_id
def resolve_workspace_id(self) -> Optional[str]:
"""Return workspace id from run data or project lookup fallback."""
workspace_id = self.workspace_id
if workspace_id and workspace_id.strip():
return workspace_id.strip()
project_id = self.automation_run_data.project_id
if not project_id:
return None
try:
project = self.speckle_client.project.get(project_id)
except Exception:
return None
workspace_id = getattr(project, "workspace_id", None)
if isinstance(workspace_id, str) and workspace_id.strip():
resolved_workspace_id = workspace_id.strip()
self.automation_run_data.workspace_id = resolved_workspace_id
return resolved_workspace_id
return None
def elapsed(self) -> float:
"""Return the elapsed time in seconds since the initialization time."""
return time.perf_counter() - self._init_time
@@ -493,29 +520,29 @@ class AutomationContext:
Args:
level: Result level.
category (str): A short tag for the event type.
affected_objects (Union[Base, List[Base]]): A single object or a list of
objects that are causing the info case.
affected_objects (Union[Base, List[Base]]): A single object, a list of
objects, or an empty list. When empty, a result case is still
appended with no object IDs (e.g. for skipped rules or version-level
messages).
message (Optional[str]): Optional message.
metadata: User provided metadata key value pairs
visual_overrides: Case specific 3D visual overrides.
"""
if isinstance(affected_objects, list):
if len(affected_objects) < 1:
raise ValueError(
f"Need atleast one object to report a(n) {level.value.upper()}"
)
object_list = affected_objects
else:
object_list = [affected_objects]
ids: Dict[str, Optional[str]] = {}
# When objects are provided, each must have an id (empty list allowed for
# version-level/skipped results).
for o in object_list:
# validate that the Base.id is not None. If its a None, throw an Exception
if not o.id:
if not getattr(o, "id", None):
raise Exception(
f"You can only attach {level} results to objects with an id."
)
ids[o.id] = o.applicationId
ids[o.id] = getattr(o, "applicationId", None)
print(
f"Created new {level.value.upper()}"
f" category: {category} caused by: {message}"
+2
View File
@@ -68,6 +68,7 @@ def create_test_automation_run(
createTestAutomationRun(automationId: $automationId) {
automationRunId
functionRunId
workspaceId
triggers {
payload {
modelId
@@ -119,6 +120,7 @@ def create_test_automation_run_data(
return AutomationRunData(
project_id=test_automation_environment.project_id,
workspace_id=test_automation_run_data.workspace_id,
speckle_server_url=test_automation_environment.server_url,
automation_id=test_automation_environment.automation_id,
automation_run_id=test_automation_run_data.automation_run_id,
+3 -1
View File
@@ -1,7 +1,7 @@
""""""
from enum import Enum
from typing import Any, Literal
from typing import Any, Literal, Optional
from pydantic import BaseModel, ConfigDict, Field
from pydantic.alias_generators import to_camel
@@ -31,6 +31,7 @@ class AutomationRunData(BaseModel):
"""Values of the project / model that triggered the run of this function."""
project_id: str
workspace_id: Optional[str] = None
speckle_server_url: str
automation_id: str
automation_run_id: str
@@ -48,6 +49,7 @@ class TestAutomationRunData(BaseModel):
automation_run_id: str
function_run_id: str
workspace_id: Optional[str] = None
triggers: list[VersionCreationTrigger]
+9 -9
View File
@@ -157,10 +157,10 @@ def test_parse_project():
def test_parse_model():
wrap = StreamWrapper(
"https://latest.speckle.systems/projects/843d07eb10/models/d9eb4918c8"
"https://app.speckle.systems/projects/8be1007be1/models/cc7578012d"
)
assert wrap.branch_name == "building wrapper"
assert wrap.branch_name == "speckle tower revit 2025"
assert wrap.type == "branch"
@@ -191,10 +191,10 @@ def test_parse_object_fe2():
def test_parse_version():
wrap = StreamWrapper(
"https://latest.speckle.systems/projects/843d07eb10/models/4e7345c838@c42d5cbac1"
"https://app.speckle.systems/projects/8be1007be1/models/cc7578012d@7199443eff"
)
wrap_quoted = StreamWrapper(
"https://latest.speckle.systems/projects/843d07eb10/models/4e7345c838%40c42d5cbac1"
"https://app.speckle.systems/projects/8be1007be1/models/cc7578012d%407199443eff"
)
assert wrap.type == "commit"
assert wrap_quoted.type == "commit"
@@ -208,11 +208,11 @@ def test_to_string():
"https://testing.speckle.dev/streams/0c6ad366c4/globals/abd3787893",
"https://testing.speckle.dev/streams/4c3ce1459c/commits/8b9b831792",
"https://testing.speckle.dev/streams/a75ab4f10f/objects/5530363e6d51c904903dafc3ea1d2ec6",
"https://latest.speckle.systems/projects/843d07eb10",
"https://latest.speckle.systems/projects/843d07eb10/models/4e7345c838",
"https://latest.speckle.systems/projects/843d07eb10/models/4e7345c838@c42d5cbac1",
"https://latest.speckle.systems/projects/843d07eb10/models/4e7345c838%40c42d5cbac1",
"https://latest.speckle.systems/projects/24c3741255/models/b48d1b10f5a732f4ca4144286391282c",
"https://app.speckle.systems/projects/8be1007be1",
"https://app.speckle.systems/projects/8be1007be1/models/cc7578012d",
"https://app.speckle.systems/projects/8be1007be1/models/cc7578012d@7199443eff",
"https://app.speckle.systems/projects/8be1007be1/models/cc7578012d%407199443eff",
"https://app.speckle.systems/projects/8be1007be1/models/9b5e57dca804a923a8d42d55dcc0191a",
]
for url in urls:
wrap = StreamWrapper(url)
@@ -0,0 +1,110 @@
"""Unit tests for AutomationContext.attach_result_to_objects contract."""
from unittest.mock import MagicMock
import pytest
from speckle_automate import AutomationContext
from speckle_automate.schema import (
AutomationRunData,
ObjectResultLevel,
VersionCreationTrigger,
VersionCreationTriggerPayload,
)
from specklepy.objects.base import Base
def _minimal_automation_context() -> AutomationContext:
run_data = AutomationRunData(
project_id="p",
speckle_server_url="http://localhost",
automation_id="a",
automation_run_id="r",
function_run_id="f",
triggers=[
VersionCreationTrigger(
trigger_type="versionCreation",
payload=VersionCreationTriggerPayload(model_id="m", version_id="v"),
)
],
)
return AutomationContext(
automation_run_data=run_data,
speckle_client=MagicMock(),
_server_transport=MagicMock(),
_speckle_token="",
)
def test_attach_result_to_objects_accepts_empty_list() -> None:
"""Empty affected_objects appends one result case with no object IDs."""
ctx = _minimal_automation_context()
assert len(ctx._automation_result.object_results) == 0
ctx.attach_result_to_objects(
ObjectResultLevel.WARNING,
"SkippedRule",
[],
message="No elements to check.",
)
assert len(ctx._automation_result.object_results) == 1
case = ctx._automation_result.object_results[0]
assert case.level == ObjectResultLevel.WARNING
assert case.category == "SkippedRule"
assert case.object_app_ids == {}
assert case.message == "No elements to check."
def test_attach_result_to_objects_with_objects_appends_case_with_ids() -> None:
"""Single or multiple objects with id produce result case with object_app_ids."""
ctx = _minimal_automation_context()
obj1 = Base()
obj1.id = "id-one"
obj1.applicationId = "app-one"
obj2 = Base()
obj2.id = "id-two"
ctx.attach_result_to_objects(
ObjectResultLevel.ERROR,
"BadType",
[obj1, obj2],
message="Invalid type.",
)
assert len(ctx._automation_result.object_results) == 1
case = ctx._automation_result.object_results[0]
assert case.level == ObjectResultLevel.ERROR
assert case.category == "BadType"
assert case.object_app_ids == {"id-one": "app-one", "id-two": None}
assert case.message == "Invalid type."
def test_attach_result_to_objects_raises_when_object_has_no_id() -> None:
"""At least one object without id raises."""
ctx = _minimal_automation_context()
obj = Base()
obj.id = None
with pytest.raises(Exception, match="results to objects with an id"):
ctx.attach_result_to_objects(
ObjectResultLevel.ERROR,
"Bad",
obj,
message="No id.",
)
assert len(ctx._automation_result.object_results) == 0
def test_attach_info_to_objects_accepts_empty_list() -> None:
"""attach_info_to_objects (convenience method) also accepts empty list."""
ctx = _minimal_automation_context()
ctx.attach_info_to_objects("VersionLevel", [], message="No levels in model.")
assert len(ctx._automation_result.object_results) == 1
case = ctx._automation_result.object_results[0]
assert case.level == ObjectResultLevel.INFO
assert case.category == "VersionLevel"
assert case.object_app_ids == {}