diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4c905c8bce..d205a4d4a0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -21,6 +21,7 @@ The types of changes are:
 - Added new column for Action Type in privacy request event logs [#5546](https://github.com/ethyca/fides/pull/5546)
 - Added `fides_consent_override` option in FidesJS SDK [#5541](https://github.com/ethyca/fides/pull/5541)
 - Added new `script` ConsentMethod in FidesJS SDK for tracking automated consent [#5541](https://github.com/ethyca/fides/pull/5541)
+- Added a new page under system integrations to run standalone dataset tests (Fidesplus) [#5549](https://github.com/ethyca/fides/pull/5549)
 
 ### Changed
 - Adding hashes to system tab URLs [#5535](https://github.com/ethyca/fides/pull/5535)
diff --git a/clients/admin-ui/src/app/store.ts b/clients/admin-ui/src/app/store.ts
index 4945966089..af87c45e3d 100644
--- a/clients/admin-ui/src/app/store.ts
+++ b/clients/admin-ui/src/app/store.ts
@@ -41,6 +41,7 @@ import { propertySlice } from "~/features/properties";
 import { systemSlice } from "~/features/system";
 import { dictSuggestionsSlice } from "~/features/system/dictionary-form/dict-suggestion.slice";
 import { taxonomySlice } from "~/features/taxonomy";
+import { datasetTestSlice } from "~/features/test-datasets";
 import { userManagementSlice } from "~/features/user-management";
 
 /**
@@ -79,6 +80,7 @@ const reducer = {
   [dataSubjectsSlice.name]: dataSubjectsSlice.reducer,
   [dataUseSlice.name]: dataUseSlice.reducer,
   [datasetSlice.name]: datasetSlice.reducer,
+  [datasetTestSlice.name]: datasetTestSlice.reducer,
   [datastoreConnectionSlice.name]: datastoreConnectionSlice.reducer,
   [discoveryDetectionSlice.name]: discoveryDetectionSlice.reducer,
   [featuresSlice.name]: featuresSlice.reducer,
diff --git a/clients/admin-ui/src/features/datastore-connections/add-connection/forms/ConnectorParametersForm.tsx b/clients/admin-ui/src/features/datastore-connections/add-connection/forms/ConnectorParametersForm.tsx
index b17fc8b9e6..e1050d9cd3 100644
--- a/clients/admin-ui/src/features/datastore-connections/add-connection/forms/ConnectorParametersForm.tsx
+++ b/clients/admin-ui/src/features/datastore-connections/add-connection/forms/ConnectorParametersForm.tsx
@@ -63,7 +63,7 @@ const ConnectorParametersForm = ({
   isSubmitting = false,
   onSaveClick,
   onTestConnectionClick,
-  testButtonLabel = "Test connection",
+  testButtonLabel = "Test integration",
 }: ConnectorParametersFormProps) => {
   const mounted = useRef(false);
   const { handleError } = useAPIHelper();
diff --git a/clients/admin-ui/src/features/datastore-connections/datastore-connection.slice.ts b/clients/admin-ui/src/features/datastore-connections/datastore-connection.slice.ts
index dc8e67f509..58ae66a793 100644
--- a/clients/admin-ui/src/features/datastore-connections/datastore-connection.slice.ts
+++ b/clients/admin-ui/src/features/datastore-connections/datastore-connection.slice.ts
@@ -380,6 +380,40 @@ export const datastoreConnectionApi = baseApi.injectEndpoints({
       }),
       invalidatesTags: () => ["Datastore Connection"],
     }),
+    testDatastoreConnectionDatasets: build.mutation<
+      { privacy_request_id: string },
+      {
+        connection_key: string;
+        dataset_key: string;
+        input_data: Record<string, any>;
+      }
+    >({
+      query: (params) => ({
+        url: `${CONNECTION_ROUTE}/${params.connection_key}/dataset/${params.dataset_key}/test`,
+        method: "POST",
+        body: params.input_data,
+      }),
+    }),
+    getDatasetInputs: build.query<
+      any,
+      { connectionKey: string; datasetKey: string }
+    >({
+      query: ({ connectionKey, datasetKey }) => ({
+        url: `${CONNECTION_ROUTE}/${connectionKey}/dataset/${datasetKey}/inputs`,
+        method: "GET",
+      }),
+      providesTags: () => ["Datastore Connection"],
+    }),
+    getDatasetReachability: build.query<
+      { reachable: boolean; details: string },
+      { connectionKey: string; datasetKey: string }
+    >({
+      query: ({ connectionKey, datasetKey }) => ({
+        url: `${CONNECTION_ROUTE}/${connectionKey}/dataset/${datasetKey}/reachability`,
+        method: "GET",
+      }),
+      providesTags: () => ["Datastore Connection"],
+    }),
   }),
 });
 
@@ -402,6 +436,9 @@ export const {
   usePatchDatastoreConnectionsMutation,
   useUpdateDatastoreConnectionSecretsMutation,
   usePatchDatastoreConnectionSecretsMutation,
+  useTestDatastoreConnectionDatasetsMutation,
+  useGetDatasetInputsQuery,
+  useGetDatasetReachabilityQuery,
 } = datastoreConnectionApi;
 
 /**
diff --git a/clients/admin-ui/src/features/datastore-connections/system_portal_config/forms/ConnectorParameters.tsx b/clients/admin-ui/src/features/datastore-connections/system_portal_config/forms/ConnectorParameters.tsx
index 86565794d1..7697b755e2 100644
--- a/clients/admin-ui/src/features/datastore-connections/system_portal_config/forms/ConnectorParameters.tsx
+++ b/clients/admin-ui/src/features/datastore-connections/system_portal_config/forms/ConnectorParameters.tsx
@@ -12,6 +12,7 @@ import {
   DatastoreConnectionSecretsResponse,
 } from "datastore-connections/types";
 import { Box, Flex, Spacer, useToast, UseToastOptions } from "fidesui";
+import router from "next/router";
 import { useMemo, useState } from "react";
 
 import { useAppDispatch, useAppSelector } from "~/app/hooks";
@@ -391,6 +392,10 @@ export const ConnectorParameters = ({
     },
   );
 
+  const handleTestDatasetsClick = () => {
+    router.push(`/systems/configure/${systemFidesKey}/test-datasets`);
+  };
+
   const {
     isSubmitting,
     isAuthorizing,
@@ -450,6 +455,7 @@ export const ConnectorParameters = ({
         isAuthorizing={isAuthorizing}
         onSaveClick={handleSubmit}
         onTestConnectionClick={handleTestConnectionClick}
+        onTestDatasetsClick={handleTestDatasetsClick}
         onAuthorizeConnectionClick={handleAuthorization}
         connectionOption={connectionOption}
         connectionConfig={connectionConfig}
diff --git a/clients/admin-ui/src/features/datastore-connections/system_portal_config/forms/ConnectorParametersForm.tsx b/clients/admin-ui/src/features/datastore-connections/system_portal_config/forms/ConnectorParametersForm.tsx
index a25d5fc6b7..dae5bce4b1 100644
--- a/clients/admin-ui/src/features/datastore-connections/system_portal_config/forms/ConnectorParametersForm.tsx
+++ b/clients/admin-ui/src/features/datastore-connections/system_portal_config/forms/ConnectorParametersForm.tsx
@@ -62,6 +62,10 @@ type ConnectorParametersFormProps = {
    * Parent callback when Test Connection is clicked
    */
   onTestConnectionClick: (value: TestConnectionResponse) => void;
+  /**
+   * Parent callback when Test Dataset is clicked
+   */
+  onTestDatasetsClick: () => void;
   /**
    * Text for the test button. Defaults to "Test connection"
    */
@@ -86,6 +90,7 @@ export const ConnectorParametersForm = ({
   isAuthorizing = false,
   onSaveClick,
   onTestConnectionClick,
+  onTestDatasetsClick,
   onAuthorizeConnectionClick,
   testButtonLabel = "Test integration",
   connectionOption,
@@ -517,6 +522,11 @@ export const ConnectorParametersForm = ({
                     {testButtonLabel}
                   </Button>
                 ) : null}
+                {isPlusEnabled && !_.isEmpty(initialDatasets) && (
+                  <Button onClick={() => onTestDatasetsClick()}>
+                    Test datasets
+                  </Button>
+                )}
                 {connectionOption.authorization_required && !authorized ? (
                   <Button
                     loading={isAuthorizing}
diff --git a/clients/admin-ui/src/features/privacy-requests/privacy-requests.slice.ts b/clients/admin-ui/src/features/privacy-requests/privacy-requests.slice.ts
index 856a4e5599..b69cf28f20 100644
--- a/clients/admin-ui/src/features/privacy-requests/privacy-requests.slice.ts
+++ b/clients/admin-ui/src/features/privacy-requests/privacy-requests.slice.ts
@@ -540,6 +540,21 @@ export const privacyRequestApi = baseApi.injectEndpoints({
         url: `privacy-request/${privacy_request_id}/access-results`,
       }),
     }),
+    getFilteredResults: build.query<
+      {
+        privacy_request_id: string;
+        status: PrivacyRequestStatus;
+        results: {
+          [key: string]: Array<Record<string, any>>;
+        };
+      },
+      { privacy_request_id: string }
+    >({
+      query: ({ privacy_request_id }) => ({
+        method: "GET",
+        url: `privacy-request/${privacy_request_id}/filtered-results`,
+      }),
+    }),
   }),
 });
 
@@ -570,6 +585,7 @@ export const {
   useCreateMessagingConfigurationSecretsMutation,
   useCreateTestConnectionMessageMutation,
   useGetPrivacyRequestAccessResultsQuery,
+  useGetFilteredResultsQuery,
 } = privacyRequestApi;
 
 export type CORSOrigins = Pick<SecurityApplicationConfig, "cors_origins">;
diff --git a/clients/admin-ui/src/features/test-datasets/DatasetEditorSection.tsx b/clients/admin-ui/src/features/test-datasets/DatasetEditorSection.tsx
new file mode 100644
index 0000000000..4dd828a49b
--- /dev/null
+++ b/clients/admin-ui/src/features/test-datasets/DatasetEditorSection.tsx
@@ -0,0 +1,255 @@
+import { FetchBaseQueryError } from "@reduxjs/toolkit/dist/query/fetchBaseQuery";
+import {
+  AntButton as Button,
+  AntSelect as Select,
+  ErrorWarningIcon,
+  GreenCheckCircleIcon,
+  Heading,
+  HStack,
+  Stack,
+  Text,
+  useToast,
+  VStack,
+} from "fidesui";
+import yaml, { YAMLException } from "js-yaml";
+import { useEffect, useMemo, useState } from "react";
+
+import { useAppDispatch, useAppSelector } from "~/app/hooks";
+import ClipboardButton from "~/features/common/ClipboardButton";
+import { getErrorMessage, isErrorResult } from "~/features/common/helpers";
+import { errorToastParams, successToastParams } from "~/features/common/toast";
+import { Editor } from "~/features/common/yaml/helpers";
+import { useUpdateDatasetMutation } from "~/features/dataset";
+import {
+  useGetConnectionConfigDatasetConfigsQuery,
+  useGetDatasetReachabilityQuery,
+} from "~/features/datastore-connections";
+import { Dataset } from "~/types/api";
+
+import { selectCurrentDataset, setCurrentDataset } from "./dataset-test.slice";
+import { removeNulls } from "./helpers";
+
+interface EditorSectionProps {
+  connectionKey: string;
+}
+
+const EditorSection = ({ connectionKey }: EditorSectionProps) => {
+  const toast = useToast();
+  const dispatch = useAppDispatch();
+  const [updateDataset] = useUpdateDatasetMutation();
+
+  const [editorContent, setEditorContent] = useState<string>("");
+  const currentDataset = useAppSelector(selectCurrentDataset);
+
+  const {
+    data: datasetConfigs,
+    isLoading: isDatasetConfigsLoading,
+    refetch: refetchDatasets,
+  } = useGetConnectionConfigDatasetConfigsQuery(connectionKey, {
+    skip: !connectionKey,
+  });
+
+  const { data: reachability, refetch: refetchReachability } =
+    useGetDatasetReachabilityQuery(
+      {
+        connectionKey,
+        datasetKey: currentDataset?.fides_key || "",
+      },
+      {
+        skip: !connectionKey || !currentDataset?.fides_key,
+      },
+    );
+
+  const datasetOptions = useMemo(
+    () =>
+      (datasetConfigs?.items || []).map((item) => ({
+        value: item.fides_key,
+        label: item.fides_key,
+      })),
+    [datasetConfigs?.items],
+  );
+
+  useEffect(() => {
+    if (datasetConfigs?.items.length) {
+      if (
+        !currentDataset ||
+        !datasetConfigs.items.find(
+          (item) => item.fides_key === currentDataset.fides_key,
+        )
+      ) {
+        const initialDataset = datasetConfigs.items[0];
+        dispatch(setCurrentDataset(initialDataset));
+      }
+    }
+  }, [datasetConfigs, currentDataset, dispatch]);
+
+  useEffect(() => {
+    if (currentDataset?.ctl_dataset) {
+      setEditorContent(yaml.dump(removeNulls(currentDataset?.ctl_dataset)));
+    }
+  }, [currentDataset]);
+
+  useEffect(() => {
+    if (connectionKey && currentDataset?.fides_key) {
+      refetchReachability();
+    }
+  }, [connectionKey, currentDataset, refetchReachability]);
+
+  const handleDatasetChange = async (value: string) => {
+    const selectedConfig = datasetConfigs?.items.find(
+      (item) => item.fides_key === value,
+    );
+    if (selectedConfig) {
+      dispatch(setCurrentDataset(selectedConfig));
+    }
+  };
+
+  const handleSave = async () => {
+    if (!currentDataset) {
+      return;
+    }
+
+    // Parse YAML first
+    let datasetValues: Dataset;
+    try {
+      datasetValues = yaml.load(editorContent) as Dataset;
+    } catch (yamlError) {
+      toast(
+        errorToastParams(
+          `YAML Parsing Error: ${
+            yamlError instanceof YAMLException
+              ? `${yamlError.reason} ${yamlError.mark ? `at line ${yamlError.mark.line}` : ""}`
+              : "Invalid YAML format"
+          }`,
+        ),
+      );
+      return;
+    }
+
+    // Then handle the API update
+    const result = await updateDataset(datasetValues);
+
+    if (isErrorResult(result)) {
+      toast(errorToastParams(getErrorMessage(result.error)));
+      return;
+    }
+
+    dispatch(
+      setCurrentDataset({
+        fides_key: currentDataset.fides_key,
+        ctl_dataset: result.data,
+      }),
+    );
+    toast(successToastParams("Successfully modified dataset"));
+    await refetchDatasets();
+  };
+
+  const handleRefresh = async () => {
+    try {
+      const { data } = await refetchDatasets();
+      const refreshedDataset = data?.items.find(
+        (item) => item.fides_key === currentDataset?.fides_key,
+      );
+      if (refreshedDataset?.ctl_dataset) {
+        setEditorContent(yaml.dump(removeNulls(refreshedDataset.ctl_dataset)));
+      }
+      toast(successToastParams("Successfully refreshed datasets"));
+    } catch (error) {
+      toast(errorToastParams(getErrorMessage(error as FetchBaseQueryError)));
+    }
+  };
+
+  return (
+    <VStack alignItems="stretch" flex="1" maxWidth="70vw" maxHeight="100vh">
+      <Heading
+        as="h3"
+        size="sm"
+        display="flex"
+        alignItems="center"
+        justifyContent="space-between"
+      >
+        <HStack>
+          <Text>Edit dataset: </Text>
+          <Select
+            id="format"
+            data-testid="export-format-select"
+            value={currentDataset?.fides_key || ""}
+            options={datasetOptions}
+            onChange={handleDatasetChange}
+            className="w-64"
+          />
+          <ClipboardButton copyText={editorContent} />
+        </HStack>
+        <HStack spacing={2}>
+          <Button
+            htmlType="submit"
+            size="small"
+            data-testid="refresh-btn"
+            onClick={handleRefresh}
+            loading={isDatasetConfigsLoading}
+          >
+            Refresh
+          </Button>
+          <Button htmlType="submit" size="small" onClick={handleSave}>
+            Save
+          </Button>
+        </HStack>
+      </Heading>
+      <Stack
+        border="1px solid"
+        borderColor="gray.200"
+        borderRadius="md"
+        justifyContent="space-between"
+        py={4}
+        pr={4}
+        data-testid="empty-state"
+        flex="1 1 auto"
+        minHeight="200px"
+      >
+        <Editor
+          defaultLanguage="yaml"
+          value={editorContent}
+          height="100%"
+          onChange={(value) => setEditorContent(value || "")}
+          onMount={() => {}}
+          options={{
+            fontFamily: "Menlo",
+            fontSize: 13,
+            minimap: { enabled: false },
+            readOnly: false,
+            hideCursorInOverviewRuler: true,
+            overviewRulerBorder: false,
+            scrollBeyondLastLine: false,
+          }}
+          theme="light"
+        />
+      </Stack>
+      <Stack
+        backgroundColor={reachability?.reachable ? "green.50" : "red.50"}
+        border="1px solid"
+        borderColor={reachability?.reachable ? "green.500" : "red.500"}
+        borderRadius="md"
+        p={2}
+        flexShrink={0}
+        mt={2}
+      >
+        <HStack alignItems="center">
+          <HStack flex="1">
+            {reachability?.reachable ? (
+              <GreenCheckCircleIcon />
+            ) : (
+              <ErrorWarningIcon />
+            )}
+            <Text fontSize="sm" whiteSpace="pre-wrap">
+              {reachability?.reachable
+                ? "Dataset is reachable"
+                : `Dataset is not reachable. ${reachability?.details}`}
+            </Text>
+          </HStack>
+        </HStack>
+      </Stack>
+    </VStack>
+  );
+};
+
+export default EditorSection;
diff --git a/clients/admin-ui/src/features/test-datasets/TestRunnerSection.tsx b/clients/admin-ui/src/features/test-datasets/TestRunnerSection.tsx
new file mode 100644
index 0000000000..5ae60dc576
--- /dev/null
+++ b/clients/admin-ui/src/features/test-datasets/TestRunnerSection.tsx
@@ -0,0 +1,247 @@
+import {
+  AntButton as Button,
+  Heading,
+  HStack,
+  Text,
+  Textarea,
+  useToast,
+  VStack,
+} from "fidesui";
+import { useEffect, useState } from "react";
+import { useSelector } from "react-redux";
+
+import { useAppDispatch } from "~/app/hooks";
+import ClipboardButton from "~/features/common/ClipboardButton";
+import { getErrorMessage } from "~/features/common/helpers";
+import QuestionTooltip from "~/features/common/QuestionTooltip";
+import { errorToastParams, successToastParams } from "~/features/common/toast";
+import {
+  useGetDatasetInputsQuery,
+  useTestDatastoreConnectionDatasetsMutation,
+} from "~/features/datastore-connections";
+import { useGetFilteredResultsQuery } from "~/features/privacy-requests";
+import { PrivacyRequestStatus } from "~/types/api";
+import { isErrorResult } from "~/types/errors";
+
+import {
+  finishTest,
+  selectCurrentDataset,
+  selectIsTestRunning,
+  selectPrivacyRequestId,
+  selectTestInputs,
+  selectTestResults,
+  setPrivacyRequestId,
+  setTestInputs,
+  setTestResults,
+  startTest,
+} from "./dataset-test.slice";
+
+interface TestResultsSectionProps {
+  connectionKey: string;
+}
+
+const TestResultsSection = ({ connectionKey }: TestResultsSectionProps) => {
+  const toast = useToast();
+  const dispatch = useAppDispatch();
+  const [testDatasets] = useTestDatastoreConnectionDatasetsMutation();
+
+  const currentDataset = useSelector(selectCurrentDataset);
+  const testResults = useSelector(selectTestResults);
+  const testInputs = useSelector(selectTestInputs);
+  const isTestRunning = useSelector(selectIsTestRunning);
+  const privacyRequestId = useSelector(selectPrivacyRequestId);
+
+  const [inputValue, setInputValue] = useState("{}");
+
+  useEffect(() => {
+    if (currentDataset) {
+      setInputValue(JSON.stringify(testInputs, null, 2));
+    }
+  }, [currentDataset, testInputs]);
+
+  // Poll for results when we have a privacy request ID
+  const { data: filteredResults } = useGetFilteredResultsQuery(
+    { privacy_request_id: privacyRequestId! },
+    {
+      skip: !privacyRequestId || !currentDataset?.fides_key,
+      pollingInterval: 2000,
+    },
+  );
+
+  // Get dataset inputs
+  const { refetch: refetchInputs } = useGetDatasetInputsQuery(
+    {
+      connectionKey,
+      datasetKey: currentDataset?.fides_key || "",
+    },
+    {
+      skip: !connectionKey || !currentDataset?.fides_key,
+      refetchOnMountOrArgChange: true,
+    },
+  );
+
+  useEffect(() => {
+    const datasetKey = currentDataset?.fides_key;
+
+    if (connectionKey && datasetKey) {
+      // First refetch to ensure we have latest data
+      refetchInputs().then((response) => {
+        // Only set inputs if we got valid data for the current dataset
+        if (response.data && currentDataset?.fides_key === datasetKey) {
+          dispatch(
+            setTestInputs({
+              datasetKey,
+              values: response.data,
+            }),
+          );
+        }
+      });
+    }
+  }, [currentDataset, connectionKey, dispatch, refetchInputs]);
+
+  // Handle filtered results updates
+  useEffect(() => {
+    const currentDatasetKey = currentDataset?.fides_key;
+
+    if (
+      !filteredResults ||
+      filteredResults.privacy_request_id !== privacyRequestId ||
+      !currentDatasetKey
+    ) {
+      return;
+    }
+
+    // Create the results action with the current dataset key
+    const resultsAction = {
+      datasetKey: currentDatasetKey,
+      values: JSON.stringify(filteredResults, null, 2),
+    };
+
+    if (filteredResults.status === PrivacyRequestStatus.COMPLETE) {
+      if (isTestRunning) {
+        dispatch(setTestResults(resultsAction));
+        dispatch(finishTest());
+        toast(successToastParams("Test run completed successfully"));
+      }
+    } else if (filteredResults.status === PrivacyRequestStatus.ERROR) {
+      dispatch(setTestResults(resultsAction));
+      dispatch(finishTest());
+      toast(errorToastParams("Test run failed"));
+    }
+  }, [
+    filteredResults,
+    privacyRequestId,
+    currentDataset,
+    isTestRunning,
+    dispatch,
+    toast,
+  ]);
+
+  const handleInputChange = (event: React.ChangeEvent<HTMLTextAreaElement>) => {
+    const newValue = event.target.value;
+    setInputValue(newValue);
+    try {
+      const parsedValue = JSON.parse(newValue);
+      if (currentDataset) {
+        dispatch(
+          setTestInputs({
+            datasetKey: currentDataset.fides_key,
+            values: parsedValue,
+          }),
+        );
+      }
+    } catch (error) {
+      // JSON will be invalid as the user types it out, this is expected
+    }
+  };
+
+  const handleTestRun = async () => {
+    if (!currentDataset?.fides_key) {
+      return;
+    }
+
+    try {
+      let parsedInput;
+      try {
+        parsedInput = JSON.parse(inputValue);
+      } catch (e) {
+        toast(errorToastParams("Invalid JSON in test input"));
+        return;
+      }
+
+      dispatch(startTest(currentDataset.fides_key));
+
+      const result = await testDatasets({
+        connection_key: connectionKey,
+        dataset_key: currentDataset.fides_key,
+        input_data: parsedInput,
+      });
+
+      if (isErrorResult(result)) {
+        toast(errorToastParams(getErrorMessage(result.error)));
+        dispatch(finishTest());
+      } else if ("data" in result) {
+        dispatch(setPrivacyRequestId(result.data.privacy_request_id));
+      } else {
+        dispatch(finishTest());
+        toast(errorToastParams("No privacy request ID in response"));
+      }
+    } catch (error) {
+      dispatch(finishTest());
+      toast(errorToastParams("Failed to start test run"));
+    }
+  };
+
+  return (
+    <VStack alignItems="stretch" flex="1" maxWidth="70vw" minHeight="0">
+      <Heading
+        as="h3"
+        size="sm"
+        display="flex"
+        alignItems="center"
+        justifyContent="space-between"
+      >
+        <HStack>
+          <Text>Test inputs (identities and references)</Text>
+          <ClipboardButton copyText={inputValue} />
+        </HStack>
+        <HStack>
+          <QuestionTooltip label="Run a test access request using the provided test input data" />
+          <Button
+            htmlType="submit"
+            size="small"
+            type="primary"
+            data-testid="run-btn"
+            onClick={handleTestRun}
+            loading={isTestRunning}
+          >
+            Run
+          </Button>
+        </HStack>
+      </Heading>
+      <Textarea
+        size="sm"
+        focusBorderColor="primary.600"
+        color="gray.800"
+        isDisabled={isTestRunning}
+        height="100%"
+        value={inputValue}
+        onChange={handleInputChange}
+      />
+      <Heading as="h3" size="sm">
+        Test results <ClipboardButton copyText={testResults} />
+      </Heading>
+      <Textarea
+        isReadOnly
+        size="sm"
+        focusBorderColor="primary.600"
+        color="gray.800"
+        isDisabled={false}
+        height="100%"
+        value={testResults}
+      />
+    </VStack>
+  );
+};
+
+export default TestResultsSection;
diff --git a/clients/admin-ui/src/features/test-datasets/dataset-test.slice.ts b/clients/admin-ui/src/features/test-datasets/dataset-test.slice.ts
new file mode 100644
index 0000000000..431a64d4bc
--- /dev/null
+++ b/clients/admin-ui/src/features/test-datasets/dataset-test.slice.ts
@@ -0,0 +1,131 @@
+import { createSlice, PayloadAction } from "@reduxjs/toolkit";
+
+import type { RootState } from "~/app/store";
+import { DatasetConfigSchema } from "~/types/api";
+
+interface DatasetTestState {
+  privacyRequestId: string | null;
+  currentDataset: DatasetConfigSchema | null;
+  testInputs: Record<string, Record<string, any>>;
+  testResults: Record<string, string>;
+  isTestRunning: boolean;
+}
+
+const initialState: DatasetTestState = {
+  privacyRequestId: null,
+  currentDataset: null,
+  testInputs: {},
+  testResults: {},
+  isTestRunning: false,
+};
+
+export const datasetTestSlice = createSlice({
+  name: "datasetTest",
+  initialState,
+  reducers: {
+    startTest: (draftState, action: PayloadAction<string>) => {
+      draftState.testResults = {
+        ...draftState.testResults,
+        [action.payload]: "",
+      };
+      draftState.isTestRunning = true;
+    },
+    setPrivacyRequestId: (draftState, action: PayloadAction<string>) => {
+      draftState.privacyRequestId = action.payload;
+    },
+    finishTest: (draftState) => {
+      draftState.privacyRequestId = null;
+      draftState.isTestRunning = false;
+    },
+    setTestInputs: (
+      draftState,
+      action: PayloadAction<{
+        datasetKey: string;
+        values: Record<string, any>;
+      }>,
+    ) => {
+      const existingValues =
+        draftState.testInputs[action.payload.datasetKey] || {};
+      const inputsData = action.payload.values || {};
+
+      // Skip processing if we're trying to set empty values over existing ones
+      if (
+        Object.keys(existingValues).length > 0 &&
+        Object.keys(inputsData).length === 0
+      ) {
+        return;
+      }
+
+      // Start with input data
+      const mergedValues = { ...inputsData };
+
+      // For each key in the existing values
+      Object.entries(existingValues).forEach(([key, existingValue]) => {
+        // If the new value is null and we have an existing non-null value, keep the existing value
+        if (
+          key in mergedValues &&
+          mergedValues[key] === null &&
+          existingValue !== null
+        ) {
+          mergedValues[key] = existingValue;
+        }
+      });
+
+      draftState.testInputs = {
+        ...draftState.testInputs,
+        [action.payload.datasetKey]: mergedValues,
+      };
+    },
+    setCurrentDataset: (
+      draftState,
+      action: PayloadAction<DatasetConfigSchema | null>,
+    ) => {
+      draftState.currentDataset = action.payload;
+      draftState.privacyRequestId = null;
+    },
+    setTestResults: (
+      draftState,
+      action: PayloadAction<{
+        datasetKey: string;
+        values: string;
+      }>,
+    ) => {
+      draftState.testResults = {
+        ...draftState.testResults,
+        [action.payload.datasetKey]: action.payload.values,
+      };
+    },
+  },
+});
+
+export const {
+  startTest,
+  setPrivacyRequestId,
+  finishTest,
+  setTestInputs,
+  setCurrentDataset,
+  setTestResults,
+} = datasetTestSlice.actions;
+
+export const selectPrivacyRequestId = (state: RootState) =>
+  state.datasetTest.privacyRequestId;
+export const selectDatasetTestPrivacyRequestId = (state: RootState) =>
+  state.datasetTest.privacyRequestId;
+export const selectCurrentDataset = (state: RootState) =>
+  state.datasetTest.currentDataset;
+export const selectTestInputs = (state: RootState) => {
+  const { currentDataset } = state.datasetTest;
+  return currentDataset
+    ? state.datasetTest.testInputs[currentDataset.fides_key] || {}
+    : {};
+};
+export const selectTestResults = (state: RootState) => {
+  const { currentDataset } = state.datasetTest;
+  return currentDataset
+    ? state.datasetTest.testResults[currentDataset.fides_key] || ""
+    : "";
+};
+export const selectIsTestRunning = (state: RootState) =>
+  state.datasetTest.isTestRunning;
+
+export const { reducer } = datasetTestSlice;
diff --git a/clients/admin-ui/src/features/test-datasets/helpers.ts b/clients/admin-ui/src/features/test-datasets/helpers.ts
new file mode 100644
index 0000000000..36630c32ca
--- /dev/null
+++ b/clients/admin-ui/src/features/test-datasets/helpers.ts
@@ -0,0 +1,13 @@
+export const removeNulls = (obj: any): any => {
+  if (Array.isArray(obj)) {
+    return obj.map((item) => removeNulls(item)).filter((item) => item !== null);
+  }
+  if (obj && typeof obj === "object") {
+    return Object.fromEntries(
+      Object.entries(obj)
+        .map(([key, value]) => [key, removeNulls(value)])
+        .filter(([, value]) => value !== null),
+    );
+  }
+  return obj;
+};
diff --git a/clients/admin-ui/src/features/test-datasets/index.ts b/clients/admin-ui/src/features/test-datasets/index.ts
new file mode 100644
index 0000000000..1cceac9eb8
--- /dev/null
+++ b/clients/admin-ui/src/features/test-datasets/index.ts
@@ -0,0 +1 @@
+export * from "./dataset-test.slice";
diff --git a/clients/admin-ui/src/pages/systems/configure/[id].tsx b/clients/admin-ui/src/pages/systems/configure/[id]/index.tsx
similarity index 100%
rename from clients/admin-ui/src/pages/systems/configure/[id].tsx
rename to clients/admin-ui/src/pages/systems/configure/[id]/index.tsx
diff --git a/clients/admin-ui/src/pages/systems/configure/[id]/test-datasets.tsx b/clients/admin-ui/src/pages/systems/configure/[id]/test-datasets.tsx
new file mode 100644
index 0000000000..8102b796a9
--- /dev/null
+++ b/clients/admin-ui/src/pages/systems/configure/[id]/test-datasets.tsx
@@ -0,0 +1,87 @@
+import { Box, HStack } from "fidesui";
+import type { NextPage } from "next";
+import { useRouter } from "next/router";
+import { useEffect } from "react";
+
+import { useAppDispatch } from "~/app/hooks";
+import FidesSpinner from "~/features/common/FidesSpinner";
+import Layout from "~/features/common/Layout";
+import { SYSTEM_ROUTE } from "~/features/common/nav/v2/routes";
+import PageHeader from "~/features/common/PageHeader";
+import {
+  setActiveSystem,
+  useGetSystemByFidesKeyQuery,
+} from "~/features/system";
+import EditorSection from "~/features/test-datasets/DatasetEditorSection";
+import TestResultsSection from "~/features/test-datasets/TestRunnerSection";
+
+// Helper functions
+const getSystemId = (query: { id?: string | string[] }): string => {
+  if (!query.id) {
+    return "";
+  }
+  return Array.isArray(query.id) ? query.id[0] : query.id;
+};
+
+const TestDatasetPage: NextPage = () => {
+  const router = useRouter();
+  const dispatch = useAppDispatch();
+
+  const systemId = getSystemId(router.query);
+
+  const { data: system, isLoading } = useGetSystemByFidesKeyQuery(systemId, {
+    skip: !systemId,
+  });
+
+  const connectionKey = system?.connection_configs?.key || "";
+
+  useEffect(() => {
+    dispatch(setActiveSystem(system));
+  }, [system, dispatch]);
+
+  if (isLoading) {
+    return (
+      <Layout title="Systems">
+        <FidesSpinner />
+      </Layout>
+    );
+  }
+
+  return (
+    <Layout
+      title="System inventory"
+      mainProps={{
+        paddingTop: 0,
+        height: "100vh",
+        display: "flex",
+        flexDirection: "column",
+      }}
+    >
+      <PageHeader
+        breadcrumbs={[
+          { title: "System inventory", link: SYSTEM_ROUTE },
+          {
+            title: system?.name || "",
+            link: `/systems/configure/${systemId}#integrations`,
+          },
+          { title: "Test datasets" },
+        ]}
+        isSticky
+      >
+        <Box position="relative" />
+      </PageHeader>
+      <HStack
+        alignItems="stretch"
+        flex="1"
+        minHeight="0"
+        spacing="4"
+        padding="4 4 4 0"
+      >
+        <EditorSection connectionKey={connectionKey} />
+        <TestResultsSection connectionKey={connectionKey} />
+      </HStack>
+    </Layout>
+  );
+};
+
+export default TestDatasetPage;
diff --git a/src/fides/api/api/v1/endpoints/dataset_endpoints.py b/src/fides/api/api/v1/endpoints/dataset_endpoints.py
index c1caea1b94..bdc8f367c6 100644
--- a/src/fides/api/api/v1/endpoints/dataset_endpoints.py
+++ b/src/fides/api/api/v1/endpoints/dataset_endpoints.py
@@ -1,5 +1,5 @@
 from datetime import datetime, timezone
-from typing import Annotated, Callable, List
+from typing import Annotated, Any, Callable, Dict, List
 
 import yaml
 from fastapi import Depends, HTTPException, Request
@@ -41,16 +41,25 @@
     to_graph_field,
     validate_masking_strategy_override,
 )
+from fides.api.models.policy import Policy
 from fides.api.oauth.utils import verify_oauth_client
 from fides.api.schemas.api import BulkUpdateFailed
 from fides.api.schemas.dataset import (
     BulkPutDataset,
     DatasetConfigCtlDataset,
     DatasetConfigSchema,
+    DatasetReachability,
     DatasetTraversalDetails,
     ValidateDatasetResponse,
     validate_data_categories_against_db,
 )
+from fides.api.schemas.privacy_request import TestPrivacyRequest
+from fides.api.schemas.redis_cache import UnlabeledIdentities
+from fides.api.service.dataset.dataset_service import (
+    get_dataset_reachability,
+    get_identities_and_references,
+    run_test_access_request,
+)
 from fides.api.util.api_router import APIRouter
 from fides.api.util.data_category import get_data_categories_from_db
 from fides.api.util.saas_util import merge_datasets
@@ -58,14 +67,18 @@
     DATASET_CREATE_OR_UPDATE,
     DATASET_DELETE,
     DATASET_READ,
+    DATASET_TEST,
 )
 from fides.common.api.v1.urn_registry import (
     CONNECTION_DATASETS,
     DATASET_BY_KEY,
+    DATASET_CONFIG_BY_KEY,
     DATASET_CONFIGS,
+    DATASET_INPUTS,
+    DATASET_REACHABILITY,
     DATASET_VALIDATE,
-    DATASETCONFIG_BY_KEY,
     DATASETS,
+    TEST_DATASET,
     V1_URL_PREFIX,
     YAML_DATASETS,
 )
@@ -572,7 +585,7 @@ def get_dataset_configs(
 
 
 @router.get(
-    DATASETCONFIG_BY_KEY,
+    DATASET_CONFIG_BY_KEY,
     dependencies=[Security(verify_oauth_client, scopes=[DATASET_READ])],
     response_model=DatasetConfigSchema,
 )
@@ -763,3 +776,110 @@ def clean_datasets(
             "failed": failed,
         },
     )
+
+
+@router.get(
+    DATASET_INPUTS,
+    dependencies=[Security(verify_oauth_client, scopes=[DATASET_READ])],
+)
+def dataset_identities_and_references(
+    *,
+    db: Session = Depends(deps.get_db),
+    connection_config: ConnectionConfig = Depends(_get_connection_config),
+    fides_key: FidesKey,
+) -> Dict[str, Any]:
+    """
+    Returns a dictionary containing the immediate identity and dataset reference
+    dependencies for the given dataset.
+    """
+
+    dataset_config = DatasetConfig.filter(
+        db=db,
+        conditions=(
+            (DatasetConfig.connection_config_id == connection_config.id)
+            & (DatasetConfig.fides_key == fides_key)
+        ),
+    ).first()
+    if not dataset_config:
+        raise HTTPException(
+            status_code=HTTP_404_NOT_FOUND,
+            detail=f"No dataset config with fides_key '{fides_key}'",
+        )
+
+    inputs = get_identities_and_references(dataset_config)
+    return {input: None for input in inputs}
+
+
+@router.get(
+    DATASET_REACHABILITY,
+    dependencies=[Security(verify_oauth_client, scopes=[DATASET_READ])],
+    response_model=DatasetReachability,
+)
+def dataset_reachability(
+    *,
+    db: Session = Depends(deps.get_db),
+    connection_config: ConnectionConfig = Depends(_get_connection_config),
+    fides_key: FidesKey,
+) -> Dict[str, Any]:
+    """
+    Returns a dictionary containing the immediate identity and dataset reference
+    dependencies for the given dataset.
+    """
+
+    dataset_config = DatasetConfig.filter(
+        db=db,
+        conditions=(
+            (DatasetConfig.connection_config_id == connection_config.id)
+            & (DatasetConfig.fides_key == fides_key)
+        ),
+    ).first()
+    if not dataset_config:
+        raise HTTPException(
+            status_code=HTTP_404_NOT_FOUND,
+            detail=f"No dataset config with fides_key '{fides_key}'",
+        )
+
+    reachable, details = get_dataset_reachability(db, dataset_config)
+    return {"reachable": reachable, "details": details}
+
+
+@router.post(
+    TEST_DATASET,
+    status_code=HTTP_200_OK,
+    dependencies=[Security(verify_oauth_client, scopes=[DATASET_TEST])],
+    response_model=TestPrivacyRequest,
+)
+def test_connection_datasets(
+    *,
+    db: Session = Depends(deps.get_db),
+    connection_config: ConnectionConfig = Depends(_get_connection_config),
+    fides_key: FidesKey,
+    unlabeled_identities: UnlabeledIdentities,
+) -> Dict[str, Any]:
+    dataset_config = DatasetConfig.filter(
+        db=db,
+        conditions=(
+            (DatasetConfig.connection_config_id == connection_config.id)
+            & (DatasetConfig.fides_key == fides_key)
+        ),
+    ).first()
+    if not dataset_config:
+        raise HTTPException(
+            status_code=HTTP_404_NOT_FOUND,
+            detail=f"No dataset config with fides_key '{fides_key}'",
+        )
+
+    access_policy = Policy.get_by(db, field="key", value="default_access_policy")
+    if not access_policy:
+        raise HTTPException(
+            status_code=HTTP_404_NOT_FOUND,
+            detail='Policy with key "default_access_policy" not found',
+        )
+
+    privacy_request = run_test_access_request(
+        db,
+        access_policy,
+        dataset_config,
+        input_data=unlabeled_identities.data,
+    )
+    return {"privacy_request_id": privacy_request.id}
diff --git a/src/fides/api/api/v1/endpoints/privacy_request_endpoints.py b/src/fides/api/api/v1/endpoints/privacy_request_endpoints.py
index f1699994e2..c8dc17a273 100644
--- a/src/fides/api/api/v1/endpoints/privacy_request_endpoints.py
+++ b/src/fides/api/api/v1/endpoints/privacy_request_endpoints.py
@@ -72,6 +72,7 @@
 )
 from fides.api.models.privacy_preference import PrivacyPreferenceHistory
 from fides.api.models.privacy_request import (
+    EXITED_EXECUTION_LOG_STATUSES,
     CheckpointActionRequired,
     ConsentRequest,
     CustomPrivacyRequestField,
@@ -107,6 +108,7 @@
     BulkSoftDeletePrivacyRequests,
     DenyPrivacyRequests,
     ExecutionLogDetailResponse,
+    FilteredPrivacyRequestResults,
     ManualWebhookData,
     PrivacyRequestAccessResults,
     PrivacyRequestCreate,
@@ -166,6 +168,7 @@
     PRIVACY_REQUEST_BULK_RETRY,
     PRIVACY_REQUEST_BULK_SOFT_DELETE,
     PRIVACY_REQUEST_DENY,
+    PRIVACY_REQUEST_FILTERED_RESULTS,
     PRIVACY_REQUEST_MANUAL_WEBHOOK_ACCESS_INPUT,
     PRIVACY_REQUEST_MANUAL_WEBHOOK_ERASURE_INPUT,
     PRIVACY_REQUEST_NOTIFICATIONS,
@@ -602,6 +605,14 @@ def _filter_privacy_request_queryset(
             )
         )
 
