Some Prefer header tests.

tests/endpoints/test_ldp.py

@@ -8,93 +8,280 @@ from rdflib import Graph
 from rdflib.namespace import RDF
 from rdflib.term import Literal, URIRef
 
+from lakesuperior.dictionaries.namespaces import ns_collection as nsc
+from lakesuperior.model.ldpr import Ldpr
+from lakesuperior.toolbox import Toolbox
 
 @pytest.fixture(scope='module')
 def random_uuid():
     return str(uuid.uuid4())
 
 
-def test_get_root_node(client, db):
-    #assert client.get(url_for('ldp.get_resource')).status_code == 200
-    assert client.get('/ldp').status_code == 200
-
-
-def test_post_resource(client):
+@pytest.mark.usefixtures('client_class')
+@pytest.mark.usefixtures('db')
+class TestLdp:
     '''
-    Check response headers for a POST operation with empty payload.
+    Test HTTP interaction with LDP endpoint.
     '''
-    res = client.post('/ldp/')
-    assert res.status_code == 201
-    assert 'Location' in res.headers
+    def test_get_root_node(self):
+        '''
+        Get the root node from two different endpoints.
 
+        The test triplestore must be initialized, hence the `db` fixture.
+        '''
+        ldp_resp = self.client.get('/ldp')
+        rest_resp = self.client.get('/rest')
 
-def test_put_empty_resource(client, random_uuid):
-    '''
-    Check response headers for a PUT operation with empty payload.
-    '''
-    res = client.put('/ldp/{}'.format(random_uuid))
-    assert res.status_code == 201
+        assert ldp_resp.status_code == 200
+        assert rest_resp.status_code == 200
+        #assert ldp_resp.data == rest_resp.data
 
 
-def test_put_ldp_rs(client):
-    '''
-    PUT a resource with RDF payload and verify.
-    '''
-    with open('tests/data/marcel_duchamp_single_subject.ttl', 'rb') as f:
-        client.put('/ldp/ldprs01', data=f, content_type='text/turtle')
+    def test_post_resource(self, client):
+        '''
+        Check response headers for a POST operation with empty payload.
+        '''
+        res = self.client.post('/ldp/')
+        assert res.status_code == 201
+        assert 'Location' in res.headers
+
+
+    def test_put_empty_resource(self, random_uuid):
+        '''
+        Check response headers for a PUT operation with empty payload.
+        '''
+        res = self.client.put('/ldp/{}'.format(random_uuid))
+        assert res.status_code == 201
+
+
+    def test_put_existing_resource(self, random_uuid):
+        '''
+        Trying to PUT an existing resource should return a 204 if the payload
+        is empty.
+        '''
+        path = '/ldp/nonidempotent01'
+        assert self.client.put(path).status_code == 201
+        assert self.client.get(path).status_code == 200
+        assert self.client.put(path).status_code == 204
+
+
+    def test_put_ldp_rs(self, client):
+        '''
+        PUT a resource with RDF payload and verify.
+        '''
+        with open('tests/data/marcel_duchamp_single_subject.ttl', 'rb') as f:
+            self.client.put('/ldp/ldprs01', data=f, content_type='text/turtle')
 
-    resp = client.get('/ldp/ldprs01', headers={'accept' : 'text/turtle'})
-    assert resp.status_code == 200
+        resp = self.client.get('/ldp/ldprs01', headers={'accept' : 'text/turtle'})
+        assert resp.status_code == 200
 
-    g = Graph().parse(data=resp.data, format='text/turtle')
-    assert URIRef('http://vocab.getty.edu/ontology#Subject') in \
-            g.objects(None, RDF.type)
+        g = Graph().parse(data=resp.data, format='text/turtle')
+        assert URIRef('http://vocab.getty.edu/ontology#Subject') in \
+                g.objects(None, RDF.type)
 
 
-def test_put_ldp_nr(client, rnd_img):
+    def test_put_ldp_nr(self, rnd_img):
+        '''
+        PUT a resource with binary payload and verify checksums.
+        '''
+        rnd_img['content'].seek(0)
+        self.client.put('/ldp/ldpnr01', data=rnd_img['content'], headers={
+                'Content-Disposition' : 'attachment; filename={}'.format(
+                rnd_img['filename'])})
+
+        resp = self.client.get('/ldp/ldpnr01', headers={'accept' : 'image/png'})
+        assert resp.status_code == 200
+        assert sha1(resp.data).hexdigest() == rnd_img['hash']
+
+
+
+@pytest.mark.usefixtures('client_class')
+@pytest.mark.usefixtures('db')
+class TestPrefHeader:
     '''
-    PUT a resource with binary payload and verify checksums.
+    Test various combinations of `Prefer` header.
     '''
