Port.io Testing Guide: Blueprints, Scorecards, and Self-Service Action Testing

import pytest
import httpx
import os

PORT_API = "https://api.getport.io/v1"
CLIENT_ID = os.environ["PORT_CLIENT_ID"]
CLIENT_SECRET = os.environ["PORT_CLIENT_SECRET"]

@pytest.fixture(scope="session")
def port_token():
    r = httpx.post(f"{PORT_API}/auth/access_token", json={
        "clientId": CLIENT_ID,
        "clientSecret": CLIENT_SECRET
    })
    assert r.status_code == 200
    return r.json()["accessToken"]

@pytest.fixture(scope="session")
def port(port_token):
    return httpx.Client(
        base_url=PORT_API,
        headers={"Authorization": f"Bearer {port_token}"},
        timeout=30.0
    )

def test_service_blueprint_exists(port):
    r = port.get("/blueprints/service")
    assert r.status_code == 200, f"Service blueprint not found: {r.text}"
    blueprint = r.json()["blueprint"]
    assert blueprint["identifier"] == "service"

def test_service_blueprint_has_required_properties(port):
    r = port.get("/blueprints/service")
    blueprint = r.json()["blueprint"]
    props = blueprint.get("schema", {}).get("properties", {})
    
    required_props = [
        "language",
        "team",
        "tier",
        "oncall_team",
        "deployment_frequency",
        "has_ci",
        "has_monitoring"
    ]
    
    for prop in required_props:
        assert prop in props, f"Service blueprint missing property: {prop}"

def test_service_blueprint_has_correct_types(port):
    r = port.get("/blueprints/service")
    props = r.json()["blueprint"]["schema"]["properties"]
    
    # Tier should be an enum
    assert props["tier"]["type"] == "string"
    assert "enum" in props["tier"], "tier should be an enum"
    assert set(props["tier"]["enum"]) >= {"tier-1", "tier-2", "tier-3"}
    
    # has_ci should be boolean
    assert props["has_ci"]["type"] == "boolean"

def test_environment_blueprint_exists(port):
    r = port.get("/blueprints/environment")
    assert r.status_code == 200

def test_service_environment_relation_defined(port):
    r = port.get("/blueprints/service")
    blueprint = r.json()["blueprint"]
    relations = blueprint.get("relations", {})
    
    assert "environment" in relations, \
        "Service blueprint should have a relation to Environment"
    assert relations["environment"]["target"] == "environment"

def test_blueprint_schema_validates():
    """Load blueprints from IaC files and validate their structure."""
    import yaml
    import glob
    
    for path in glob.glob("port/blueprints/*.yaml"):
        with open(path) as f:
            blueprint = yaml.safe_load(f)
        
        assert "identifier" in blueprint, f"{path}: missing identifier"
        assert "title" in blueprint, f"{path}: missing title"
        assert "schema" in blueprint, f"{path}: missing schema"
        assert "properties" in blueprint["schema"], f"{path}: missing schema.properties"

def test_service_scorecard_exists(port):
    r = port.get("/blueprints/service/scorecards")
    assert r.status_code == 200
    scorecards = r.json()["scorecards"]
    
    scorecard_ids = [sc["identifier"] for sc in scorecards]
    assert "production_readiness" in scorecard_ids, \
        "Production readiness scorecard not found"

def test_production_readiness_scorecard_levels(port):
    r = port.get("/blueprints/service/scorecards/production_readiness")
    assert r.status_code == 200
    
    scorecard = r.json()["scorecard"]
    levels = [l["title"] for l in scorecard.get("levels", [])]
    
    # Levels should progress from basic to advanced
    assert "Bronze" in levels
    assert "Silver" in levels
    assert "Gold" in levels

def test_scorecard_bronze_rules_achievable(port):
    """Bronze rules should be achievable by new services."""
    r = port.get("/blueprints/service/scorecards/production_readiness")
    scorecard = r.json()["scorecard"]
    
    bronze_rules = [
        rule for rule in scorecard.get("rules", [])
        if rule.get("level") == "Bronze"
    ]
    
    # Bronze should only require basic things: has_ci, team, tier
    bronze_rule_props = [
        r["query"]["rules"][0].get("property") 
        for r in bronze_rules
        if r.get("query", {}).get("rules")
    ]
    
    # No Bronze rule should require deployment_frequency (hard to measure for new services)
    assert "deployment_frequency" not in bronze_rule_props, \
        "deployment_frequency should not be a Bronze requirement"

