huanld/automate-function-bilt-workshop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Directory organisation

Browse Source
This commit is contained in:
Jonathon Broughton
2024-05-06 04:37:03 +00:00
parent 37d2c889bf
commit 6b4f002f63
20 changed files with 632 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Exercise_0/__int__.py → Workshop/Exercise_0/__int__.py
View File
Exercise_1/__init__.py → Workshop/Exercise_1/__init__.py
View File
Exercise_1/automate_function.py → Workshop/Exercise_1/automate_function.py
View File
Exercise_2/__init__.py → Workshop/Exercise_2/__init__.py
View File
Exercise_2/automate_function.py → Workshop/Exercise_2/automate_function.py
View File
Exercise_3/__init__.py → Workshop/Exercise_3/__init__.py
View File
Exercise_3/automate_function.py → Workshop/Exercise_3/automate_function.py
View File
Exercise_3/rules.py → Workshop/Exercise_3/rules.py
View File
Exercise_4/__init__.py → Workshop/Exercise_4/__init__.py
View File
Exercise_4/automate_function.py → Workshop/Exercise_4/automate_function.py
View File
Exercise_4/rules.py → Workshop/Exercise_4/rules.py
View File
Exercise_5/__init__.py → Workshop/Exercise_5/__init__.py
View File
Workshop/__init__.py
View File
tests/__init__.py
View File
tests/conftest.py
+24
View File
@@ -0,0 +1,24 @@
import os
from dotenv import load_dotenv
def pytest_configure(config):
load_dotenv(dotenv_path=".env")
token_var = "SPECKLE_TOKEN"
server_var = "SPECKLE_SERVER_URL"
token = os.getenv(token_var)
server = os.getenv(server_var)
if not token:
raise ValueError(f"Cannot run tests without a {token_var} environment variable")
if not server:
raise ValueError(
f"Cannot run tests without a {server_var} environment variable"
)
# Set the token as an attribute on the config object
config.SPECKLE_TOKEN = token
config.SPECKLE_SERVER_URL = server
tests/test_exercise_1.py
+146
View File
@@ -0,0 +1,146 @@
import os
import secrets
import string
import pytest
from gql import gql
from speckle_automate import (
AutomationContext,
AutomationRunData,
AutomationStatus,
run_function,
)
from specklepy.api.client import SpeckleClient
from Workshop.Exercise_1.automate_function import FunctionInputs, automate_function
def crypto_random_string(length: int) -> str:
"""Generate a semi crypto random string of a given length."""
alphabet = string.ascii_letters + string.digits
return "".join(secrets.choice(alphabet) for _ in range(length))
def register_new_automation(
project_id: str,
model_id: str,
speckle_client: SpeckleClient,
automation_id: str,
automation_name: str,
automation_revision_id: str,
):
"""Register a new automation in the speckle server."""
query = gql(
"""
mutation CreateAutomation(
$projectId: String!
$modelId: String!
$automationName: String!
$automationId: String!
$automationRevisionId: String!
) {
automationMutations {
create(
input: {
projectId: $projectId
modelId: $modelId
automationName: $automationName
automationId: $automationId
automationRevisionId: $automationRevisionId
}
)
}
}
"""
)
params = {
"projectId": project_id,
"modelId": model_id,
"automationName": automation_name,
"automationId": automation_id,
"automationRevisionId": automation_revision_id,
}
speckle_client.httpclient.execute(query, params)
@pytest.fixture()
def speckle_token() -> str:
"""Provide a speckle token for the test suite."""
env_var = "SPECKLE_TOKEN"
token = os.getenv(env_var)
if not token:
raise ValueError(f"Cannot run tests without a {env_var} environment variable")
return token
@pytest.fixture()
def speckle_server_url() -> str:
"""Provide a speckle server url for the test suite, default to localhost."""
return os.getenv("SPECKLE_SERVER_URL", "http://127.0.0.1:3000")
@pytest.fixture()
def test_client(speckle_server_url: str, speckle_token: str) -> SpeckleClient:
"""Initialize a SpeckleClient for testing."""
test_client = SpeckleClient(
speckle_server_url, speckle_server_url.startswith("https")
)
test_client.authenticate_with_token(speckle_token)
return test_client
@pytest.fixture()
# fixture to mock the AutomationRunData that would be generated by a full Automation Run
def fake_automation_run_data(request, test_client: SpeckleClient) -> AutomationRunData:
server_url = request.config.SPECKLE_SERVER_URL
project_id = "ce0f229748"
model_id = "5e44c03b0b"
function_name = "BILT Exercise 1"
automation_id = crypto_random_string(10)
automation_name = "Local Test"
automation_revision_id = crypto_random_string(10)
register_new_automation(
project_id,
model_id,
test_client,
automation_id,
automation_name,
automation_revision_id,
)
fake_run_data = AutomationRunData(
project_id=project_id,
model_id=model_id,
branch_name="exercise 1",
version_id="136fe819e0",
speckle_server_url=server_url,
# These ids would be available with a valid registered Automation definition.
automation_id=automation_id,
automation_revision_id=automation_revision_id,
automation_run_id=crypto_random_string(12),
# These ids would be available with a valid registered Function definition. Can also be faked.
function_id="12345",
function_name=function_name,
function_logo=None,
)
return fake_run_data
@pytest.fixture
def context(fake_automation_run_data: AutomationRunData, speckle_token: str):
return AutomationContext.initialize(fake_automation_run_data, speckle_token)
def test_function_run(fake_automation_run_data: AutomationRunData, speckle_token: str):
"""Run an integration test for the automate function."""
context = AutomationContext.initialize(fake_automation_run_data, speckle_token)
automate_sdk = run_function(
context, automate_function, FunctionInputs(commentPhrase="Tested Locally")
)
assert automate_sdk.run_status == AutomationStatus.SUCCEEDED
tests/test_exercise_2.py
+143
View File
@@ -0,0 +1,143 @@
import os
import secrets
import string
import pytest
from gql import gql
from speckle_automate import (
AutomationContext,
AutomationRunData,
AutomationStatus,
run_function,
)
from specklepy.api.client import SpeckleClient
from Workshop.Exercise_2.automate_function import FunctionInputs, automate_function
def crypto_random_string(length: int) -> str:
"""Generate a semi crypto random string of a given length."""
alphabet = string.ascii_letters + string.digits
return "".join(secrets.choice(alphabet) for _ in range(length))
def register_new_automation(
project_id: str,
model_id: str,
speckle_client: SpeckleClient,
automation_id: str,
automation_name: str,
automation_revision_id: str,
):
"""Register a new automation in the speckle server."""
query = gql(
"""
mutation CreateAutomation(
$projectId: String!
$modelId: String!
$automationName: String!
$automationId: String!
$automationRevisionId: String!
) {
automationMutations {
create(
input: {
projectId: $projectId
modelId: $modelId
automationName: $automationName
automationId: $automationId
automationRevisionId: $automationRevisionId
}
)
}
}
"""
)
params = {
"projectId": project_id,
"modelId": model_id,
"automationName": automation_name,
"automationId": automation_id,
"automationRevisionId": automation_revision_id,
}
speckle_client.httpclient.execute(query, params)
@pytest.fixture()
def speckle_token() -> str:
"""Provide a speckle token for the test suite."""
env_var = "SPECKLE_TOKEN"
token = os.getenv(env_var)
if not token:
raise ValueError(f"Cannot run tests without a {env_var} environment variable")
return token
@pytest.fixture()
def speckle_server_url() -> str:
"""Provide a speckle server url for the test suite, default to localhost."""
return os.getenv("SPECKLE_SERVER_URL", "http://127.0.0.1:3000")
@pytest.fixture()
def test_client(speckle_server_url: str, speckle_token: str) -> SpeckleClient:
"""Initialize a SpeckleClient for testing."""
test_client = SpeckleClient(
speckle_server_url, speckle_server_url.startswith("https")
)
test_client.authenticate_with_token(speckle_token)
return test_client
@pytest.fixture()
# fixture to mock the AutomationRunData that would be generated by a full Automation Run
def fake_automation_run_data(request, test_client: SpeckleClient) -> AutomationRunData:
server_url = request.config.SPECKLE_SERVER_URL
project_id = "ce0f229748"
model_id = "7658a4f311"
function_name = "Local Test"
automation_id = crypto_random_string(10)
automation_name = "BILT Exercise 2"
automation_revision_id = crypto_random_string(10)
register_new_automation(
project_id,
model_id,
test_client,
automation_id,
automation_name,
automation_revision_id,
)
fake_run_data = AutomationRunData(
project_id=project_id,
model_id=model_id,
branch_name="exercise 2",
version_id="bbfa404f93",
speckle_server_url=server_url,
# These ids would be available with a valid registered Automation definition.
automation_id=automation_id,
automation_revision_id=automation_revision_id,
automation_run_id=crypto_random_string(12),
# These ids would be available with a valid registered Function definition. Can also be faked.
function_id="12345",
function_name=function_name,
function_logo=None,
)
return fake_run_data
def test_function_run(fake_automation_run_data: AutomationRunData, speckle_token: str):
"""Run an integration test for the automate function."""
context = AutomationContext.initialize(fake_automation_run_data, speckle_token)
automate_sdk = run_function(
context,
automate_function,
FunctionInputs(commentPhrase="Tested Locally", numberOfElements=10000),
)
assert automate_sdk.run_status == AutomationStatus.SUCCEEDED
tests/test_exercise_3.py
+143
View File
@@ -0,0 +1,143 @@
import os
import secrets
import string
import pytest
from gql import gql
from speckle_automate import (
AutomationContext,
AutomationRunData,
AutomationStatus,
run_function,
)
from specklepy.api.client import SpeckleClient
from Workshop.Exercise_3.automate_function import FunctionInputs, automate_function
def crypto_random_string(length: int) -> str:
"""Generate a semi crypto random string of a given length."""
alphabet = string.ascii_letters + string.digits
return "".join(secrets.choice(alphabet) for _ in range(length))
def register_new_automation(
project_id: str,
model_id: str,
speckle_client: SpeckleClient,
automation_id: str,
automation_name: str,
automation_revision_id: str,
):
"""Register a new automation in the speckle server."""
query = gql(
"""
mutation CreateAutomation(
$projectId: String!
$modelId: String!
$automationName: String!
$automationId: String!
$automationRevisionId: String!
) {
automationMutations {
create(
input: {
projectId: $projectId
modelId: $modelId
automationName: $automationName
automationId: $automationId
automationRevisionId: $automationRevisionId
}
)
}
}
"""
)
params = {
"projectId": project_id,
"modelId": model_id,
"automationName": automation_name,
"automationId": automation_id,
"automationRevisionId": automation_revision_id,
}
speckle_client.httpclient.execute(query, params)
@pytest.fixture()
def speckle_token() -> str:
"""Provide a speckle token for the test suite."""
env_var = "SPECKLE_TOKEN"
token = os.getenv(env_var)
if not token:
raise ValueError(f"Cannot run tests without a {env_var} environment variable")
return token
@pytest.fixture()
def speckle_server_url() -> str:
"""Provide a speckle server url for the test suite, default to localhost."""
return os.getenv("SPECKLE_SERVER_URL", "http://127.0.0.1:3000")
@pytest.fixture()
def test_client(speckle_server_url: str, speckle_token: str) -> SpeckleClient:
"""Initialize a SpeckleClient for testing."""
test_client = SpeckleClient(
speckle_server_url, speckle_server_url.startswith("https")
)
test_client.authenticate_with_token(speckle_token)
return test_client
@pytest.fixture()
# fixture to mock the AutomationRunData that would be generated by a full Automation Run
def fake_automation_run_data(request, test_client: SpeckleClient) -> AutomationRunData:
server_url = request.config.SPECKLE_SERVER_URL
project_id = "ce0f229748"
model_id = "71fc3bfac6"
function_name = "Local Test"
automation_id = crypto_random_string(10)
automation_name = "BILT Exercise 3"
automation_revision_id = crypto_random_string(10)
register_new_automation(
project_id,
model_id,
test_client,
automation_id,
automation_name,
automation_revision_id,
)
fake_run_data = AutomationRunData(
project_id=project_id,
model_id=model_id,
branch_name="exercise 3",
version_id="227572481f",
speckle_server_url=server_url,
# These ids would be available with a valid registered Automation definition.
automation_id=automation_id,
automation_revision_id=automation_revision_id,
automation_run_id=crypto_random_string(12),
# These ids would be available with a valid registered Function definition. Can also be faked.
function_id="12345",
function_name=function_name,
function_logo=None,
)
return fake_run_data
def test_function_run(fake_automation_run_data: AutomationRunData, speckle_token: str):
"""Run an integration test for the automate function."""
context = AutomationContext.initialize(fake_automation_run_data, speckle_token)
automate_sdk = run_function(
context,
automate_function,
FunctionInputs(category="Doors", property="SPECKLE_Classification"),
)
assert automate_sdk.run_status == AutomationStatus.SUCCEEDED
tests/test_exercise_4.py
+149
View File
@@ -0,0 +1,149 @@
import os
import secrets
import string
import pytest
from gql import gql
from speckle_automate import (
AutomationContext,
AutomationRunData,
AutomationStatus,
run_function,
)
from specklepy.api.client import SpeckleClient
from Utilities.helpers import speckle_print
from Workshop.Exercise_4.automate_function import FunctionInputs, automate_function
def crypto_random_string(length: int) -> str:
"""Generate a semi crypto random string of a given length."""
alphabet = string.ascii_letters + string.digits
return "".join(secrets.choice(alphabet) for _ in range(length))
def register_new_automation(
project_id: str,
model_id: str,
speckle_client: SpeckleClient,
automation_id: str,
automation_name: str,
automation_revision_id: str,
):
"""Register a new automation in the speckle server."""
query = gql(
"""
mutation CreateAutomation(
$projectId: String!
$modelId: String!
$automationName: String!
$automationId: String!
$automationRevisionId: String!
) {
automationMutations {
create(
input: {
projectId: $projectId
modelId: $modelId
automationName: $automationName
automationId: $automationId
automationRevisionId: $automationRevisionId
}
)
}
}
"""
)
params = {
"projectId": project_id,
"modelId": model_id,
"automationName": automation_name,
"automationId": automation_id,
"automationRevisionId": automation_revision_id,
}
speckle_client.httpclient.execute(query, params)
@pytest.fixture()
def speckle_token() -> str:
"""Provide a speckle token for the test suite."""
env_var = "SPECKLE_TOKEN"
token = os.getenv(env_var)
if not token:
raise ValueError(f"Cannot run tests without a {env_var} environment variable")
return token
@pytest.fixture()
def speckle_server_url() -> str:
"""Provide a speckle server url for the test suite, default to localhost."""
return os.getenv("SPECKLE_SERVER_URL", "http://127.0.0.1:3000")
@pytest.fixture()
def test_client(speckle_server_url: str, speckle_token: str) -> SpeckleClient:
"""Initialize a SpeckleClient for testing."""
test_client = SpeckleClient(
speckle_server_url, speckle_server_url.startswith("https")
)
test_client.authenticate_with_token(speckle_token)
return test_client
@pytest.fixture()
# fixture to mock the AutomationRunData that would be generated by a full Automation Run
def fake_automation_run_data(request, test_client: SpeckleClient) -> AutomationRunData:
server_url = request.config.SPECKLE_SERVER_URL
project_id = "ce0f229748"
model_id = "6dadd92d7f"
function_name = "Local Test"
automation_id = crypto_random_string(10)
automation_name = "BILT Exercise 4"
automation_revision_id = crypto_random_string(10)
register_new_automation(
project_id,
model_id,
test_client,
automation_id,
automation_name,
automation_revision_id,
)
fake_run_data = AutomationRunData(
project_id=project_id,
model_id=model_id,
branch_name="exercise 4",
version_id="8fe89b87d9",
speckle_server_url=server_url,
# These ids would be available with a valid registered Automation definition.
automation_id=automation_id,
automation_revision_id=automation_revision_id,
automation_run_id=crypto_random_string(12),
# These ids would be available with a valid registered Function definition. Can also be faked.
function_id="12345",
function_name=function_name,
function_logo=None,
)
return fake_run_data
def test_function_run(fake_automation_run_data: AutomationRunData, speckle_token: str):
"""Run an integration test for the automate function."""
context = AutomationContext.initialize(fake_automation_run_data, speckle_token)
default_url: str = (
"https://docs.google.com/spreadsheets/d/e/2PACX-1vSFmjLfqxPKXJHg-wEs1cp_nJEJJhESGVTLCvWLG_"
"IgIuRZ4CmMDCSceOYFvuo8IqcmT4sj9qPiLfCx/pub?gid=0&single=true&output=tsv"
)
automate_sdk = run_function(
context,
automate_function,
FunctionInputs(spreadsheet_url=default_url),
)
assert automate_sdk.run_status == AutomationStatus.SUCCEEDED
tests/test_function.py → tests/tests.py
+27 -49
View File
@@ -1,10 +1,7 @@
"""Run integration tests with a speckle server."""
import os
import secrets
import string
from specklepy.logging.exceptions import SpeckleException
import pytest
from gql import gql
from speckle_automate import (
@@ -13,10 +10,7 @@ from speckle_automate import (
AutomationStatus,
run_function,
)
from specklepy.api import operations
from specklepy.api.client import SpeckleClient
from specklepy.objects.base import Base
from specklepy.transports.server import ServerTransport
from main import FunctionInputs, automate_function
@@ -96,33 +90,16 @@ def test_client(speckle_server_url: str, speckle_token: str) -> SpeckleClient:
@pytest.fixture()
def test_object() -> Base:
"""Create a Base model for testing."""
root_object = Base()
root_object.foo = "bar"
return root_object
# fixture to mock the AutomationRunData that would be generated by a full Automation Run
def fake_automation_run_data(request, test_client: SpeckleClient) -> AutomationRunData:
server_url = request.config.SPECKLE_SERVER_URL
project_id = "ce0f229748"
model_id = "634a39fa45"
function_name = "BILT Riga Workshop"
@pytest.fixture()
def automation_run_data(
test_object: Base, test_client: SpeckleClient, speckle_server_url: str
) -> AutomationRunData:
"""Set up an automation context for testing."""
project_id = test_client.stream.create("Automate function e2e test")
branch_name = "main"
model = test_client.branch.get(project_id, branch_name, commits_limit=1)
model_id: str = model.id
root_obj_id = operations.send(
test_object, [ServerTransport(project_id, test_client)]
)
version_id = test_client.commit.create(project_id, root_obj_id)
if isinstance(version_id, SpeckleException):
raise version_id
automation_name = crypto_random_string(10)
automation_id = crypto_random_string(10)
automation_name = "Local Test"
automation_revision_id = crypto_random_string(10)
register_new_automation(
@@ -134,35 +111,36 @@ def automation_run_data(
automation_revision_id,
)
automation_run_id = crypto_random_string(10)
function_id = crypto_random_string(10)
function_revision = crypto_random_string(10)
return AutomationRunData(
fake_run_data = AutomationRunData(
project_id=project_id,
model_id=model_id,
branch_name=branch_name,
version_id=version_id,
speckle_server_url=speckle_server_url,
branch_name="basic",
version_id="df0f86a3fd",
speckle_server_url=server_url,
# These ids would be available with a valid registered Automation definition.
automation_id=automation_id,
automation_revision_id=automation_revision_id,
automation_run_id=automation_run_id,
function_id=function_id,
function_name=crypto_random_string(10),
automation_run_id=crypto_random_string(12),
# These ids would be available with a valid registered Function definition. Can also be faked.
function_id="12345",
function_name=function_name,
function_logo=None,
)
return fake_run_data
def test_function_run(automation_run_data: AutomationRunData, speckle_token: str):
def test_function_run(fake_automation_run_data: AutomationRunData, speckle_token: str):
"""Run an integration test for the automate function."""
automation_context = AutomationContext.initialize(
automation_run_data, speckle_token
)
context = AutomationContext.initialize(fake_automation_run_data, speckle_token)
automate_sdk = run_function(
automation_context,
context,
automate_function,
FunctionInputs(
forbidden_speckle_type="Base", whisper_message="testing automatically"
),
FunctionInputs(commentPhrase="Test Locally"),
# FunctionInputs(
# tolerance=0.1, tolerance_unit="mm", static_model_name="simple beams"
# ),
)
assert automate_sdk.run_status == AutomationStatus.FAILED
assert automate_sdk.run_status == AutomationStatus.SUCCEEDED