+    # Filter out test privacy requests
+    query = query.filter(
+        or_(
+            PrivacyRequest.source != PrivacyRequestSource.dataset_test,
+            PrivacyRequest.source.is_(None),
+        )
+    )
+
     # Filter out deleted requests
     if not include_deleted_requests:
         query = query.filter(PrivacyRequest.deleted_at.is_(None))
@@ -2611,3 +2622,43 @@ def get_access_results_urls(
         return PrivacyRequestAccessResults(access_result_urls=[])
 
     return privacy_request.access_result_urls
+
+
+@router.get(
+    PRIVACY_REQUEST_FILTERED_RESULTS,
+    dependencies=[
+        Security(verify_oauth_client, scopes=[PRIVACY_REQUEST_READ_ACCESS_RESULTS])
+    ],
+    status_code=HTTP_200_OK,
+    response_model=FilteredPrivacyRequestResults,
+)
+def get_test_privacy_request_results(
+    privacy_request_id: str,
+    db: Session = Depends(deps.get_db),
+) -> Dict[str, Any]:
+    """Get filtered results for a test privacy request and update its status if complete."""
+    privacy_request = get_privacy_request_or_error(db, privacy_request_id)
+
+    if privacy_request.source != PrivacyRequestSource.dataset_test:
+        raise HTTPException(
+            status_code=HTTP_403_FORBIDDEN,
+            detail="Results can only be retrieved for test privacy requests.",
+        )
+
+    # Check completion status of all tasks
+    statuses = [task.status for task in privacy_request.access_tasks]
+    all_completed = all(status in EXITED_EXECUTION_LOG_STATUSES for status in statuses)
+
+    # Update request status if all tasks are done
+    if all_completed:
+        has_errors = ExecutionLogStatus.error in statuses
+        privacy_request.status = (
+            PrivacyRequestStatus.error if has_errors else PrivacyRequestStatus.complete
+        )
+        privacy_request.save(db=db)
+
+    return {
+        "privacy_request_id": privacy_request.id,
+        "status": privacy_request.status,
+        "results": privacy_request.get_raw_access_results(),
+    }
diff --git a/src/fides/api/graph/graph.py b/src/fides/api/graph/graph.py
index 393a09eafe..e99cfc550d 100644
--- a/src/fides/api/graph/graph.py
+++ b/src/fides/api/graph/graph.py
@@ -200,7 +200,8 @@ def __init__(self, *datasets: GraphDataset) -> None:
                             "Referenced object {} does not exist", dest_field_address
                         )
                         raise ValidationError(
-                            f"Referred to object {dest_field_address} does not exist"
+                            f"Referred to object {dest_field_address} does not exist",
+                            errors=[dest_field_address.value],
                         )
                     self.edges.add(
                         Edge.create_edge(
diff --git a/src/fides/api/graph/traversal.py b/src/fides/api/graph/traversal.py
index e7686fb26d..82256812c7 100644
--- a/src/fides/api/graph/traversal.py
+++ b/src/fides/api/graph/traversal.py
@@ -255,9 +255,9 @@ def __init__(self, graph: DatasetGraph, data: Dict[str, Any]):
                 )
             )
 