def test_entity_scorecard_evaluation(port):
    """Create a test entity and verify its scorecard is evaluated correctly."""
    # Create a minimal service entity (should achieve at least Bronze)
    entity = {
        "identifier": "test-scorecard-service",
        "title": "Test Scorecard Service",
        "properties": {
            "language": "python",
            "team": "platform",
            "tier": "tier-3",
            "has_ci": True,
            "has_monitoring": False,  # Missing — should not pass Silver
            "oncall_team": "platform",
        }
    }
    
    create_r = port.post("/blueprints/service/entities", json=entity)
    assert create_r.status_code in (200, 201), f"Failed to create entity: {create_r.text}"
    
    try:
        # Get entity with scorecard results
        r = port.get(
            "/blueprints/service/entities/test-scorecard-service",
            params={"include": ["scorecard_results"]}
        )
        assert r.status_code == 200
        
        scorecard_results = r.json()["entity"].get("scorecards", {})
        prod_readiness = scorecard_results.get("production_readiness", {})
        
        # Should achieve Bronze but not Silver (missing has_monitoring)
        assert prod_readiness.get("level") in ("Bronze", "Incomplete"), \
            f"Expected Bronze/Incomplete, got: {prod_readiness.get('level')}"
    
    finally:
        # Cleanup
        port.delete("/blueprints/service/entities/test-scorecard-service")

def test_create_service_action_exists(port):
    r = port.get("/blueprints/service/actions")
    assert r.status_code == 200
    
    actions = r.json()["actions"]
    action_ids = [a["identifier"] for a in actions]
    assert "create_service" in action_ids

def test_create_service_action_has_required_inputs(port):
    r = port.get("/blueprints/service/actions/create_service")
    assert r.status_code == 200
    
    action = r.json()["action"]
    input_props = action.get("userInputs", {}).get("properties", {})
    
    required_inputs = ["service_name", "team", "language", "tier"]
    for inp in required_inputs:
        assert inp in input_props, f"create_service action missing input: {inp}"

def test_create_service_action_validates_inputs(port):
    """Triggering the action with invalid inputs should fail validation, not create bad state."""
    payload = {
        "identifier": "test-run-invalid",
        "properties": {
            "service_name": "",  # Empty name — should be rejected
            "team": "platform",
            "language": "invalid-lang",  # Not in enum
            "tier": "tier-99"  # Invalid tier
        }
    }
    
    r = port.post("/blueprints/service/actions/create_service/runs", json=payload)
    # Should be 400 (validation error) or the workflow itself fails immediately
    # Port may accept the run and fail it, or reject at the API level
    if r.status_code == 200:
        run_id = r.json()["run"]["id"]
        # Poll for failure
        for _ in range(10):
            run_r = port.get(f"/actions/runs/{run_id}")
            status = run_r.json()["run"]["status"]
            if status == "FAILURE":
                return  # Expected
            elif status == "SUCCESS":
                pytest.fail("Action succeeded with invalid inputs — missing validation")
            import time; time.sleep(2)
        pytest.fail("Action run did not complete within timeout")
    else:
        assert r.status_code == 400, f"Expected 400 for invalid inputs, got {r.status_code}"

def test_action_trigger_creates_audit_log(port):
    """Every action trigger should create an audit log entry."""
    # Trigger a safe, read-only action (like "refresh service data")
    r = port.post("/blueprints/service/actions/refresh_service/runs", json={
        "identifier": "test-audit",
        "properties": {}
    })
    
    if r.status_code != 200:
        pytest.skip("refresh_service action not configured")
    
    run_id = r.json()["run"]["id"]
    
    # Check audit log
    audit_r = port.get("/audit-log", params={"actionIdentifier": "refresh_service"})
    assert audit_r.status_code == 200
    
    runs = audit_r.json()["runs"]
    run_ids = [run["id"] for run in runs]
    assert run_id in run_ids, "Action trigger not found in audit log"

