Draft: Tests for non-purge delete

Summary

Added tests for delete, removed some inter-test dependencies, small refactors.

Depends on SDK and Service changes.

Check List for the Author

• All automated tests pass
• Reference related issues
• Appropriate user and developer documentation for new code
• Annotations in code (Gitlab comments)

Check List for the Reviewer

• I understand the intent of this MR
• All automated tests pass
• Appropriate user and developer documentation for new code
• The test environment setup works and the intended behavior is reproducible in the test environment
• In-code documentation and comments are up-to-date.

manager_test/manager_fdo/test_delete_fdo.py

@@ -26,9 +26,207 @@ from ..manager_logging.test_list_log_events \
     import PATH_LOCAL as PATH_LOGGING
 
 PATH_FILE = str(Path(__file__).relative_to(root_dir))
+PATH_LOCAL_DELETE = PATH_FILE + "::TestClassDeleteFDO::"
 PATH_LOCAL_PURGE = PATH_FILE + "::TestClassPurgeFDO::"
 
 
+@pytest.mark.order(after=[
+    f"{PATH_CREATE_CHECKS}test_create_fdo_outside_systems_check",
+    f"{PATH_RESOLVE}test_http_resolve_fdo",
+    f"{PATH_RESOLVE}test_client_resolve_fdo",
+    f"{PATH_LOGGING}test_http_log_createfdo"])
+class TestClassDeleteFDO:                           # pylint: disable=fixme
+    """Basic Tests for /fdo delete calls with purge=false"""
+    # ToDo: Cleanup dependencies - some might be redundant now that FDOs are
+    #       not shared
+
+    @pytest.mark.parametrize("service_url", [pytest.param(service["url"], marks=[
+        pytest.mark.dependency(scope='session', depends=[
+            f'{PATH_LISTREPO}test_http_repositories[{service["url"]}]'])])
+        for service in services])
+    def test_http_delete_fdo(self, service_url: str):
+        """Tests delete of a fdo with a delete request to /fdo/... with
+        purge=false.
+
+        Parameters
+        ----------
+        service_url : str
+                      URL of the service to be tested without trailing slash,
+                      for example "https://manager.testbed.pid.gwdg.de/api/v1"
+        """
+        urls = []
+        # delete a fdo in every available repository
+        try:
+            response = requests.get(service_url + "/repositories", timeout=10)
+            repositories = json.loads(response.content.decode())['data']
+            if len(repositories) == 0:
+                pytest.skip("Cannot delete an FDO without repository.")
+            for repo in repositories:
+                if repo['id'] == "Linkahead":     # ToDo: Remove when implemented
+                    pytest.skip(
+                        'Linkahead does not yet support deletion of FDOs.')
+                # Create FDO
+                url_md, _ = man_util.create_fdo(service_url, repo['id'])
+                urls.append(url_md)
+                data_md, metadata_md = man_util.get_data_metadata(url_md)
+                # Test
+                query_time = datetime.datetime.now(datetime.timezone.utc)
+                response = requests.delete(f"{url_md}?delete_MD=true",
+                                           timeout=10)
+                return_time = datetime.datetime.now(datetime.timezone.utc)
+                self.assertions_helper(response, query_time, return_time,
+                                       url_md, data_md, metadata_md, True)
+                # Create FDO
+                url_no_md, _ = man_util.create_fdo(service_url, repo['id'])
+                urls.append(url_no_md)
+                data_no_md, metadata_no_md = man_util.get_data_metadata(
+                    url_no_md)
+                # Test
+                query_time = datetime.datetime.now(datetime.timezone.utc)
+                response = requests.delete(f"{url_no_md}", timeout=10)
+                return_time = datetime.datetime.now(datetime.timezone.utc)
+                self.assertions_helper(response, query_time, return_time,
+                                       url_no_md, data_no_md, metadata_no_md)
+        finally:
+            for url in urls:
+                man_util.purge_fdo(url)
+
+    @pytest.mark.parametrize("service_url", [pytest.param(service["url"], marks=[
+        pytest.mark.dependency(scope='session', depends=[
+            f'{PATH_LOCAL_DELETE}test_http_delete_fdo[{service["url"]}]',
+            f'{PATH_LISTREPO}test_client_repositories[{service["url"]}]'])])
+        for service in services])
+    def test_client_delete_fdo(self, service_url: str):
+        """Tests delete of a fdo with an unauthenticated client.
+
+        Parameters
+        ----------
+        service_url : str
+                      URL of the service to be tested without trailing slash,
+                      for example "https://manager.testbed.pid.gwdg.de/api/v1"
+        """
+        urls = []
+        # delete a fdo in every available repository
+        try:
+            with Client(base_url=service_url) as client:
+                response = list_repositories.sync_detailed(client=client)
+                repositories = json.loads(response.content.decode())['data']
+                if len(repositories) == 0:
+                    pytest.skip("Cannot purge an FDO without repository.")
+                for repo in repositories:
+                    if repo['id'] == "Linkahead":     # ToDo: Remove when implemented
+                        pytest.skip(
+                            'Linkahead does not yet support deletion of FDOs.')
+                    # Create FDO
+                    url_md, _ = man_util.create_fdo(service_url, repo['id'])
+                    urls.append(url_md)
+                    data_md, metadata_md = man_util.get_data_metadata(url_md)
+                    # Test
+                    pid = url_md.split(f'{service_url}/fdo/')[1]
+                    prefix, suffix = pid.split('/')
+                    query_time = datetime.datetime.now(datetime.timezone.utc)
+                    response = base_delete.sync_detailed(client=client, prefix=prefix,
+                                                         suffix=suffix, delete_md=True)
+                    return_time = datetime.datetime.now(datetime.timezone.utc)
+                    self.assertions_helper(response, query_time, return_time,
+                                           url_md, data_md, metadata_md, True)
+                    # Create FDO
+                    url_no_md, _ = man_util.create_fdo(service_url, repo['id'])
+                    urls.append(url_no_md)
+                    data_no_md, metadata_no_md = man_util.get_data_metadata(
+                        url_no_md)
+                    # Test
+                    pid = url_no_md.split(f'{service_url}/fdo/')[1]
+                    prefix, suffix = pid.split('/')
+                    query_time = datetime.datetime.now(datetime.timezone.utc)
+                    response = base_delete.sync_detailed(client=client,
+                                                         prefix=prefix, suffix=suffix)
+                    return_time = datetime.datetime.now(datetime.timezone.utc)
+                    self.assertions_helper(response, query_time, return_time,
+                                           url_no_md, data_no_md, metadata_no_md)
+        finally:
+            for url in urls:
+                man_util.purge_fdo(url)
+
+    @pytest.mark.parametrize("service_url", [pytest.param(service["url"], marks=[
+        pytest.mark.dependency(scope='session', depends=[
+            f'{PATH_LOCAL_DELETE}test_client_delete_fdo[{service["url"]}]'])])
+        for service in services])
+    def test_client_delete_unknown_fdo(self, service_url: str):
+        """Tests delete of a nonexistent fdo.
+
+        Parameters
+        ----------
+        service_url : str
+                      URL of the service to be tested without trailing slash,
+                      for example "https://manager.testbed.pid.gwdg.de/api/v1"
+        """
+        with Client(base_url=service_url) as client:
+            prefix, suffix = "invalid.prefix", "ThisIsNotASuffix145670935"
+            query_time = datetime.datetime.now(datetime.timezone.utc)
+            response = base_delete.sync_detailed(client=client, prefix=prefix,
+                                                 suffix=suffix)
+            return_time = datetime.datetime.now(datetime.timezone.utc)
+            # Correct Error
+            assert response.status_code == HTTPStatus.NOT_FOUND
+            # Same procedure for delete_md=True
+            check_timeout(query_time, return_time, response.headers['date'])
+            query_time = datetime.datetime.now(datetime.timezone.utc)
+            response = base_delete.sync_detailed(client=client, prefix=prefix,
+                                                 suffix=suffix, delete_md=True)
+            return_time = datetime.datetime.now(datetime.timezone.utc)
+            assert response.status_code == HTTPStatus.NOT_FOUND
+            check_timeout(query_time, return_time, response.headers['date'])
+
+    def assertions_helper(self,
+                          response: requests.Response | client_types.Response,
+                          query_time: datetime.datetime,
+                          return_time: datetime.datetime,
+                          fdo_location: str,
+                          data_pid: str,
+                          metadata_pid: str,
+                          delete_md: bool = False):
+        """Asserts correct behaviour of /fdo purge calls.
+        This includes response code and reasonable response time.
+
+        Parameters
+        ----------
+        response     : requests.Response | fdo_[...]_client.types.Response
+                       Response from the /fdo request.
+        query_time   : datetime.datetime
+                       Time from just before the request to /fdo was sent,
+                       using datetime.datetime.now(datetime.timezone.utc)
+        return_time  : datetime.datetime
+                       Time from just after the request to /fdo was sent,
+                       using datetime.datetime.now(datetime.timezone.utc)
+        fdo_location : str
+                       Url of the deleted fdo.
+        data_pid     : str
+                       PID of the deleted data.
+        metadata_pid : str
+                       PID of the deleted metadata.
+        delete_md    : bool: optional, default False
+                       Boolean indicating whether the metadata was supposed
+                       to also be deleted.
+        """
+        # No errors
+        assert response.status_code == HTTPStatus.NO_CONTENT
+        # Answer within one second
+        check_timeout(query_time, return_time, response.headers['date'], 10)
+        # Correct changes:
+        # ToDo: As soon as access of attributes and files is possible from
+        #       the /fdo endpoint, also check content changes
+        response_fdo = requests.get(fdo_location, timeout=10)
+        assert response_fdo.status_code == HTTPStatus.OK
+        base_url = '/'.join(fdo_location.split('/')[:-2])+'/'
+        if data_pid is not None:
+            response_data = requests.get(base_url+data_pid, timeout=10)
+            assert response_data.status_code == HTTPStatus.OK
+        if metadata_pid is not None:
+            response_md = requests.get(base_url+metadata_pid, timeout=10)
+            assert response_md.status_code == HTTPStatus.OK
+
+
 @pytest.mark.order(after=[
     f"{PATH_CREATE_CHECKS}test_create_fdo_outside_systems_check",
     f"{PATH_RESOLVE}test_http_resolve_fdo",
@@ -62,12 +260,15 @@ class TestClassPurgeFDO:                           # pylint: disable=fixme
                 if repo['id'] == "Linkahead":     # ToDo: Remove when implemented
                     pytest.skip(
                         'Linkahead does not yet support deletion of FDOs.')
-                urls.append(man_util.create_fdo(service_url, repo['id'])[0])
+                url, _ = man_util.create_fdo(service_url, repo['id'])
+                urls.append(url)
+                data, metadata = man_util.get_data_metadata(url)
                 query_time = datetime.datetime.now(datetime.timezone.utc)
                 response = requests.delete(f"{urls[-1]}?purge=true",
                                            timeout=10)
                 return_time = datetime.datetime.now(datetime.timezone.utc)
-                self.assertions_helper(response, query_time, return_time)
+                self.assertions_helper(response, query_time, return_time,
+                                       url, data, metadata)
         finally:
             for url in urls:
                 man_util.purge_fdo(url)
@@ -98,15 +299,17 @@ class TestClassPurgeFDO:                           # pylint: disable=fixme
                     if repo['id'] == "Linkahead":     # ToDo: Remove when implemented
                         pytest.skip(
                             'Linkahead does not yet support deletion of FDOs.')
-                    urls.append(man_util.create_fdo(
-                        service_url, repo['id'])[0])
-                    pid = urls[-1].split(f'{service_url}/fdo/')[1]
+                    url, _ = man_util.create_fdo(service_url, repo['id'])
+                    urls.append(url)
+                    data, metadata = man_util.get_data_metadata(url)
+                    pid = url.split(f'{service_url}/fdo/')[1]
                     prefix, suffix = pid.split('/')
                     query_time = datetime.datetime.now(datetime.timezone.utc)
-                    response = base_delete.sync_detailed(client=client,
-                                                         prefix=prefix, suffix=suffix, purge=True)
+                    response = base_delete.sync_detailed(client=client, prefix=prefix,
+                                                         suffix=suffix, purge=True)
                     return_time = datetime.datetime.now(datetime.timezone.utc)
-                    self.assertions_helper(response, query_time, return_time)
+                    self.assertions_helper(response, query_time, return_time,
+                                           url, data, metadata)
         finally:
             for url in urls:
                 man_util.purge_fdo(url)
@@ -138,25 +341,234 @@ class TestClassPurgeFDO:                           # pylint: disable=fixme
     def assertions_helper(self,
                           response: requests.Response | client_types.Response,
                           query_time: datetime.datetime,
-                          return_time: datetime.datetime):
+                          return_time: datetime.datetime,
+                          fdo_location: str,
+                          data_pid: str,
+                          metadata_pid: str):
         """Asserts correct behaviour of /fdo purge calls.
         This includes response code and reasonable response time.
 
         Parameters
         ----------
-        response    : requests.Response | fdo_[...]_client.types.Response
-                      Response from the /fdo request.
-        query_time  : datetime.datetime
-                      Time from just before the request to /fdo was sent,
-                      using datetime.datetime.now(datetime.timezone.utc)
-        return_time : datetime.datetime
-                      Time from just after the request to /fdo was sent,
-                      using datetime.datetime.now(datetime.timezone.utc)
+        response     : requests.Response | fdo_[...]_client.types.Response
+                       Response from the /fdo request.
+        query_time   : datetime.datetime
+                       Time from just before the request to /fdo was sent,
+                       using datetime.datetime.now(datetime.timezone.utc)
+        return_time  : datetime.datetime
+                       Time from just after the request to /fdo was sent,
+                       using datetime.datetime.now(datetime.timezone.utc)
+        fdo_location : str
+                       Url of the deleted fdo.
+        data_pid     : str
+                       PID of the deleted data.
+        metadata_pid : str
+                       PID of the deleted metadata.
         """
         # No errors
         assert response.status_code == HTTPStatus.NO_CONTENT
         # Answer within one second
         check_timeout(query_time, return_time, response.headers['date'], 10)
+        # FDO deleted
+        response_fdo = requests.get(fdo_location, timeout=10)
+        assert response_fdo.status_code == HTTPStatus.NOT_FOUND
+        base_url = '/'.join(fdo_location.split('/')[:-2])+'/'
+        if data_pid is not None:
+            response_data = requests.get(base_url+data_pid, timeout=10)
+            assert response_data.status_code == HTTPStatus.NOT_FOUND
+        if metadata_pid is not None:
+            response_md = requests.get(base_url+metadata_pid, timeout=10)
+            assert response_md.status_code == HTTPStatus.NOT_FOUND
+
+
+@pytest.mark.order(after=["TestClassDeleteFDO"])
+class TestClassDeleteFDOChecks:                        # pylint: disable=fixme
+    """Tests whether FDO delete is reflected in outside systems"""
+
+    @pytest.mark.parametrize("service", [pytest.param(service, marks=[
+        pytest.mark.dependency(scope='session', depends=[])])
+        for service in services])
+    def test_purge_fdo_outside_systems_check(self, service: dict):
+        """Tests successful deletion of a fdo by checking directly with
+        the repo if possible.
+
+        Parameters
+        ----------
+        service:    dict
+                    Dictionary containing info of the service to be tested,
+                    including url and information about available repos.
+        """
+        urls = []
+        try:
+            any_checks_made = False
+            response = requests.get(
+                service['url'] + "/repositories", timeout=10)
+            repositories = json.loads(response.content.decode())['data']
+            for repo in repositories:
+                if repo['id'] == "Linkahead":     # ToDo: Remove when implemented
+                    # Linkahead does not yet support deletion of FDOs.
+                    continue
+                try:
+                    # Setup Data
+                    url_md, _ = man_util.create_fdo(service['url'], repo['id'])
+                    urls.append(url_md)
+                    data_md, metadata_md = man_util.get_data_metadata(url_md)
+                    url_no_md, _ = man_util.create_fdo(
+                        service['url'], repo['id'])
+                    urls.append(url_no_md)
+                    data_no_md, metadata_no_md = man_util.get_data_metadata(
+                        url_no_md)
+                    # ToDo: Save files from the metadata of the delete_MD=false
+                    #       call here to later check that they were not changed
+                    # Deletion
+                    requests.delete(f"{url_md}?delete_MD=true", timeout=10)
+                    with Client(base_url=service['url']) as client:
+                        prefix, suffix = url_no_md.split('/')[-2:]
+                        base_delete.sync_detailed(
+                            prefix, suffix, client=client)
+                except Exception as e:     # pylint: disable=broad-exception-caught
+                    pytest.skip(f"Setup unsuccessful with Exception: {str(e)}")
+                repo = service["repos"][repo['id']]
+                pid_md = url_md.split(f"{service['url']}/fdo/")[1]
+                pid_no_md = url_no_md.split(f"{service['url']}/fdo/")[1]
+                # If possible, check with repo
+                if "retrieve" in repo["links"] and repo["links"]["retrieve"]:
+                    api_config = repo["links"]["retrieve"]
+                    match api_config["api_version"]:
+                        case "Cordra-V1":
+                            cordra_url = api_config["url"]
+                            # Checks with metadata deletion
+                            self.assertions_helper_cordra(cordra_url, pid_md)
+                            any_checks_made = True
+                            if data_md is not None:
+                                self.assertions_helper_cordra(
+                                    cordra_url, data_md, is_fdo=False)
+                            if metadata_md is not None:
+                                self.assertions_helper_cordra(
+                                    cordra_url, metadata_md, is_fdo=False)
+                            # Checks without metadata deletion
+                            self.assertions_helper_cordra(
+                                cordra_url, pid_no_md)
+                            if data_no_md is not None:
+                                self.assertions_helper_cordra(
+                                    cordra_url, data_no_md, is_fdo=False)
+                            if metadata_no_md is not None:
+                                self.assertions_helper_cordra(
+                                    cordra_url, metadata_no_md, is_fdo=False, deleted=False)
+                # If possible, check with handle system
+                if "handle" in repo["links"] and repo["links"]["handle"] != "":
+                    handle_url = repo["links"]["handle"]["url"]
+                    # Checks with metadata deletion
+                    self.assertions_helper_handle(handle_url, pid_md)
+                    any_checks_made = True
+                    if data_md is not None:
+                        self.assertions_helper_handle(
+                            handle_url, data_md, is_fdo=False)
+                    if metadata_md is not None:
+                        self.assertions_helper_handle(
+                            handle_url, metadata_md, is_fdo=False)
+                    # Checks without metadata deletion
+                    self.assertions_helper_handle(handle_url, pid_no_md)
+                    if data_no_md is not None:
+                        self.assertions_helper_handle(
+                            handle_url, data_no_md, is_fdo=False)
+                    if metadata_no_md is not None:
+                        self.assertions_helper_handle(
+                            handle_url, metadata_no_md, is_fdo=False, deleted=False)
+        finally:
+            for url in urls:
+                man_util.purge_fdo(url)
+        if not any_checks_made:
+            pytest.skip('No checks with outside systems were possible.')
+
+    def assertions_helper_cordra(self,
+                                 cordra_url: str,
+                                 pid: str,
+                                 deleted: bool = True,
+                                 is_fdo: bool = True):
+        """Asserts that the Cordra response matches expected output.
+
+        Parameters
+        ----------
+        cordra_url    : str
+                        Url to the Cordra API endpoint
+        pid           : str
+                        PID of DO to be tested
+        deleted       : bool: optional, default True
+                        Boolean indicating whether the content is supposed to
+                        have been deleted.
+        is_fdo        : bool: optional, default True
+                        Boolean indicating whether to check fdo or component
+                        deletion.
+        """
+        response = requests.get(
+            cordra_url, params={"targetId": pid}, timeout=10)
+        # No errors
+        assert response.status_code == HTTPStatus.OK
+        # Correct changes
+        content = json.loads(response.content.decode())[
+            'attributes']['content']
+        if not deleted:
+            if "FDO_Status" in content:
+                assert content["FDO_Status"] != "deleted"
+                assert "URL_Status" not in content
+            if "DO_Status" in content:
+                assert content["DO_Status"] != "deleted"
+                assert "URL_Status" not in content
+            return
+        if is_fdo:
+            assert "FDO_Status" in content
+            assert content["FDO_Status"] == "deleted"
+            assert "URL_Status" not in content
+        else:
+            assert "DO_Status" in content.keys()
+            assert content["DO_Status"] == "deleted"
+            assert "URL_Status" in content
+            assert "file_size" not in content
+
+    def assertions_helper_handle(self,
+                                 handle_url: str,
+                                 pid: str,
+                                 deleted: bool = True,
+                                 is_fdo: bool = True):
+        """Asserts that the Cordra response matches expected output.
+
+        Parameters
+        ----------
+        handle_url    : str
+                        Url to the Handle API endpoint
+        pid           : str
+                        PID of DO to be tested
+        deleted       : bool: optional, default True
+                        Boolean indicating whether the content is supposed to
+                        have been deleted.
+        is_fdo        : bool: optional, default True
+                        Boolean indicating whether to check fdo or component
+                        deletion.
+        """
+        response = requests.get(f'{handle_url}{pid}', timeout=10)
+        # No errors
+        assert response.status_code == HTTPStatus.OK
+        # Correct changes
+        content = json.loads(response.content.decode())
+        attributes = {entry["type"]: entry["data"]["value"]
+                      for entry in content["values"]}
+        if not deleted:
+            if "FDO_Status" in attributes:
+                assert attributes["FDO_Status"] != "deleted"
+                assert "URL_Status" not in attributes
+            if "DO_Status" in attributes:
+                assert attributes["DO_Status"] != "deleted"
+                assert "URL_Status" not in attributes
+            return
+        if is_fdo:
+            assert "FDO_Status" in attributes
+            assert attributes["FDO_Status"] == "deleted"
+            assert "URL_Status" not in attributes
+        else:
+            assert "DO_Status" in attributes
+            assert attributes["DO_Status"] == "deleted"
+            assert "URL_Status" in attributes
 
 
 @pytest.mark.order(after=["TestClassPurgeFDO"])
@@ -196,7 +608,6 @@ class TestClassPurgeFDOChecks:                        # pylint: disable=fixme
                     man_util.purge_fdo(url)
                 except Exception as e:     # pylint: disable=broad-exception-caught
                     pytest.skip(f"Setup unsuccessful with Exception: {str(e)}")
-                pid = url.split(f"{service['url']}/fdo/")[1]
                 repo = service["repos"][repo['id']]
                 pid = url.split(f"{service['url']}/fdo/")[1]
                 # If possible, check with repo