Enforce and catch single-subject rule violations in PATCH, but allow fragment URIs and the

resource FQURIs in the subject.

lakesuperior/endpoints/ldp.py

@@ -286,14 +286,14 @@ def put_resource(uuid):
                 disposition=disposition)
     except InvalidResourceError as e:
         return str(e), 409
-    except ServerManagedTermError as e:
+    except (ServerManagedTermError, SingleSubjectError) as e:
         return str(e), 412
     except IncompatibleLdpTypeError as e:
         return str(e), 415
 
     try:
         ret = rsrc.put()
-    except (InvalidResourceError, ResourceExistsError ) as e:
+    except (InvalidResourceError, ResourceExistsError) as e:
         return str(e), 409
     except TombstoneError as e:
         return _tombstone_response(e, uuid)
@@ -327,7 +327,7 @@ def patch_resource(uuid):
         return str(e), 404
     except TombstoneError as e:
         return _tombstone_response(e, uuid)
-    except ServerManagedTermError as e:
+    except (ServerManagedTermError, SingleSubjectError) as e:
         return str(e), 412
 
     return '', 204, headers

lakesuperior/model/ldp_rs.py

@@ -1,5 +1,7 @@
 from copy import deepcopy
 
+from flask import g
+
 from lakesuperior.dictionaries.namespaces import ns_collection as nsc
 from lakesuperior.model.ldpr import Ldpr, atomic
 
@@ -40,7 +42,11 @@ class LdpRs(Ldpr):
 
         @param update_str (string) SPARQL-Update staements.
         '''
-        delta = self._sparql_delta(update_str.replace('<>', self.urn.n3()))
+        local_update_str = update_str.replace('<>', self.urn.n3()).replace(
+                g.webroot + '/', nsc['fcres']).replace(
+                g.webroot, nsc['fcres'])
+        delta = self._sparql_delta(local_update_str)
+        self._ensure_single_subject_rdf(delta[0] | delta[1])
 
         return self._modify_rsrc(self.RES_UPDATED, *delta)
 

lakesuperior/model/ldpr.py

@@ -744,6 +744,9 @@ class Ldpr(metaclass=ABCMeta):
         Ensure that a RDF payload for a POST or PUT has a single resource.
         '''
         for s in set(gr.subjects()):
+            # Remove fragment and query components
+            if '#' in s:
+               s = URIRef(s.split('#')[0])
             if not s == self.urn:
                 raise SingleSubjectError(s, self.uuid)
 

tests/endpoints/test_ldp.py

@@ -259,6 +259,36 @@ class TestLdp:
         assert gr[ URIRef(uri) : nsc['dc'].title : Literal('Ciao') ]
 
 
+    def test_patch_ssr(self):
+        '''
+        Test patching a resource violating the single-subject rule.
+        '''
+        path = '/ldp/test_patch_ssr'
+        self.client.put(path)
+
+        uri = g.webroot + '/test_patch_ssr'
+
+        nossr_qry = 'INSERT { <http://bogus.org> a <urn:ns:A> . } WHERE {}'
+        abs_qry = 'INSERT {{ <{}> a <urn:ns:A> . }} WHERE {{}}'.format(uri)
+        frag_qry = 'INSERT {{ <{}#frag> a <urn:ns:A> . }} WHERE {{}}'\
+                .format(uri)
+
+        assert self.client.patch(
+            path, data=nossr_qry,
+            headers={'content-type' : 'application/sparql-update'}
+        ).status_code == 412
+
+        assert self.client.patch(
+            path, data=abs_qry,
+            headers={'content-type' : 'application/sparql-update'}
+        ).status_code == 204
+
+        assert self.client.patch(
+            path, data=frag_qry,
+            headers={'content-type' : 'application/sparql-update'}
+        ).status_code == 204
+
+
     def test_patch_ldp_nr_metadata(self):
         '''
         Test patching a LDP-NR metadata resource, both from the fcr:metadata