def test_github_integration_syncing(port):
    """GitHub integration should have synced recently."""
    r = port.get("/integrations")
    assert r.status_code == 200
    
    integrations = r.json()["integrations"]
    github_integrations = [i for i in integrations if i["type"] == "github"]
    
    assert github_integrations, "No GitHub integration found"
    
    for integration in github_integrations:
        last_sync = integration.get("lastSyncTime")
        assert last_sync is not None, f"Integration {integration['id']} never synced"
        
        from datetime import datetime, timezone, timedelta
        sync_time = datetime.fromisoformat(last_sync.replace("Z", "+00:00"))
        age = datetime.now(timezone.utc) - sync_time
        
        assert age < timedelta(hours=2), \
            f"GitHub integration last synced {age} ago — staleness threshold exceeded"

def test_kubernetes_entities_present(port):
    """Kubernetes integration should have populated cluster entities."""
    r = port.get("/blueprints/cluster/entities")
    assert r.status_code == 200
    
    entities = r.json()["entities"]
    assert len(entities) > 0, "No cluster entities found — Kubernetes integration may not be syncing"

def test_service_entities_have_github_data(port):
    """Services should have GitHub-sourced properties populated."""
    r = port.get("/blueprints/service/entities", params={"limit": 10})
    entities = r.json()["entities"]
    
    services_with_repo = [
        e for e in entities 
        if e.get("properties", {}).get("github_repo")
    ]
    
    # At least some services should have a linked repo
    assert len(services_with_repo) > 0, \
        "No services have github_repo property — GitHub integration may not be working"

import yaml
import glob

def test_all_blueprints_have_icons():
    """Blueprints should have icons for good UX in the catalog."""
    for path in glob.glob("port/blueprints/*.yaml"):
        with open(path) as f:
            blueprint = yaml.safe_load(f)
        assert blueprint.get("icon"), f"{path}: blueprint missing icon"

def test_no_duplicate_blueprint_identifiers():
    identifiers = []
    for path in glob.glob("port/blueprints/*.yaml"):
        with open(path) as f:
            bp = yaml.safe_load(f)
        identifier = bp.get("identifier")
        assert identifier not in identifiers, \
            f"Duplicate blueprint identifier '{identifier}' in {path}"
        identifiers.append(identifier)

def test_relations_reference_existing_blueprints():
    """Relation targets must reference blueprints that exist."""
    blueprint_ids = set()
    blueprints_data = {}
    
    for path in glob.glob("port/blueprints/*.yaml"):
        with open(path) as f:
            bp = yaml.safe_load(f)
        blueprint_ids.add(bp["identifier"])
        blueprints_data[bp["identifier"]] = bp
    
    for bp_id, bp in blueprints_data.items():
        for rel_name, rel in bp.get("relations", {}).items():
            target = rel.get("target")
            assert target in blueprint_ids, \
                f"Blueprint '{bp_id}' relation '{rel_name}' references unknown blueprint '{target}'"

def test_scorecard_rules_reference_existing_properties():
    """Scorecard rules should only reference properties that exist on the blueprint."""
    for bp_path in glob.glob("port/blueprints/*.yaml"):
        with open(bp_path) as f:
            bp = yaml.safe_load(f)
        
        bp_id = bp["identifier"]
        defined_props = set(bp.get("schema", {}).get("properties", {}).keys())
        
        # Load scorecards for this blueprint
        for sc_path in glob.glob(f"port/scorecards/{bp_id}/*.yaml"):
            with open(sc_path) as f:
                scorecard = yaml.safe_load(f)
            
            for rule in scorecard.get("rules", []):
                for condition in rule.get("query", {}).get("rules", []):
                    prop = condition.get("property", "").lstrip("$.")
                    if prop and prop not in ("identifier", "title", "createdAt"):
                        assert prop in defined_props, \
                            f"Scorecard rule references unknown property '{prop}' on blueprint '{bp_id}'"

# .github/workflows/port-tests.yml
name: Port.io Tests

on:
  pull_request:
    paths:
      - 'port/**'

jobs:
  test:
    runs-on: ubuntu-latest

    env:
      PORT_CLIENT_ID: ${{ secrets.PORT_CLIENT_ID }}
      PORT_CLIENT_SECRET: ${{ secrets.PORT_CLIENT_SECRET }}

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install pytest httpx pyyaml jsonschema

      - name: Run config validation tests (no API needed)
        run: pytest tests/port/test_config_validation.py -v

      - name: Run API integration tests
        run: pytest tests/port/test_api.py -v --tb=short
        continue-on-error: true  # Don't fail on API flakiness in CI

      - name: Apply Port config changes
        if: github.ref == 'refs/heads/main'
        run: |
          pip install port-pulumi
          cd port && pulumi up --yes