-    rnd_img['content'].seek(0)
-    client.put('/ldp/ldpnr01', data=rnd_img['content'], headers={
-            'Content-Disposition' : 'attachment; filename={}'.format(
-            rnd_img['filename'])})
+    @pytest.fixture(scope='class')
+    def cont_structure(self):
+        '''
+        Create a container structure to be used for subsequent requests.
+        '''
+        parent_path = '/ldp/test_parent'
+        self.client.put(parent_path)
+        self.client.put(parent_path + '/child1')
+        self.client.put(parent_path + '/child2')
+        self.client.put(parent_path + '/child3')
 
-    resp = client.get('/ldp/ldpnr01', headers={'accept' : 'image/png'})
-    assert resp.status_code == 200
-    assert sha1(resp.data).hexdigest() == rnd_img['hash']
+        return {
+            'path' : parent_path,
+            'response' : self.client.get(parent_path),
+            'subject' : URIRef(Toolbox().base_url + '/test_parent')
+        }
 
 
-def test_put_existing_resource(client, db, random_uuid):
-    '''
-    Trying to PUT an existing resource should:
+    def test_put_prefer_handling(self, random_uuid):
+        '''
+        Trying to PUT an existing resource should:
 
-    - Return a 204 if the payload is empty
-    - Return a 204 if the payload is RDF, server-managed triples are included
-      and the 'Prefer' header is set to 'handling=lenient'
-    - Return a 412 (ServerManagedTermError) if the payload is RDF,
-      server-managed triples are included and handling is set to 'strict'
-    '''
-    assert client.get('/ldp/{}'.format(random_uuid)).status_code == 200
-    assert client.put('/ldp/{}'.format(random_uuid)).status_code == 204
-    with open('tests/data/rdf_payload_w_srv_mgd_trp.ttl', 'rb') as f:
-        rsp_len = client.put(
-            '/ldp/{}'.format(random_uuid),
-            headers={
-                'Prefer' : 'handling=lenient',
-                'Content-Type' : 'text/turtle',
-            },
-            data=f
-        )
-    assert rsp_len.status_code == 204
-    with open('tests/data/rdf_payload_w_srv_mgd_trp.ttl', 'rb') as f:
-        rsp_strict = client.put(
-            '/ldp/{}'.format(random_uuid),
-            headers={
-                'Prefer' : 'handling=strict',
-                'Content-Type' : 'text/turtle',
-            },
-            data=f
-        )
-    assert rsp_strict.status_code == 412
+        - Return a 204 if the payload is empty
+        - Return a 204 if the payload is RDF, server-managed triples are
+          included and the 'Prefer' header is set to 'handling=lenient'
+        - Return a 412 (ServerManagedTermError) if the payload is RDF,
+          server-managed triples are included and handling is set to 'strict'
+        '''
+        path = '/ldp/put_pref_header01'
+        assert self.client.put(path).status_code == 201
+        assert self.client.get(path).status_code == 200
+        assert self.client.put(path).status_code == 204
+        with open('tests/data/rdf_payload_w_srv_mgd_trp.ttl', 'rb') as f:
+            rsp_len = self.client.put(
+                '/ldp/{}'.format(random_uuid),
+                headers={
+                    'Prefer' : 'handling=lenient',
+                    'Content-Type' : 'text/turtle',
+                },
+                data=f
+            )
+        assert rsp_len.status_code == 204
+        with open('tests/data/rdf_payload_w_srv_mgd_trp.ttl', 'rb') as f:
+            rsp_strict = self.client.put(
+                path,
+                headers={
+                    'Prefer' : 'handling=strict',
+                    'Content-Type' : 'text/turtle',
+                },
+                data=f
+            )
+        assert rsp_strict.status_code == 412
+
+
+    def test_embed_children(self, cont_structure):
+        '''
+        verify the "embed children" prefer header.
+        '''
+        parent_path = cont_structure['path']
+        cont_resp = cont_structure['response']
+        cont_subject = cont_structure['subject']
+
+        minimal_resp = self.client.get(parent_path, headers={
+            'Prefer' : 'return=minimal',
+        })
+
+        incl_embed_children_resp = self.client.get(parent_path, headers={
+            'Prefer' : 'return=representation; include={}'\
+                    .format(Ldpr.EMBED_CHILD_RES_URI),
+        })
+        omit_embed_children_resp = self.client.get(parent_path, headers={
+            'Prefer' : 'return=representation; omit={}'\
+                    .format(Ldpr.EMBED_CHILD_RES_URI),
+        })
+
+        assert omit_embed_children_resp.data == cont_resp.data
+
+        incl_g = Graph().parse(
+                data=incl_embed_children_resp.data, format='turtle')
+        omit_g = Graph().parse(
+                data=omit_embed_children_resp.data, format='turtle')
+
+        children = set(incl_g[cont_subject : nsc['ldp'].contains])
+        assert len(children) == 3
+
+        children = set(incl_g[cont_subject : nsc['ldp'].contains])
+        for child_uri in children:
+            assert set(incl_g[ child_uri : : ])
+            assert not set(omit_g[ child_uri : : ])
+
+
+    def test_return_children(self, cont_structure):
+        '''
+        verify the "return children" prefer header.
+        '''
+        parent_path = cont_structure['path']
+        cont_resp = cont_structure['response']
+        cont_subject = cont_structure['subject']
+
+        incl_children_resp = self.client.get(parent_path, headers={
+            'Prefer' : 'return=representation; include={}'\
+                    .format(Ldpr.RETURN_CHILD_RES_URI),
+        })
+        omit_children_resp = self.client.get(parent_path, headers={
+            'Prefer' : 'return=representation; omit={}'\
+                    .format(Ldpr.RETURN_CHILD_RES_URI),
+        })
+
+        assert incl_children_resp.data == cont_resp.data
+
+        incl_g = Graph().parse(data=incl_children_resp.data, format='turtle')
+        omit_g = Graph().parse(data=omit_children_resp.data, format='turtle')
+
+        children = incl_g[cont_subject : nsc['ldp'].contains]
+        for child_uri in children:
+            assert not omit_g[ cont_subject : nsc['ldp'].contains : child_uri ]
+
+
+    def test_inbound_rel(self, cont_structure):
+        '''
+        verify the "inboud relationships" prefer header.
+        '''
+        parent_path = cont_structure['path']
+        cont_resp = cont_structure['response']
+        cont_subject = cont_structure['subject']
+
+        incl_inbound_resp = self.client.get(parent_path, headers={
+            'Prefer' : 'return=representation; include={}'\
+                    .format(Ldpr.RETURN_INBOUND_REF_URI),
+        })
+        omit_inbound_resp = self.client.get(parent_path, headers={
+            'Prefer' : 'return=representation; omit={}'\
+                    .format(Ldpr.RETURN_INBOUND_REF_URI),
+        })
+
+        assert omit_inbound_resp.data == cont_resp.data
+
+        incl_g = Graph().parse(data=incl_inbound_resp.data, format='turtle')
+        omit_g = Graph().parse(data=omit_inbound_resp.data, format='turtle')
+
+        assert set(incl_g[ : : cont_subject ])
+        assert not set(omit_g[ : : cont_subject ])
+
+
+    def test_srv_mgd_triples(self, cont_structure):
+        '''
+        verify the "server managed triples" prefer header.
+        '''
+        parent_path = cont_structure['path']
+        cont_resp = cont_structure['response']
+        cont_subject = cont_structure['subject']
+
+        incl_srv_mgd_resp = self.client.get(parent_path, headers={
+            'Prefer' : 'return=representation; include={}'\
+                    .format(Ldpr.RETURN_SRV_MGD_RES_URI),
+        })
+        omit_srv_mgd_resp = self.client.get(parent_path, headers={
+            'Prefer' : 'return=representation; omit={}'\
+                    .format(Ldpr.RETURN_SRV_MGD_RES_URI),
+        })
+
+        assert incl_srv_mgd_resp.data == cont_resp.data
+
+        incl_g = Graph().parse(data=incl_srv_mgd_resp.data, format='turtle')
+        omit_g = Graph().parse(data=omit_srv_mgd_resp.data, format='turtle')
+
+        for pred in {
+            nsc['fcrepo'].created,
+            nsc['fcrepo'].createdBy,
+            nsc['fcrepo'].lastModified,
+            nsc['fcrepo'].lastModifiedBy,
+            nsc['ldp'].contains,
+        }:
+            assert set(incl_g[ cont_subject : pred : ])
+            assert not set(omit_g[ cont_subject : pred : ])
+
+        for type in {
+                nsc['fcrepo'].Resource,
+                nsc['ldp'].Container,
+                nsc['ldp'].Resource,
+        }:
+            assert incl_g[ cont_subject : RDF.type : type ]
+            assert not omit_g[ cont_subject : RDF.type : type ]