def METHOD_NAME(self, graphql_context):
    instance = graphql_context.instance
    def _fetch_limits(key: str):
        results = execute_dagster_graphql(
            graphql_context,
            GET_CONCURRENCY_LIMITS_QUERY,
            variables={
                "concurrencyKey": key,
            },
        )
        assert results.data
        assert "instance" in results.data
        assert "concurrencyLimits" in results.data["instance"]
        return results.data["instance"]["concurrencyLimits"]
    def _set_limits(key: str, limit: int):
        execute_dagster_graphql(
            graphql_context,
            SET_CONCURRENCY_LIMITS_MUTATION,
            variables={
                "concurrencyKey": key,
                "limit": limit,
            },
        )
    # default limits are empty
    assert _fetch_limits("foo") == []
    # set a limit
    _set_limits("foo", 10)
    assert _fetch_limits("foo") == [
        {"concurrencyKey": "foo", "slotCount": 10, "activeRunIds": [], "activeSlotCount": 0}
    ]
    # claim a slot
    run_id = "fake_run_id"
    instance.event_log_storage.claim_concurrency_slot("foo", run_id, "fake_step_key")
    assert _fetch_limits("foo") == [
        {
            "concurrencyKey": "foo",
            "slotCount": 10,
            "activeRunIds": [run_id],
            "activeSlotCount": 1,
        }
    ]
    # set a new limit
    _set_limits("foo", 5)
    assert _fetch_limits("foo") == [
        {
            "concurrencyKey": "foo",
            "slotCount": 5,
            "activeRunIds": [run_id],
            "activeSlotCount": 1,
        }
    ]
    # free a slot
    instance.event_log_storage.free_concurrency_slots_for_run(run_id)
    assert _fetch_limits("foo") == [
        {"concurrencyKey": "foo", "slotCount": 5, "activeRunIds": [], "activeSlotCount": 0}
    ]