Testing custom extensions

# This test case always passes
def test_always_pass(context):
    return testing.ok("I always pass")

# This test case always fails
def test_always_fail(context):
    return testing.fail("I always fail")

# This function is ignored by the test runner because it doesn't start with 'test_'
def ignored():
    pass

$ docker run \
    -v $(pwd):/tests:ro \
    CERBOS_DISTRIBUTION_REPO/synapse/synapse:latest \
    test /tests

test_suite = struct(
   synapse_config = testing.load_synapse_config("synapse.yaml"),
)

test_suite = struct(
   synapse_config = {
        "pdp": {
            "inProcess": {
                "storage": {
                    "driver": "disk",
                    "disk": {
                        "directory": "${SYNAPSE_ROOT}/policies"
                    }
                }
            }
        }
   }
)

test_suite = struct(
   name = "Tests for extension foo",
   synapse_config = testing.load_synapse_config("synapse.yaml")
)

test_suite = struct(
   name = "Tests for extension foo",
   synapse_config = testing.load_synapse_config("synapse.yaml"),
   skip = True
)

def always_fail(context):
    return testing.fail("I always fail")

test_suite = struct(
   name = "Tests for extension foo",
   synapse_config = testing.load_synapse_config("synapse.yaml"),
   test_cases = {
      "always_pass": lambda context: testing.ok("I always pss"), (1)
      "always_fail": always_fail (2)
   }
)

test_suite = struct(
   name = "Illustrate how to exercise different extension types",
   synapse_config = testing.load_synapse_config("synapse.yaml"),
)

# Send a Cerbos CheckResources request to exercise configured proxy extensions
def test_check_resources(context):
    request = struct(
        request_id = "test",
        principal = struct(id = "john", roles = ["employee"]),
        resources = [struct(
            actions = ["view"],
            resource = struct(
                kind = "invoice",
                id = "XX125",
                attr = { "owner": "john", "department": "IT", "geography": "GB"  }
            )
        )]
    )

    have = context.check_resources(request)
    return testing.assert(have.results[0].actions["view"] == "EFFECT_ALLOW")

# Send a Cerbos PlanResources request to exercise configured proxy extensions
def test_plan_resources(context):
    request = struct(
        request_id = "test",
        principal = struct(id = "john", roles = ["employee"]),
        actions = ["view"],
        resource = struct(kind = "invoice")
    )

    have = context.plan_resources(request)
    return testing.assert(set(have.actions) == set(["view"]) and have.filter.kind == "KIND_CONDITIONAL")

# Send an AuthZEN access evaluation request to exercise configured proxy extensions
def test_access_evaluation(context):
    request = struct(
        subject = struct(
            type = "user",
            id = "john",
            properties = {"cerbos.roles": ["employee"]}
        ),
        resource = struct(
            type = "invoice",
            id = "XX125",
            properties =  { "owner": "john", "department": "IT", "geography": "GB"  }
        ),
        action = struct(name = "view")
    )

    have = context.access_evaluation(request)
    return testing.assert(have.decision)

# Send an AuthZEN access evaluation batch request to exercise configured proxy extensions
def test_access_evaluation_batch(context):
    request = struct(
        subject = struct(
            type = "user",
            id = "john",
            properties = {"cerbos.roles": ["employee"]}
        ),
        resource = struct(
            type = "invoice",
            id = "XX125",
            properties =  { "owner": "john", "department": "IT", "geography": "GB"  }
        ),
        evaluations = [struct(action = struct(name = "view"))]
    )

    have = context.access_evaluation_batch(request)
    return testing.assert(have.evaluations[0].decision)

# Send an extenal authz request to the Envoy extension
def test_envoy_check(context):
    request = struct(
        attributes = struct(
            source = struct(
                principal = "spiffe://cerbos.dev/cerbie"
            ),
            request = struct(
                http = struct(
                   id = "foo",
                   method = "GET",
                   path = "/foo?op=check_response"
                ),
            ),
        ),
    )

    have = context.envoy_check(request)
    return testing.assert(have.status.code == 0)

# Send an HTTP request to a configured route extension
def test_route_extension(context):
   have = context.http_get("/ext/foo", {"op": "http_response"})
   return testing.assert(have.status_code == 200 and have.body() == "Welcome")

testData: (1)
  req_john: (2)
    checkResourcesRequest: (3)
      requestId: "john"
      principal:
        id: "john"
        roles: ["employee"]
      resources:
        - resource:
            id: "XX125"
            kind: "invoice"
            attr:
              owner: "john"
              department: "Music"
              geography: "GB"
          actions: ["view"]

  resp_john:
    checkResourcesResponse:
      requestId: "john"
      results:
        - resource:
            id: "XX125"
            kind: "invoice"
          actions:
            view: EFFECT_ALLOW

testdata = testing.load_testdata("check-resources-test-data.yaml") (1)

def test_external_test_data(context):
    request = testdata["req_john"] (2)
    have = context.check_resources(request)
    want = testdata["resp_john"]
    return testing.assert(have.results[0].actions["view"] == want.results[0].actions["view"])

def build_test_case(request, expected):
    def test_case(context):
        actual = context.check_resources(request)
        for (i, expected_result) in enumerate(expected.results):
            for (action, expected_effect) in dict(expected_result.actions).items():
                if actual.results[i].actions[action] != expected_effect:
                    return testing.fail("Action {} of result {} should be {} but is {}".format(action, i, expected_effect, actual.results[i].actions[action]))


    return test_case

def build_test_case_list():
    testdata = testing.load_testdata("check-resources-test-data.yaml")
    test_cases = {}
    for (name, data) in testdata.items():
        if name.startswith("req_"):
            test_name = name.removeprefix("req_")
            expected_resp = testdata.get("resp_" + test_name)
            if expected_resp:
                test_cases[test_name] = build_test_case(data, expected_resp)

    return test_cases


test_suite = struct(
    name = "CheckResources tests",
    synapse_config = testing.load_synapse_config("synapse.yaml"),
    test_cases = build_test_case_list() (1)
)

load("re", "re")

def test_regex(context):
    return testing.assert(re.match("f.*", "foo"))

# Discover all the test suites under /tests and run them
$ docker run \
    -v $(pwd):/tests:ro \
    CERBOS_DISTRIBUTION_REPO/synapse/synapse:latest \
    test /tests

# Run only extensionx_test.star and extensiony_test.star suites
$ docker run \
    -v $(pwd):/tests:ro \
    CERBOS_DISTRIBUTION_REPO/synapse/synapse:latest \
    test /tests/extensionx/extensionx_test.star /tests/extensiony/extensiony_test.star

# Run only test cases whose names begin with "enrich"
$ docker run \
    -v $(pwd):/tests:ro \
    CERBOS_DISTRIBUTION_REPO/synapse/synapse:latest \
    test --match='^enrich' /tests

$ docker run \
    -v $(pwd):/tests:ro \
    CERBOS_DISTRIBUTION_REPO/synapse/synapse:latest \
    test --verbose /tests

$ docker run \
    -v $(pwd):/tests:ro \
    CERBOS_DISTRIBUTION_REPO/synapse/synapse:latest \
    starlark repl /tests/fail_test.star