Add Policy schema, repository and API resource (#924)

Co-authored-by: Matt <mithmatt@github.com>

bootstrap/sql/mysql/v001__create_db_connection_info.sql

@@ -255,6 +255,22 @@ CREATE TABLE IF NOT EXISTS thread_entity (
     PRIMARY KEY (id)
 );
 
+--
+-- Policies related tables
+--
+CREATE TABLE IF NOT EXISTS policy_entity (
+    id VARCHAR(36) GENERATED ALWAYS AS (json ->> '$.id') STORED NOT NULL,
+    fullyQualifiedName VARCHAR(256) GENERATED ALWAYS AS (json ->> '$.fullyQualifiedName') NOT NULL,
+    json JSON NOT NULL,
+    updatedAt TIMESTAMP GENERATED ALWAYS AS (TIMESTAMP(STR_TO_DATE(json ->> '$.updatedAt', '%Y-%m-%dT%T.%fZ'))) NOT NULL,
+    updatedBy VARCHAR(256) GENERATED ALWAYS AS (json ->> '$.updatedBy') NOT NULL,
+    timestamp BIGINT,
+    PRIMARY KEY (id),
+    UNIQUE KEY unique_name(fullyQualifiedName),
+    INDEX (updatedBy),
+    INDEX (updatedAt)
+);
+
 --
 -- User, Team, and bots
 --

catalog-rest-service/src/main/java/org/openmetadata/catalog/Entity.java

@@ -39,6 +39,9 @@ public final class Entity {
   public static final String BOTS = "bots";
   public static final String LOCATION = "location";
 
+  // Policies
+  public static final String POLICY = "policy";
+
   // Team/user
   public static final String USER = "user";
   public static final String TEAM = "team";

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/CollectionDAO.java

@@ -34,6 +34,7 @@ import org.openmetadata.catalog.entity.data.Pipeline;
 import org.openmetadata.catalog.entity.data.Report;
 import org.openmetadata.catalog.entity.data.Table;
 import org.openmetadata.catalog.entity.data.Topic;
+import org.openmetadata.catalog.entity.policies.Policy;
 import org.openmetadata.catalog.entity.services.DashboardService;
 import org.openmetadata.catalog.entity.services.DatabaseService;
 import org.openmetadata.catalog.entity.services.MessagingService;
@@ -55,6 +56,7 @@ import org.openmetadata.catalog.jdbi3.MetricsRepository.MetricsEntityInterface;
 import org.openmetadata.catalog.jdbi3.ModelRepository.ModelEntityInterface;
 import org.openmetadata.catalog.jdbi3.PipelineRepository.PipelineEntityInterface;
 import org.openmetadata.catalog.jdbi3.PipelineServiceRepository.PipelineServiceEntityInterface;
+import org.openmetadata.catalog.jdbi3.PolicyRepository.PolicyEntityInterface;
 import org.openmetadata.catalog.jdbi3.ReportRepository.ReportEntityInterface;
 import org.openmetadata.catalog.jdbi3.StorageServiceRepository.StorageServiceEntityInterface;
 import org.openmetadata.catalog.jdbi3.TableRepository.TableEntityInterface;
@@ -124,6 +126,9 @@ public interface CollectionDAO {
   @CreateSqlObject
   BotsDAO botsDAO();
 
+  @CreateSqlObject
+  PolicyDAO policyDAO();
+
   @CreateSqlObject
   DatabaseServiceDAO dbServiceDAO();
 
@@ -524,6 +529,24 @@ public interface CollectionDAO {
     }
   }
 
+  interface PolicyDAO extends EntityDAO<Policy> {
+    @Override
+    default String getTableName() { return "policy_entity"; }
+
+    @Override
+    default Class<Policy> getEntityClass() {
+      return Policy.class;
+    }
+
+    @Override
+    default String getNameColumn() { return "fullyQualifiedName"; }
+
+    @Override
+    default EntityReference getEntityReference(Policy entity) {
+      return new PolicyEntityInterface(entity).getEntityReference();
+    }
+  }
+
   interface ReportDAO extends EntityDAO<Report> {
     @Override
     default String getTableName() { return "report_entity"; }

catalog-rest-service/src/main/java/org/openmetadata/catalog/jdbi3/PolicyRepository.java

@@ -0,0 +1,268 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements. See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License. You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.catalog.jdbi3;
+
+import lombok.extern.slf4j.Slf4j;
+import org.jdbi.v3.sqlobject.transaction.Transaction;
+import org.openmetadata.catalog.Entity;
+import org.openmetadata.catalog.entity.policies.Policy;
+import org.openmetadata.catalog.exception.EntityNotFoundException;
+import org.openmetadata.catalog.resources.policies.PolicyResource;
+import org.openmetadata.catalog.type.ChangeDescription;
+import org.openmetadata.catalog.type.EntityReference;
+import org.openmetadata.catalog.type.TagLabel;
+import org.openmetadata.catalog.util.EntityInterface;
+import org.openmetadata.catalog.util.EntityUtil;
+import org.openmetadata.catalog.util.EntityUtil.Fields;
+import org.openmetadata.catalog.util.JsonUtils;
+
+import java.io.IOException;
+import java.net.URI;
+import java.text.ParseException;
+import java.util.Date;
+import java.util.List;
+import java.util.UUID;
+
+import static org.openmetadata.catalog.exception.CatalogExceptionMessage.entityNotFound;
+
+@Slf4j
+public class PolicyRepository extends EntityRepository<Policy> {
+    private static final Fields POLICY_UPDATE_FIELDS = new Fields(PolicyResource.FIELD_LIST,
+            "displayName,description,owner,policyUrl,enabled");
+    private static final Fields POLICY_PATCH_FIELDS = new Fields(PolicyResource.FIELD_LIST,
+            "displayName,description,owner,policyUrl,enabled");
+    private final CollectionDAO dao;
+
+    public PolicyRepository(CollectionDAO dao) {
+        super(Policy.class, dao.policyDAO(), dao, POLICY_PATCH_FIELDS, POLICY_UPDATE_FIELDS);
+        this.dao = dao;
+    }
+
+    public static String getFQN(Policy policy) {
+        return (policy.getName());
+    }
+
+    @Transaction
+    public void delete(UUID id) {
+        if (dao.relationshipDAO().findToCount(id.toString(), Relationship.CONTAINS.ordinal(), Entity.POLICY) > 0) {
+            throw new IllegalArgumentException("Policy is not empty");
+        }
+        if (dao.policyDAO().delete(id) <= 0) {
+            throw EntityNotFoundException.byMessage(entityNotFound(Entity.POLICY, id));
+        }
+        dao.relationshipDAO().deleteAll(id.toString());
+    }
+
+    @Transaction
+    public EntityReference getOwnerReference(Policy policy) throws IOException {
+        return EntityUtil.populateOwner(dao.userDAO(), dao.teamDAO(), policy.getOwner());
+    }
+
+    @Override
+    public Policy setFields(Policy policy, Fields fields) throws IOException {
+        policy.setDisplayName(fields.contains("displayName") ? policy.getDisplayName() : null);
+        policy.setDescription(fields.contains("description") ? policy.getDescription() : null);
+        policy.setOwner(fields.contains("owner") ? getOwner(policy) : null);
+        policy.setPolicyUrl(fields.contains("policyUrl") ? policy.getPolicyUrl() : null);
+        policy.setEnabled(fields.contains("enabled") ? policy.getEnabled() : null);
+        return policy;
+    }
+
+    @Override
+    public void restorePatchAttributes(Policy original, Policy updated) throws IOException, ParseException {
+    }
+
+    @Override
+    public EntityInterface<Policy> getEntityInterface(Policy entity) {
+        return new PolicyEntityInterface(entity);
+    }
+
+
+    @Override
+    public void validate(Policy policy) throws IOException {
+        policy.setFullyQualifiedName(getFQN(policy));
+
+        // Check if owner is valid and set the relationship
+        EntityUtil.populateOwner(dao.userDAO(), dao.teamDAO(), policy.getOwner());
+    }
+
+    @Override
+    public void store(Policy policy, boolean update) throws IOException {
+        // Relationships and fields such as href are derived and not stored as part of json
+        EntityReference owner = policy.getOwner();
+        URI href = policy.getHref();
+
+        // Don't store owner and href as JSON. Build it on the fly based on relationships
+        policy.withOwner(null).withHref(null);
+
+        if (update) {
+            dao.policyDAO().update(policy.getId(), JsonUtils.pojoToJson(policy));
+        } else {
+            dao.policyDAO().insert(policy);
+        }
+
+        // Restore the relationships
+        policy.withOwner(owner).withHref(href);
+    }
+
+    @Override
+    public void storeRelationships(Policy policy) throws IOException {
+        // Add policy owner relationship
+        setOwner(policy, policy.getOwner());
+    }
+
+    @Override
+    public EntityUpdater getUpdater(Policy original, Policy updated, boolean patchOperation) throws IOException {
+        return new PolicyUpdater(original, updated, patchOperation);
+    }
+
+    private EntityReference getOwner(Policy policy) throws IOException {
+        return policy == null ? null : EntityUtil.populateOwner(policy.getId(), dao.relationshipDAO(), dao.userDAO(),
+                dao.teamDAO());
+    }
+
+    public void setOwner(Policy policy, EntityReference owner) {
+        EntityUtil.setOwner(dao.relationshipDAO(), policy.getId(), Entity.POLICY, owner);
+        policy.setOwner(owner);
+    }
+
+    static class PolicyEntityInterface implements EntityInterface<Policy> {
+        private final Policy entity;
+
+        PolicyEntityInterface(Policy entity) {
+            this.entity = entity;
+        }
+
+        @Override
+        public UUID getId() {
+            return entity.getId();
+        }
+
+        @Override
+        public String getDescription() {
+            return entity.getDescription();
+        }
+
+        @Override
+        public String getDisplayName() {
+            return entity.getDisplayName();
+        }
+
+        @Override
+        public EntityReference getOwner() {
+            return entity.getOwner();
+        }
+
+        @Override
+        public String getFullyQualifiedName() {
+            return entity.getFullyQualifiedName();
+        }
+
+        @Override
+        public List<TagLabel> getTags() {
+            // Policy does not have tags.
+            return null;
+        }
+
+        @Override
+        public Double getVersion() {
+            return entity.getVersion();
+        }
+
+        @Override
+        public String getUpdatedBy() {
+            return entity.getUpdatedBy();
+        }
+
+        @Override
+        public Date getUpdatedAt() {
+            return entity.getUpdatedAt();
+        }
+
+        @Override
+        public URI getHref() {
+            return entity.getHref();
+        }
+
+        @Override
+        public List<EntityReference> getFollowers() {
+            // Policy does not have followers.
+            return null;
+        }
+
+        @Override
+        public EntityReference getEntityReference() {
+            return new EntityReference().withId(getId()).withName(getFullyQualifiedName())
+                    .withDescription(getDescription()).withDisplayName(getDisplayName()).withType(Entity.POLICY);
+        }
+
+        @Override
+        public Policy getEntity() {
+            return entity;
+        }
+
+        @Override
+        public void setId(UUID id) {
+            entity.setId(id);
+        }
+
+        @Override
+        public void setDescription(String description) {
+            entity.setDescription(description);
+        }
+
+        @Override
+        public void setDisplayName(String displayName) {
+            entity.setDisplayName(displayName);
+        }
+
+        @Override
+        public void setUpdateDetails(String updatedBy, Date updatedAt) {
+            entity.setUpdatedBy(updatedBy);
+            entity.setUpdatedAt(updatedAt);
+        }
+
+        @Override
+        public void setChangeDescription(Double newVersion, ChangeDescription changeDescription) {
+            entity.setVersion(newVersion);
+            entity.setChangeDescription(changeDescription);
+        }
+
+        @Override
+        public void setTags(List<TagLabel> tags) {
+            // Policy does not have tags.
+        }
+    }
+
+    /**
+     * Handles entity updated from PUT and POST operation.
+     */
+    public class PolicyUpdater extends EntityUpdater {
+        public PolicyUpdater(Policy original, Policy updated, boolean patchOperation) {
+            super(original, updated, patchOperation);
+        }
+
+        @Override
+        public void entitySpecificUpdate() throws IOException {
+            updatePolicyUrl(original.getEntity(), updated.getEntity());
+        }
+
+        private void updatePolicyUrl(Policy original, Policy updated) {
+            recordChange("policyUrl", original.getPolicyUrl(), updated.getPolicyUrl());
+        }
+    }
+}

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/policies/PolicyResource.java

@@ -0,0 +1,296 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements. See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License. You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.catalog.resources.policies;
+
+import com.google.inject.Inject;
+import io.swagger.annotations.Api;
+import io.swagger.v3.oas.annotations.ExternalDocumentation;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.media.Content;
+import io.swagger.v3.oas.annotations.media.ExampleObject;
+import io.swagger.v3.oas.annotations.media.Schema;
+import io.swagger.v3.oas.annotations.parameters.RequestBody;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import org.openmetadata.catalog.api.policies.CreatePolicy;
+import org.openmetadata.catalog.entity.policies.Policy;
+import org.openmetadata.catalog.jdbi3.CollectionDAO;
+import org.openmetadata.catalog.jdbi3.PolicyRepository;
+import org.openmetadata.catalog.resources.Collection;
+import org.openmetadata.catalog.security.CatalogAuthorizer;
+import org.openmetadata.catalog.security.SecurityUtil;
+import org.openmetadata.catalog.type.EntityReference;
+import org.openmetadata.catalog.util.EntityUtil;
+import org.openmetadata.catalog.util.EntityUtil.Fields;
+import org.openmetadata.catalog.util.RestUtil;
+import org.openmetadata.catalog.util.RestUtil.PutResponse;
+import org.openmetadata.catalog.util.ResultList;
+
+import javax.json.JsonPatch;
+import javax.validation.Valid;
+import javax.validation.constraints.Max;
+import javax.validation.constraints.Min;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.DefaultValue;
+import javax.ws.rs.GET;
+import javax.ws.rs.PATCH;
+import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.SecurityContext;
+import javax.ws.rs.core.UriInfo;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.security.GeneralSecurityException;
+import java.text.ParseException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+
+@Path("/v1/policies")
+@Api(value = "Policies collection", tags = "Policies collection")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+@Collection(name = "policies")
+public class PolicyResource {
+    public static final String POLICY_COLLECTION_PATH = "v1/policies/";
+    private final PolicyRepository dao;
+    private final CatalogAuthorizer authorizer;
+
+    public static void addHref(UriInfo uriInfo, EntityReference ref) {
+        ref.withHref(RestUtil.getHref(uriInfo, POLICY_COLLECTION_PATH, ref.getId()));
+    }
+
+    public static List<Policy> addHref(UriInfo uriInfo, List<Policy> policies) {
+        Optional.ofNullable(policies).orElse(Collections.emptyList()).forEach(i -> addHref(uriInfo, i));
+        return policies;
+    }
+
+    public static Policy addHref(UriInfo uriInfo, Policy policy) {
+        policy.setHref(RestUtil.getHref(uriInfo, POLICY_COLLECTION_PATH, policy.getId()));
+        EntityUtil.addHref(uriInfo, policy.getOwner());
+        return policy;
+    }
+
+    @Inject
+    public PolicyResource(CollectionDAO dao, CatalogAuthorizer authorizer) {
+        Objects.requireNonNull(dao, "PolicyRepository must not be null");
+        this.dao = new PolicyRepository(dao);
+        this.authorizer = authorizer;
+    }
+
+    public static class PolicyList extends ResultList<Policy> {
+        @SuppressWarnings("unused")
+        PolicyList() {
+            // Empty constructor needed for deserialization
+        }
+
+        public PolicyList(List<Policy> data, String beforeCursor, String afterCursor, int total)
+                throws GeneralSecurityException, UnsupportedEncodingException {
+            super(data, beforeCursor, afterCursor, total);
+        }
+    }
+
+    static final String FIELDS = "displayName,description,owner,policyUrl,enabled";
+    public static final List<String> FIELD_LIST = Arrays.asList(FIELDS.replaceAll(" ", "")
+            .split(","));
+
+    @GET
+    @Valid
+    @Operation(summary = "List Policies", tags = "policies",
+            description = "Get a list of policies. Use `fields` parameter to get only necessary fields. " +
+                    "Use cursor-based pagination to limit the number " +
+                    "entries in the list using `limit` and `before` or `after` query params.",
+            responses = {
+                    @ApiResponse(responseCode = "200", description = "List of policies",
+                            content = @Content(mediaType = "application/json",
+                                    schema = @Schema(implementation = PolicyList.class)))
+            })
+    public ResultList<Policy> list(@Context UriInfo uriInfo,
+                                   @Context SecurityContext securityContext,
+                                   @Parameter(description = "Fields requested in the returned resource",
+                                           schema = @Schema(type = "string", example = FIELDS))
+                                   @QueryParam("fields") String fieldsParam,
+                                   @Parameter(description = "Limit the number policies returned. (1 to 1000000, " +
+                                           "default = 10)")
+                                   @DefaultValue("10")
+                                   @Min(1)
+                                   @Max(1000000)
+                                   @QueryParam("limit") int limitParam,
+                                   @Parameter(description = "Returns list of policies before this cursor",
+                                           schema = @Schema(type = "string"))
+                                   @QueryParam("before") String before,
+                                   @Parameter(description = "Returns list of policies after this cursor",
+                                           schema = @Schema(type = "string"))
+                                   @QueryParam("after") String after
+    ) throws IOException, GeneralSecurityException, ParseException {
+        RestUtil.validateCursors(before, after);
+        Fields fields = new Fields(FIELD_LIST, fieldsParam);
+
+        ResultList<Policy> policies;
+        if (before != null) { // Reverse paging
+            policies = dao.listBefore(fields, null, limitParam, before); // Ask for one extra entry
+        } else { // Forward paging or first page
+            policies = dao.listAfter(fields, null, limitParam, after);
+        }
+        addHref(uriInfo, policies.getData());
+        return policies;
+    }
+
+    @GET
+    @Path("/{id}")
+    @Operation(summary = "Get a policy", tags = "policies",
+            description = "Get a policy by `id`.",
+            responses = {
+                    @ApiResponse(responseCode = "200", description = "The policy",
+                            content = @Content(mediaType = "application/json",
+                                    schema = @Schema(implementation = Policy.class))),
+                    @ApiResponse(responseCode = "404", description = "Policy for instance {id} is not found")
+            })
+    public Policy get(@Context UriInfo uriInfo,
+                      @Context SecurityContext securityContext,
+                      @PathParam("id") String id,
+                      @Parameter(description = "Fields requested in the returned resource",
+                              schema = @Schema(type = "string", example = FIELDS))
+                      @QueryParam("fields") String fieldsParam) throws IOException, ParseException {
+        Fields fields = new Fields(FIELD_LIST, fieldsParam);
+        return addHref(uriInfo, dao.get(id, fields));
+    }
+
+    @GET
+    @Path("/name/{fqn}")
+    @Operation(summary = "Get a policy by name", tags = "policies",
+            description = "Get a policy by fully qualified name.",
+            responses = {
+                    @ApiResponse(responseCode = "200", description = "The policy",
+                            content = @Content(mediaType = "application/json",
+                                    schema = @Schema(implementation = Policy.class))),
+                    @ApiResponse(responseCode = "404", description = "Policy for instance {id} is not found")
+            })
+    public Policy getByName(@Context UriInfo uriInfo, @PathParam("fqn") String fqn,
+                            @Context SecurityContext securityContext,
+                            @Parameter(description = "Fields requested in the returned resource",
+                                    schema = @Schema(type = "string", example = FIELDS))
+                            @QueryParam("fields") String fieldsParam) throws IOException, ParseException {
+        Fields fields = new Fields(FIELD_LIST, fieldsParam);
+        Policy policy = dao.getByName(fqn, fields);
+        return addHref(uriInfo, policy);
+    }
+
+
+    @POST
+    @Operation(summary = "Create a policy", tags = "policies",
+            description = "Create a new policy.",
+            responses = {
+                    @ApiResponse(responseCode = "200", description = "The policy",
+                            content = @Content(mediaType = "application/json",
+                                    schema = @Schema(implementation = CreatePolicy.class))),
+                    @ApiResponse(responseCode = "400", description = "Bad request")
+            })
+    public Response create(@Context UriInfo uriInfo, @Context SecurityContext securityContext,
+                           @Valid CreatePolicy create) throws IOException, ParseException {
+        SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
+        Policy policy = new Policy()
+                .withId(UUID.randomUUID())
+                .withName(create.getName())
+                .withDisplayName(create.getDisplayName())
+                .withDescription(create.getDescription())
+                .withOwner(create.getOwner())
+                .withPolicyUrl(create.getPolicyUrl())
+                .withPolicyType(create.getPolicyType())
+                .withUpdatedBy(securityContext.getUserPrincipal().getName())
+                .withUpdatedAt(new Date());
+        policy = addHref(uriInfo, dao.create(policy));
+        return Response.created(policy.getHref()).entity(policy).build();
+    }
+
+    @PATCH
+    @Path("/{id}")
+    @Operation(summary = "Update a policy", tags = "policies",
+            description = "Update an existing policy using JsonPatch.",
+            externalDocs = @ExternalDocumentation(description = "JsonPatch RFC",
+                    url = "https://tools.ietf.org/html/rfc6902"))
+    @Consumes(MediaType.APPLICATION_JSON_PATCH_JSON)
+    public Policy updateDescription(@Context UriInfo uriInfo,
+                                    @Context SecurityContext securityContext,
+                                    @PathParam("id") String id,
+                                    @RequestBody(description = "JsonPatch with array of operations",
+                                            content = @Content(mediaType = MediaType.APPLICATION_JSON_PATCH_JSON,
+                                                    examples = {@ExampleObject("[" +
+                                                            "{op:remove, path:/a}," +
+                                                            "{op:add, path: /b, value: val}" +
+                                                            "]")}))
+                                            JsonPatch patch) throws IOException, ParseException {
+        Fields fields = new Fields(FIELD_LIST, FIELDS);
+        Policy policy = dao.get(id, fields);
+        SecurityUtil.checkAdminRoleOrPermissions(authorizer, securityContext, dao.getOwnerReference(policy));
+        policy = dao.patch(UUID.fromString(id), securityContext.getUserPrincipal().getName(), patch);
+        return addHref(uriInfo, policy);
+    }
+
+    @PUT
+    @Operation(summary = "Create or update a policy", tags = "policies",
+            description = "Create a new policy, if it does not exist or update an existing policy.",
+            responses = {
+                    @ApiResponse(responseCode = "200", description = "The policy",
+                            content = @Content(mediaType = "application/json",
+                                    schema = @Schema(implementation = Policy.class))),
+                    @ApiResponse(responseCode = "400", description = "Bad request")
+            })
+    public Response createOrUpdate(@Context UriInfo uriInfo,
+                                   @Context SecurityContext securityContext,
+                                   @Valid CreatePolicy create) throws IOException, ParseException {
+        Policy policy = new Policy()
+                .withId(UUID.randomUUID())
+                .withName(create.getName())
+                .withDisplayName(create.getDisplayName())
+                .withDescription(create.getDescription())
+                .withOwner(create.getOwner())
+                .withPolicyUrl(create.getPolicyUrl())
+                .withPolicyType(create.getPolicyType())
+                .withUpdatedBy(securityContext.getUserPrincipal().getName())
+                .withUpdatedAt(new Date());
+
+        PutResponse<Policy> response = dao.createOrUpdate(policy);
+        policy = addHref(uriInfo, response.getEntity());
+        return Response.status(response.getStatus()).entity(policy).build();
+    }
+
+    @DELETE
+    @Path("/{id}")
+    @Operation(summary = "Delete a Policy", tags = "policy",
+            description = "Delete a policy by `id`.",
+            responses = {
+                    @ApiResponse(responseCode = "200", description = "OK"),
+                    @ApiResponse(responseCode = "404", description = "policy for instance {id} is not found")
+            })
+    public Response delete(@Context UriInfo uriInfo, @PathParam("id") String id) {
+        dao.delete(UUID.fromString(id));
+        return Response.ok().build();
+    }
+}

catalog-rest-service/src/main/java/org/openmetadata/catalog/util/EntityUtil.java

@@ -25,7 +25,6 @@ import org.openmetadata.catalog.exception.CatalogExceptionMessage;
 import org.openmetadata.catalog.exception.EntityNotFoundException;
 import org.openmetadata.catalog.jdbi3.CollectionDAO;
 import org.openmetadata.catalog.jdbi3.CollectionDAO.EntityRelationshipDAO;
-import org.openmetadata.catalog.jdbi3.CollectionDAO.LocationDAO;
 import org.openmetadata.catalog.jdbi3.CollectionDAO.TagDAO;
 import org.openmetadata.catalog.jdbi3.CollectionDAO.TeamDAO;
 import org.openmetadata.catalog.jdbi3.CollectionDAO.UsageDAO;
@@ -39,6 +38,7 @@ import org.openmetadata.catalog.resources.feeds.MessageParser.EntityLink;
 import org.openmetadata.catalog.resources.models.ModelResource;
 import org.openmetadata.catalog.resources.pipelines.PipelineResource;
 import org.openmetadata.catalog.resources.locations.LocationResource;
+import org.openmetadata.catalog.resources.policies.PolicyResource;
 import org.openmetadata.catalog.resources.services.dashboard.DashboardServiceResource;
 import org.openmetadata.catalog.resources.services.database.DatabaseServiceResource;
 import org.openmetadata.catalog.resources.services.messaging.MessagingServiceResource;
@@ -162,6 +162,8 @@ public final class EntityUtil {
       DashboardResource.addHref(uriInfo, ref);
     } else if (entity.equalsIgnoreCase(Entity.MODEL)) {
       ModelResource.addHref(uriInfo, ref);
+    } else if (entity.equalsIgnoreCase(Entity.POLICY)) {
+      PolicyResource.addHref(uriInfo, ref);
     } else if (entity.equalsIgnoreCase(Entity.PIPELINE)) {
       PipelineResource.addHref(uriInfo, ref);
     } else if (entity.equalsIgnoreCase(Entity.LOCATION)) {
@@ -288,6 +290,8 @@ public final class EntityUtil {
       return dao.modelDAO().findEntityReferenceById(id);
     } else if (entity.equalsIgnoreCase(Entity.LOCATION)) {
       return dao.locationDAO().findEntityReferenceById(id);
+    } else if (entity.equalsIgnoreCase(Entity.POLICY)) {
+      return dao.policyDAO().findEntityReferenceById(id);
     }
     throw EntityNotFoundException.byMessage(CatalogExceptionMessage.entityTypeNotFound(entity));
   }
@@ -312,6 +316,8 @@ public final class EntityUtil {
       return dao.pipelineDAO().findEntityReferenceByName(fqn);
     } else if (entity.equalsIgnoreCase(Entity.MODEL)) {
       return dao.modelDAO().findEntityReferenceByName(fqn);
+    } else if (entity.equalsIgnoreCase(Entity.POLICY)) {
+      return dao.policyDAO().findEntityReferenceByName(fqn);
     } else if (entity.equalsIgnoreCase(Entity.USER)) {
       return dao.userDAO().findEntityReferenceByName(fqn);
     } else if (entity.equalsIgnoreCase(Entity.TEAM)) {

catalog-rest-service/src/main/resources/json/schema/api/policies/createPolicy.json

@@ -0,0 +1,41 @@
+{
+  "$id": "https://open-metadata.org/schema/api/data/createPolicy.json",
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "Create Policy Entity Request",
+  "description": "Create Policy Entity Request",
+  "type": "object",
+  "properties": {
+    "name": {
+      "description": "Name that identifies this Policy.",
+      "type": "string",
+      "minLength": 1,
+      "maxLength": 64
+    },
+    "displayName": {
+      "description": "Title for this Policy.",
+      "type": "string"
+    },
+    "description": {
+      "description": "A short description of the Policy, comprehensible to regular users.",
+      "type": "string"
+    },
+    "owner": {
+      "description": "Owner of this Policy.",
+      "$ref": "../../type/entityReference.json"
+    },
+    "policyUrl": {
+      "description": "Link to a well documented definition of this Policy.",
+      "type": "string",
+      "format": "uri"
+    },
+    "policyType": {
+      "$ref": "../../entity/policies/policy.json#/definitions/policyType"
+    }
+  },
+  "required": [
+    "name",
+    "description",
+    "owner",
+    "policyType"
+  ]
+}

catalog-rest-service/src/main/resources/json/schema/entity/policies/policy.json

@@ -0,0 +1,97 @@
+{
+  "$id": "https://open-metadata.org/schema/entity/data/policy.json",
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "Policy",
+  "description": "This schema defines the Policy entity. A Policy defines lifecycle or access control that needs to be applied across different Data Entities.",
+  "type": "object",
+  "definitions": {
+    "policyType": {
+      "javaType": "org.openmetadata.catalog.type.PolicyType",
+      "description": "This schema defines the type used for describing different types of policies.",
+      "type": "string",
+      "enum": [
+        "AccessControl",
+        "Lifecycle"
+      ],
+      "javaEnums": [
+        {
+          "name": "AccessControl"
+        },
+        {
+          "name": "Lifecycle"
+        }
+      ]
+    }
+  },
+  "properties": {
+    "id": {
+      "description": "Unique identifier that identifies this Policy.",
+      "$ref": "../../type/basic.json#/definitions/uuid"
+    },
+    "name": {
+      "description": "Name that identifies this Policy.",
+      "type": "string",
+      "minLength": 1,
+      "maxLength": 64
+    },
+    "fullyQualifiedName": {
+      "description": "Name that uniquely identifies a Policy.",
+      "type": "string",
+      "minLength": 1,
+      "maxLength": 128
+    },
+    "displayName": {
+      "description": "Title for this Policy.",
+      "type": "string"
+    },
+    "description": {
+      "description": "A short description of the Policy, comprehensible to regular users.",
+      "type": "string"
+    },
+    "owner": {
+      "description": "Owner of this Policy.",
+      "$ref": "../../type/entityReference.json",
+      "default": null
+    },
+    "policyUrl": {
+      "description": "Link to a well documented definition of this Policy.",
+      "type": "string",
+      "format": "uri",
+      "default": null
+    },
+    "href": {
+      "description": "Link to the resource corresponding to this entity.",
+      "$ref": "../../type/basic.json#/definitions/href"
+    },
+    "policyType": {
+      "$ref": "#/definitions/policyType"
+    },
+    "enabled": {
+      "description": "Is the policy enabled.",
+      "type": "boolean",
+      "default": true
+    },
+    "version" : {
+      "description": "Metadata version of the Policy.",
+      "$ref": "../../type/entityHistory.json#/definitions/entityVersion"
+    },
+    "updatedAt" : {
+      "description": "Last update time corresponding to the new version of the Policy.",
+      "$ref": "../../type/basic.json#/definitions/dateTime"
+    },
+    "updatedBy" : {
+      "description": "User who made the update.",
+      "type": "string"
+    },
+    "changeDescription": {
+      "description" : "Change that led to this version of the entity.",
+      "$ref": "../../type/entityHistory.json#/definitions/changeDescription"
+    }
+  },
+  "required": [
+    "id",
+    "name",
+    "owner",
+    "policyType"
+  ]
+}

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/policies/PolicyResourceTest.java

@@ -0,0 +1,566 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements. See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License. You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.catalog.resources.policies;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.client.HttpResponseException;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.openmetadata.catalog.CatalogApplicationTest;
+import org.openmetadata.catalog.Entity;
+import org.openmetadata.catalog.api.policies.CreatePolicy;
+import org.openmetadata.catalog.entity.policies.Policy;
+import org.openmetadata.catalog.entity.teams.Team;
+import org.openmetadata.catalog.entity.teams.User;
+import org.openmetadata.catalog.exception.CatalogExceptionMessage;
+import org.openmetadata.catalog.resources.policies.PolicyResource.PolicyList;
+import org.openmetadata.catalog.resources.teams.TeamResourceTest;
+import org.openmetadata.catalog.resources.teams.UserResourceTest;
+import org.openmetadata.catalog.type.EntityReference;
+import org.openmetadata.catalog.type.PolicyType;
+import org.openmetadata.catalog.util.JsonUtils;
+import org.openmetadata.catalog.util.TestUtils;
+import org.openmetadata.catalog.util.TestUtils.UpdateType;
+import org.openmetadata.common.utils.JsonSchemaUtil;
+
+import javax.json.JsonPatch;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.Response.Status;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+import java.util.UUID;
+
+import static javax.ws.rs.core.Response.Status.BAD_REQUEST;
+import static javax.ws.rs.core.Response.Status.CONFLICT;
+import static javax.ws.rs.core.Response.Status.CREATED;
+import static javax.ws.rs.core.Response.Status.FORBIDDEN;
+import static javax.ws.rs.core.Response.Status.NOT_FOUND;
+import static javax.ws.rs.core.Response.Status.OK;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.openmetadata.catalog.exception.CatalogExceptionMessage.entityNotFound;
+import static org.openmetadata.catalog.util.TestUtils.UpdateType.MINOR_UPDATE;
+import static org.openmetadata.catalog.util.TestUtils.UpdateType.NO_CHANGE;
+import static org.openmetadata.catalog.util.TestUtils.adminAuthHeaders;
+import static org.openmetadata.catalog.util.TestUtils.assertEntityPagination;
+import static org.openmetadata.catalog.util.TestUtils.assertResponse;
+import static org.openmetadata.catalog.util.TestUtils.authHeaders;
+
+@Slf4j
+public class PolicyResourceTest extends CatalogApplicationTest {
+    public static User USER1;
+    public static EntityReference USER_OWNER1;
+    public static Team TEAM1;
+    public static EntityReference TEAM_OWNER1;
+
+
+    @BeforeAll
+    public static void setup(TestInfo test) throws HttpResponseException {
+        USER1 = UserResourceTest.createUser(UserResourceTest.create(test),
+                authHeaders("test@open-metadata.org"));
+        USER_OWNER1 = new EntityReference().withId(USER1.getId()).withType("user");
+        TEAM1 = TeamResourceTest.createTeam(TeamResourceTest.create(test), adminAuthHeaders());
+        TEAM_OWNER1 = new EntityReference().withId(TEAM1.getId()).withType("team");
+    }
+
+    @Test
+    public void post_policyWithLongName_400_badRequest(TestInfo test) {
+        CreatePolicy create = create(test).withName(TestUtils.LONG_ENTITY_NAME);
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () -> createPolicy(create,
+                adminAuthHeaders()));
+        assertResponse(exception, BAD_REQUEST, "[name size must be between 1 and 64]");
+    }
+
+    @Test
+    public void post_PolicyWithoutName_400_badRequest(TestInfo test) {
+        CreatePolicy create = create(test).withName("");
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () ->
+                createPolicy(create, adminAuthHeaders()));
+        assertResponse(exception, BAD_REQUEST, "[name size must be between 1 and 64]");
+    }
+
+    @Test
+    public void post_PolicyWithoutOwner_400_badRequest(TestInfo test) {
+        CreatePolicy create = create(test).withOwner(null);
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () ->
+                createPolicy(create, adminAuthHeaders()));
+        assertResponse(exception, BAD_REQUEST, "[owner must not be null]");
+    }
+
+    @Test
+    public void post_PolicyWithoutPolicyType_400_badRequest(TestInfo test) {
+        CreatePolicy create = create(test).withPolicyType(null);
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () ->
+                createPolicy(create, adminAuthHeaders()));
+        assertResponse(exception, BAD_REQUEST, "[policyType must not be null]");
+    }
+
+    @Test
+    public void post_PolicyAlreadyExists_409_conflict(TestInfo test) throws HttpResponseException {
+        CreatePolicy create = create(test);
+        createPolicy(create, adminAuthHeaders());
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () ->
+                createPolicy(create, adminAuthHeaders()));
+        assertResponse(exception, CONFLICT, CatalogExceptionMessage.ENTITY_ALREADY_EXISTS);
+    }
+
+    @Test
+    public void post_validPolicies_as_admin_200_OK(TestInfo test) throws HttpResponseException {
+        // Create valid policy
+        CreatePolicy create = create(test);
+        createAndCheckPolicy(create, adminAuthHeaders());
+        create.withName(getPolicyName(test, 1)).withDescription("description");
+        createAndCheckPolicy(create, adminAuthHeaders());
+    }
+
+    @Test
+    public void post_PolicyWithUserOwner_200_ok(TestInfo test) throws HttpResponseException {
+        CreatePolicy create = create(test).withOwner(USER_OWNER1);
+        createAndCheckPolicy(create, adminAuthHeaders());
+    }
+
+    @Test
+    public void post_PolicyWithTeamOwner_200_ok(TestInfo test) throws HttpResponseException {
+        CreatePolicy create = create(test).withOwner(TEAM_OWNER1);
+        createAndCheckPolicy(create, adminAuthHeaders());
+    }
+
+    @Test
+    public void post_Policy_as_non_admin_401(TestInfo test) {
+        CreatePolicy create = create(test);
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () ->
+                createPolicy(create, authHeaders("test@open-metadata.org")));
+        assertResponse(exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test'} is not admin");
+    }
+
+    @Test
+    public void post_PolicyWithInvalidOwnerType_4xx(TestInfo test) {
+        EntityReference owner = new EntityReference().withId(TEAM1.getId()); /* No owner type is set */
+
+        CreatePolicy create = create(test).withOwner(owner);
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () ->
+                createPolicy(create, adminAuthHeaders()));
+        TestUtils.assertResponseContains(exception, BAD_REQUEST, "type must not be null");
+    }
+
+    @Test
+    public void post_PolicyWithNonExistentOwner_4xx(TestInfo test) {
+        EntityReference owner = new EntityReference().withId(TestUtils.NON_EXISTENT_ENTITY).withType("user");
+        CreatePolicy create = create(test).withOwner(owner);
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () ->
+                createPolicy(create, adminAuthHeaders()));
+        assertResponse(exception, NOT_FOUND, entityNotFound("User", TestUtils.NON_EXISTENT_ENTITY));
+    }
+
+    @Test
+    public void get_PolicyListWithInvalidLimitOffset_4xx() {
+        // Limit must be >= 1 and <= 1000,000
+        HttpResponseException exception = assertThrows(HttpResponseException.class, ()
+                -> listPolicies(null, -1, null, null, adminAuthHeaders()));
+        assertResponse(exception, BAD_REQUEST, "[query param limit must be greater than or equal to 1]");
+
+        exception = assertThrows(HttpResponseException.class, ()
+                -> listPolicies(null, 0, null, null, adminAuthHeaders()));
+        assertResponse(exception, BAD_REQUEST, "[query param limit must be greater than or equal to 1]");
+
+        exception = assertThrows(HttpResponseException.class, ()
+                -> listPolicies(null, 1000001, null, null, adminAuthHeaders()));
+        assertResponse(exception, BAD_REQUEST, "[query param limit must be less than or equal to 1000000]");
+    }
+
+    @Test
+    public void get_PolicyListWithInvalidPaginationCursors_4xx() {
+        // Passing both before and after cursors is invalid
+        HttpResponseException exception = assertThrows(HttpResponseException.class, ()
+                -> listPolicies(null, 1, "", "", adminAuthHeaders()));
+        assertResponse(exception, BAD_REQUEST, "Only one of before or after query parameter allowed");
+    }
+
+    @Test
+    public void get_PolicyListWithValidLimitOffset_4xx(TestInfo test) throws HttpResponseException {
+        // Create a large number of Policies
+        int maxPolicies = 40;
+        for (int i = 0; i < maxPolicies; i++) {
+            createPolicy(create(test, i), adminAuthHeaders());
+        }
+
+        // List all Policies
+        PolicyList allPolicies = listPolicies(null, 1000000, null,
+                null, adminAuthHeaders());
+        int totalRecords = allPolicies.getData().size();
+        printPolicies(allPolicies);
+
+        // List limit number Policies at a time at various offsets and ensure right results are returned
+        for (int limit = 1; limit < maxPolicies; limit++) {
+            String after = null;
+            String before;
+            int pageCount = 0;
+            int indexInAllPolicies = 0;
+            PolicyList forwardPage;
+            PolicyList backwardPage;
+            do { // For each limit (or page size) - forward scroll till the end
+                log.info("Limit {} forward scrollCount {} afterCursor {}", limit, pageCount, after);
+                forwardPage = listPolicies(null, limit, null, after, adminAuthHeaders());
+                printPolicies(forwardPage);
+                after = forwardPage.getPaging().getAfter();
+                before = forwardPage.getPaging().getBefore();
+                assertEntityPagination(allPolicies.getData(), forwardPage, limit, indexInAllPolicies);
+
+                if (pageCount == 0) {  // CASE 0 - First page is being returned. There is no before cursor
+                    assertNull(before);
+                } else {
+                    // Make sure scrolling back based on before cursor returns the correct result
+                    backwardPage = listPolicies(null, limit, before, null, adminAuthHeaders());
+                    assertEntityPagination(allPolicies.getData(), backwardPage, limit, (indexInAllPolicies - limit));
+                }
+
+                indexInAllPolicies += forwardPage.getData().size();
+                pageCount++;
+            } while (after != null);
+
+            // We have now reached the last page - test backward scroll till the beginning
+            pageCount = 0;
+            indexInAllPolicies = totalRecords - limit - forwardPage.getData().size();
+            do {
+                log.info("Limit {} backward scrollCount {} beforeCursor {}", limit, pageCount, before);
+                forwardPage = listPolicies(null, limit, before, null, adminAuthHeaders());
+                printPolicies(forwardPage);
+                before = forwardPage.getPaging().getBefore();
+                assertEntityPagination(allPolicies.getData(), forwardPage, limit, indexInAllPolicies);
+                pageCount++;
+                indexInAllPolicies -= forwardPage.getData().size();
+            } while (before != null);
+        }
+    }
+
+    private void printPolicies(PolicyList list) {
+        list.getData().forEach(Policy -> log.info("DB {}", Policy.getFullyQualifiedName()));
+        log.info("before {} after {} ", list.getPaging().getBefore(), list.getPaging().getAfter());
+    }
+
+    @Test
+    public void put_PolicyUpdateWithNoChange_200(TestInfo test) throws HttpResponseException {
+        // Create a Policy with POST
+        CreatePolicy request = create(test)
+                .withDescription("des")
+                .withPolicyType(PolicyType.Lifecycle);
+        Policy policy = createAndCheckPolicy(request, adminAuthHeaders());
+
+        // Update Policy two times successfully with PUT requests
+        policy = updateAndCheckPolicy(policy, request, OK, adminAuthHeaders(), NO_CHANGE);
+        updateAndCheckPolicy(policy, request, OK, adminAuthHeaders(), NO_CHANGE);
+    }
+
+    @Test
+    public void put_PolicyCreate_200(TestInfo test) throws HttpResponseException {
+        // Create a new Policy with PUT
+        CreatePolicy request = create(test).withOwner(USER_OWNER1);
+        updateAndCheckPolicy(null, request.withName(test.getDisplayName()).withDescription("description"),
+                CREATED, adminAuthHeaders(), NO_CHANGE);
+    }
+
+    @Test
+    public void put_PolicyEmptyDescriptionUpdate_200(TestInfo test) throws HttpResponseException {
+        // Create table with empty description
+        CreatePolicy request = create(test).withDescription("");
+        Policy policy = createAndCheckPolicy(request, adminAuthHeaders());
+
+        // Update empty description with a new description
+        updateAndCheckPolicy(policy, request.withDescription("newDescription"), OK, adminAuthHeaders(), MINOR_UPDATE);
+    }
+
+    @Test
+    public void put_PolicyNonEmptyDescriptionUpdate_200(TestInfo test) throws HttpResponseException {
+        CreatePolicy request = create(test).withDescription("description");
+        createAndCheckPolicy(request, adminAuthHeaders());
+
+        // Updating description is ignored when backend already has description
+        Policy db = updatePolicy(request.withDescription("newDescription"), OK, adminAuthHeaders());
+        assertEquals("description", db.getDescription());
+    }
+
+    @Test
+    public void put_PolicyUpdateOwner_200(TestInfo test) throws HttpResponseException {
+        CreatePolicy request = create(test).withDescription("");
+        Policy policy = createAndCheckPolicy(request, adminAuthHeaders());
+
+        // Change ownership from USER_OWNER1 to TEAM_OWNER1
+        policy = updateAndCheckPolicy(policy, request.withOwner(TEAM_OWNER1), OK, adminAuthHeaders(), MINOR_UPDATE);
+        assertNotNull(policy.getOwner());
+    }
+
+    @Test
+    public void get_nonExistentPolicy_404_notFound() {
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () ->
+                getPolicy(TestUtils.NON_EXISTENT_ENTITY, adminAuthHeaders()));
+        assertResponse(exception, NOT_FOUND, entityNotFound(Entity.POLICY, TestUtils.NON_EXISTENT_ENTITY));
+    }
+
+    @Test
+    public void get_PolicyWithDifferentFields_200_OK(TestInfo test) throws HttpResponseException {
+        CreatePolicy create = create(test);
+        Policy policy = createAndCheckPolicy(create, adminAuthHeaders());
+        validateGetWithDifferentFields(policy, false);
+    }
+
+    @Test
+    public void get_PolicyByNameWithDifferentFields_200_OK(TestInfo test) throws HttpResponseException {
+        CreatePolicy create = create(test);
+        Policy policy = createAndCheckPolicy(create, adminAuthHeaders());
+        validateGetWithDifferentFields(policy, true);
+    }
+
+    @Test
+    public void patch_PolicyAttributes_200_ok(TestInfo test) throws HttpResponseException, JsonProcessingException {
+        Policy policy = createPolicy(create(test), adminAuthHeaders());
+        assertNull(policy.getPolicyUrl());
+        assertTrue(policy.getEnabled());
+
+        URI uri = null;
+        try {
+            uri = new URI("http://www.example.com/policy1");
+        } catch (URISyntaxException e) {
+            fail("could not construct URI for test");
+        }
+
+        policy = getPolicy(policy.getId(), "displayName,description,owner,policyUrl,enabled", adminAuthHeaders());
+
+        // Add policyUrl which was previously null and set enabled to false
+        patchPolicyAttributesAndCheck(policy, uri, false, adminAuthHeaders(), MINOR_UPDATE);
+    }
+
+    @Test
+    public void delete_emptyPolicy_200_ok(TestInfo test) throws HttpResponseException {
+        Policy policy = createPolicy(create(test), adminAuthHeaders());
+        deletePolicy(policy.getId(), adminAuthHeaders());
+    }
+
+    @Test
+    public void delete_nonEmptyPolicy_4xx() {
+        // TODO
+    }
+
+    @Test
+    public void delete_nonExistentPolicy_404() {
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () ->
+                deletePolicy(TestUtils.NON_EXISTENT_ENTITY, adminAuthHeaders()));
+        assertResponse(exception, NOT_FOUND, entityNotFound(Entity.POLICY, TestUtils.NON_EXISTENT_ENTITY));
+    }
+
+    public static Policy createAndCheckPolicy(CreatePolicy create, Map<String, String> authHeaders) throws
+            HttpResponseException {
+        String updatedBy = TestUtils.getPrincipal(authHeaders);
+        Policy policy = createPolicy(create, authHeaders);
+        validatePolicy(policy, create.getDisplayName(), create.getDescription(), create.getOwner(), updatedBy);
+        return getAndValidate(policy.getId(), create, authHeaders, updatedBy);
+    }
+
+    public static Policy updateAndCheckPolicy(Policy before, CreatePolicy create, Status status,
+                                              Map<String, String> authHeaders, UpdateType updateType)
+            throws HttpResponseException {
+        String updatedBy = TestUtils.getPrincipal(authHeaders);
+        Policy updatedPolicy = updatePolicy(create, status, authHeaders);
+        validatePolicy(updatedPolicy, create.getDescription(), create.getOwner(), updatedBy);
+        if (before == null) {
+            assertEquals(0.1, updatedPolicy.getVersion()); // First version created
+        } else {
+            TestUtils.validateUpdate(before.getVersion(), updatedPolicy.getVersion(), updateType);
+        }
+
+        return getAndValidate(updatedPolicy.getId(), create, authHeaders, updatedBy);
+    }
+
+    // Make sure in GET operations the returned Policy has all the required information passed during creation
+    public static Policy getAndValidate(UUID policyID, CreatePolicy create, Map<String, String> authHeaders,
+                                        String expectedUpdatedBy) throws HttpResponseException {
+        // GET the newly created Policy by ID and validate
+        Policy policy = getPolicy(policyID, "displayName,description,owner,policyUrl,enabled", authHeaders);
+        validatePolicy(policy, create.getDescription(), create.getOwner(), expectedUpdatedBy);
+
+        // GET the newly created Policy by name and validate
+        String fqn = policy.getFullyQualifiedName();
+        policy = getPolicyByName(fqn, "displayName,description,owner,policyUrl,enabled", authHeaders);
+        return validatePolicy(policy, create.getDescription(), create.getOwner(), expectedUpdatedBy);
+    }
+
+    public static Policy updatePolicy(CreatePolicy create, Status status, Map<String, String> authHeaders)
+            throws HttpResponseException {
+        return TestUtils.put(getResource("policies"), create, Policy.class, status, authHeaders);
+    }
+
+    public static Policy createPolicy(CreatePolicy create, Map<String, String> authHeaders) throws
+            HttpResponseException {
+        return TestUtils.post(getResource("policies"), create, Policy.class, authHeaders);
+    }
+
+    /**
+     * Validate returned fields GET .../policies/{id}?fields="..." or GET .../policies/name/{fqn}?fields="..."
+     */
+    private void validateGetWithDifferentFields(Policy policy, boolean byName) throws HttpResponseException {
+        // .../policies?fields=owner
+        String fields = "owner";
+        policy = byName ? getPolicyByName(policy.getFullyQualifiedName(), fields, adminAuthHeaders()) :
+                getPolicy(policy.getId(), fields, adminAuthHeaders());
+        assertNotNull(policy.getOwner());
+
+        // .../policies?fields=owner,displayName
+        fields = "owner,displayName";
+        policy = byName ? getPolicyByName(policy.getFullyQualifiedName(), fields, adminAuthHeaders()) :
+                getPolicy(policy.getId(), fields, adminAuthHeaders());
+        assertNotNull(policy.getOwner());
+
+        // .../policies?fields=owner,displayName,policyUrl
+        fields = "owner,displayName,policyUrl";
+        policy = byName ? getPolicyByName(policy.getFullyQualifiedName(), fields, adminAuthHeaders()) :
+                getPolicy(policy.getId(), fields, adminAuthHeaders());
+        assertNotNull(policy.getOwner());
+    }
+
+    private static Policy validatePolicy(Policy policy, String expectedDisplayName, String expectedDescription,
+                                         EntityReference expectedOwner, String expectedUpdatedBy) {
+        Policy newPolicy = validatePolicy(policy, expectedDescription, expectedOwner, expectedUpdatedBy);
+        assertEquals(expectedDisplayName, newPolicy.getDisplayName());
+        return newPolicy;
+    }
+
+    private static Policy validatePolicy(Policy policy, URI expectedPolicyUrl, Boolean expectedEnabled,
+                                         String expectedUpdatedBy) {
+        assertNotNull(policy.getId());
+        assertNotNull(policy.getHref());
+        assertEquals(expectedPolicyUrl, policy.getPolicyUrl());
+        assertEquals(expectedEnabled, policy.getEnabled());
+        assertEquals(expectedUpdatedBy, policy.getUpdatedBy());
+        return policy;
+    }
+
+    private static Policy validatePolicy(Policy policy, String expectedDescription, EntityReference expectedOwner,
+                                         String expectedUpdatedBy) {
+        assertNotNull(policy.getId());
+        assertNotNull(policy.getHref());
+        assertEquals(expectedDescription, policy.getDescription());
+        assertEquals(expectedUpdatedBy, policy.getUpdatedBy());
+
+        // Validate owner
+        if (expectedOwner != null) {
+            TestUtils.validateEntityReference(policy.getOwner());
+            assertEquals(expectedOwner.getId(), policy.getOwner().getId());
+            assertEquals(expectedOwner.getType(), policy.getOwner().getType());
+            assertNotNull(policy.getOwner().getHref());
+        }
+
+        return policy;
+    }
+
+    private Policy patchPolicyAttributesAndCheck(Policy before, URI policyUrl, Boolean enabled,
+                                                 Map<String, String> authHeaders, UpdateType updateType)
+            throws JsonProcessingException, HttpResponseException {
+        String updatedBy = TestUtils.getPrincipal(authHeaders);
+        String policyJson = JsonUtils.pojoToJson(before);
+
+        // Update the attributes
+        before.setPolicyUrl(policyUrl);
+        before.setEnabled(enabled);
+
+        // Validate information returned in patch response has the updates
+        Policy updatedPolicy = patchPolicy(policyJson, before, authHeaders);
+        validatePolicy(updatedPolicy, policyUrl, enabled, updatedBy);
+        TestUtils.validateUpdate(before.getVersion(), updatedPolicy.getVersion(), updateType);
+
+        // GET the table and Validate information returned
+        Policy getPolicy = getPolicy(before.getId(), "policyUrl,enabled", authHeaders);
+        validatePolicy(getPolicy, policyUrl, enabled, updatedBy);
+        return updatedPolicy;
+    }
+
+    private Policy patchPolicy(UUID policyId, String originalJson, Policy updatedPolicy,
+                               Map<String, String> authHeaders)
+            throws JsonProcessingException, HttpResponseException {
+        String updatedPolicyJson = JsonUtils.pojoToJson(updatedPolicy);
+        JsonPatch patch = JsonSchemaUtil.getJsonPatch(originalJson, updatedPolicyJson);
+        return TestUtils.patch(getResource("policies/" + policyId), patch, Policy.class, authHeaders);
+    }
+
+    private Policy patchPolicy(String originalJson, Policy updatedPolicy, Map<String, String> authHeaders)
+            throws JsonProcessingException, HttpResponseException {
+        return patchPolicy(updatedPolicy.getId(), originalJson, updatedPolicy, authHeaders);
+    }
+
+    public static void getPolicy(UUID id, Map<String, String> authHeaders) throws HttpResponseException {
+        getPolicy(id, null, authHeaders);
+    }
+
+    public static Policy getPolicy(UUID id, String fields, Map<String, String> authHeaders)
+            throws HttpResponseException {
+        WebTarget target = getResource("policies/" + id);
+        target = fields != null ? target.queryParam("fields", fields) : target;
+        return TestUtils.get(target, Policy.class, authHeaders);
+    }
+
+    public static Policy getPolicyByName(String fqn, String fields, Map<String, String> authHeaders)
+            throws HttpResponseException {
+        WebTarget target = getResource("policies/name/" + fqn);
+        target = fields != null ? target.queryParam("fields", fields) : target;
+        return TestUtils.get(target, Policy.class, authHeaders);
+    }
+
+    public static PolicyList listPolicies(String fields, Integer limitParam,
+                                          String before, String after, Map<String, String> authHeaders)
+            throws HttpResponseException {
+        WebTarget target = getResource("policies");
+        target = fields != null ? target.queryParam("fields", fields) : target;
+        target = limitParam != null ? target.queryParam("limit", limitParam) : target;
+        target = before != null ? target.queryParam("before", before) : target;
+        target = after != null ? target.queryParam("after", after) : target;
+        return TestUtils.get(target, PolicyList.class, authHeaders);
+    }
+
+    private void deletePolicy(UUID id, Map<String, String> authHeaders) throws HttpResponseException {
+        TestUtils.delete(getResource("policies/" + id), authHeaders);
+
+        // Ensure deleted Policy does not exist
+        HttpResponseException exception = assertThrows(HttpResponseException.class, () -> getPolicy(id, authHeaders));
+        assertResponse(exception, NOT_FOUND, entityNotFound(Entity.POLICY, id));
+    }
+
+    public static String getPolicyName(TestInfo test) {
+        return String.format("policy_%s", test.getDisplayName());
+    }
+
+    public static String getPolicyName(TestInfo test, int index) {
+        return String.format("policy%d_%s", index, test.getDisplayName());
+    }
+
+    public static CreatePolicy create(TestInfo test) {
+        return new CreatePolicy()
+                .withName(getPolicyName(test))
+                .withDescription("description")
+                .withPolicyType(PolicyType.AccessControl)
+                .withOwner(USER_OWNER1);
+    }
+
+    public static CreatePolicy create(TestInfo test, int index) {
+        return create(test).withName(getPolicyName(test, index));
+    }
+
+}