Skip to content
Merged
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
95 changes: 95 additions & 0 deletions tests/integration/test_integration_vespa_cloud.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,25 @@

from vespa.package import (
EmptyDeploymentConfiguration,
ServicesConfiguration,
Validation,
ValidationID,
sample_package,
)
from vespa.configuration.services import (
container,
content,
document,
documents,
node,
nodes,
redundancy,
search,
secrets,
services,
)
from vespa.configuration.vt import vt
import hashlib

APP_INIT_TIMEOUT = 900

Expand Down Expand Up @@ -630,3 +645,83 @@ def tearDown(self) -> None:
# Wait a little bit to make sure the deployment is finished
time.sleep(10)
self.vespa_cloud.delete(instance=self.instance, environment=self.environment)


# @unittest.skip("Creates cloud resources — run manually to verify vault access flow")
@unittest.skip(
"Requires interactive (Auth0) login — API key auth cannot set vault access rules. "
"To run manually: comment out this @unittest.skip decorator."
)
class TestDeployAddsVaultAccessCloud(unittest.TestCase):
"""
End-to-end test: deploy an app with secrets to verify vault access rule setup.

Requires:
- Interactive (Auth0) login (not API key auth) — needed for vault rule modifications
- A pre-existing vault and secret in the tenant
- The app does not need to exist beforehand
"""

def setUp(self) -> None:
self.tenant = "thttest04"
# generate a random application name
self.app_name = "test" + hashlib.md5(str(time.time()).encode()).hexdigest()[:8]
self.vault_name = "pyvespa-testvault"
self.secret_name = "my-api-key"

schema = Schema(
name="doc",
document=Document(
fields=[
Field(name="text", type="string", indexing=["index", "summary"]),
]
),
)
services_config = ServicesConfiguration(
application_name=self.app_name,
services_config=services(
container(id=f"{self.app_name}_container", version="1.0")(
secrets(
vt(
tag="apiKey",
vault=self.vault_name,
name=self.secret_name,
),
),
search(),
),
content(id=f"{self.app_name}_content", version="1.0")(
redundancy("1"),
documents(document(type_="doc", mode="index")),
nodes(node(distribution_key="0", hostalias="node1")),
),
),
)
self.app_package = ApplicationPackage(
name=self.app_name,
schema=[schema],
services_config=services_config,
)
self.vespa_cloud = VespaCloud(
tenant=self.tenant,
application=self.app_name,
application_package=self.app_package,
)

def test_deploy_with_secrets(self):
"""Deploy an app with vault secrets — verifies vault access rules are auto-configured."""
# Verify services.xml contains the secrets/vault reference
services_xml = self.app_package.services_to_text
self.assertIn("<secrets>", services_xml)
self.assertIn(self.vault_name, services_xml)

# Verify vault names are parsed correctly
vault_names = VespaCloud._parse_vault_names_from_services_xml(services_xml)
self.assertEqual(vault_names, {self.vault_name})

# Deploy — _ensure_vault_access_for_dev runs automatically
app = self.vespa_cloud.deploy()
self.assertIsNotNone(app)

def tearDown(self) -> None:
self.vespa_cloud.delete()
Loading