-        self.__verify_traversal()
+        self._verify_traversal()
 
-    def __verify_traversal(self) -> None:
+    def _verify_traversal(self) -> None:
         """Verify that a valid traversal exists. This method simply assembles a traversal
         and raises an error on any traversal failure conditions."""
         self.traverse(
diff --git a/src/fides/api/models/privacy_request.py b/src/fides/api/models/privacy_request.py
index 5a1326c2e6..22dc84dbdd 100644
--- a/src/fides/api/models/privacy_request.py
+++ b/src/fides/api/models/privacy_request.py
@@ -183,12 +183,14 @@ class PrivacyRequestSource(str, EnumType):
     - Request Manager: Request submitted from the Admin UI's Request manager page
     - Consent Webhook: Request created as a side-effect of a consent webhook request (bidirectional consent)
     - Fides.js: Request created as a side-effect of a privacy preference update from Fides.js
+    - Dataset Test: Standalone dataset test
     """
 
     privacy_center = "Privacy Center"
     request_manager = "Request Manager"
     consent_webhook = "Consent Webhook"
     fides_js = "Fides.js"
+    dataset_test = "Dataset Test"
 
 
 class CallbackType(EnumType):
diff --git a/src/fides/api/schemas/dataset.py b/src/fides/api/schemas/dataset.py
index d749cdc8d0..75265ebd89 100644
--- a/src/fides/api/schemas/dataset.py
+++ b/src/fides/api/schemas/dataset.py
@@ -131,3 +131,12 @@ class DryRunDatasetResponse(FidesSchema):
 
     collectionAddress: CollectionAddressResponse
     query: Any
+
+
+class DatasetReachability(FidesSchema):
+    """
+    Response containing reachability details for a single dataset
+    """
+
+    reachable: bool
+    details: Optional[str]
diff --git a/src/fides/api/schemas/privacy_request.py b/src/fides/api/schemas/privacy_request.py
index 215c7c7d67..69ebcb2df5 100644
--- a/src/fides/api/schemas/privacy_request.py
+++ b/src/fides/api/schemas/privacy_request.py
@@ -388,3 +388,17 @@ class PrivacyRequestAccessResults(FidesSchema):
     """Schema for the access results of a PrivacyRequest"""
 
     access_result_urls: List[str]
+
+
+class TestPrivacyRequest(FidesSchema):
+    """Schema containing the data for a test privacy request"""
+
+    privacy_request_id: str
+
+
+class FilteredPrivacyRequestResults(FidesSchema):
+    """Schema representing the status and results of a test privacy request"""
+
+    privacy_request_id: str
+    status: PrivacyRequestStatus
+    results: Dict[str, Any]
diff --git a/src/fides/api/schemas/redis_cache.py b/src/fides/api/schemas/redis_cache.py
index 2561bb5ed0..dd96193b64 100644
--- a/src/fides/api/schemas/redis_cache.py
+++ b/src/fides/api/schemas/redis_cache.py
@@ -1,8 +1,18 @@
 import uuid
 from typing import Any, Dict, List, Optional, Union
 
-from pydantic import ConfigDict, EmailStr, Field, StrictInt, StrictStr, field_validator
-
+from pydantic import (
+    ConfigDict,
+    EmailStr,
+    Field,
+    StrictInt,
+    StrictStr,
+    ValidationError,
+    field_validator,
+    model_validator,
+)
+
+from fides.api.common_exceptions import BadRequest
 from fides.api.custom_types import PhoneNumber
 from fides.api.schemas.base_class import FidesSchema
 
@@ -116,6 +126,50 @@ def labeled_dict(
         return d
 
 
+class UnlabeledIdentities(FidesSchema):
+    """
+    A model for validating identity dictionaries where standard fields use Identity's validation
+    but custom fields just need to be valued.
+    """
+
+    data: Dict[str, Any]
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_identities(cls, values: Dict[str, Any]) -> Dict[str, Any]:
+        if not isinstance(values, dict):
+            raise BadRequest("Inputs must be JSON formatted")
+
+        if not values:
+            raise BadRequest("No inputs provided")
+
+        standard_fields = {}
+        custom_fields = {}
+
+        # Separate standard and custom fields
+        for field, value in values.items():
+            if field in Identity.model_fields:
+                if value is None:
+                    raise BadRequest(f'Input "{field}" cannot be empty')
+                standard_fields[field] = value
+            else:
+                if value is None:
+                    raise BadRequest(f'Input "{field}" cannot be empty')
+                custom_fields[field] = value
+
+        # Validate standard fields using Identity
+        try:
+            Identity(**standard_fields)
+        except ValidationError as e:
+            error_detail = e.errors()[0]
+            field = error_detail.get("loc")[0]  # type: ignore[assignment, index]
+            clean_message = error_detail.get("msg")
+            raise BadRequest(f'"{field}" {clean_message}')
+
+        # Return the combined validated data
+        return {"data": {**standard_fields, **custom_fields}}
+
+
 class CustomPrivacyRequestField(FidesSchema):
     """Schema for custom privacy request fields."""
 
diff --git a/src/fides/api/service/dataset/__init__.py b/src/fides/api/service/dataset/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/src/fides/api/service/dataset/dataset_service.py b/src/fides/api/service/dataset/dataset_service.py
new file mode 100644
index 0000000000..0c4cb6ab2e
--- /dev/null
+++ b/src/fides/api/service/dataset/dataset_service.py
@@ -0,0 +1,177 @@
+from copy import deepcopy
+from typing import Any, Dict, Optional, Set, Tuple
+
+from sqlalchemy.orm import Session
+
+from fides.api.common_exceptions import UnreachableNodesError, ValidationError
+from fides.api.graph.config import GraphDataset
+from fides.api.graph.graph import DatasetGraph
+from fides.api.graph.traversal import Traversal
+from fides.api.models.datasetconfig import DatasetConfig
+from fides.api.models.policy import Policy
+from fides.api.models.privacy_request import (
+    PrivacyRequest,
+    PrivacyRequestSource,
+    PrivacyRequestStatus,
+)
+from fides.api.schemas.redis_cache import Identity, LabeledIdentity
+from fides.api.task.create_request_tasks import run_access_request
+
+
+def get_dataset_reachability(
+    db: Session,
+    dataset_config: DatasetConfig,
+) -> Tuple[bool, Optional[str]]:
+    """
+    Determines if the given dataset is reachable along with an error message
+    """
+
+    # Get all the dataset configs that are not associated with a disabled connection
+    datasets = DatasetConfig.all(db=db)
+    dataset_graphs = [
+        dataset_config.get_graph()
+        for dataset_config in datasets
+        if not dataset_config.connection_config.disabled
+    ]
+
+    # We still want to check reachability even if our dataset config's connection is disabled.
+    # We also consider the siblings, because if the connection is enabled, then all the
+    # datasets will be enabled with it.
+    sibling_dataset_graphs = []
+    if dataset_config.connection_config.disabled:
+        sibling_datasets = dataset_config.connection_config.datasets
+        sibling_dataset_graphs = [
+            dataset_config.get_graph() for dataset_config in sibling_datasets
+        ]
+
+    try:
+        dataset_graph = DatasetGraph(*dataset_graphs, *sibling_dataset_graphs)
+    except ValidationError as exc:
+        return (
+            False,
+            f'The following dataset references do not exist "{", ".join(exc.errors)}"',
+        )
+
+    # dummy data is enough to determine traversability
+    identity_seed: Dict[str, str] = {
+        k: "something" for k in dataset_graph.identity_keys.values()
+    }
+
+    try:
+        Traversal(dataset_graph, identity_seed)
+    except UnreachableNodesError as exc:
+        return (
+            False,
+            f'The following collections are not reachable "{", ".join(exc.errors)}"',
+        )
+
+    return True, None
+
+
+def get_identities_and_references(
+    dataset_config: DatasetConfig,
+) -> Set[str]:
+    """
+    Returns all identity and dataset references in the dataset.
+    If a field has multiple references only the first reference will be considered.
+    """
+
+    result: Set[str] = set()
+    dataset: GraphDataset = dataset_config.get_graph()
+    for collection in dataset.collections:
+        # Process the identities in the collection
+        result.update(collection.identities().values())
+        for _, field_refs in collection.references().items():
+            # Take first reference only, we only care that this collection is reachable,
+            # how we get there doesn't matter for our current use case
+            ref, edge_direction = field_refs[0]
+            if edge_direction == "from" and ref.dataset != dataset_config.fides_key:
+                result.add(ref.value)
+    return result
+
+
+def run_test_access_request(
+    db: Session,
+    policy: Policy,
+    dataset_config: DatasetConfig,
+    input_data: Dict[str, Any],
+) -> PrivacyRequest:
+    """
+    Creates a privacy request with a source of "Dataset test" that runs an access request for a single dataset.
+    The input data is used to mock any external dataset values referenced by our dataset so that it can run
+    complete isolation.
+    """
+
+    # Create a privacy request with a source of "Dataset test"
+    # so it's not shown to the user.
+    privacy_request = PrivacyRequest.create(
+        db=db,
+        data={
+            "policy_id": policy.id,
+            "source": PrivacyRequestSource.dataset_test,
+            "status": PrivacyRequestStatus.in_processing,
+        },
+    )
+
+    # Remove periods and colons to avoid them being parsed as path delimiters downstream.
+    escaped_input_data = {
+        key.replace(".", "_").replace(":", "_"): value
+        for key, value in input_data.items()
+    }
+
+    # Manually cache the input data as identity data.
+    # We're doing a bit of trickery here to avoid asking for labels for custom identities.
+    predefined_fields = Identity.model_fields.keys()
+    input_identity = {
+        key: (
+            value
+            if key in predefined_fields
+            else LabeledIdentity(label=key, value=value)
+        )
+        for key, value in escaped_input_data.items()
+    }
+    privacy_request.cache_identity(input_identity)
+
+    graph_dataset = dataset_config.get_graph()
+    modified_graph_dataset = _replace_references_with_identities(
+        dataset_config.fides_key, graph_dataset
+    )
+
+    dataset_graph = DatasetGraph(modified_graph_dataset)
+    connection_config = dataset_config.connection_config
+
+    # Finally invoke the existing DSR 3.0 access request task
+    run_access_request(
+        privacy_request,
+        policy,
+        dataset_graph,
+        [connection_config],
+        escaped_input_data,
+        db,
+        privacy_request_proceed=False,
+    )
+    return privacy_request
+
+
+def _replace_references_with_identities(
+    dataset_key: str, graph_dataset: GraphDataset
+) -> GraphDataset:
+    """
+    Replace external field references with identity values for testing.
+
+    Creates a copy of the graph dataset and replaces dataset references with
+    equivalent identity references that can be seeded directly. This allows
+    testing a single dataset in isolation without needing to load data from
+    referenced external datasets.
+    """
+
+    modified_graph_dataset = deepcopy(graph_dataset)
+
+    for collection in modified_graph_dataset.collections:
+        for field in collection.fields:
+            for ref, edge_direction in field.references[:]:
+                if edge_direction == "from" and ref.dataset != dataset_key:
+                    field.identity = f"{ref.dataset}_{ref.collection}_{'_'.join(ref.field_path.levels)}"
+                    field.references.remove((ref, "from"))
+
+    return modified_graph_dataset
diff --git a/src/fides/api/task/graph_task.py b/src/fides/api/task/graph_task.py
index c2e15580eb..6b78b57297 100644
--- a/src/fides/api/task/graph_task.py
+++ b/src/fides/api/task/graph_task.py
@@ -59,6 +59,7 @@
 )
 from fides.api.util.consent_util import add_errored_system_status_for_consent_reporting
 from fides.api.util.logger import Pii
+from fides.api.util.logger_context_utils import LoggerContextKeys
 from fides.api.util.saas_util import FIDESOPS_GROUPED_INPUTS
 from fides.config import CONFIG
 
@@ -575,21 +576,25 @@ def skip_if_action_disabled(self, action_type: ActionType) -> None:
     @retry(action_type=ActionType.access, default_return=[])
     def access_request(self, *inputs: List[Row]) -> List[Row]:
         """Run an access request on a single node."""
-        formatted_input_data: NodeInput = self.pre_process_input_data(
-            *inputs, group_dependent_fields=True
-        )
-        output: List[Row] = self.connector.retrieve_data(
-            self.execution_node,
-            self.resources.policy,
-            self.resources.request,
-            self.resources.privacy_request_task,
-            formatted_input_data,
-        )
-        filtered_output: List[Row] = self.access_results_post_processing(
-            self.pre_process_input_data(*inputs, group_dependent_fields=False), output
-        )
-        self.log_end(ActionType.access)
-        return filtered_output
+        with logger.contextualize(
+            **{LoggerContextKeys.privacy_request_id.value: self.resources.request.id}
+        ):
+            formatted_input_data: NodeInput = self.pre_process_input_data(
+                *inputs, group_dependent_fields=True
+            )
+            output: List[Row] = self.connector.retrieve_data(
+                self.execution_node,
+                self.resources.policy,
+                self.resources.request,
+                self.resources.privacy_request_task,
+                formatted_input_data,
+            )
+            filtered_output: List[Row] = self.access_results_post_processing(
+                self.pre_process_input_data(*inputs, group_dependent_fields=False),
+                output,
+            )
+            self.log_end(ActionType.access)
+            return filtered_output
 
     @retry(action_type=ActionType.erasure, default_return=0)
     def erasure_request(
diff --git a/src/fides/common/api/scope_registry.py b/src/fides/common/api/scope_registry.py
index 06cf06e1d3..a6183d2909 100644
--- a/src/fides/common/api/scope_registry.py
+++ b/src/fides/common/api/scope_registry.py
@@ -60,6 +60,7 @@
 SYSTEM = "system"
 SYSTEM_MANAGER = "system_manager"
 TAXONOMY = "taxonomy"
+TEST = "test"
 TRANSFER = "transfer"
 UPDATE = "update"
 UPLOAD_DATA = "upload_data"
@@ -129,6 +130,7 @@
 DATASET_CREATE_OR_UPDATE = f"{DATASET}:{CREATE_OR_UPDATE}"
 DATASET_DELETE = f"{DATASET}:{DELETE}"
 DATASET_READ = f"{DATASET}:{READ}"
+DATASET_TEST = f"{DATASET}:{TEST}"
 
 ENCRYPTION_EXEC = f"{ENCRYPTION}:{EXEC}"
 
@@ -275,6 +277,7 @@
     DATASET_CREATE_OR_UPDATE: "Create or modify datasets",
     DATASET_DELETE: "Delete datasets",
     DATASET_READ: "View datasets",
+    DATASET_TEST: "Run a standalone privacy request test for a dataset",
     ENCRYPTION_EXEC: "Encrypt data",
     MESSAGING_TEMPLATE_UPDATE: "Update messaging templates",
     EVALUATION_CREATE: "Create evaluation",
diff --git a/src/fides/common/api/v1/urn_registry.py b/src/fides/common/api/v1/urn_registry.py
index 85401adfa9..e9aa00ed75 100644
--- a/src/fides/common/api/v1/urn_registry.py
+++ b/src/fides/common/api/v1/urn_registry.py
@@ -86,6 +86,9 @@
 PRIVACY_REQUEST_REQUEUE = "/privacy-request/{privacy_request_id}/requeue"
 REQUEST_TASK_CALLBACK = "/request-task/callback"
 PRIVACY_REQUEST_ACCESS_RESULTS = "/privacy-request/{privacy_request_id}/access-results"
+PRIVACY_REQUEST_FILTERED_RESULTS = (
+    "/privacy-request/{privacy_request_id}/filtered-results"
+)
 
 PRIVACY_REQUEST_VERIFY_IDENTITY = "/privacy-request/{privacy_request_id}/verify"
 PRIVACY_REQUEST_RESUME = "/privacy-request/{privacy_request_id}/resume"
@@ -158,7 +161,10 @@
 CONNECTION_DATASETS = CONNECTION_BY_KEY + DATASETS
 DATASET_CONFIGS = CONNECTION_BY_KEY + DATASET_CONFIG
 DATASET_BY_KEY = CONNECTION_BY_KEY + DATASETS + "/{fides_key}"
-DATASETCONFIG_BY_KEY = CONNECTION_BY_KEY + DATASET_CONFIG + "/{fides_key}"
+DATASET_CONFIG_BY_KEY = CONNECTION_BY_KEY + DATASET_CONFIG + "/{fides_key}"
+DATASET_INPUTS = DATASET_BY_KEY + "/inputs"
+DATASET_REACHABILITY = DATASET_BY_KEY + "/reachability"
+TEST_DATASET = DATASET_BY_KEY + "/test"
 
 # YAML Collection URLs
 YAML_DATASETS = YAML + CONNECTION_DATASETS
diff --git a/tests/conftest.py b/tests/conftest.py
index df39ae7bbe..d513472ee3 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1602,3 +1602,32 @@ def load_default_data_uses(db):
         # loaded, in which case the create will throw an error. so we first check existence.
         if DataUse.get_by(db, field="name", value=data_use.name) is None:
             DataUse.create(db=db, data=data_use.model_dump(mode="json"))
+
+
+@pytest.fixture
+def owner_auth_header(owner_user):
+    return generate_role_header_for_user(owner_user, owner_user.client.roles)
+
+
+@pytest.fixture
+def contributor_auth_header(contributor_user):
+    return generate_role_header_for_user(
+        contributor_user, contributor_user.client.roles
+    )
+
+
+@pytest.fixture
+def viewer_auth_header(viewer_user):
+    return generate_role_header_for_user(viewer_user, viewer_user.client.roles)
+
+
+@pytest.fixture
+def approver_auth_header(approver_user):
+    return generate_role_header_for_user(approver_user, approver_user.client.roles)
+
+
+@pytest.fixture
+def viewer_and_approver_auth_header(viewer_and_approver_user):
+    return generate_role_header_for_user(
+        viewer_and_approver_user, viewer_and_approver_user.client.roles
+    )
diff --git a/tests/ops/api/v1/endpoints/test_dataset_endpoints.py b/tests/ops/api/v1/endpoints/test_dataset_endpoints.py
index 7fe31d2895..f744d59e47 100644
--- a/tests/ops/api/v1/endpoints/test_dataset_endpoints.py
+++ b/tests/ops/api/v1/endpoints/test_dataset_endpoints.py
@@ -27,9 +27,9 @@
 from fides.common.api.v1.urn_registry import (
     CONNECTION_DATASETS,
     DATASET_BY_KEY,
+    DATASET_CONFIG_BY_KEY,
     DATASET_CONFIGS,
     DATASET_VALIDATE,
-    DATASETCONFIG_BY_KEY,
     DATASETS,
     V1_URL_PREFIX,
     YAML_DATASETS,
@@ -2261,8 +2261,8 @@ def get_dataset_config_url(
     connection_config: Optional[ConnectionConfig] = None,
     dataset_config: Optional[DatasetConfig] = None,
 ) -> str:
-    """Helper to construct the DATASETCONFIG_BY_KEY URL, substituting valid/invalid keys in the path"""
-    path = V1_URL_PREFIX + DATASETCONFIG_BY_KEY
+    """Helper to construct the DATASET_CONFIG_BY_KEY URL, substituting valid/invalid keys in the path"""
+    path = V1_URL_PREFIX + DATASET_CONFIG_BY_KEY
     connection_key = "nonexistent_key"
     if connection_config:
         connection_key = connection_config.key
diff --git a/tests/ops/api/v1/endpoints/test_dataset_test_endpoints.py b/tests/ops/api/v1/endpoints/test_dataset_test_endpoints.py
new file mode 100644
index 0000000000..c5056aec61
--- /dev/null
+++ b/tests/ops/api/v1/endpoints/test_dataset_test_endpoints.py
@@ -0,0 +1,329 @@
+import pytest
+from starlette.status import HTTP_200_OK, HTTP_400_BAD_REQUEST, HTTP_403_FORBIDDEN
+from starlette.testclient import TestClient
+
+from fides.api.db.seed import get_client_id, load_default_access_policy
+from fides.api.util.data_category import get_user_data_categories
+from fides.common.api.scope_registry import (
+    DATASET_CREATE_OR_UPDATE,
+    DATASET_READ,
+    DATASET_TEST,
+)
+from tests.ops.api.v1.endpoints.test_dataset_endpoints import get_connection_dataset_url
+
+
+class TestDatasetInputs:
+    def test_dataset_inputs_not_authenticated(
+        self, dataset_config, connection_config, api_client
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        response = api_client.get(dataset_url + "/inputs", headers={})
+        assert response.status_code == 401
+
+    def test_dataset_inputs_wrong_scope(
+        self,
+        dataset_config,
+        connection_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = generate_auth_header(scopes=[DATASET_CREATE_OR_UPDATE])
+        response = api_client.get(dataset_url + "/inputs", headers=auth_header)
+        assert response.status_code == 403
+
+    @pytest.mark.parametrize(
+        "auth_header,expected_status",
+        [
+            ("owner_auth_header", HTTP_200_OK),
+            ("contributor_auth_header", HTTP_200_OK),
+            ("viewer_and_approver_auth_header", HTTP_200_OK),
+            ("viewer_auth_header", HTTP_200_OK),
+            ("approver_auth_header", HTTP_403_FORBIDDEN),
+        ],
+    )
+    def test_dataset_inputs_with_roles(
+        self,
+        dataset_config,
+        connection_config,
+        auth_header,
+        expected_status,
+        test_client: TestClient,
+        request,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = request.getfixturevalue(auth_header)
+        response = test_client.get(
+            dataset_url + "/inputs",
+            headers=auth_header,
+        )
+        assert response.status_code == expected_status
+
+    def test_dataset_inputs_connection_does_not_exist(
+        self,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(None, None)
+        auth_header = generate_auth_header(scopes=[DATASET_READ])
+        response = api_client.get(
+            dataset_url + "/inputs",
+            headers=auth_header,
+        )
+        assert response.status_code == 404
+
+    def test_dataset_inputs_dataset_does_not_exist(
+        self,
+        connection_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, None)
+        auth_header = generate_auth_header(scopes=[DATASET_READ])
+        response = api_client.get(
+            dataset_url + "/inputs",
+            headers=auth_header,
+        )
+        assert response.status_code == 404
+
+    def test_dataset_inputs(
+        self,
+        connection_config,
+        dataset_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = generate_auth_header(scopes=[DATASET_READ])
+        response = api_client.get(
+            dataset_url + "/inputs",
+            headers=auth_header,
+        )
+        assert response.status_code == HTTP_200_OK
+        assert response.json() == {"email": None}
+
+
+class TestDatasetReachability:
+    def test_dataset_reachability_not_authenticated(
+        self, dataset_config, connection_config, api_client
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        response = api_client.get(dataset_url + "/reachability", headers={})
+        assert response.status_code == 401
+
+    def test_dataset_test_wrong_scope(
+        self,
+        dataset_config,
+        connection_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = generate_auth_header(scopes=[DATASET_CREATE_OR_UPDATE])
+        response = api_client.get(dataset_url + "/reachability", headers=auth_header)
+        assert response.status_code == 403
+
+    @pytest.mark.parametrize(
+        "auth_header,expected_status",
+        [
+            ("owner_auth_header", HTTP_200_OK),
+            ("contributor_auth_header", HTTP_200_OK),
+            ("viewer_and_approver_auth_header", HTTP_200_OK),
+            ("viewer_auth_header", HTTP_200_OK),
+            ("approver_auth_header", HTTP_403_FORBIDDEN),
+        ],
+    )
+    def test_dataset_reachability_with_roles(
+        self,
+        dataset_config,
+        connection_config,
+        auth_header,
+        expected_status,
+        test_client: TestClient,
+        request,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = request.getfixturevalue(auth_header)
+        response = test_client.get(
+            dataset_url + "/reachability",
+            headers=auth_header,
+        )
+        assert response.status_code == expected_status
+
+    def test_dataset_reachability_connection_does_not_exist(
+        self,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(None, None)
+        auth_header = generate_auth_header(scopes=[DATASET_READ])
+        response = api_client.get(
+            dataset_url + "/reachability",
+            headers=auth_header,
+        )
+        assert response.status_code == 404
+
+    def test_dataset_reachability_dataset_does_not_exist(
+        self,
+        connection_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, None)
+        auth_header = generate_auth_header(scopes=[DATASET_READ])
+        response = api_client.get(
+            dataset_url + "/reachability",
+            headers=auth_header,
+        )
+        assert response.status_code == 404
+
+    def test_dataset_reachability(
+        self,
+        connection_config,
+        dataset_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = generate_auth_header(scopes=[DATASET_READ])
+        response = api_client.get(
+            dataset_url + "/reachability",
+            headers=auth_header,
+        )
+        assert response.status_code == HTTP_200_OK
+        assert set(response.json().keys()) == {"reachable", "details"}
+
+
+@pytest.mark.integration_external
+@pytest.mark.integration_postgres
+class TestDatasetTest:
+    @pytest.fixture(scope="function")
+    def default_access_policy(self, db) -> None:
+        load_default_access_policy(db, get_client_id(db), get_user_data_categories())
+
+    def test_dataset_test_not_authenticated(
+        self, dataset_config, connection_config, api_client
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        response = api_client.post(dataset_url + "/test", headers={})
+        assert response.status_code == 401
+
+    def test_dataset_test_wrong_scope(
+        self,
+        dataset_config,
+        connection_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = generate_auth_header(scopes=[DATASET_CREATE_OR_UPDATE])
+        response = api_client.post(dataset_url + "/test", headers=auth_header)
+        assert response.status_code == 403
+
+    @pytest.mark.usefixtures("default_access_policy")
+    @pytest.mark.parametrize(
+        "auth_header,expected_status",
+        [
+            ("owner_auth_header", HTTP_200_OK),
+            ("contributor_auth_header", HTTP_200_OK),
+            ("viewer_and_approver_auth_header", HTTP_403_FORBIDDEN),
+            ("viewer_auth_header", HTTP_403_FORBIDDEN),
+            ("approver_auth_header", HTTP_403_FORBIDDEN),
+        ],
+    )
+    def test_dataset_test_with_roles(
+        self,
+        dataset_config,
+        connection_config,
+        auth_header,
+        expected_status,
+        test_client: TestClient,
+        request,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = request.getfixturevalue(auth_header)
+        response = test_client.post(
+            dataset_url + "/test",
+            headers=auth_header,
+            json={"email": "user@example.com"},
+        )
+        assert response.status_code == expected_status
+
+    def test_dataset_test_connection_does_not_exist(
+        self,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(None, None)
+        auth_header = generate_auth_header(scopes=[DATASET_TEST])
+        response = api_client.post(
+            dataset_url + "/test",
+            headers=auth_header,
+            json={"email": "user@example.com"},
+        )
+        assert response.status_code == 404
+
+    def test_dataset_test_dataset_does_not_exist(
+        self,
+        connection_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, None)
+        auth_header = generate_auth_header(scopes=[DATASET_TEST])
+        response = api_client.post(
+            dataset_url + "/test",
+            headers=auth_header,
+            json={"email": "user@example.com"},
+        )
+        assert response.status_code == 404
+
+    @pytest.mark.parametrize(
+        "payload, expected_response",
+        [
+            ("user@example.com", "Inputs must be JSON formatted"),
+            ({}, "No inputs provided"),
+            ({"loyalty_id": None}, 'Input "loyalty_id" cannot be empty'),
+            (
+                {"email": "user"},
+                '"email" value is not a valid email address: An email address must have an @-sign.',
+            ),
+        ],
+    )
+    @pytest.mark.usefixtures("default_access_policy")
+    def test_dataset_test_invalid_payloads(
+        self,
+        connection_config,
+        dataset_config,
+        api_client: TestClient,
+        generate_auth_header,
+        payload,
+        expected_response,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = generate_auth_header(scopes=[DATASET_TEST])
+        response = api_client.post(
+            dataset_url + "/test",
+            headers=auth_header,
+            json=payload,
+        )
+        assert response.status_code == HTTP_400_BAD_REQUEST
+        assert response.json()["detail"] == expected_response
+
+    @pytest.mark.usefixtures("default_access_policy", "postgres_integration_db")
+    def test_dataset_test(
+        self,
+        connection_config,
+        dataset_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = generate_auth_header(scopes=[DATASET_TEST])
+        response = api_client.post(
+            dataset_url + "/test",
+            headers=auth_header,
+            json={"email": "jane@example.com"},
+        )
+        assert response.status_code == HTTP_200_OK
+        assert "privacy_request_id" in response.json().keys()
diff --git a/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py b/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py
index fd7da45594..5ddbcd8f60 100644
--- a/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py
+++ b/tests/ops/api/v1/endpoints/test_privacy_request_endpoints.py
@@ -12,6 +12,7 @@
 from dateutil.parser import parse
 from fastapi import HTTPException, status
 from fastapi_pagination import Params
+from starlette.status import HTTP_200_OK, HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
 from starlette.testclient import TestClient
 
 from fides.api.api.v1.endpoints.privacy_request_endpoints import (
@@ -23,6 +24,7 @@
     JWE_PAYLOAD_CLIENT_ID,
     JWE_PAYLOAD_ROLES,
 )
+from fides.api.db.seed import get_client_id, load_default_access_policy
 from fides.api.graph.config import CollectionAddress
 from fides.api.graph.graph import DatasetGraph
 from fides.api.models.application_config import ApplicationConfig
@@ -58,6 +60,7 @@
 from fides.api.task.graph_runners import access_runner
 from fides.api.tasks import MESSAGING_QUEUE_NAME
 from fides.api.util.cache import get_encryption_cache_key, get_masking_secret_cache_key
+from fides.api.util.data_category import get_user_data_categories
 from fides.api.util.fuzzy_search_utils import (
     get_should_refresh_automaton,
     manually_reset_automaton,
@@ -65,12 +68,14 @@
 )
 from fides.common.api.scope_registry import (
     DATASET_CREATE_OR_UPDATE,
+    DATASET_TEST,
     PRIVACY_REQUEST_CALLBACK_RESUME,
     PRIVACY_REQUEST_CREATE,
     PRIVACY_REQUEST_DELETE,
     PRIVACY_REQUEST_NOTIFICATIONS_CREATE_OR_UPDATE,
     PRIVACY_REQUEST_NOTIFICATIONS_READ,
     PRIVACY_REQUEST_READ,
+    PRIVACY_REQUEST_READ_ACCESS_RESULTS,
     PRIVACY_REQUEST_REVIEW,
     PRIVACY_REQUEST_TRANSFER,
     PRIVACY_REQUEST_UPLOAD_DATA,
@@ -85,6 +90,7 @@
     PRIVACY_REQUEST_BULK_RETRY,
     PRIVACY_REQUEST_BULK_SOFT_DELETE,
     PRIVACY_REQUEST_DENY,
+    PRIVACY_REQUEST_FILTERED_RESULTS,
     PRIVACY_REQUEST_MANUAL_WEBHOOK_ACCESS_INPUT,
     PRIVACY_REQUEST_MANUAL_WEBHOOK_ERASURE_INPUT,
     PRIVACY_REQUEST_NOTIFICATIONS,
@@ -106,6 +112,7 @@
 )
 from fides.config import CONFIG
 from tests.conftest import generate_role_header_for_user
+from tests.ops.api.v1.endpoints.test_dataset_endpoints import get_connection_dataset_url
 
 page_size = Params().size
 
@@ -8355,3 +8362,100 @@ def test_get_access_results_contributor_but_disabled(
         )
         response = api_client.get(url, headers=auth_header)
         assert response.status_code == 403
+
+
+@pytest.mark.integration_external
+@pytest.mark.integration_postgres
+class TestPrivacyRequestFilteredResults:
+    @pytest.fixture(scope="function")
+    def default_access_policy(self, db) -> None:
+        load_default_access_policy(db, get_client_id(db), get_user_data_categories())
+
+    def test_filtered_results_not_authenticated(
+        self, privacy_request, api_client
+    ) -> None:
+        url = V1_URL_PREFIX + PRIVACY_REQUEST_FILTERED_RESULTS.format(
+            privacy_request_id=privacy_request.id
+        )
+        response = api_client.get(url, headers={})
+        assert response.status_code == 401
+
+    def test_filtered_results_wrong_scope(
+        self,
+        privacy_request,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        url = V1_URL_PREFIX + PRIVACY_REQUEST_FILTERED_RESULTS.format(
+            privacy_request_id=privacy_request.id
+        )
+        auth_header = generate_auth_header(scopes=[DATASET_CREATE_OR_UPDATE])
+        response = api_client.get(url, headers=auth_header)
+        assert response.status_code == 403
+
+    @pytest.mark.usefixtures("default_access_policy")
+    @pytest.mark.parametrize(
+        "auth_header,expected_status",
+        [
+            ("owner_auth_header", HTTP_200_OK),
+            ("contributor_auth_header", HTTP_200_OK),
+            ("viewer_and_approver_auth_header", HTTP_403_FORBIDDEN),
+            ("viewer_auth_header", HTTP_403_FORBIDDEN),
+            ("approver_auth_header", HTTP_403_FORBIDDEN),
+        ],
+    )
+    def test_filtered_results_with_roles(
+        self,
+        db,
+        privacy_request,
+        auth_header,
+        expected_status,
+        test_client: TestClient,
+        request,
+    ) -> None:
+        # this endpoint is only for test privacy requests
+        privacy_request.source = PrivacyRequestSource.dataset_test
+        db.commit()
+
+        url = V1_URL_PREFIX + PRIVACY_REQUEST_FILTERED_RESULTS.format(
+            privacy_request_id=privacy_request.id
+        )
+        auth_header = request.getfixturevalue(auth_header)
+        response = test_client.get(
+            url,
+            headers=auth_header,
+        )
+        assert response.status_code == expected_status
+
+    @pytest.mark.usefixtures("default_access_policy", "postgres_integration_db")
+    def test_filtered_results(
+        self,
+        connection_config,
+        dataset_config,
+        api_client: TestClient,
+        generate_auth_header,
+    ) -> None:
+        dataset_url = get_connection_dataset_url(connection_config, dataset_config)
+        auth_header = generate_auth_header(scopes=[DATASET_TEST])
+        response = api_client.post(
+            dataset_url + "/test",
+            headers=auth_header,
+            json={"email": "jane@example.com"},
+        )
+        assert response.status_code == HTTP_200_OK
+
+        privacy_request_id = response.json()["privacy_request_id"]
+        url = V1_URL_PREFIX + PRIVACY_REQUEST_FILTERED_RESULTS.format(
+            privacy_request_id=privacy_request_id
+        )
+        auth_header = generate_auth_header(scopes=[PRIVACY_REQUEST_READ_ACCESS_RESULTS])
+        response = api_client.get(
+            url,
+            headers=auth_header,
+        )
+        assert response.status_code == HTTP_200_OK
+        assert set(response.json().keys()) == {
+            "privacy_request_id",
+            "status",
+            "results",
+        }
diff --git a/tests/ops/service/dataset/conftest.py b/tests/ops/service/dataset/conftest.py
new file mode 100644
index 0000000000..d2f2cd0a0f
--- /dev/null
+++ b/tests/ops/service/dataset/conftest.py
@@ -0,0 +1,153 @@
+import os
+from typing import Any, Dict, Generator
+
+import pytest
+from sqlalchemy.orm import Session
+
+from fides.api.models.connectionconfig import ConnectionConfig
+from fides.api.models.datasetconfig import DatasetConfig
+from fides.api.models.sql_models import Dataset as CtlDataset
+from tests.fixtures.application_fixtures import load_dataset
+
+
+def load_example_dataset(filename: str) -> Dict[str, Any]:
+    """Helper function specifically for loading example dataset files"""
+
+    test_dir = os.path.dirname(__file__)
+    file_path = os.path.join(test_dir, "example_datasets", filename)
+
+    if not os.path.exists(file_path):
+        raise FileNotFoundError(
+            f"Example dataset file not found: {file_path}. "
+            f"Make sure the file exists in {os.path.join(test_dir, 'example_datasets')}"
+        )
+
+    return load_dataset(file_path)
+
+
+@pytest.fixture
+def single_identity_dataset_config(
+    db: Session,
+    connection_config: ConnectionConfig,
+) -> Generator:
+    dataset = load_example_dataset("single_identity.yml")[0]
+    ctl_dataset = CtlDataset.create_from_dataset_dict(db, dataset)
+
+    dataset_config = DatasetConfig.create(
+        db=db,
+        data={
+            "connection_config_id": connection_config.id,
+            "fides_key": dataset["fides_key"],
+            "ctl_dataset_id": ctl_dataset.id,
+        },
+    )
+    yield dataset_config
+    dataset_config.delete(db=db)
+    ctl_dataset.delete(db=db)
+
+
+@pytest.fixture
+def single_identity_with_internal_dependency_dataset_config(
+    db: Session,
+    connection_config: ConnectionConfig,
+) -> Generator:
+    dataset = load_example_dataset("single_identity_with_internal_dependency.yml")[0]
+    ctl_dataset = CtlDataset.create_from_dataset_dict(db, dataset)
+
+    dataset_config = DatasetConfig.create(
+        db=db,
+        data={
+            "connection_config_id": connection_config.id,
+            "fides_key": dataset["fides_key"],
+            "ctl_dataset_id": ctl_dataset.id,
+        },
+    )
+    yield dataset_config
+    dataset_config.delete(db=db)
+    ctl_dataset.delete(db=db)
+
+
+@pytest.fixture
+def multiple_identities_dataset_config(
+    db: Session,
+    connection_config: ConnectionConfig,
+) -> Generator:
+    dataset = load_example_dataset("multiple_identities.yml")[0]
+    ctl_dataset = CtlDataset.create_from_dataset_dict(db, dataset)
+
+    dataset_config = DatasetConfig.create(
+        db=db,
+        data={
+            "connection_config_id": connection_config.id,
+            "fides_key": dataset["fides_key"],
+            "ctl_dataset_id": ctl_dataset.id,
+        },
+    )
+    yield dataset_config
+    dataset_config.delete(db=db)
+    ctl_dataset.delete(db=db)
+
+
+@pytest.fixture
+def multiple_identities_with_external_dependencies_dataset_config(
+    db: Session,
+    connection_config: ConnectionConfig,
+) -> Generator:
+    dataset = load_example_dataset("multiple_identities_with_external_dependency.yml")[
+        0
+    ]
+    ctl_dataset = CtlDataset.create_from_dataset_dict(db, dataset)
+
+    dataset_config = DatasetConfig.create(
+        db=db,
+        data={
+            "connection_config_id": connection_config.id,
+            "fides_key": dataset["fides_key"],
+            "ctl_dataset_id": ctl_dataset.id,
+        },
+    )
+    yield dataset_config
+    dataset_config.delete(db=db)
+    ctl_dataset.delete(db=db)
+
+
+@pytest.fixture
+def optional_identities_dataset_config(
+    db: Session,
+    connection_config: ConnectionConfig,
+) -> Generator:
+    dataset = load_example_dataset("optional_identities.yml")[0]
+    ctl_dataset = CtlDataset.create_from_dataset_dict(db, dataset)
+
+    dataset_config = DatasetConfig.create(
+        db=db,
+        data={
+            "connection_config_id": connection_config.id,
+            "fides_key": dataset["fides_key"],
+            "ctl_dataset_id": ctl_dataset.id,
+        },
+    )
+    yield dataset_config
+    dataset_config.delete(db=db)
+    ctl_dataset.delete(db=db)
+
+
+@pytest.fixture
+def no_identities_dataset_config(
+    db: Session,
+    connection_config: ConnectionConfig,
+) -> Generator:
+    dataset = load_example_dataset("no_identities.yml")[0]
+    ctl_dataset = CtlDataset.create_from_dataset_dict(db, dataset)
+
+    dataset_config = DatasetConfig.create(
+        db=db,
+        data={
+            "connection_config_id": connection_config.id,
+            "fides_key": dataset["fides_key"],
+            "ctl_dataset_id": ctl_dataset.id,
+        },
+    )
+    yield dataset_config
+    dataset_config.delete(db=db)
+    ctl_dataset.delete(db=db)
diff --git a/tests/ops/service/dataset/example_datasets/multiple_identities.yml b/tests/ops/service/dataset/example_datasets/multiple_identities.yml
new file mode 100644
index 0000000000..053afb3ced
--- /dev/null
+++ b/tests/ops/service/dataset/example_datasets/multiple_identities.yml
@@ -0,0 +1,43 @@
+dataset:
+  - fides_key: multiple_identities
+    name: Multiple identities
+    description: A dataset with two collections, each with their own identity dependency
+    collections:
+      - name: customer
+        fields:
+          - name: address_id
+            data_categories: [system.operations]
+          - name: created
+            data_categories: [system.operations]
+          - name: email
+            data_categories: [user.contact.email]
+            fides_meta:
+              identity: email
+              data_type: string
+          - name: id
+            data_categories: [user.unique_id]
+            fides_meta:
+              primary_key: True
+          - name: name
+            data_categories: [user.name]
+            fides_meta:
+              data_type: string
+              length: 40
+      - name: loyalty
+        fields:
+          - name: id
+            data_categories: [user.unique_id]
+            fides_meta:
+              identity: loyalty_id
+          - name: name
+            data_categories: [user.name]
+            fides_meta:
+              data_type: string
+          - name: points
+            data_categories: [user.content]
+            fides_meta:
+              data_type: integer
+          - name: tier
+            data_categories: [system.operations]
+            fides_meta:
+              data_type: string
diff --git a/tests/ops/service/dataset/example_datasets/multiple_identities_with_external_dependency.yml b/tests/ops/service/dataset/example_datasets/multiple_identities_with_external_dependency.yml
new file mode 100644
index 0000000000..fdfcd32bfc
--- /dev/null
+++ b/tests/ops/service/dataset/example_datasets/multiple_identities_with_external_dependency.yml
@@ -0,0 +1,38 @@
+dataset:
+  - fides_key: multiple_identities_with_external_dependency
+    name: Multiple identities with external dependency
+    description: A dataset with one collection that relies on an identity, and another collection that relies on an external dataset reference tied to a different identity
+    collections:
+      - name: loyalty
+        fields:
+          - name: id
+            data_categories: [user.unique_id]
+            fides_meta:
+              identity: loyalty_id
+          - name: name
+            data_categories: [user.name]
+            fides_meta:
+              data_type: string
+          - name: points
+            data_categories: [user.content]
+            fides_meta:
+              data_type: integer
+          - name: tier
+            data_categories: [system.operations]
+            fides_meta:
+              data_type: string
+      - name: orders
+        fields:
+          - name: customer_id
+            data_categories: [user.unique_id]
+            fides_meta:
+              references:
+                - dataset: single_identity
+                  field: customer.id
+                  direction: from
+          - name: id
+            data_categories: [system.operations]
+            fides_meta:
+              primary_key: True
+          - name: shipping_address_id
+            data_categories: [system.operations]
diff --git a/tests/ops/service/dataset/example_datasets/no_identities.yml b/tests/ops/service/dataset/example_datasets/no_identities.yml
new file mode 100644
index 0000000000..fac879de99
--- /dev/null
+++ b/tests/ops/service/dataset/example_datasets/no_identities.yml
@@ -0,0 +1,22 @@
+dataset:
+  - fides_key: single_identity
+    name: Single identity
+    description: A dataset with a single collection that relies on an identity
+    collections:
+      - name: customer
+        fields:
+          - name: address_id
+            data_categories: [system.operations]
+          - name: created
+            data_categories: [system.operations]
+          - name: email
+            data_categories: [user.contact.email]
+          - name: id
+            data_categories: [user.unique_id]
+            fides_meta:
+              primary_key: True
+          - name: name
+            data_categories: [user.name]
+            fides_meta:
+              data_type: string
+              length: 40
diff --git a/tests/ops/service/dataset/example_datasets/optional_identities.yml b/tests/ops/service/dataset/example_datasets/optional_identities.yml
new file mode 100644
index 0000000000..867a8aef68
--- /dev/null
+++ b/tests/ops/service/dataset/example_datasets/optional_identities.yml
@@ -0,0 +1,26 @@
+dataset:
+  - fides_key: optional_identities
+    name: Optional identities
+    description: A dataset with a single collection that relies on either identity
+    collections:
+      - name: customer
+        fields:
+          - name: address_id
+            data_categories: [system.operations]
+          - name: created
+            data_categories: [system.operations]
+          - name: email
+            data_categories: [user.contact.email]
+            fides_meta:
+              identity: email
+              data_type: string
+          - name: id
+            data_categories: [user.unique_id]
+            fides_meta:
+              identity: user_id
+              data_type: string
+          - name: name
+            data_categories: [user.name]
+            fides_meta:
+              data_type: string
+              length: 40
diff --git a/tests/ops/service/dataset/example_datasets/single_identity.yml b/tests/ops/service/dataset/example_datasets/single_identity.yml
new file mode 100644
index 0000000000..19cdc7df3e
--- /dev/null
+++ b/tests/ops/service/dataset/example_datasets/single_identity.yml
@@ -0,0 +1,25 @@
+dataset:
+  - fides_key: single_identity
+    name: Single identity
+    description: A dataset with a single collection that relies on an identity
+    collections:
+      - name: customer
+        fields:
+          - name: address_id
+            data_categories: [system.operations]
+          - name: created
+            data_categories: [system.operations]
+          - name: email
+            data_categories: [user.contact.email]
+            fides_meta:
+              identity: email
+              data_type: string
+          - name: id
+            data_categories: [user.unique_id]
+            fides_meta:
+              primary_key: True
+          - name: name
+            data_categories: [user.name]
+            fides_meta:
+              data_type: string
+              length: 40
diff --git a/tests/ops/service/dataset/example_datasets/single_identity_with_internal_dependency.yml b/tests/ops/service/dataset/example_datasets/single_identity_with_internal_dependency.yml
new file mode 100644
index 0000000000..708aefbaf0
--- /dev/null
+++ b/tests/ops/service/dataset/example_datasets/single_identity_with_internal_dependency.yml
@@ -0,0 +1,44 @@
+dataset:
+  - fides_key: single_identity_with_internal_dependency
+    name: Single identity with internal dependency
+    description: A dataset with a collection that relies on an identity, and another collection with an internal dependency on the first collection
+    collections:
+      - name: customer
+        fields:
+          - name: address_id
+            data_categories: [system.operations]
+          - name: created
+            data_categories: [system.operations]
+          - name: email
+            data_categories: [user.contact.email]
+            fides_meta:
+              identity: email
+              data_type: string
+          - name: id
+            data_categories: [user.unique_id]
+            fides_meta:
+              primary_key: True
+          - name: name
+            data_categories: [user.name]
+            fides_meta:
+              data_type: string
+              length: 40
+      - name: address
+        fields:
+          - name: city
+            data_categories: [user.contact.address.city]
+          - name: house
+            data_categories: [user.contact.address.street]
+          - name: id
+            data_categories: [system.operations]
+            fides_meta:
+              references:
+                - dataset: single_identity_with_internal_dependency
+                  field: customer.address_id
+                  direction: from
+          - name: state
+            data_categories: [user.contact.address.state]
+          - name: street
+            data_categories: [user.contact.address.street]
+          - name: zip
+            data_categories: [user.contact.address.postal_code]
diff --git a/tests/ops/service/dataset/test_dataset_service.py b/tests/ops/service/dataset/test_dataset_service.py
new file mode 100644
index 0000000000..c1e56f8d01
--- /dev/null
+++ b/tests/ops/service/dataset/test_dataset_service.py
@@ -0,0 +1,234 @@
+import datetime
+from typing import List, Set
+
+import pytest
+from pytest import FixtureRequest
+from sqlalchemy.orm import Session
+
+from fides.api.models.connectionconfig import ConnectionConfig
+from fides.api.models.datasetconfig import DatasetConfig
+from fides.api.models.policy import Policy
+from fides.api.schemas.policy import ActionType
+from fides.api.service.dataset.dataset_service import (
+    get_dataset_reachability,
+    get_identities_and_references,
+    run_test_access_request,
+)
+from tests.conftest import wait_for_tasks_to_complete
+
+
+class TestGetDatasetReachability:
+
+    @pytest.mark.parametrize(
+        "dataset_config, expected",
+        [
+            ("single_identity_dataset_config", (True, None)),
+            (
+                "single_identity_with_internal_dependency_dataset_config",
+                (True, None),
+            ),
+            (
+                "multiple_identities_dataset_config",
+                (True, None),
+            ),
+            (
+                "multiple_identities_with_external_dependencies_dataset_config",
+                (
+                    False,
+                    'The following dataset references do not exist "single_identity:customer:id"',
+                ),
+            ),
+            (
+                "optional_identities_dataset_config",
+                (True, None),
+            ),
+            (
+                "no_identities_dataset_config",
+                (
+                    False,
+                    'The following collections are not reachable "single_identity:customer"',
+                ),
+            ),
+        ],
+    )
+    def test_get_dataset_reachability(
+        self,
+        db: Session,
+        dataset_config: List[str],
+        expected: bool,
+        connection_config: ConnectionConfig,
+        request: pytest.FixtureRequest,
+    ):
+        request.getfixturevalue(dataset_config)
+
+        reachable = get_dataset_reachability(
+            db,
+            connection_config.datasets[0],
+        )
+        assert reachable == expected
+
+
+class TestGetIdentitiesAndReferences:
+
+    @pytest.mark.parametrize(
+        "dataset_config, expected_required_identities",
+        [
+            ("single_identity_dataset_config", {"email"}),
+            (
+                "single_identity_with_internal_dependency_dataset_config",
+                {"email"},
+            ),
+            (
+                "multiple_identities_dataset_config",
+                {"email", "loyalty_id"},
+            ),
+            (
+                "multiple_identities_with_external_dependencies_dataset_config",
+                {"loyalty_id", "single_identity:customer:id"},
+            ),
+            (
+                "optional_identities_dataset_config",
+                {"email", "user_id"},
+            ),
+            (
+                "no_identities_dataset_config",
+                set(),
+            ),
+        ],
+    )
+    def test_get_identities_and_references(
+        self,
+        dataset_config: List[str],
+        expected_required_identities: Set[str],
+        connection_config: ConnectionConfig,
+        request: FixtureRequest,
+    ):
+        request.getfixturevalue(dataset_config)
+
+        required_identities = get_identities_and_references(
+            connection_config.datasets[0],
+        )
+        assert required_identities == expected_required_identities
+
+
+@pytest.mark.integration_external
+@pytest.mark.integration_postgres
+class TestRunTestAccessRequest:
+    @pytest.mark.usefixtures("postgres_integration_db")
+    def test_run_test_access_request(
+        self,
+        db: Session,
+        policy: Policy,
+        postgres_example_test_dataset_config: DatasetConfig,
+    ):
+        dataset_config = postgres_example_test_dataset_config
+        privacy_request = run_test_access_request(
+            db,
+            policy,
+            dataset_config,
+            {"email": "jane@example.com"},
+        )
+        wait_for_tasks_to_complete(db, privacy_request, ActionType.access)
+        assert privacy_request.get_raw_access_results() == {
+            "postgres_example_test_dataset:customer": [
+                {
+                    "address_id": 4,
+                    "created": datetime.datetime(2020, 4, 1, 11, 47, 42),
+                    "email": "jane@example.com",
+                    "id": 3,
+                    "name": "Jane Customer",
+                }
+            ],
+            "postgres_example_test_dataset:employee": [],
+            "postgres_example_test_dataset:report": [],
+            "postgres_example_test_dataset:visit": [],
+            "postgres_example_test_dataset:orders": [
+                {"customer_id": 3, "id": "ord_ddd-eee", "shipping_address_id": 4}
+            ],
+            "postgres_example_test_dataset:login": [
+                {"customer_id": 3, "id": 8, "time": datetime.datetime(2021, 1, 6, 1, 0)}
+            ],
+            "postgres_example_test_dataset:payment_card": [
+                {
+                    "billing_address_id": 4,
+                    "ccn": 373719391,
+                    "code": 222,
+                    "customer_id": 3,
+                    "id": "pay_ccc-ccc",
+                    "name": "Example Card 3",
+                    "preferred": False,
+                }
+            ],
+            "postgres_example_test_dataset:service_request": [],
+            "postgres_example_test_dataset:order_item": [],
+            "postgres_example_test_dataset:address": [
+                {
+                    "city": "Example Mountain",
+                    "house": 1111,
+                    "id": 4,
+                    "state": "TX",
+                    "street": "Example Place",
+                    "zip": "54321",
+                }
+            ],
+            "postgres_example_test_dataset:product": [],
+        }
+
+    @pytest.mark.usefixtures("postgres_integration_db")
+    def test_run_sample_access_request_with_custom_id(
+        self,
+        db: Session,
+        policy: Policy,
+        postgres_example_test_extended_dataset_config: DatasetConfig,
+    ):
+        dataset_config = postgres_example_test_extended_dataset_config
+        privacy_request = run_test_access_request(
+            db,
+            policy,
+            dataset_config,
+            {"loyalty_id": "CH-1"},
+        )
+        wait_for_tasks_to_complete(db, privacy_request, ActionType.access)
+        assert privacy_request.get_raw_access_results() == {
+            "postgres_example_test_extended_dataset:loyalty": [
+                {
+                    "id": "CH-1",
+                    "name": "Jane Customer",
+                    "points": 100,
+                    "tier": "Cookie Rookie",
+                }
+            ]
+        }
+
+    @pytest.mark.usefixtures(
+        "postgres_integration_db",
+    )
+    def test_run_test_access_request_with_dataset_reference(
+        self,
+        db: Session,
+        policy: Policy,
+        multiple_identities_with_external_dependencies_dataset_config: DatasetConfig,
+    ):
+        dataset_config = multiple_identities_with_external_dependencies_dataset_config
+        privacy_request = run_test_access_request(
+            db,
+            policy,
+            dataset_config,
+            {"loyalty_id": "CH-1", "single_identity:customer:id": 1},
+        )
+        wait_for_tasks_to_complete(db, privacy_request, ActionType.access)
+        assert privacy_request.get_raw_access_results() == {
+            "multiple_identities_with_external_dependency:loyalty": [
+                {
+                    "id": "CH-1",
+                    "name": "Jane Customer",
+                    "points": 100,
+                    "tier": "Cookie Rookie",
+                }
+            ],
+            "multiple_identities_with_external_dependency:orders": [
+                {"customer_id": 1, "id": "ord_aaa-aaa", "shipping_address_id": 2},
+                {"customer_id": 1, "id": "ord_ccc-ccc", "shipping_address_id": 1},
+                {"customer_id": 1, "id": "ord_ddd-ddd", "shipping_address_id": 1},
+            ],
+        }